Links and monitors
This commit is contained in:
parent
bb3d822988
commit
f8a7d95e55
|
@ -1,7 +1,13 @@
|
||||||
package org.syndicate_lang.actors;
|
package org.syndicate_lang.actors;
|
||||||
|
|
||||||
|
import java.util.HashMap;
|
||||||
|
import java.util.HashSet;
|
||||||
|
import java.util.Map;
|
||||||
|
import java.util.Set;
|
||||||
import java.util.concurrent.*;
|
import java.util.concurrent.*;
|
||||||
import java.util.concurrent.atomic.AtomicLong;
|
import java.util.concurrent.atomic.AtomicLong;
|
||||||
|
import java.util.function.Consumer;
|
||||||
|
import java.util.function.Supplier;
|
||||||
import java.util.logging.Level;
|
import java.util.logging.Level;
|
||||||
import java.util.logging.Logger;
|
import java.util.logging.Logger;
|
||||||
|
|
||||||
|
@ -21,6 +27,9 @@ public class Actor implements Executor {
|
||||||
private boolean _alive = true;
|
private boolean _alive = true;
|
||||||
private Throwable _exitReason = null;
|
private Throwable _exitReason = null;
|
||||||
private boolean _isCounted = true;
|
private boolean _isCounted = true;
|
||||||
|
private Set<Actor> _links = null;
|
||||||
|
private Map<Object, Remote<IMonitorHandler>> _monitors = null;
|
||||||
|
private Consumer<Actor> _exitTrap = null;
|
||||||
|
|
||||||
public static Actor current() {
|
public static Actor current() {
|
||||||
return _currentActor.get();
|
return _currentActor.get();
|
||||||
|
@ -48,6 +57,15 @@ public class Actor implements Executor {
|
||||||
return new Actor().proxyFor(o);
|
return new Actor().proxyFor(o);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
public static<T> Promise<Remote<T>> boot(Supplier<T> f) {
|
||||||
|
Promise<Remote<T>> p = new Promise<>();
|
||||||
|
Actor a = new Actor();
|
||||||
|
a.execute(
|
||||||
|
() -> p.resolveCalling(() -> Actor.ref(f.get())),
|
||||||
|
() -> p.rejectWith(new ActorTerminated(a)));
|
||||||
|
return p;
|
||||||
|
}
|
||||||
|
|
||||||
public String getName() {
|
public String getName() {
|
||||||
return _name;
|
return _name;
|
||||||
}
|
}
|
||||||
|
@ -134,6 +152,21 @@ public class Actor implements Executor {
|
||||||
} else {
|
} else {
|
||||||
getLogger().log(Level.SEVERE, "Actor terminated with error", reason);
|
getLogger().log(Level.SEVERE, "Actor terminated with error", reason);
|
||||||
}
|
}
|
||||||
|
Set<Actor> linkedPeers = _links;
|
||||||
|
if (linkedPeers != null) {
|
||||||
|
_links = null;
|
||||||
|
for (var peer : linkedPeers) {
|
||||||
|
peer.notifyExit(this);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
Map<Object, Remote<IMonitorHandler>> monitoringPeers = _monitors;
|
||||||
|
if (monitoringPeers != null) {
|
||||||
|
_monitors = null;
|
||||||
|
for (var entry : monitoringPeers.entrySet()) {
|
||||||
|
final var ref = entry.getKey();
|
||||||
|
entry.getValue().async((h) -> h.handleMonitor(this, ref));
|
||||||
|
}
|
||||||
|
}
|
||||||
_releaseCount();
|
_releaseCount();
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
@ -188,4 +221,64 @@ public class Actor implements Executor {
|
||||||
System.setProperty("java.util.logging.SimpleFormatter.format",
|
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");
|
"%1$tY-%1$tm-%1$td %1$tH:%1$tM:%1$tS.%1$tL %4$s %3$s %5$s%6$s%n");
|
||||||
}
|
}
|
||||||
|
|
||||||
|
public void link() {
|
||||||
|
final Actor peer = Actor.current();
|
||||||
|
this.linkPeer(peer);
|
||||||
|
peer.linkPeer(this);
|
||||||
|
}
|
||||||
|
|
||||||
|
public void unlink() {
|
||||||
|
final Actor peer = Actor.current();
|
||||||
|
this.unlinkPeer(peer);
|
||||||
|
peer.unlinkPeer(this);
|
||||||
|
}
|
||||||
|
|
||||||
|
private synchronized void linkPeer(Actor peer) {
|
||||||
|
if (this._alive) {
|
||||||
|
if (_links == null) _links = new HashSet<>();
|
||||||
|
_links.add(peer);
|
||||||
|
} else {
|
||||||
|
peer.notifyExit(this);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
private void notifyExit(final Actor exitingPeer) {
|
||||||
|
this.execute(() -> {
|
||||||
|
this.unlinkPeer(exitingPeer);
|
||||||
|
if (this._exitTrap != null) {
|
||||||
|
this._exitTrap.accept(exitingPeer);
|
||||||
|
} else {
|
||||||
|
this._stop(exitingPeer.getExitReason() == null, new ActorTerminated(exitingPeer));
|
||||||
|
}
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
|
private synchronized void unlinkPeer(Actor peer) {
|
||||||
|
if (_links != null) {
|
||||||
|
_links.remove(peer);
|
||||||
|
if (_links.isEmpty()) {
|
||||||
|
_links = null;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
public void trapExits(Consumer<Actor> handler) {
|
||||||
|
this._exitTrap = handler;
|
||||||
|
}
|
||||||
|
|
||||||
|
public synchronized Object monitor(Consumer<Actor> handler) {
|
||||||
|
Object ref = new Object();
|
||||||
|
monitor(ref, (actor, _ref) -> handler.accept(actor));
|
||||||
|
return ref;
|
||||||
|
}
|
||||||
|
|
||||||
|
public synchronized void monitor(final Object ref, IMonitorHandler handler) {
|
||||||
|
if (this._alive) {
|
||||||
|
if (_monitors == null) _monitors = new HashMap<>();
|
||||||
|
_monitors.put(ref, Actor.ref(handler));
|
||||||
|
} else {
|
||||||
|
Actor.ref(handler).async((h) -> h.handleMonitor(this, ref));
|
||||||
|
}
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
|
@ -0,0 +1,5 @@
|
||||||
|
package org.syndicate_lang.actors;
|
||||||
|
|
||||||
|
public interface IMonitorHandler {
|
||||||
|
void handleMonitor(Actor exitingPeer, Object ref);
|
||||||
|
}
|
|
@ -5,13 +5,17 @@ import java.util.List;
|
||||||
import java.util.concurrent.*;
|
import java.util.concurrent.*;
|
||||||
import java.util.function.Consumer;
|
import java.util.function.Consumer;
|
||||||
import java.util.function.Function;
|
import java.util.function.Function;
|
||||||
|
import java.util.function.Supplier;
|
||||||
|
|
||||||
public class Promise<T> implements Future<T> {
|
public class Promise<T> implements Future<T> {
|
||||||
public static enum State {
|
|
||||||
|
public static final TimeoutException TIMEOUT_WAITING_FOR_PROMISE_RESOLUTION = new TimeoutException("Waiting for promise resolution");
|
||||||
|
|
||||||
|
public enum State {
|
||||||
PENDING,
|
PENDING,
|
||||||
FULFILLED,
|
FULFILLED,
|
||||||
REJECTED
|
REJECTED
|
||||||
};
|
}
|
||||||
|
|
||||||
private volatile State _state = State.PENDING;
|
private volatile State _state = State.PENDING;
|
||||||
private T _value = null;
|
private T _value = null;
|
||||||
|
@ -65,17 +69,20 @@ public class Promise<T> implements Future<T> {
|
||||||
this.resolveWith(null);
|
this.resolveWith(null);
|
||||||
}
|
}
|
||||||
|
|
||||||
public synchronized void resolveWith(T t) {
|
public void resolveWith(T t) {
|
||||||
if (this.isPending()) {
|
List<Consumer<T>> worklist = null;
|
||||||
this._value = t;
|
synchronized (this) {
|
||||||
this._state = State.FULFILLED;
|
if (this.isPending()) {
|
||||||
var worklist = _resolvers;
|
this._value = t;
|
||||||
_resolvers = null;
|
this._state = State.FULFILLED;
|
||||||
_rejecters = null;
|
worklist = _resolvers;
|
||||||
if (worklist != null) {
|
_resolvers = null;
|
||||||
for (var callback : worklist) {
|
_rejecters = null;
|
||||||
callback.accept(t);
|
}
|
||||||
}
|
}
|
||||||
|
if (worklist != null) {
|
||||||
|
for (var callback : worklist) {
|
||||||
|
callback.accept(t);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
@ -84,21 +91,24 @@ public class Promise<T> implements Future<T> {
|
||||||
if (this == t) {
|
if (this == t) {
|
||||||
throw new IllegalArgumentException("cannot chain promise immediately to itself");
|
throw new IllegalArgumentException("cannot chain promise immediately to itself");
|
||||||
}
|
}
|
||||||
t.whenFulfilled((v) -> this.resolveWith(v));
|
t.whenFulfilled(this::resolveWith);
|
||||||
t.whenRejected((e) -> this.rejectWith(e));
|
t.whenRejected(this::rejectWith);
|
||||||
}
|
}
|
||||||
|
|
||||||
public void rejectWith(Throwable e) {
|
public void rejectWith(Throwable e) {
|
||||||
if (this.isPending()) {
|
List<Consumer<Throwable>> worklist = null;
|
||||||
this._reason = e;
|
synchronized (this) {
|
||||||
this._state = State.REJECTED;
|
if (this.isPending()) {
|
||||||
var worklist = _rejecters;
|
this._reason = e;
|
||||||
_resolvers = null;
|
this._state = State.REJECTED;
|
||||||
_rejecters = null;
|
worklist = _rejecters;
|
||||||
if (worklist != null) {
|
_resolvers = null;
|
||||||
for (var callback : worklist) {
|
_rejecters = null;
|
||||||
callback.accept(e);
|
}
|
||||||
}
|
}
|
||||||
|
if (worklist != null) {
|
||||||
|
for (var callback : worklist) {
|
||||||
|
callback.accept(e);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
@ -107,29 +117,31 @@ public class Promise<T> implements Future<T> {
|
||||||
return this.andThen(ok, null);
|
return this.andThen(ok, null);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
public void resolveCalling(Supplier<T> f) {
|
||||||
|
try {
|
||||||
|
this.resolveWith(f.get());
|
||||||
|
} catch (Throwable e) {
|
||||||
|
this.rejectWith(e);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
public synchronized<R> Promise<R> andThen(final Function<T, R> ok, final Function<Throwable, R> fail) {
|
public synchronized<R> Promise<R> andThen(final Function<T, R> ok, final Function<Throwable, R> fail) {
|
||||||
Actor a0 = Actor.current();
|
Actor a0 = Actor.current();
|
||||||
final Actor a = a0 != null ? a0 : new Actor();
|
final Actor a = a0 != null ? a0 : new Actor();
|
||||||
|
|
||||||
Promise<R> p = new Promise<>();
|
Promise<R> p = new Promise<>();
|
||||||
this.whenFulfilled((t) -> a.execute(() -> {
|
this.whenFulfilled((t) -> a.execute(
|
||||||
try {
|
() -> p.resolveCalling(() -> ok.apply(t)),
|
||||||
p.resolveWith(ok.apply(t));
|
() -> p.rejectWith(new ActorTerminated(a))));
|
||||||
} catch (Throwable e) {
|
this.whenRejected((e) -> a.execute(
|
||||||
p.rejectWith(e);
|
() -> {
|
||||||
}
|
if (fail == null) {
|
||||||
}, () -> p.rejectWith(new ActorTerminated(a))));
|
p.rejectWith(e);
|
||||||
this.whenRejected((e) -> a.execute(() -> {
|
} else {
|
||||||
try {
|
p.resolveCalling(() -> fail.apply(e));
|
||||||
if (fail == null) {
|
}
|
||||||
p.rejectWith(e);
|
},
|
||||||
} else {
|
() -> p.rejectWith(new ActorTerminated(a))));
|
||||||
p.resolveWith(fail.apply(e));
|
|
||||||
}
|
|
||||||
} catch (Throwable e2) {
|
|
||||||
p.rejectWith(e2);
|
|
||||||
}
|
|
||||||
}, () -> p.rejectWith(new ActorTerminated(a))));
|
|
||||||
return p;
|
return p;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -155,14 +167,30 @@ public class Promise<T> implements Future<T> {
|
||||||
}
|
}
|
||||||
|
|
||||||
public T _await(long delay, TimeUnit unit) throws TimeoutException, InterruptedException {
|
public T _await(long delay, TimeUnit unit) throws TimeoutException, InterruptedException {
|
||||||
Semaphore s = new Semaphore(0);
|
Actor a = Actor.current();
|
||||||
this.whenFulfilled((_t) -> s.release());
|
if (a == null) {
|
||||||
this.whenRejected((_e) -> s.release());
|
Semaphore s = new Semaphore(0);
|
||||||
if (delay == -1) {
|
this.whenFulfilled((_t) -> s.release());
|
||||||
s.acquire();
|
this.whenRejected((_e) -> s.release());
|
||||||
|
if (delay == -1) {
|
||||||
|
s.acquire();
|
||||||
|
} else {
|
||||||
|
if (!s.tryAcquire(delay, unit)) throw TIMEOUT_WAITING_FOR_PROMISE_RESOLUTION;
|
||||||
|
}
|
||||||
} else {
|
} else {
|
||||||
if (!s.tryAcquire(delay, unit)) {
|
this.whenFulfilled((_t) -> { synchronized (a) { a.notifyAll(); } });
|
||||||
throw new TimeoutException();
|
this.whenRejected((_e) -> { synchronized (a) { a.notifyAll(); } });
|
||||||
|
synchronized (a) {
|
||||||
|
if (delay == -1) {
|
||||||
|
while (this.isPending()) a.wait();
|
||||||
|
} else {
|
||||||
|
long targetTime = System.currentTimeMillis() + unit.toMillis(delay);
|
||||||
|
while (this.isPending()) {
|
||||||
|
long now = System.currentTimeMillis();
|
||||||
|
if (now >= targetTime) throw TIMEOUT_WAITING_FOR_PROMISE_RESOLUTION;
|
||||||
|
a.wait(targetTime - now);
|
||||||
|
}
|
||||||
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
if (this.isFulfilled()) {
|
if (this.isFulfilled()) {
|
||||||
|
|
|
@ -1,7 +1,5 @@
|
||||||
package org.syndicate_lang.actors.example.example2;
|
package org.syndicate_lang.actors.example.example2;
|
||||||
|
|
||||||
import org.syndicate_lang.actors.Actor;
|
|
||||||
|
|
||||||
import java.util.List;
|
import java.util.List;
|
||||||
|
|
||||||
public class Forwarder implements IForwarder {
|
public class Forwarder implements IForwarder {
|
||||||
|
|
|
@ -4,7 +4,4 @@ import org.syndicate_lang.actors.Actor;
|
||||||
|
|
||||||
public interface IForwarder {
|
public interface IForwarder {
|
||||||
void handleMessage(int hopCount);
|
void handleMessage(int hopCount);
|
||||||
default void shutdown() {
|
|
||||||
Actor.current().stop();
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
|
@ -1,6 +1,7 @@
|
||||||
package org.syndicate_lang.actors.example.example2;
|
package org.syndicate_lang.actors.example.example2;
|
||||||
|
|
||||||
import org.syndicate_lang.actors.Actor;
|
import org.syndicate_lang.actors.Actor;
|
||||||
|
import org.syndicate_lang.actors.Remote;
|
||||||
|
|
||||||
import java.util.ArrayList;
|
import java.util.ArrayList;
|
||||||
import java.util.List;
|
import java.util.List;
|
||||||
|
@ -9,8 +10,6 @@ import static java.lang.Integer.parseInt;
|
||||||
|
|
||||||
public class Main implements IForwarder {
|
public class Main implements IForwarder {
|
||||||
|
|
||||||
private List<IForwarder> _actors;
|
|
||||||
|
|
||||||
public static void main(String[] args) throws InterruptedException {
|
public static void main(String[] args) throws InterruptedException {
|
||||||
Actor.convenientLogging();
|
Actor.convenientLogging();
|
||||||
Actor.forObject(new Main(parseInt(args[0]), parseInt(args[1]))).syncVoid(Main::boot).await();
|
Actor.forObject(new Main(parseInt(args[0]), parseInt(args[1]))).syncVoid(Main::boot).await();
|
||||||
|
@ -29,27 +28,22 @@ public class Main implements IForwarder {
|
||||||
|
|
||||||
public void boot() {
|
public void boot() {
|
||||||
Actor.log().info("Available processors: " + Runtime.getRuntime().availableProcessors());
|
Actor.log().info("Available processors: " + Runtime.getRuntime().availableProcessors());
|
||||||
this._actors = new ArrayList<>();
|
List<IForwarder> _actors = new ArrayList<>();
|
||||||
IForwarder me = Actor.ref(this).asyncProxy(IForwarder.class);
|
final IForwarder me = Actor.ref(this).asyncProxy(IForwarder.class);
|
||||||
for (int i = 0; i < _nActors; i++) {
|
for (int i = 0; i < _nActors; i++) {
|
||||||
this._actors.add(Actor.forObject(
|
Remote<IForwarder> a = Actor.forObject(new Forwarder(i, _actors, me, this._nRounds));
|
||||||
new Forwarder(i, this._actors, me, this._nRounds))
|
a.getActor().link();
|
||||||
.asyncProxy(IForwarder.class));
|
_actors.add(a.asyncProxy(IForwarder.class));
|
||||||
// Actor.log().info(this._actors.get(this._actors.size()-1).toString());
|
|
||||||
}
|
}
|
||||||
Actor.log().info("Start");
|
Actor.log().info("Start");
|
||||||
this._actors.forEach((a) -> a.handleMessage(0));
|
_actors.forEach((a) -> a.handleMessage(0));
|
||||||
}
|
}
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
public void handleMessage(int hopCount) {
|
public void handleMessage(int hopCount) {
|
||||||
this._remainingToReceive--;
|
this._remainingToReceive--;
|
||||||
// Actor.log().info(String.format("hopCount: %d, remainingToReceive: %d",
|
|
||||||
// hopCount,
|
|
||||||
// this._remainingToReceive));
|
|
||||||
if (this._remainingToReceive == 0) {
|
if (this._remainingToReceive == 0) {
|
||||||
this._actors.forEach(IForwarder::shutdown);
|
Actor.current().stop();
|
||||||
this.shutdown();
|
|
||||||
Actor.log().info("Stop after " + (_nActors * _nRounds) + " messages");
|
Actor.log().info("Stop after " + (_nActors * _nRounds) + " messages");
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
Loading…
Reference in New Issue