Skip to content

Commit

Permalink
mostly require caller to be specified on actor calls (#281)
Browse files Browse the repository at this point in the history
  • Loading branch information
@dgrove-oss
dgrove-oss authored Mar 22, 2022
1 parent 000eb66 commit f854211
Show file tree
Hide file tree
Showing 13 changed files with 67 additions and 82 deletions.
4 changes: 2 additions & 2 deletions ...actors-dp-java-reactive/src/main/java/com/ibm/research/kar/example/philosophers/Cafe.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -34,7 +34,7 @@
public class Cafe extends ActorSkeleton {
@Remote
public Uni<JsonValue> occupancy(JsonString table) {
return Actors.call(Actors.ref("Table", table.getString()), "occupancy");
return Actors.call(this, Actors.ref("Table", table.getString()), "occupancy");
}

@Remote
Expand All @@ -50,7 +50,7 @@ public Uni<JsonString> seatTable(JsonNumber n, JsonNumber servings) {

@Remote
public Uni<JsonString> seatTable(JsonNumber n, JsonNumber servings, JsonString requestId) {
return Actors.call(Actors.ref("Table", requestId.getString()), "prepare", Json.createValue(this.getId()), n, servings)
return Actors.call(this, Actors.ref("Table", requestId.getString()), "prepare", Json.createValue(this.getId()), n, servings)
.chain(() -> Uni.createFrom().item(requestId));
}
}
8 changes: 4 additions & 4 deletions ...dp-java-reactive/src/main/java/com/ibm/research/kar/example/philosophers/Philosopher.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -95,7 +95,7 @@ public Uni<TailCall> joinTable(JsonString table, JsonString firstFork, JsonStrin

@Remote
public Uni<TailCall> getFirstFork(JsonNumber attempt) {
return Actors.call(Actors.ref("Fork", this.firstFork.getString()), "pickUp", Json.createValue(this.getId()))
return Actors.call(this, Actors.ref("Fork", this.firstFork.getString()), "pickUp", Json.createValue(this.getId()))
.chain(acquired -> {
if (acquired.equals(JsonValue.TRUE)) {
return Actors.tailCall(this, "getSecondFork", Json.createValue(1));
Expand All @@ -111,7 +111,7 @@ public Uni<TailCall> getFirstFork(JsonNumber attempt) {

@Remote
public Uni<TailCall> getSecondFork(JsonNumber attempt) {
return Actors.call(Actors.ref("Fork", this.secondFork.getString()), "pickUp", Json.createValue(this.getId()))
return Actors.call(this, Actors.ref("Fork", this.secondFork.getString()), "pickUp", Json.createValue(this.getId()))
.chain(acquired -> {
if (acquired.equals(JsonValue.TRUE)) {
return Actors.tailCall(this, "eat", this.servingsEaten);
Expand All @@ -128,8 +128,8 @@ public Uni<TailCall> getSecondFork(JsonNumber attempt) {
@Remote
public Uni<TailCall> eat(JsonNumber serving) {
if (VERBOSE) System.out.println(this.getId() + " ate serving number " + serving);
return Actors.call(Actors.ref("Fork", this.secondFork.getString()), "putDown", Json.createValue(this.getId()))
.chain(() -> Actors.call(Actors.ref("Fork", this.firstFork.getString()), "putDown", Json.createValue(this.getId())))
return Actors.call(this, Actors.ref("Fork", this.secondFork.getString()), "putDown", Json.createValue(this.getId()))
.chain(() -> Actors.call(this, Actors.ref("Fork", this.firstFork.getString()), "putDown", Json.createValue(this.getId())))
.chain(() -> {
this.servingsEaten = Json.createValue(serving.intValue() + 1);
return Actors.State.set(this, "servingsEaten", this.servingsEaten);
Expand Down
4 changes: 2 additions & 2 deletions examples/actors-dp-java/src/main/java/com/ibm/research/kar/example/philosophers/Cafe.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -32,7 +32,7 @@
public class Cafe extends ActorSkeleton {
@Remote
public JsonValue occupancy(JsonString table) {
return Actors.call(Actors.ref("Table", table.getString()), "occupancy");
return Actors.call(this, Actors.ref("Table", table.getString()), "occupancy");
}

@Remote
Expand All @@ -48,7 +48,7 @@ public JsonString seatTable(JsonNumber n, JsonNumber servings) {

@Remote
public JsonString seatTable(JsonNumber n, JsonNumber servings, JsonString requestId) {
Actors.call(Actors.ref("Table", requestId.getString()), "prepare", Json.createValue(this.getId()), n, servings);
Actors.call(this, Actors.ref("Table", requestId.getString()), "prepare", Json.createValue(this.getId()), n, servings);
return requestId;
}
}
8 changes: 4 additions & 4 deletions ...s/actors-dp-java/src/main/java/com/ibm/research/kar/example/philosophers/Philosopher.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -93,7 +93,7 @@ public TailCall joinTable(JsonString table, JsonString firstFork, JsonString sec

@Remote
public TailCall getFirstFork(JsonNumber attempt) {
if (Actors.call(Actors.ref("Fork", this.firstFork.getString()), "pickUp", Json.createValue(this.getId())).equals(JsonValue.TRUE)) {
if (Actors.call(this, Actors.ref("Fork", this.firstFork.getString()), "pickUp", Json.createValue(this.getId())).equals(JsonValue.TRUE)) {
return new TailCall(this, "getSecondFork", Json.createValue(1));
} else {
if (attempt.intValue() > 5) {
Expand All @@ -106,7 +106,7 @@ public TailCall getFirstFork(JsonNumber attempt) {

@Remote
public TailCall getSecondFork(JsonNumber attempt) {
if (Actors.call(Actors.ref("Fork", this.secondFork.getString()), "pickUp", Json.createValue(this.getId())).equals(JsonValue.TRUE)) {
if (Actors.call(this, Actors.ref("Fork", this.secondFork.getString()), "pickUp", Json.createValue(this.getId())).equals(JsonValue.TRUE)) {
return new TailCall(this, "eat", this.servingsEaten);
} else {
if (attempt.intValue() > 5) {
Expand All @@ -120,8 +120,8 @@ public TailCall getSecondFork(JsonNumber attempt) {
@Remote
public TailCall eat(JsonNumber serving) {
if (VERBOSE) System.out.println(this.getId()+" ate serving number "+serving);
Actors.call(Actors.ref("Fork", this.secondFork.getString()), "putDown", Json.createValue(this.getId()));
Actors.call(Actors.ref("Fork", this.firstFork.getString()), "putDown", Json.createValue(this.getId()));
Actors.call(this, Actors.ref("Fork", this.secondFork.getString()), "putDown", Json.createValue(this.getId()));
Actors.call(this, Actors.ref("Fork", this.firstFork.getString()), "putDown", Json.createValue(this.getId()));
this.servingsEaten = Json.createValue(serving.intValue() + 1);
Actors.State.set(this, "servingsEaten", this.servingsEaten);
if (serving.intValue() < this.targetServings.intValue()) {
Expand Down
12 changes: 6 additions & 6 deletions examples/actors-dp-js/philosophers.js
View file
Original file line number Diff line number Diff line change
Expand Up @@ -67,7 +67,7 @@ class Philosopher {
}

async getFirstFork (attempt) {
if (await actor.call(actor.proxy('Fork', this.firstFork), 'pickUp', this.kar.id)) {
if (await actor.call(this, actor.proxy('Fork', this.firstFork), 'pickUp', this.kar.id)) {
return actor.tailCall(this, 'getSecondFork', 1)
} else {
if (attempt > 5) {
Expand All @@ -79,7 +79,7 @@ class Philosopher {
}

async getSecondFork (attempt) {
if (await actor.call(actor.proxy('Fork', this.secondFork), 'pickUp', this.kar.id)) {
if (await actor.call(this, actor.proxy('Fork', this.secondFork), 'pickUp', this.kar.id)) {
return actor.tailCall(this, 'eat', this.servingsEaten)
} else {
if (attempt > 5) {
Expand All @@ -92,8 +92,8 @@ class Philosopher {

async eat (serving) {
if (verbose) console.log(`${this.kar.id} ate serving number ${serving}`)
await actor.call(actor.proxy('Fork', this.secondFork), 'putDown', this.kar.id)
await actor.call(actor.proxy('Fork', this.firstFork), 'putDown', this.kar.id)
await actor.call(this, actor.proxy('Fork', this.secondFork), 'putDown', this.kar.id)
await actor.call(this, actor.proxy('Fork', this.firstFork), 'putDown', this.kar.id)
this.servingsEaten = serving + 1
await actor.state.set(this, 'servingsEaten', this.servingsEaten)
if (this.servingsEaten < this.targetServings) {
Expand Down Expand Up @@ -163,11 +163,11 @@ class Table {

class Cafe {
async occupancy (table) {
return actor.call(actor.proxy('Table', table), 'occupancy')
return actor.call(this, actor.proxy('Table', table), 'occupancy')
}

async seatTable (n = 5, servings = 20, requestId = uuidv4()) {
await actor.call(actor.proxy('Table', requestId), 'prepare', this.kar.id, n, servings, requestId)
await actor.call(this, actor.proxy('Table', requestId), 'prepare', this.kar.id, n, servings, requestId)
return requestId
}
}
Expand Down
2 changes: 1 addition & 1 deletion examples/actors-ykt/ykt.js
View file
Original file line number Diff line number Diff line change
Expand Up @@ -138,7 +138,7 @@ class Site {
this.company = state.company
await actor.state.set(this, 'cleanShutdown', false)
if (this.company !== undefined && state.cleanShutdown !== true) {
const employees = await actor.call(actor.proxy('Company', this.company), 'activeEmployees', this.name)
const employees = await actor.call(this, actor.proxy('Company', this.company), 'activeEmployees', this.name)
for (const sn of employees) {
this.workers[sn] = States.ONBOARDING // Not accurate, but will be corrected on next workerUpdate
}
Expand Down
5 changes: 3 additions & 2 deletions ...ples/misc/actor-timeout-java/src/main/java/com/ibm/research/kar/example/timeout/Test.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,7 @@
package com.ibm.research.kar.example.timeout;

import static com.ibm.research.kar.Kar.Actors.call;
import static com.ibm.research.kar.Kar.Actors.rootCall;
import static com.ibm.research.kar.Kar.Actors.tell;

import java.time.Duration;
Expand Down Expand Up @@ -49,7 +50,7 @@ public class Test extends ActorSkeleton {

@Remote public void B() {
System.out.println("Entering method B");
call(this, "A"); // synchronous call to self in a new session -> deadlock
rootCall(this, "A"); // synchronous call to self in a new session -> deadlock
System.out.println("Exiting method B");
}

Expand All @@ -59,7 +60,7 @@ public class Test extends ActorSkeleton {

@Remote public void externalA(JsonString target) {
ActorRef other = ref("Test", target.getString());
call(other, "A");
rootCall(other, "A");
}

@Remote public void echo(JsonString msg, JsonNumber count) {
Expand Down
4 changes: 2 additions & 2 deletions ...rver-middle/src/main/java/com/ibm/research/kar/example/timeout/middle/MiddleServices.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -62,7 +62,7 @@ public JsonValue doubler(JsonValue body) {

System.out.println("Awake; invoking first actor");
ActorRef myBackend = Kar.Actors.ref("SlowAdder", "Singleton");
JsonValue firstPart = Kar.Actors.call(myBackend, "add", Json.createValue(data), Json.createValue(delay));
JsonValue firstPart = Kar.Actors.rootCall(myBackend, "add", Json.createValue(data), Json.createValue(delay));

System.out.println("Received response " + firstPart + "; now sleeping for " + delay + " seconds.");
try {
Expand All @@ -72,7 +72,7 @@ public JsonValue doubler(JsonValue body) {
}

System.out.println("Awake; invoking second actor");
JsonValue secondPart = Kar.Actors.call(myBackend, "add", Json.createValue(data), Json.createValue(delay));
JsonValue secondPart = Kar.Actors.rootCall(myBackend, "add", Json.createValue(data), Json.createValue(delay));
System.out.println("Received response " + secondPart + "; now sleeping for " + delay + " seconds.");
try {
Thread.sleep(delay * 1000);
Expand Down
47 changes: 11 additions & 36 deletions sdk-java/kar-runtime-liberty/src/main/java/com/ibm/research/kar/Kar.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -521,7 +521,7 @@ public static void tell(ActorRef actor, String path, JsonValue... args) {

/**
* Synchronous actor invocation where the invoked method will execute as part of
* the current session.
* the current chain of calls.
*
* @param caller The calling actor.
* @param actor The target actor.
Expand All @@ -532,8 +532,7 @@ public static void tell(ActorRef actor, String path, JsonValue... args) {
public static JsonValue call(ActorInstance caller, ActorRef actor, String path, JsonValue... args)
throws ActorMethodNotFoundException, ActorMethodInvocationException {
try {
Response response = sidecar.actorCall(actor.getType(), actor.getId(), path, caller.getSession(),
packArgs(args));
Response response = sidecar.actorCall(actor.getType(), actor.getId(), path, caller.getSession(), packArgs(args));
return callProcessResponse(response);
} catch (WebApplicationException e) {
if (e.getResponse() != null && e.getResponse().getStatus() == 404) {
Expand All @@ -550,44 +549,17 @@ public static JsonValue call(ActorInstance caller, ActorRef actor, String path,
}

/**
* Synchronous actor invocation where the invoked method will execute as part of
* the specified session.
*
* @param session The session in which to execute the actor method
* @param actor The target actor.
* @param path The actor method to invoke.
* @param args The arguments with which to invoke the actor method.
* @return The result of the invoked actor method.
*/
public static JsonValue call(String session, ActorRef actor, String path, JsonValue... args)
throws ActorMethodNotFoundException, ActorMethodInvocationException, ActorMethodTimeoutException {
try {
Response response = sidecar.actorCall(actor.getType(), actor.getId(), path, session, packArgs(args));
return callProcessResponse(response);
} catch (WebApplicationException e) {
if (e.getResponse() != null && e.getResponse().getStatus() == 404) {
String msg = responseToString(e.getResponse());
throw new ActorMethodNotFoundException(
msg != null ? msg : "Not found: " + actor.getType() + "[" + actor.getId() + "]." + path, e);
} else if (e.getResponse() != null && e.getResponse().getStatus() == 408) {
throw new ActorMethodTimeoutException(
"Method timeout: " + actor.getType() + "[" + actor.getId() + "]." + path);
} else {
throw e;
}
}
}

/**
* Synchronous actor invocation where the invoked method will execute in a new
* session.
* Synchronous actor invocation where the execution of the invoked callee method
* will create a new chain of calls. Typically this method should only be used
* when the caller is not an actor as it does not create a caller-callee dependency
* that can be managed by KAR during failure recovery.
*
* @param actor The target Actor.
* @param path The actor method to invoke.
* @param args The arguments with which to invoke the actor method.
* @return The result of the invoked actor method.
*/
public static JsonValue call(ActorRef actor, String path, JsonValue... args)
public static JsonValue rootCall(ActorRef actor, String path, JsonValue... args)
throws ActorMethodNotFoundException, ActorMethodInvocationException {
try {
Response response = sidecar.actorCall(actor.getType(), actor.getId(), path, null, packArgs(args));
Expand All @@ -608,7 +580,10 @@ public static JsonValue call(ActorRef actor, String path, JsonValue... args)

/**
* Asynchronous actor invocation with eventual access to the result of the
* invocation.
* invocation which will execute in a new top-level chain of calls.
* Typically this method should only be used
* when the caller is not an actor because this API does not create a
* caller-callee dependency that can be managed by KAR during failure recovery.
*
* @param actor The target Actor.
* @param path The actor method to invoke.
Expand Down
25 changes: 6 additions & 19 deletions sdk-java/kar-runtime-quarkus/src/main/java/com/ibm/research/kar/Kar.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -401,7 +401,7 @@ public static Uni<Void> tell(ActorRef actor, String path, JsonValue... args) {

/**
* Synchronous actor invocation where the invoked method will execute as part of
* the current session.
* the current chain of calls.
*
* @param caller The calling actor.
* @param actor The target actor.
Expand All @@ -415,30 +415,17 @@ public static Uni<JsonValue> call(ActorInstance caller, ActorRef actor, String p
}

/**
* Synchronous actor invocation where the invoked method will execute as part of
* the specified session.
*
* @param session The session in which to execute the actor method
* @param actor The target actor.
* @param path The actor method to invoke.
* @param args The arguments with which to invoke the actor method.
* @return The result of the invoked actor method.
*/
public static Uni<JsonValue> call(String session, ActorRef actor, String path, JsonValue... args) {
return sidecar.actorCall(actor.getType(), actor.getId(), path, session, packArgs(args))
.chain(resp -> callProcessResponse(resp, actor, path));
}

/**
* Synchronous actor invocation where the invoked method will execute in a new
* session.
* Synchronous actor invocation where the execution of the invoked callee method
* will create a new chain of calls. Typically this method should only be used
* when the caller is not an actor as it does not create a caller-callee dependency
* that can be managed by KAR during failure recovery.
*
* @param actor The target Actor.
* @param path The actor method to invoke.
* @param args The arguments with which to invoke the actor method.
* @return The result of the invoked actor method.
*/
public static Uni<JsonValue> call(ActorRef actor, String path, JsonValue... args) {
public static Uni<JsonValue> rootCall(ActorRef actor, String path, JsonValue... args) {
return sidecar.actorCall(actor.getType(), actor.getId(), path, packArgs(args))
.chain(resp -> callProcessResponse(resp, actor, path));
}
Expand Down
11 changes: 8 additions & 3 deletions sdk-java/kar-runtime-quarkus/src/main/java/com/ibm/research/kar/quarkus/ActorEndpoints.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -186,10 +186,12 @@ public Uni<Response> invokeActorMethod(String type, String id, String sessionid,
actorObj.setSession(sessionid);
Object result = actorMethod.invokeWithArguments(actuals);
if (result == null || actorMethod.type().returnType().equals(Void.TYPE)) {
actorObj.setSession(priorSession);
return Uni.createFrom().item(Response.status(NO_CONTENT).build());
} else if (result instanceof Uni<?>) {
return ((Uni<?>)result)
.chain(res -> {
actorObj.setSession(priorSession);
if (res == null) {
return Uni.createFrom().item(Response.status(NO_CONTENT).build());
} else {
Expand All @@ -198,16 +200,19 @@ public Uni<Response> invokeActorMethod(String type, String id, String sessionid,
return Uni.createFrom().item(resp);
}
})
.onFailure().recoverWithItem(t -> encodeInvocationError(t));
.onFailure().recoverWithItem(t -> {
actorObj.setSession(priorSession);
return encodeInvocationError(t);
});
} else {
actorObj.setSession(priorSession);
JsonObject response = encodeInvocationResult(result);
Response resp = Response.ok().type(KAR_ACTOR_JSON).entity(response.toString()).build();
return Uni.createFrom().item(resp);
}
} catch (Throwable t) {
return Uni.createFrom().item(encodeInvocationError(t));
} finally {
actorObj.setSession(priorSession);
return Uni.createFrom().item(encodeInvocationError(t));
}
}

Expand Down
2 changes: 1 addition & 1 deletion sdk-js/index.d.ts
View file
Original file line number Diff line number Diff line change
Expand Up @@ -158,7 +158,7 @@ export namespace actor {
* @param path The actor method to invoke.
* @param args The arguments with which to invoke the actor method.
*/
export function call (callee: Actor, path: string, ...args: any[]): Promise<any>;
export function rootCall (callee: Actor, path: string, ...args: any[]): Promise<any>;

/**
* Asynchronously remove all user-level and runtime state of an Actor.
Expand Down
17 changes: 17 additions & 0 deletions sdk-js/index.js
View file
Original file line number Diff line number Diff line change
Expand Up @@ -144,6 +144,22 @@ function actorProxy (type, id) { return { kar: { type, id } } }
const actorTell = (actor, path, ...args) => post(`actor/${actor.kar.type}/${actor.kar.id}/call/${path}`, args, { 'Content-Type': 'application/kar+json', Pragma: 'async' })

function actorCall (...args) {
if (typeof args[1] === 'string') {
// call (callee:Actor, path:string, ...args:any[]):Promise<any>;
// TODO: Once we release an SDK with rootCall, then remove this option
const ta = args.shift()
const path = args.shift()
return postActor(`actor/${ta.kar.type}/${ta.kar.id}/call/${path}`, args, { 'Content-Type': 'application/kar+json' })
} else {
// call (from:Actor, callee:Actor, path:string, ...args:any[]):Promise<any>;
const sa = args.shift()
const ta = args.shift()
const path = args.shift()
return postActor(`actor/${ta.kar.type}/${ta.kar.id}/call/${path}?session=${sa.kar.session}`, args, { 'Content-Type': 'application/kar+json' })
}
}

function actorRootCall (...args) {
if (typeof args[1] === 'string') {
// call (callee:Actor, path:string, ...args:any[]):Promise<any>;
const ta = args.shift()
Expand Down Expand Up @@ -386,6 +402,7 @@ module.exports = {
proxy: actorProxy,
tell: actorTell,
call: actorCall,
rootCall: actorRootCall,
asyncCall: actorAsyncCall,
remove: actorDelete,
tailCall: actorEncodeTailCall,
Expand Down

0 comments on commit f854211

Please sign in to comment.