syndicate-java/src/main/java/org/syndicate_lang/actors/Actor.java

package org.syndicate_lang.actors;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Function;
import java.util.logging.Level;
import java.util.logging.Logger;

/**
 * I represent the shared execution concepts for a collection of objects; I am roughly analogous to the E concept of a Vat.
 */
public class Actor implements Executor {
    private final static AtomicLong _count = new AtomicLong(0);
    private final static AtomicLong _actorId = new AtomicLong(0);
    protected final static ExecutorService _executor = Executors.newWorkStealingPool();
    protected final static ScheduledExecutorService _scheduledExecutor = Executors.newScheduledThreadPool(Runtime.getRuntime().availableProcessors());

    private final String _name;
    private final Logger _logger;

    private boolean _alive = true;
    private Throwable _exitReason = null;
    private boolean _isCounted = true;
    private Map<Long, Ref> _outbound;

    private final static class WorkItem extends AtomicReference<WorkItem> {
        Runnable work;
        Runnable ifNotAlive;

        public WorkItem(Runnable work, Runnable ifNotAlive) {
            this.work = work;
            this.ifNotAlive = ifNotAlive;
        }

        public final void clear() {
            this.work = null;
            this.ifNotAlive = null;
        }
    }

    private WorkItem head = new WorkItem(null, null);
    private final AtomicReference<WorkItem> tail = new AtomicReference<>(head);
    private final AtomicLong workItemCount = new AtomicLong(0);

    public Actor() {
        this(null, null);
    }

    public Actor(String debugName) {
        this(debugName, null);
    }

    public Actor(String debugName, Map<Long, Ref> outbound) {
        this._name = debugName == null ? "" + _actorId.incrementAndGet() : debugName;
        this._logger = Logger.getLogger(this.getClass().getSimpleName() + "(" + this._name + ")");
        this._outbound = outbound == null ? new HashMap<>() : outbound;
        _count.incrementAndGet();
    }

    public static Ref forEntity(IEntity o) {
        return new Actor().ref(o);
    }

    public static Promise<Ref> boot(Function<Turn, IEntity> f) {
        final Promise<Ref> p = new Promise<>();
        final Actor a = new Actor();
        a.execute(
                () -> Turn.forActor(a, t -> p.resolveCalling(() -> a.ref(f.apply(t)))),
                () -> p.rejectWith(new ActorTerminated(a)));
        return p;
    }

    public String getName() {
        return _name;
    }

    public Throwable getExitReason() {
        return _exitReason;
    }

    public Logger log() {
        return _logger;
    }

    public synchronized boolean isDaemon() {
        return _alive && !_isCounted;
    }

    public synchronized Actor daemonize() {
        this._releaseCount();
        return this;
    }

    private void _releaseCount() {
        if (_isCounted) {
            _isCounted = false;
            synchronized (_count) {
                if (_count.decrementAndGet() == 0) {
                    _count.notifyAll();
                }
            }
        }
    }

    public void _injectOutbound(Long handle, Ref peer) {
        _outbound.put(handle, peer);
    }

    public Ref _lookupOutbound(Long handle) {
        return _outbound.get(handle);
    }

    public Ref _extractOutbound(Long handle) {
        return _outbound.remove(handle);
    }

    public String toString() {
        return super.toString() + "(" + this._name + ")";
    }

    public<T> Ref ref(IEntity o) {
        return new Ref(this, o);
    }

    private void _performSync(Runnable work, Runnable ifNotAlive) {
        synchronized (this) {
            _perform(work, ifNotAlive);
        }
    }

    private void _perform(Runnable work, Runnable ifNotAlive) {
        if (!_alive) {
            if (ifNotAlive != null) ifNotAlive.run();
        } else {
            try {
                work.run();
            } catch (Throwable exn) {
                this._stop(false, exn);
            }
        }
    }

    public Promise<?> stop() {
        return stop(true, null);
    }

    public Promise<?> stop(Throwable reason) {
        return stop(true, reason);
    }
    public Promise<?> stop(boolean normally, Throwable reason) {
        Promise<?> p = new Promise<>();
        this.execute(() -> {
            this._stop(normally, reason);
            p.resolve();
        }, p::resolve);
        return p;
    }

    private synchronized void _stop(boolean normally, Throwable reason) {
        if (_alive) {
            _alive = false;
            _exitReason = reason;
            if (normally) {
                log().log(Level.FINE, "Actor stopped", reason);
            } else {
                log().log(Level.SEVERE, "Actor terminated with error", reason);
            }
            Turn.forActor(this, t -> _outbound.forEach(t::_retract_), true);
            _releaseCount();
        }
    }

    @Override
    public void execute(Runnable work) {
        this.execute(work, null);
    }

    public void execute(Runnable work, Runnable ifNotAlive) {
        {
            WorkItem i = new WorkItem(work, ifNotAlive);
            tail.getAndSet(i).set(i);
        }
        if (workItemCount.getAndIncrement() == 0) {
            _executor.execute(() -> {
                synchronized (this) {
                    long batch = workItemCount.get();
                    while (batch > 0) {
                        for (int count = 0; count < batch; count++) {
                            WorkItem i = null;
                            while (i == null) i = head.get();
                            head = i;
                            this._perform(i.work, i.ifNotAlive);
                            i.clear();
                        }
                        batch = workItemCount.addAndGet(-batch);
                    }
                }
            });
        }
    }

    public void later(long delayMilliseconds, Runnable work) {
        this.later(delayMilliseconds, work, null);
    }

    public void later(long delayMilliseconds, Runnable work, Runnable ifNotAlive) {
        if (delayMilliseconds == 0) {
            this.execute(work, ifNotAlive);
        } else {
            _scheduledExecutor.schedule(() -> this._performSync(work, ifNotAlive), delayMilliseconds, TimeUnit.MILLISECONDS);
        }
    }

    public PeriodicTimer every(long periodMilliseconds, Runnable f) {
        return every(0, periodMilliseconds, f);
    }

    public PeriodicTimer every(long initialDelayMilliseconds, long periodMilliseconds, Runnable f) {
        return new PeriodicTimer(
                _scheduledExecutor.scheduleAtFixedRate(
                        () -> this._performSync(f, null),
                        initialDelayMilliseconds,
                        periodMilliseconds,
                        TimeUnit.MILLISECONDS));
    }

    public static void awaitAll() throws InterruptedException {
        while (_count.get() > 0) {
            synchronized (_count) {
                _count.wait(1000);
            }
        }
        _executor.shutdown();
        _scheduledExecutor.shutdown();
        //noinspection ResultOfMethodCallIgnored
        _executor.awaitTermination(5, TimeUnit.MINUTES);
        //noinspection ResultOfMethodCallIgnored
        _scheduledExecutor.awaitTermination(5, TimeUnit.MINUTES);
    }

    public static void convenientLogging() {
        System.setProperty("java.util.logging.SimpleFormatter.format",
                "%1$tY-%1$tm-%1$td %1$tH:%1$tM:%1$tS.%1$tL %4$s %3$s %5$s%6$s%n");
    }
}