Interrogating Actors with Promises

Problem

Pony gives us an excellent abstraction for actors. We can define fields within those actors to maintain state and rely on the single-threaded nature of inbound message processing to ensure safe access to those fields. A problem arises when one actor wants to access the internal state of another actor.

Let's say that you want to collect values obtained from multiple actors without having to create a giant state machine. To illustrate this problem, we'll use an actor called AccountAggregate that is maintaining an internal balance.

This actor might look something like this:

actor AccountAggregate let _account: String var _balance: U64 new create (account: String , starting_balance: U64 ) => _account = account _balance = starting_balance be handle_tx_event(tx: TransactionEvent ) => _balance = _balance + tx.amount()

In our sample problem, the system might be holding onto hundreds of instances of the AccountAggregate actor, each with its own balance. What if we want to make a quick tour through all of these actors and ask them for their balances for display on a dashboard of some kind? We can't access the individual fields of the actor.

We can try to write a method like this that returns the internal state:

fun balance (): U64 => _balance

Adding this method compiles. But what happens if we attempt to use this method?

let bal = savings.balance()

This line of code doesn't compile. This is because the receiver (savings) is a tag (an opaque reference that allows neither read nor write, only send). Our options are getting more and more limited, it seems.

Solution

Unlike some other languages with native actor patterns, we don't have primitives to ask for values or await responses from actors in Pony. As mentioned in the access pattern, we can send a lambda value to the actor which allows for internal state to be captured as a parameter, but there might be a cleaner way to deal with this one problem: Promises.

A promise lets us declare that we realize that some value will either be fulfilled or rejected sometime in the future by whatever has been tasked with that promise. Since a Promise is an actor, we can send a promise to an actor as a tag without breaking any of the safety rules of actors and messaging.

In the simplest case, we can have the AccountAggregate actor fulfill the promise inside a behavior:

be balance(p: Promise [ U64 ]) => p(_balance)

We can then send the promise to the aggregate with the following code:

let p = Promise [ U64 ] agg.balance(p)

This is somewhat useful, but the value of the promise is lost. We still want to be able to respond to the value used to fulfill the promise somehow. We can do this with promise chaining:

let p = Promise [ U64 ] p.next[ None ]( Outputter ~output(env)) agg.balance(p)

This gets us a little closer to what we want. Now, when the aggregate actor fulfills the promise, the result of that fulfillment will be sent as a parameter to the partially-applied output function on the Outputter primitive.

Getting better, but not good enough. What we really want to be able to do is query multiple actors to get the account summary data and then send all of that data (preferably bundled up in a nice array) to a destination actor that can then display and/or process the information. For this we're going to need an intermediary - something that awaits promise fulfillment and adds to a collection when fulfilled. Once this intermediary has received all expected fulfillments, it can then fulfill a single promise of the collection. This intermediary promise can be created using the Promises.join function.

Now we can create multiple promises to send to multiple bank accounts:

let accounts = [ "0001" ; "0002" ; "0003" ; "0004" ] let create_summary_promise = {(account: String ): Promise [ AccountSummary ] => let aggregate = AccountAggregate (account, 6000 ) aggregate.handle_tx_event( recover TransactionEvent ( 351 ) end ) aggregate.handle_tx_event( recover TransactionEvent ( 224 ) end ) let p = Promise [ AccountSummary ] aggregate.summarize(p) p } Promises [ AccountSummary ].join( Iter [ String ](accounts.values()) .map[ Promise [ AccountSummary ]](consume create_summary_promise)) .next[ None ]( recover this ~receive_collection() end )

Our bank account aggregate can be modified to include an account summary with the summarize method:

be summarize(p: Promise [ AccountSummary ]) => p( recover AccountSummary (_balance, _account) end )

Finally, we add the behavior to our Main actor that will respond to the list of account summaries:

be receive_collection(coll: Array [ AccountSummary ] ) => _env.out.print( "received account summaries:" ) for summary in coll.values() do _env.out.print( "Account " + summary.accountnumber() + ": $" + summary.currentbalance().string()) end

Putting it all together, we can now write code like the following that creates multiple actors and queries their internal state in a completely asynchronous fashion:

use "itertools" use "promises" class val TransactionEvent let _amount : U64 new create (amount: U64 ) => _amount = amount fun transaction_amount () : U64 => _amount class val AccountSummary let _balance : U64 let _account : String new create (balance: U64 , account: String ) => _balance = balance _account = account fun currentbalance () : U64 => _balance fun accountnumber () : String => _account actor AccountAggregate let _account: String var _balance: U64 new create (account: String , starting_balance: U64 ) => _account = account _balance = starting_balance be handle_tx_event(tx: TransactionEvent ) => _balance = _balance + tx.transaction_amount() be summarize(p: Promise [ AccountSummary ]) => p( recover AccountSummary (_balance, _account) end ) actor Main let _env: Env new create (env: Env ) => _env = env let accounts = [ "0001" ; "0002" ; "0003" ; "0004" ] let create_summary_promise = {(account: String ): Promise [ AccountSummary ] => let aggregate = AccountAggregate (account, 6000 ) aggregate.handle_tx_event( recover TransactionEvent ( 351 ) end ) aggregate.handle_tx_event( recover TransactionEvent ( 224 ) end ) let p = Promise [ AccountSummary ] aggregate.summarize(p) p } Promises [ AccountSummary ].join( Iter [ String ](accounts.values()) .map[ Promise [ AccountSummary ]](consume create_summary_promise)) .next[ None ]( recover this ~receive_collection() end ) be receive_collection(coll: Array [ AccountSummary ] ) => _env.out.print( "received account summaries:" ) for summary in coll.values() do _env.out.print( "Account " + summary.accountnumber() + ": $" + summary.currentbalance().string()) end

Discussion

Actor systems have been around for quite some time now, but most developers don't default to modeling their problems as actor patterns. Most of us want to solve this problem with synchronous code that looks like this:

for acct in _accounts.values() do _summaries.push(acct.summarize()) end

The problem with this is that as our real-world problems get more complex, simple loops like this are just not powerful enough. In bigger, more complex models, there is often a cost to asking an actor for its internal state. It might not be a pre-calculated field. Instead, invoking summarize might make calls to external systems, databases, or microservices.

Naively running through the summarization method in a for loop could cause a consumer to wait an indeterminate amount of time. By sending out a flood of promises, we can let each of the actors fulfill the promise on their own time and we'll get the results back far sooner than if we'd done the requests synchronously. This also gives us an added degree of reliability - by sending out these promises, we can also set a timeout in the collector so that we can build in things like a "circuit breaker" and return data indicating that we couldn't summarize all of the accounts.

In conclusion, Pony's actor system is incredibly powerful and some of that power comes from its deliberate restrictions. Learning how to embrace the actor model in combination with promises can provide an elegant solution to complex problems.