Github
文档
Introduction - Actor framework featuring actors and agents
Sento is a 'message passing' library/framework with actors similar to Erlang or Akka. It supports creating systems that should work reactive, require parallel computing and event based message handling.
Sento features:
- Actors with
ask(?) andtell(!) operations.askcan be asynchronous or synchronous. - Agents: Agents are a specialization of Actors for wrapping state with a standardized interface of
init,getandset. There are also specialized Agents for Common Lisps array and hash-map data structures. - Router: Router offers a similar interface as Actor with
askandtellbut collects multiple Actors for load-balancing. - EventStream: all Actors and Agents are connected to an EventStream and can subscribe to messages or publish messages. This is similar to an event-bus.
- Tasks: a simple API for concurrency.
Intro
(Please also checkout the API documentation for further information) (for migrations from Sento v2, please check below migration guide)
Projects using Sento (for example usage):
- Chipi automation tool: Actors used for foundational primitives like 'items' and 'persistences'.
- KNX-conn: Used for asynchronous reading/writing from/to KNX bus
- Hunchentoot taskmanager: High throughput Hunchentoot task manager
Creating an actor-system
The first thing you wanna do is to create an actor system. In simple terms, an actor system is a container where all actors live in. So at any time the actor system knows which actors exist.
To create an actor system we can first change package to :sento-user because it imports the majority of necessary namespaces fopr convenience. Then, do:
(defvar *system* (make-actor-system))
When we look at *system* in the repl we see some information of the actor system:
#<ACTOR-SYSTEM config: (DISPATCHERS
(SHARED (WORKERS 4 STRATEGY RANDOM))
TIMEOUT-TIMER
(RESOLUTION 500 MAX-SIZE 1000)
EVENTSTREAM
(DISPATCHER-ID SHARED)
SCHEDULER
(ENABLED TRUE RESOLUTION 100 MAX-SIZE 500)
), user actors: 0, internal actors: 5>
The actor-system has, by default, four shared message dispatcher workers. Depending on how busy the system tends to be this default can be increased. Those four workers are part of the 'internal actors'. The 5th actor drives the event-stream (later more on that, but in a nutshell it's something like an event bus).
There are none 'user actors' yet, and the 'config' is the default config specifying the number of message dispatch workers (4) and the strategy they use to balance throughput, 'random' here.
Using a custom config is it possible to change much of those defaults. For instance, create custom dispatchers, i.e. a dedicated dispatcher used for the 'Tasks' api (see later for more info). The event-stream by default uses the global 'shared' dispatcher. Changing the config it would be possible to have the event-stream actor use a :pinned dispatcher (more on dispatchers later) to optimize throughput. Etc.
Actors live in the actor system, but more concrete in an actor-context. An actor-context contains a collection (of actors) and represents a Common Lisp protocol that defines a set of generic functions for creating, removing and finding actors in an actor-context. The actor system itself is also implementing the actor-context protocol, so it also acts as such and hence the protocol ac (actor-context) is used to operate on the actor system.
I.e. to shutdown the actor system one has to execute: (ac:shutdown *system*).
Creating and using actors
Now we want to create actors.
(actor-of *system* :name "answerer"
:receive
(lambda (msg)
(let ((output (format nil "Hello ~a" msg)))
(reply output))))
This creates an actor in *system*. Notice that the actor is not assigned to a variable (but you can). It is now registered in the system. Using function ac:find-actors you'll be able to find it again. Of course it makes sense to store important actors that are frequently used in a defparameter variable.
The :receive key argument to actor-of is a function which implements the message processing behaviour of an actor. The parameter to the 'receive' function is just the received message (msg).
actor-of also allows to specify the initial state, a name, and a custom actor type via key parameters. By default a standard actor of type 'actor is created. It is possible to subclass 'actor and specify your own. It is further possible to specify an 'after initialization' function, using the :init key, and 'after destroy' function using :destroy keyword. :init can, for example, be used to subscribe to the event-stream for listening to important messages.
The return value of 'receive' function is only used when using the synchronous ask-s function to 'ask' the actor. Using ask (equivalent: ?) the return value is ignored. If an answer should be provided to an asking actor, or if replying is part of an interface contract, then reply should be used.
The above actor was stored to a variable *answerer*. We can evaluate this in repl and see:
#<ACTOR path: /user/answerer, cell: #<ACTOR answerer, running: T, state: NIL, message-box: #<SENTO.MESSAGEB:MESSAGE-BOX/DP mesgb-1356, processed messages: 1, max-queue-size: 0, queue: #<SENTO.QUEUE:QUEUE-UNBOUNDED 82701A6D13>>>>
We'll see the 'path' of the actor. The prefix '/user' means that the actor was created in a user actor context of the actor system. Further we see whether the actor is 'running', its 'state' and the used 'message-box' type, by default it uses an unbounded queue.
Now, when sending a message using 'ask' pattern to the above actor like so:
(? *answerer* "FooBar")
we'll get a 'future' as result, because ?/ask is asynchronous.
#<FUTURE promise: #<BLACKBIRD-BASE:PROMISE
finished: NIL
errored: NIL
forward: NIL 80100E8B7B>>
We can check for a 'future' result. By now the answer from the *answerer* (via reply) should be available:
USER> (fresult *)
"Hello FooBar"
If the reply had not been received yet, fresult would return :not-ready. So, fresult doesn't block, it is necessary to repeatedly probe using fresult until result is other than :not-ready.
A nicer and asynchronous way without querying is to use fcompleted. Using fcompleted you setup a callback function that is called with the result when it is available. Like this:
(fcompleted
(? *answerer* "Buzz")
(result)
(format t "The answer is: ~a~%" result))
Which will asynchronously print "The answer is: Hello Buzz" after a short while.
This will also work when the ask/? was used with a timeout, in which case result will be a tuple of (:handler-error . <ask-timeout condition>) if the operation timed out.
Creating child actors
To build actor hierarchies one has to create actors in actors. This is of course possible. There are two options for this.
- Actors are created as part of
actor-ofs:initfunction like so:
(actor-of *system*
:name "answerer-with-child"
:receive
(lambda (msg)
(let ((output (format nil "Hello ~a" msg)))
(reply output)))
:init
(lambda (self)
(actor-of self
:name "child-answerer"
:receive
(lambda (msg)
(let ((output (format nil "Hello-child ~a" msg)))
(format nil "~a~%" output))))))
Notice the context for creating 'child-answerer', it is self, which is 'answerer-with-child'.
- Or it is possible externally like so:
(actor-of *answerer* :name "child-answerer"
:receive
(lambda (msg)
(let ((output (format nil "~a" "Hello-child ~a" msg)))
(format nil "~a~%" output))))
This uses *answerer* context as parameter of actor-of. But has the same effect as above.
Now we can check if there is an actor in context of 'answerer-with-child':
USER> (all-actors *actor-with-child*)
(#<ACTOR path: /user/answerer-with-child/child-answerer, cell: #<ACTOR child-answerer, running: T, state: NIL, message-box: #<SENTO.MESSAGEB:MESSAGE-BOX/DP mesgb-1374, processed messages: 0, max-queue-size: 0, queue: #<SENTO.QUEUE:QUEUE-UNBOUNDED 8200A195FB>>>>)
The 'path' is what we expected: '/user/answerer-with-child/child-answerer'.
Ping Pong
Another example that only works with tell/! (fire and forget).
We have those two actors.
The 'ping' actor:
(defparameter *ping*
(actor-of *system*
:receive
(lambda (msg)
(cond
((consp msg)
(case (car msg)
(:start-ping
(progn
(format t "Starting ping...~%")
(! (cdr msg) :ping *self*)))))
((eq msg :pong)
(progn
(format t "pong~%")
(sleep 2)
(reply :ping)))))))
And the 'pong' actor:
(defparameter *pong*
(actor-of *system*
:receive
(lambda (msg)
(case msg
(:ping
(progn
(format t "ping~%")
(sleep 2)
(reply :pong)))))))
The 'ping' actor understands a :start-ping message which is a cons and has as cdr the 'pong' actor instance.
It also understands a :pong message as received from 'pong' actor.
The 'pong' actor only understands a :ping message. Each of the actors respond with either :ping or :pong respectively after waiting 2 seconds.
We trigger the ping-pong by doing:
(! *ping* `(:start-ping . ,*pong*))
And then see in the console like:
Starting ping...
ping
pong
ping
...
To stop the ping-pong one just has to send (! *ping* :stop) to one of them.
:stop will completely stop the actors message processing, and the actor will not be useable anymore.
Synchronous ask
At last an example for the synchronous 'ask', ask-s. It is insofar similar to ask that it provides a result to the caller. However, it is not bound to reply as with ask. Here, the return value of the 'receive' function is returned to the caller, and ask-s will block until 'receive' function returns.
Beware that ask-s will dead-lock your actor when ask-s is used to call itself.
Let's make an example:
(defparameter *s-asker*
(actor-of *system*
:receive
(lambda (msg)
(cond
((stringp msg)
(format nil "Hello ~a" msg))
(t (format nil "Unknown message!"))))))
So we can do:
USER> (ask-s *s-asker* "Foo")
"Hello Foo"
USER> (ask-s *s-asker* 'foo)
"Unknown message!"
Dispatchers :pinned vs. :shared
Dispatchers are somewhat alike thread pools. Dispatchers of the :shared type are a pool of workers. Workers are actors using a :pinned dispatcher. :pinned just means that an actor spawns its own mailbox thread.
So :pinned and :shared are types of dispatchers. :pinned spawns its own mailbox thread, :shared uses a worker pool to handle the mailbox messages.
By default an actor created using actor-of uses a :shared dispatcher type which uses the shared message dispatcher that is automatically setup in the system.
When creating an actor it is possible to specify the dispatcher-id. This parameter specifies which 'dispatcher' should handle the mailbox queue/messages.
Please see below for more info on dispatchers.
Finding actors in the context
If actors are not directly stored in a dynamic or lexical context they can still be looked up and used. The actor-context protocol contains a function find-actors which can lookup actors in various ways. Checkout the API documentation.
Mapping futures with fmap
Let's asume we have such a simple actor that just increments the value passed to it.
(defparameter *incer*
(actor-of *system*
:receive (lambda (value)
(reply (1+ value)))))
Since ask returns a future it is possible to map multiple ask operations like this:
(-> (ask *incer* 0)
(fmap (result)
(ask *incer* result))
(fmap (result)
(ask *incer* result))
(fcompleted (result)
(format t "result: ~a~%" result)
(assert (= result 3))))
ask-s and ask with timeout
A timeout (in seconds) can be specified for both ask-s and
ask and is done like so:
To demonstrate this we could setup an example 'sleeper' actor:
(ac:actor-of *system*
:receive
(lambda (msg)
(sleep 5)))
If we store this to *sleeper* and do the following, the
ask-s will return a handler-error with an
ask-timeout condition.
(act:ask-s *sleeper* "Foo" :time-out 2)
(:HANDLER-ERROR . #<CL-GSERVER.UTILS:ASK-TIMEOUT #x30200319F97D>)
This works similar with the ask only that the future will
be fulfilled with the handler-error cons.
To get a readable error message of the condition we can do:
CL-USER> (format t "~a" (cdr *))
A timeout set to 2 seconds occurred. Cause:
#<BORDEAUX-THREADS:TIMEOUT #x302002FAB73D>
Note that ask-s uses the calling thread for the timeout checks.
ask uses a wheel timer to handle timeouts. The default resolution for ask timeouts is 500ms with a maximum size of wheel slots (registered timeouts) of 1000. What this means is that you can have timeouts of a multiple of 500ms and 1000 ask operations with timeouts. This default can be tweaked when creating an actor-system, see API documentation for more details.
Long running and asynchronous operations in receive
Be careful with doing long running computations in the receive function message handler, because it will block message processing. It is advised to use a third-party thread-pool or a library like lparallel to do the computations with, and return early from the receive message handler.
The computation result can be 'awaited' for in an asynchronous manner and a response to *sender* can be sent manually (via reply). The sender of the original message is set to the dynamic variable *sender*.
Due to an asynchronous callback of a computation running is a separate thread, the *sender* must be copied into a lexical environment because at the time of when the callback is executed the *sender* can have a different value.
For instance, if there is a potentially long running and asynchronous operation happening in 'receive', the original sender must be captured and the async operation executed in a lexical context, like so (receive function):
(lambda (msg)
(case msg
(:do-lengthy-op
(let ((sender *sender*))
;; do lengthy computation
(reply :my-later-reply sender)))
(otherwise
;; do other non async stuff
(reply :my-reply))))
Notice that for the lengthy operation the sender must be captured because if the lengthy operation is asynchronous 'receive' function is perhaps called for another message where *sender* is different. In that case sender must be supplied explicitly for reply.
See this test for more info.
NOTE: you should not change actor state from within an asynchronously executed operation in receive. This is not
