CAF User Manual¶
C++ Actor Framework version 0.17.5.
Contents¶
Introduction¶
Before diving into the API of CAF, we discuss the concepts behind it and explain the terminology used in this manual.
Actor Model¶
The actor model describes concurrent entities—actors—that do not share state and communicate only via asynchronous message passing. Decoupling concurrently running software components via message passing avoids race conditions by design. Actors can create—spawn—new actors and monitor each other to build fault-tolerant, hierarchical systems. Since message passing is network transparent, the actor model applies to both concurrency and distribution.
Implementing applications on top of low-level primitives such as mutexes and semaphores has proven challenging and error-prone. In particular when trying to implement applications that scale up to many CPU cores. Queueing, starvation, priority inversion, and false sharing are only a few of the issues that can decrease performance significantly in mutex-based concurrency models. In the extreme, an application written with the standard toolkit can run slower when adding more cores.
The actor model has gained momentum over the last decade due to its high level of abstraction and its ability to scale dynamically from one core to many cores and from one node to many nodes. However, the actor model has not yet been widely adopted in the native programming domain. With CAF, we contribute a library for actor programming in C++ as open-source software to ease native development of concurrent as well as distributed systems. In this regard, CAF follows the C++ philosophy building the highest abstraction possible without sacrificing performance.
Terminology¶
CAF is inspired by other implementations based on the actor model such as Erlang or Akka. It aims to provide a modern C++ API allowing for type-safe as well as dynamically typed messaging. While there are similarities to other implementations, we made many different design decisions that lead to slight differences when comparing CAF to other actor frameworks.
Dynamically Typed Actor¶
A dynamically typed actor accepts any kind of message and dispatches on its content dynamically at the receiver. This is the traditional messaging style found in implementations like Erlang or Akka. The upside of this approach is (usually) faster prototyping and less code. This comes at the cost of requiring excessive testing.
Statically Typed Actor¶
CAF achieves static type-checking for actors by defining abstract messaging interfaces. Since interfaces define both input and output types, CAF is able to verify messaging protocols statically. The upside of this approach is much higher robustness to code changes and fewer possible runtime errors. This comes at an increase in required source code, as developers have to define and use messaging interfaces.
Actor References¶
CAF uses reference counting for actors. The three ways to store a reference to an actor are addresses, handles, and pointers. Note that address does not refer to a memory region in this context.
Address¶
Each actor has a (network-wide) unique logical address. This identifier is
represented by actor_addr, which allows to identify and monitor an actor.
Unlike other actor frameworks, CAF does not allow users to send messages to
addresses. This limitation is due to the fact that the address does not contain
any type information. Hence, it would not be safe to send it a message, because
the receiving actor might use a statically typed interface that does not accept
the given message. Because an actor_addr fills the role of an identifier, it
has weak reference semantics (see Reference Counting).
Handle¶
An actor handle contains the address of an actor along with its type information
and is required for sending messages to actors. The distinction between handles
and addresses—which is unique to CAF when comparing it to other actor
systems—is a consequence of the design decision to enforce static type
checking for all messages. Dynamically typed actors use actor handles, while
statically typed actors use typed_actor<...> handles. Both types have
strong reference semantics (see Reference Counting).
Pointer¶
In a few instances, CAF uses strong_actor_ptr to refer to an actor using
strong reference semantics (see Reference Counting) without knowing the
proper handle type. Pointers must be converted to a handle via actor_cast
(see Converting Actor References with actor_cast) prior to sending messages. A strong_actor_ptr can be
null.
Spawning¶
Spawning an actor means to create and run a new actor.
Monitor¶
A monitored actor sends a down message (see Down Handler) to all actors monitoring it as part of its termination. This allows actors to supervise other actors and to take actions when one of the supervised actors fails, i.e., terminates with a non-normal exit reason.
Link¶
A link is a bidirectional connection between two actors. Each actor sends an exit message (see Exit Handler) to all of its links as part of its termination. Unlike down messages, exit messages cause the receiving actor to terminate as well when receiving a non-normal exit reason per default. This allows developers to create a set of actors with the guarantee that either all or no actors are alive. Actors can override the default handler to implement error recovery strategies.
Experimental Features¶
Sections that discuss experimental features are highlighted with experimental. The API of such features is not stable. This means even minor updates to CAF can come with breaking changes to the API or even remove a feature completely. However, we encourage developers to extensively test such features and to start discussions to uncover flaws, report bugs, or tweaking the API in order to improve a feature or streamline it to cover certain use cases.
Overview¶
Compiling CAF requires CMake and a C++11-compatible compiler. To get and compile the sources on UNIX-like systems, type the following in a terminal:
git clone https://github.com/actor-framework/actor-framework
cd actor-framework
./configure
make -C build
make -C build test [optional]
make -C build install [as root, optional]
If the output of make test indicates an error, please submit a bug report
that includes (a) your compiler version, (b) your OS, and (c) the content of the
file build/Testing/Temporary/LastTest.log.
The configure script provides several build options for advanced configuration.
Please run ./configure -h to see all available options. If you are building
CAF only as a dependency, disabling the unit tests and the examples can safe you
some time during the build.
Note
The configure script provides a convenient way for creating a build
directory and calling CMake. Users that are familiar with CMake can of course
also use CMake directly and avoid using the configure script entirely.
On Windows, we recomment using the CMake GUI to generate a Visual Studio project file for CAF.
Features¶
- Lightweight, fast and efficient actor implementations
- Network transparent messaging
- Error handling based on Erlang’s failure model
- Pattern matching for messages as internal DSL to ease development
- Thread-mapped actors for soft migration of existing applications
- Publish/subscribe group communication
Minimal Compiler Versions¶
- GCC 4.8
- Clang 3.4
- Visual Studio 2015, Update 3
Supported Operating Systems¶
- Linux
- Mac OS X
- Windows (static library only)
Hello World Example¶
#include <string>
#include <iostream>
#include "caf/all.hpp"
using std::endl;
using std::string;
using namespace caf;
behavior mirror(event_based_actor* self) {
// return the (initial) actor behavior
return {
// a handler for messages containing a single string
// that replies with a string
[=](const string& what) -> string {
// prints "Hello World!" via aout (thread-safe cout wrapper)
aout(self) << what << endl;
// reply "!dlroW olleH"
return string(what.rbegin(), what.rend());
}
};
}
void hello_world(event_based_actor* self, const actor& buddy) {
// send "Hello World!" to our buddy ...
self->request(buddy, std::chrono::seconds(10), "Hello World!").then(
// ... wait up to 10s for a response ...
[=](const string& what) {
// ... and print it
aout(self) << what << endl;
}
);
}
void caf_main(actor_system& system) {
// create a new actor that calls 'mirror()'
auto mirror_actor = system.spawn(mirror);
// create another actor that calls 'hello_world(mirror_actor)';
system.spawn(hello_world, mirror_actor);
// system will wait until both actors are destroyed before leaving main
}
// creates a main function for us that calls our caf_main
CAF_MAIN()
Type Inspection¶
CAF is designed with distributed systems in mind. Hence, all message types must
be serializable and need a platform-neutral, unique name (see
Adding Custom Message Types). Using a message type that is not serializable
causes a compiler error (see Whitelisting Unsafe Message Types). CAF serializes
individual elements of a message by using the inspection API. This API allows
users to provide code for serialization as well as string conversion, usually
with a single free function. The signature for a class my_class is always as
follows:
template <class Inspector>
typename Inspector::result_type inspect(Inspector& f, my_class& x) {
return f(...);
}
The function inspect passes meta information and data fields to the
variadic call operator of the inspector. The following example illustrates an
implementation of inspect for a simple POD called foo.
struct foo {
std::vector<int> a;
int b;
};
template <class Inspector>
typename Inspector::result_type inspect(Inspector& f, foo& x) {
return f(meta::type_name("foo"), x.a, x.b);
}
The inspector recursively inspects all data fields and has builtin support for
std::tuple, std::pair, arrays, as well as containers that provide the
member functions size, empty, begin and end.
We consciously made the inspect API as generic as possible to allow for extensibility. This allows users to use CAF’s types in other contexts, to implement parsers, etc.
Note
When converting a user-defined type to a string, CAF calls user-defined
to_string functions and prefers those over inspect.
Inspector Concept¶
The following concept class shows the requirements for inspectors. The
placeholder T represents any user-defined type. Usually error or
error_code.
Inspector {
using result_type = T;
static constexpr bool reads_state = ...;
static constexpr bool writes_state = ...;
template <class... Ts>
result_type operator()(Ts&&...);
}
A saving Inspector is required to handle constant lvalue and rvalue
references. A loading Inspector must only accept mutable lvalue
references to data fields, but still allow for constant lvalue references and
rvalue references to annotations.
Inspectors that only visit data fields (such as a serializer) sets
reads_state to true and writes_state to false. Inspectors that
override data fields (such as a deserializer) assign the inverse values.
These compile-time constants can be used in if constexpr statements or to
select different inspect overloads with enable_if.
Annotations¶
Annotations allow users to fine-tune the behavior of inspectors by providing
addition meta information about a type. All annotations live in the namespace
caf::meta and derive from caf::meta::annotation. An
inspector can query whether a type T is an annotation with
caf::meta::is_annotation<T>::value. Annotations are passed to the
call operator of the inspector along with data fields. The following list shows
all annotations supported by CAF:
type_name(n): Display type name asnin human-friendly output (position before data fields).hex_formatted(): Format the following data field in hex format.omittable(): Omit the following data field in human-friendly output.omittable_if_empty(): Omit the following data field in human-friendly output if it is empty.omittable_if_none(): Omit the following data field in human-friendly output if it equalsnone.save_callback(f): Callfifreads_state == true. Pass this callback after the data fields.load_callback(f): Callfwrites_state == true. Pass this callback after the data fields.
Backwards and Third-party Compatibility¶
CAF evaluates common free function other than inspect in order to simplify
users to integrate CAF into existing code bases.
Serializers and deserializers call user-defined serialize functions. Both
types support operator& as well as operator() for individual data
fields. A serialize function has priority over inspect.
Whitelisting Unsafe Message Types¶
Message types that are not serializable cause compile time errors when used in
actor communication. For messages that never cross the network, this errors can
be suppressed by whitelisting types with CAF_ALLOW_UNSAFE_MESSAGE_TYPE. The
macro is defined as follows.
#define CAF_ALLOW_UNSAFE_MESSAGE_TYPE(type_name) \
namespace caf { \
template <> \
struct allowed_unsafe_message_type<type_name> : std::true_type {}; \
}
Keep in mind that unsafe means that your program runs into undefined
behavior (or segfaults) when you break your promise and try to serialize
messages that contain unsafe message types.
Saving and Loading with Getters and Setters¶
Many classes shield their member variables with getter and setter functions.
This can be addressed by declaring the inspect function as friend, but
only when not dealing with 3rd party library types. For example, consider the
following class foo with getter and setter functions and no public access to
its members.
class foo {
public:
foo(int a0 = 0, int b0 = 0) : a_(a0), b_(b0) {
// nop
}
foo(const foo&) = default;
foo& operator=(const foo&) = default;
int a() const {
return a_;
}
void set_a(int val) {
a_ = val;
}
int b() const {
return b_;
}
void set_b(int val) {
b_ = val;
}
private:
int a_;
int b_;
};
Since there is no access to the data fields a_ and b_ (and assuming no
changes to foo are possible), we can serialize or deserialize from/to local
variables and use a load callback to write back to the object when loading
state.
template <class Inspector>
typename Inspector::result_type inspect(Inspector& f, foo& x) {
auto a = x.a();
auto b = x.b();
auto load = meta::load_callback([&]() -> error {
// Write back to x when loading values from the inspector.
x.set_a(a);
x.set_b(b);
return none;
});
return f(meta::type_name("foo"), a, b, load);
}
For more complicated cases, we can also split the inspect overload as follows:
template <class Inspector>
typename std::enable_if<Inspector::reads_state,
typename Inspector::result_type>::type
inspect(Inspector& f, my_class& x) {
// ... serializing ...
}
template <class Inspector>
typename std::enable_if<Inspector::writes_state,
typename Inspector::result_type>::type
inspect(Inspector& f, my_class& x) {
// ... deserializing ...
}
Message Handlers¶
Actors can store a set of callbacks—usually implemented as lambda
expressions—using either behavior or message_handler.
The former stores an optional timeout, while the latter is composable.
Definition and Composition¶
As the name implies, a behavior defines the response of an actor to
messages it receives. The optional timeout allows an actor to dynamically
change its behavior when not receiving message after a certain amount of time.
message_handler x1{
[](int i) { /*...*/ },
[](double db) { /*...*/ },
[](int a, int b, int c) { /*...*/ },
};
In our first example, x1 models a behavior accepting messages that consist
of either exactly one int, or one double, or three int values. Any
other message is not matched and gets forwarded to the default handler (see
Default Handler).
message_handler x2{
[](double db) { /*...*/ },
[](double db) { /* - unreachable - */ },
};
Our second example illustrates an important characteristic of the matching
mechanism. Each message is matched against the callbacks in the order they are
defined. The algorithm stops at the first match. Hence, the second callback in
x2 is unreachable.
message_handler x3 = x1.or_else(x2);
message_handler x4 = x2.or_else(x1);
Message handlers can be combined using or_else. This composition is
not commutative, as our third examples illustrates. The resulting message
handler will first try to handle a message using the left-hand operand and will
fall back to the right-hand operand if the former did not match. Thus,
x3 behaves exactly like x1. This is because the second
callback in x1 will consume any message with a single
double and both callbacks in x2 are thus unreachable.
The handler x4 will consume messages with a single
double using the first callback in x2, essentially
overriding the second callback in x1.
behavior bhvr{
[](int i) { /*...*/ },
[](double db) { /*...*/ },
[](int a, int b, int c) { /*...*/ },
after(std::chrono::seconds(10)) >> [] { /*..*/ },
};
Similar to message_handler, behavior stores a set of functions.
Additionally, behaviors accept optional timeouts as last argument to the
constructor. In the example above, the behavior calls the timeout handler after
receiving no messages for 10 seconds. This timeout triggers repeatedly, i.e.,
the timeout handler gets called again after 20 seconds when not receiving any
messages, then again after 30 seconds, and so on.
Atoms¶
Defining message handlers in terms of callbacks is convenient, but requires a simple way to annotate messages with meta data. Imagine an actor that provides a mathematical service for integers. It receives two integers, performs a user-defined operation and returns the result. Without additional context, the actor cannot decide whether it should multiply or add the integers, for example.
Thus, the operation must be encoded into the message. The Erlang programming language introduced an approach to use non-numerical constants, so-called atoms, which have an unambiguous, special-purpose type and do not have the runtime overhead of string constants.
Atoms in CAF are mapped to integer values at compile time. This mapping is
guaranteed to be collision-free and invertible, but limits atom literals to ten
characters and prohibits special characters. Legal characters are _0-9A-Za-z
and the whitespace character. Atoms are created using the constexpr function
atom, as the following example illustrates.
atom_value a1 = atom("add");
atom_value a2 = atom("multiply");
Warning
The compiler cannot enforce the restrictions at compile time, except for a
length check. The assertion atom("!?") != atom("?!") is not true, because
each invalid character translates to a whitespace character.
While the atom_value is computed at compile time, it is not
uniquely typed and thus cannot be used in the signature of a callback. To
accomplish this, CAF offers compile-time atom constants.
using add_atom = atom_constant<atom("add")>;
using multiply_atom = atom_constant<atom("multiply")>;
Using these constants, we can now define message passing interfaces in a convenient way:
behavior do_math{
[](add_atom, int a, int b) {
return a + b;
},
[](multiply_atom, int a, int b) {
return a * b;
}
};
// caller side: send(math_actor, add_atom::value, 1, 2)
Atom constants define a static member value. Please note that this
static value member does not have the type
atom_value, unlike std::integral_constant for example.
Actors¶
Actors in CAF represent a lightweight abstraction for units of computations. They are active objects in the sense that they own their state and do not allow others to access it. The only way to modify the state of an actor is sending messages to it.
CAF provides several actor implementations, each covering a particular use case. The available implementations differ in three characteristics:
- dynamically or statically typed
- class-based or function-based
- using asynchronous event handlers or blocking receives
These three characteristics can be combined freely, with one exception:
statically typed actors are always event-based. For example, an actor can have
dynamically typed messaging, implement a class, and use blocking receives. The
common base class for all user-defined actors is called local_actor.
Dynamically typed actors are more familiar to developers coming from Erlang or Akka. They (usually) enable faster prototyping but require extensive unit testing. Statically typed actors require more source code but enable the compiler to verify communication between actors. Since CAF supports both, developers can freely mix both kinds of actors to get the best of both worlds. A good rule of thumb is to make use of static type checking for actors that are visible across multiple translation units.
Actors that utilize the blocking receive API always require an exclusive thread of execution. Event-based actors, on the other hand, are usually scheduled cooperatively and are very lightweight with a memory footprint of only few hundred bytes. Developers can exclude (detach) event-based actors that potentially starve others from the cooperative scheduling while spawning it. A detached actor lives in its own thread of execution.
Environment / Actor Systems¶
All actors live in an actor_system representing an actor environment
including Scheduler, Registry, and optional components (modules)
such as a Middleman. A single process can have multiple actor_system
instances, but this is usually not recommended (a use case for multiple systems
is to strictly separate two or more sets of actors by running them in different
schedulers). For configuration and fine-tuning options of actor systems see
Configuring Actor Applications. A distributed CAF application consists of two or more
connected actor systems. We also refer to interconnected actor_system
instances as a distributed actor system.
Common Actor Base Types¶
The following pseudo-UML depicts the class diagram for actors in CAF. Irrelevant member functions and classes as well as mixins are omitted for brevity. Selected individual classes are presented in more detail in the following sections.
Class local_actor¶
The class local_actor is the root type for all user-defined actors
in CAF. It defines all common operations. However, users of the library
usually do not inherit from this class directly. Proper base classes for
user-defined actors are event_based_actor or
blocking_actor. The following table also includes member function
inherited from monitorable_actor and abstract_actor.
| Types | |
mailbox_type |
A concurrent, many-writers-single-reader queue type. |
| Constructors | |
(actor_config&) |
Constructs the actor using a config. |
| Observers | |
actor_addr address() |
Returns the address of this actor. |
actor_system& system() |
Returns context()->system(). |
actor_system& home_system() |
Returns the system that spawned this actor. |
execution_unit* context() |
Returns underlying thread or current scheduler worker. |
| Customization Points | |
on_exit() |
Performs cleanup steps. |
initialize() |
Installs an initial behavior. |
const char* name() |
Returns a debug name for this actor type. |
| Actor Management | |
link_to(other) |
Links to other (see Link). |
unlink_from(other) |
Remove the link to other. |
monitor(other) |
Adds a monitor to other (see Monitor). |
demonitor(other) |
Removes a monitor from whom. |
spawn(F fun, xs...) |
Spawns a new actor from fun. |
spawn<T>(xs...) |
Spawns a new actor of type T. |
| Message Processing | |
T make_response_promise<Ts...>() |
Allows an actor to delay its response message. |
T response(xs...) |
Convenience function for creating fulfilled promises. |
Class scheduled_actor¶
All scheduled actors inherit from scheduled_actor. This includes
statically and dynamically typed event-based actors as well as brokers
Network I/O with Brokers.
| Types | |
pointer |
scheduled_actor* |
exception_handler |
function<error (pointer, std::exception_ptr&)> |
default_handler |
function<result<message> (pointer, message_view&)> |
error_handler |
function<void (pointer, error&)> |
down_handler |
function<void (pointer, down_msg&)> |
exit_handler |
function<void (pointer, exit_msg&)> |
| Constructors | |
(actor_config&) |
Constructs the actor using a config. |
| Termination | |
quit() |
Stops this actor with normal exit reason. |
quit(error x) |
Stops this actor with error x. |
| Special-purpose Handlers | |
set_exception_handler(F f) |
Installs f for converting exceptions to errors (see Errors). |
set_down_handler(F f) |
Installs f to handle down messages (see Down Handler). |
set_exit_handler(F f) |
Installs f to handle exit messages (see Exit Handler). |
set_error_handler(F f) |
Installs f to handle error messages (see Error Handler). |
set_default_handler(F f) |
Installs f as fallback message handler (see Default Handler). |
Class blocking_actor¶
A blocking actor always lives in its own thread of execution. They are not as
lightweight as event-based actors and thus do not scale up to large numbers.
The primary use case for blocking actors is to use a scoped_actor
for ad-hoc communication to selected actors. Unlike scheduled actors, CAF does
not dispatch system messages to special-purpose handlers. A blocking
actors receives all messages regularly through its mailbox. A blocking
actor is considered done only after it returned from act (or
from the implementation in function-based actors). A scoped_actor
sends its exit messages as part of its destruction.
| Constructors | |
(actor_config&) |
Constructs the actor using a config. |
| Customization Points | |
void act() |
Implements the behavior of the actor. |
| Termination | |
const error& fail_state() |
Returns the current exit reason. |
fail_state(error x) |
Sets the current exit reason. |
| Actor Management | |
wait_for(Ts... xs) |
Blocks until all actors xs... are done. |
await_all_other_actors_done() |
Blocks until all other actors are done. |
| Message Handling | |
receive(Ts... xs) |
Receives a message using the callbacks xs.... |
receive_for(T& begin, T end) |
See Receive Loops. |
receive_while(F stmt) |
See Receive Loops. |
do_receive(Ts... xs) |
See Receive Loops. |
Messaging Interfaces¶
Statically typed actors require abstract messaging interfaces to allow the
compiler to type-check actor communication. Interfaces in CAF are defined using
the variadic template typed_actor<...>, which defines the proper
actor handle at the same time. Each template parameter defines one
input/output pair via
replies_to<X1,...,Xn>::with<Y1,...,Yn>. For inputs that do not
generate outputs, reacts_to<X1,...,Xn> can be used as shortcut for
replies_to<X1,...,Xn>::with<void>. In the same way functions cannot
be overloaded only by their return type, interfaces cannot accept one input
twice (possibly mapping it to different outputs). The example below defines a
messaging interface for a simple calculator.
using calculator_actor
= typed_actor<replies_to<add_atom, int32_t, int32_t>::with<int32_t>,
replies_to<sub_atom, int32_t, int32_t>::with<int32_t>>;
It is not required to create a type alias such as calculator_actor,
but it makes dealing with statically typed actors much easier. Also, a central
alias definition eases refactoring later on.
Interfaces have set semantics. This means the following two type aliases
i1 and i2 are equal:
using i1 = typed_actor<replies_to<A>::with<B>, replies_to<C>::with<D>>;
using i2 = typed_actor<replies_to<C>::with<D>, replies_to<A>::with<B>>;
Further, actor handles of type A are assignable to handles of type
B as long as B is a subset of A.
For convenience, the class typed_actor<...> defines the member
types shown below to grant access to derived types.
| Types | |
behavior_type |
A statically typed set of message handlers. |
base |
Base type for actors, i.e., typed_event_based_actor<...>. |
pointer |
A pointer of type base*. |
stateful_base<T> |
See stateful-actor. |
stateful_pointer<T> |
A pointer of type stateful_base<T>*. |
extend<Ts...> |
Extend this typed actor with Ts.... |
extend_with<Other> |
Extend this typed actor with all cases from Other. |
Spawning Actors¶
Both statically and dynamically typed actors are spawned from an
actor_system using the member function spawn. The
function either takes a function as first argument or a class as first template
parameter. For example, the following functions and classes represent actors.
// prototypes and forward declarations
behavior calculator_fun(event_based_actor* self);
void blocking_calculator_fun(blocking_actor* self);
calculator_actor::behavior_type typed_calculator_fun();
class calculator;
class blocking_calculator;
class typed_calculator;
Spawning an actor for each implementation is illustrated below.
auto a1 = system.spawn(blocking_calculator_fun);
auto a2 = system.spawn(calculator_fun);
auto a3 = system.spawn(typed_calculator_fun);
auto a4 = system.spawn<blocking_calculator>();
auto a5 = system.spawn<calculator>();
auto a6 = system.spawn<typed_calculator>();
Additional arguments to spawn are passed to the constructor of a class or
used as additional function arguments, respectively. In the example above, none
of the three functions takes any argument other than the implicit (optional)
self pointer.
Function-based Actors¶
When using a function or function object to implement an actor, the first
argument can be used to capture a pointer to the actor itself. The type
of this pointer is usually event_based_actor* or
blocking_actor*. The proper pointer type for any
typed_actor handle T can be obtained via
T::pointer (see Messaging Interfaces).
Blocking actors simply implement their behavior in the function body. The actor is done once it returns from that function.
Event-based actors can either return a behavior (see Message Handlers)
that is used to initialize the actor or explicitly set the initial behavior by
calling self->become(...). Due to the asynchronous, event-based nature of
this kind of actor, the function usually returns immediately after setting a
behavior (message handler) for the next incoming message. Hence, variables on
the stack will be out of scope once a message arrives. Managing state in
function-based actors can be done either via rebinding state with become,
using heap-located data referenced via std::shared_ptr or by using the
stateful actor abstraction (see Stateful Actors).
The following three functions implement the prototypes shown in spawn and illustrate one blocking actor and two event-based actors (statically and dynamically typed).
// function-based, dynamically typed, event-based API
behavior calculator_fun(event_based_actor*) {
return {
[](add_atom, int32_t a, int32_t b) { return a + b; },
[](sub_atom, int32_t a, int32_t b) { return a - b; },
};
}
// function-based, dynamically typed, blocking API
void blocking_calculator_fun(blocking_actor* self) {
bool running = true;
self->receive_while(running)( //
[](add_atom, int32_t a, int32_t b) { return a + b; },
[](sub_atom, int32_t a, int32_t b) { return a - b; },
[&](exit_msg& em) {
if (em.reason) {
self->fail_state(std::move(em.reason));
running = false;
}
});
}
// function-based, statically typed, event-based API
calculator_actor::behavior_type typed_calculator_fun() {
return {
[](add_atom, int32_t a, int32_t b) { return a + b; },
[](sub_atom, int32_t a, int32_t b) { return a - b; },
};
}
Class-based Actors¶
Implementing an actor using a class requires the following:
- Provide a constructor taking a reference of type
actor_config&as first argument, which is forwarded to the base class. The config is passed implicitly to the constructor when callingspawn, which also forwards any number of additional arguments to the constructor. - Override
make_behaviorfor event-based actors andactfor blocking actors.
Implementing actors with classes works for all kinds of actors and allows simple management of state via member variables. However, composing states via inheritance can get quite tedious. For dynamically typed actors, composing states is particularly hard, because the compiler cannot provide much help.
The following three classes implement the prototypes shown in Spawning Actors and again feature one blocking actor and two event-based actors (statically and dynamically typed).
// class-based, dynamically typed, event-based API
class calculator : public event_based_actor {
public:
calculator(actor_config& cfg) : event_based_actor(cfg) {
// nop
}
behavior make_behavior() override {
return calculator_fun(this);
}
};
// class-based, dynamically typed, blocking API
class blocking_calculator : public blocking_actor {
public:
blocking_calculator(actor_config& cfg) : blocking_actor(cfg) {
// nop
}
void act() override {
blocking_calculator_fun(this);
}
};
// class-based, statically typed, event-based API
class typed_calculator : public calculator_actor::base {
public:
typed_calculator(actor_config& cfg) : calculator_actor::base(cfg) {
// nop
}
behavior_type make_behavior() override {
return typed_calculator_fun();
}
};
Stateful Actors¶
The stateful actor API makes it easy to maintain state in function-based
actors. It is also safer than putting state in member variables, because the
state ceases to exist after an actor is done and is not delayed until the
destructor runs. For example, if two actors hold a reference to each other via
member variables, they produce a cycle and neither will get destroyed. Using
stateful actors instead breaks the cycle, because references are destroyed when
an actor calls self->quit() (or is killed externally). The
following example illustrates how to implement stateful actors with static
typing as well as with dynamic typing.
using cell = typed_actor<reacts_to<put_atom, int32_t>,
replies_to<get_atom>::with<int32_t>>;
struct cell_state {
int32_t value = 0;
};
cell::behavior_type type_checked_cell(cell::stateful_pointer<cell_state> self) {
return {
[=](put_atom, int32_t val) { self->state.value = val; },
[=](get_atom) { return self->state.value; },
};
}
behavior unchecked_cell(stateful_actor<cell_state>* self) {
return {
[=](put_atom, int32_t val) { self->state.value = val; },
[=](get_atom) { return self->state.value; },
};
}
Stateful actors are spawned in the same way as any other function-based actor (see Function-based Actors).
Attaching Cleanup Code to Actors¶
Users can attach cleanup code to actors. This code is executed immediately if the actor has already exited. Otherwise, the actor will execute it as part of its termination. The following example attaches a function object to actors for printing a custom string on exit.
// Utility function to print an exit message with custom name.
void print_on_exit(const actor& hdl, const std::string& name) {
hdl->attach_functor([=](const error& reason) {
cout << name << " exited: " << to_string(reason) << endl;
});
}
It is possible to attach code to remote actors. However, the cleanup code will run on the local machine.
Blocking Actors¶
Blocking actors always run in a separate thread and are not scheduled by CAF.
Unlike event-based actors, blocking actors have explicit, blocking receive
functions. Further, blocking actors do not handle system messages automatically
via special-purpose callbacks (see Default and System Message Handlers). This gives users
full control over the behavior of blocking actors. However, blocking actors
still should follow conventions of the actor system. For example, actors should
unconditionally terminate after receiving an exit_msg with reason
exit_reason::kill.
Receiving Messages¶
The function receive sequentially iterates over all elements in the
mailbox beginning with the first. It takes a message handler that is applied to
the elements in the mailbox until an element was matched by the handler. An
actor calling receive is blocked until it successfully dequeued a
message from its mailbox or an optional timeout occurs. Messages that are not
matched by the behavior are automatically skipped and remain in the mailbox.
self->receive (
[](int x) { /* ... */ }
);
Catch-all Receive Statements¶
Blocking actors can use inline catch-all callbacks instead of setting a default handler (see Default Handler). A catch-all case must be the last callback before the optional timeout, as shown in the example below.
self->receive(
[&](float x) {
// ...
},
[&](const down_msg& x) {
// ...
},
[&](const exit_msg& x) {
// ...
},
others >> [](message_view& x) -> result<message> {
// report unexpected message back to client
return sec::unexpected_message;
}
);
Receive Loops¶
Message handler passed to receive are temporary object at runtime.
Hence, calling receive inside a loop creates an unnecessary amount
of short-lived objects. CAF provides predefined receive loops to allow for
more efficient code.
// BAD
std::vector<int> results;
for (size_t i = 0; i < 10; ++i)
receive (
[&](int value) {
results.push_back(value);
}
);
// GOOD
std::vector<int> results;
size_t i = 0;
receive_for(i, 10) (
[&](int value) {
results.push_back(value);
}
);
// BAD
size_t received = 0;
while (received < 10) {
receive (
[&](int) {
++received;
}
);
} ;
// GOOD
size_t received = 0;
receive_while([&] { return received < 10; }) (
[&](int) {
++received;
}
);
// BAD
size_t received = 0;
do {
receive (
[&](int) {
++received;
}
);
} while (received < 10);
// GOOD
size_t received = 0;
do_receive (
[&](int) {
++received;
}
).until([&] { return received >= 10; });
The examples above illustrate the correct usage of the three loops
receive_for, receive_while and do_receive(...).until. It is possible
to nest receives and receive loops.
bool running = true;
self->receive_while([&] { return running; }) (
[&](int value1) {
self->receive (
[&](float value2) {
aout(self) << value1 << " => " << value2 << endl;
}
);
},
// ...
);
Scoped Actors¶
The class scoped_actor offers a simple way of communicating with CAF actors
from non-actor contexts. It overloads operator-> to return a
blocking_actor*. Hence, it behaves like the implicit self pointer in
functor-based actors, only that it ceases to exist at scope end.
void test(actor_system& system) {
scoped_actor self{system};
// spawn some actor
auto aut = self->spawn(my_actor_impl);
self->send(aut, "hi there");
// self will be destroyed automatically here; any
// actor monitoring it will receive down messages etc.
}
Message Passing¶
Message passing in CAF is always asynchronous. Further, CAF neither guarantees message delivery nor message ordering in a distributed setting. CAF uses TCP per default, but also enables nodes to send messages to other nodes without having a direct connection. In this case, messages are forwarded by intermediate nodes and can get lost if one of the forwarding nodes fails. Likewise, forwarding paths can change dynamically and thus cause messages to arrive out of order.
The messaging layer of CAF has three primitives for sending messages: send,
request, and delegate. The former simply enqueues a message to the
mailbox the receiver. The latter two are discussed in more detail in
Requests and Delegating Messages.
Structure of Mailbox Elements¶
When enqueuing a message to the mailbox of an actor, CAF wraps the content of
the message into a mailbox_element (shown below) to add meta data and
processing paths.
The sender is stored as a strong_actor_ptr (see Pointer) and
denotes the origin of the message. The message ID is either 0—invalid—or a
positive integer value that allows the sender to match a response to its
request. The stages vector stores the path of the message. Response
messages, i.e., the returned values of a message handler, are sent to
stages.back() after calling stages.pop_back(). This allows CAF to build
pipelines of arbitrary size. If no more stage is left, the response reaches the
sender. Finally, content() grants access to the type-erased tuple storing
the message itself.
Mailbox elements are created by CAF automatically and are usually invisible to the programmer. However, understanding how messages are processed internally helps understanding the behavior of the message passing layer.
It is worth mentioning that CAF usually wraps the mailbox element and its content into a single object in order to reduce the number of memory allocations.
Copy on Write¶
CAF allows multiple actors to implicitly share message contents, as long as no actor performs writes. This allows groups (see Group Communication) to send the same content to all subscribed actors without any copying overhead.
Actors copy message contents whenever other actors hold references to it and if one or more arguments of a message handler take a mutable reference.
Requirements for Message Types¶
Message types in CAF must meet the following requirements:
- Serializable or inspectable (see Type Inspection)
- Default constructible
- Copy constructible
Requirement 2 is a consequence of requirement 1, because CAF needs to be able to
create an object of a type before it can call serialize or inspect on
it. Requirement 3 allows CAF to implement Copy on Write (see
Copy on Write).
Default and System Message Handlers¶
CAF has a couple of system-level message types such as down_msg and
exit_msg that all actor should handle regardless of there current state.
Consequently, event-based actors handle such messages in special-purpose message
handlers. Additionally, event-based actors have a fallback handler for unmatched
messages. Note that blocking actors have neither of those special-purpose
handlers (see Blocking Actors).
Down Handler¶
Actors can monitor the lifetime of other actors by calling
self->monitor(other). This will cause the runtime system of CAF to send a
down_msg for other if it dies. Actors drop down messages unless they
provide a custom handler via set_down_handler(f), where f is a function
object with signature void (down_msg&) or
void (scheduled_actor*, down_msg&). The latter signature allows users to
implement down message handlers as free function.
Exit Handler¶
Bidirectional monitoring with a strong lifetime coupling is established by
calling self->link_to(other). This will cause the runtime to send an
exit_msg if either this or other dies. Per default, actors terminate
after receiving an exit_msg unless the exit reason is
exit_reason::normal. This mechanism propagates failure states in an actor
system. Linked actors form a sub system in which an error causes all actors to
fail collectively. Actors can override the default handler via
set_exit_handler(f), where f is a function object with signature void
(exit_message&) or void (scheduled_actor*, exit_message&).
Error Handler¶
Actors send error messages to others by returning an error (see
Errors) from a message handler. Similar to exit messages, error messages
usually cause the receiving actor to terminate, unless a custom handler was
installed via set_error_handler(f), where f is a function object with
signature void (error&) or void (scheduled_actor*, error&).
Additionally, request accepts an error handler as second argument to handle
errors for a particular request (see Error Handling in Requests). The default handler
is used as fallback if request is used without error handler.
Default Handler¶
The default handler is called whenever the behavior of an actor did not match
the input. Actors can change the default handler by calling
set_default_handler. The expected signature of the function object
is result<message> (scheduled_actor*, message_view&), whereas the
self pointer can again be omitted. The default handler can return a
response message or cause the runtime to skip the input message to allow
an actor to handle it in a later state. CAF provides the following built-in
implementations: reflect, reflect_and_quit,
print_and_drop, drop, and skip. The former
two are meant for debugging and testing purposes and allow an actor to simply
return an input. The next two functions drop unexpected messages with or
without printing a warning beforehand. Finally, skip leaves the
input message in the mailbox. The default is print_and_drop.
Requests¶
A main feature of CAF is its ability to couple input and output types via the
type system. For example, a typed_actor<replies_to<int>::with<int>>
essentially behaves like a function. It receives a single int as
input and responds with another int. CAF embraces this functional
take on actors by simply creating response messages from the result of message
handlers. This allows CAF to match request to response messages
and to provide a convenient API for this style of communication.
Sending Requests and Handling Responses¶
Actors send request messages by calling
request(receiver, timeout, content...). This function returns an
intermediate object that allows an actor to set a one-shot handler for the
response message.
Event-based actors can use either request(...).then or
request(...).await. The former multiplexes the one-shot handler with the
regular actor behavior and handles requests as they arrive. The latter suspends
the regular actor behavior until all awaited responses arrive and handles
requests in LIFO order.
Blocking actors always use request(...).receive, which blocks until the
one-shot handler was called. Actors receive a sec::request_timeout (see
System Error Codes) error message (see Error Handler) if a timeout occurs. Users
can set the timeout to infinite for unbound operations. This is only
recommended if the receiver is known to run locally.
In our following example, we use the simple cell actors shown below as
communication endpoints. The first part of the example illustrates how
event-based actors can use either then or await. The second half of the
example shows a blocking actor making use of receive. Note that blocking
actors have no special-purpose handler for error messages and therefore are
required to pass a callback for error messages when handling response messages.
using cell = typed_actor<reacts_to<put_atom, int32_t>,
replies_to<get_atom>::with<int32_t>>;
struct cell_state {
int32_t value = 0;
};
cell::behavior_type cell_impl(cell::stateful_pointer<cell_state> self,
int32_t x0) {
self->state.value = x0;
return {
[=](put_atom, int32_t val) { self->state.value = val; },
[=](get_atom) { return self->state.value; },
};
}
void waiting_testee(event_based_actor* self, vector<cell> cells) {
for (auto& x : cells)
self->request(x, seconds(1), get_atom_v).await([=](int32_t y) {
aout(self) << "cell #" << x.id() << " -> " << y << endl;
});
}
void multiplexed_testee(event_based_actor* self, vector<cell> cells) {
for (auto& x : cells)
self->request(x, seconds(1), get_atom_v).then([=](int32_t y) {
aout(self) << "cell #" << x.id() << " -> " << y << endl;
});
}
void blocking_testee(blocking_actor* self, vector<cell> cells) {
for (auto& x : cells)
self->request(x, seconds(1), get_atom_v)
.receive(
[&](int32_t y) {
aout(self) << "cell #" << x.id() << " -> " << y << endl;
},
[&](error& err) {
aout(self) << "cell #" << x.id() << " -> " << to_string(err) << endl;
});
}
In our example, we spawn five cells and assign the values 0, 1, 4, 9, and 16:
for (auto i = 0; i < 5; ++i)
cells.emplace_back(system.spawn(cell_impl, i * i));
When passing the cells vector to our three different testee
implementations, we observe three outputs. Our waiting_testee actor will
always print:
cell #9 -> 16
cell #8 -> 9
cell #7 -> 4
cell #6 -> 1
cell #5 -> 0
This is because await puts the one-shots handlers onto a stack and
enforces LIFO order by re-ordering incoming response messages.
The multiplexed_testee implementation does not print its results in
a predicable order. Response messages arrive in arbitrary order and are handled
immediately.
Finally, the blocking_testee implementation will always print:
cell #5 -> 0
cell #6 -> 1
cell #7 -> 4
cell #8 -> 9
cell #9 -> 16
Both event-based approaches send all requests, install a series of one-shot handlers, and then return from the implementing function. In contrast, the blocking function waits for a response before sending the next request.
Error Handling in Requests¶
Requests allow CAF to unambiguously correlate request and response messages.
This is also true if the response is an error message. Hence, CAF allows to add
an error handler as optional second parameter to then and await (this
parameter is mandatory for receive). If no such handler is defined, the
default error handler (see Error Handler) get called instead.
As an example, we consider a simple divider that returns an error on a division by zero. This examples uses a custom error category (see Errors).
enum class math_error : uint8_t {
division_by_zero = 1,
};
CAF_ERROR_CODE_ENUM(math_error)
std::string to_string(math_error x) {
switch (x) {
case math_error::division_by_zero:
return "division_by_zero";
default:
return "-unknown-error-";
}
}
using divider = typed_actor<replies_to<div_atom, double, double>::with<double>>;
divider::behavior_type divider_impl() {
return {
[](div_atom, double x, double y) -> result<double> {
if (y == 0.0)
return math_error::division_by_zero;
return x / y;
},
};
}
When sending requests to the divider, we use a custom error handlers to report errors to the user.
auto div = system.spawn(divider_impl);
scoped_actor self{system};
self->request(div, std::chrono::seconds(10), div_atom_v, x, y)
.receive(
[&](double z) { aout(self) << x << " / " << y << " = " << z << endl; },
[&](const error& err) {
aout(self) << "*** cannot compute " << x << " / " << y << " => "
<< to_string(err) << endl;
});
Delaying Messages¶
Messages can be delayed by using the function delayed_send, as
illustrated in the following time-based loop example.
// Uses a message-based loop to iterate over all animation steps.
behavior dancing_kirby(event_based_actor* self) {
// Let's get started.
auto i = std::begin(animation_steps);
auto e = std::end(animation_steps);
self->send(self, update_atom_v);
return {
[=](update_atom) mutable {
// We're done when reaching the past-the-end position.
if (i == e) {
cout << endl;
self->quit();
return;
}
// Print current animation step.
draw_kirby(*i);
// Animate next step in 150ms.
++i;
self->delayed_send(self, std::chrono::milliseconds(150), update_atom_v);
},
};
}
Delegating Messages¶
Actors can transfer responsibility for a request by using delegate. This
enables the receiver of the delegated message to reply as usual (by returning a
value from its message handler) and the original sender of the message will
receive the response. The following diagram illustrates request delegation from
actor B to actor C.
Returning the result of delegate(...) from a message handler, as
shown in the example below, suppresses the implicit response message and allows
the compiler to check the result type when using statically typed actors.
using calc = typed_actor<replies_to<add_atom, int32_t, int32_t>::with<int32_t>>;
void actor_a(event_based_actor* self, const calc& worker) {
self->request(worker, std::chrono::seconds(10), add_atom_v, 1, 2)
.then([=](int32_t result) { //
aout(self) << "1 + 2 = " << result << std::endl;
});
}
calc::behavior_type actor_b(calc::pointer self, const calc& worker) {
return {
[=](add_atom add, int32_t x, int32_t y) {
return self->delegate(worker, add, x, y);
},
};
}
calc::behavior_type actor_c() {
return {
[](add_atom, int32_t x, int32_t y) { return x + y; },
};
}
Response Promises¶
Response promises allow an actor to send and receive other messages prior to
replying to a particular request. Actors create a response promise using
self->make_response_promise<Ts...>(), where Ts is a
template parameter pack describing the promised return type. Dynamically typed
actors simply call self->make_response_promise(). After retrieving
a promise, an actor can fulfill it by calling the member function
deliver(...), as shown in the following example.
using adder
= typed_actor<replies_to<add_atom, int32_t, int32_t>::with<int32_t>>;
adder::behavior_type worker() {
return {
[](add_atom, int32_t a, int32_t b) { return a + b; },
};
}
adder::behavior_type calculator_master(adder::pointer self) {
auto w = self->spawn(worker);
return {
[=](add_atom x, int32_t y, int32_t z) -> result<int32_t> {
auto rp = self->make_response_promise<int32_t>();
self->request(w, infinite, x, y, z).then([=](int32_t result) mutable {
rp.deliver(result);
});
return rp;
},
};
}
Message Priorities¶
By default, all messages have the default priority, i.e.,
message_priority::normal. Actors can send urgent messages by setting the
priority explicitly: send<message_priority::high>(dst,...). Urgent messages
are put into a different queue of the receiver’s mailbox to avoid long delays
for urgent communication.
Scheduler¶
The CAF runtime maps N actors to M threads on the local machine. Applications build with CAF scale by decomposing tasks into many independent steps that are spawned as actors. In this way, sequential computations performed by individual actors are small compared to the total runtime of the application, and the attainable speedup on multi-core hardware is maximized in agreement with Amdahl’s law.
Decomposing tasks implies that actors are often short-lived. Assigning a
dedicated thread to each actor would not scale well. Instead, CAF includes a
scheduler that dynamically assigns actors to a pre-dimensioned set of worker
threads. Actors are modeled as lightweight state machines. Whenever a waiting
actor receives a message, it changes its state to ready and is scheduled for
execution. CAF cannot interrupt running actors because it is implemented in user
space. Consequently, actors that use blocking system calls such as I/O functions
can suspend threads and create an imbalance or lead to starvation. Such
“uncooperative” actors can be explicitly detached by the programmer by using the
detached spawn option, e.g., system.spawn<detached>(my_actor_fun).
The performance of actor-based applications depends on the scheduling algorithm in use and its configuration. Different application scenarios require different trade-offs. For example, interactive applications such as shells or GUIs want to stay responsive to user input at all times, while batch processing applications demand only to perform a given task in the shortest possible time.
Aside from managing actors, the scheduler bridges actor and non-actor code. For this reason, the scheduler distinguishes between external and internal events. An external event occurs whenever an actor is spawned from a non-actor context or an actor receives a message from a thread that is not under the control of the scheduler. Internal events are send and spawn operations from scheduled actors.
Policies¶
The scheduler consists of a single coordinator and a set of workers. The coordinator is needed by the public API to bridge actor and non-actor contexts, but is not necessarily an active software entity.
The scheduler of CAF is fully customizable by using a policy-based design. The
following class shows a concept class that lists all required member
types and member functions. A policy provides the two data structures
coordinator_data and worker_data that add additional
data members to the coordinator and its workers respectively, e.g., work
queues. This grants developers full control over the state of the scheduler.
struct scheduler_policy {
struct coordinator_data;
struct worker_data;
void central_enqueue(Coordinator* self, resumable* job);
void external_enqueue(Worker* self, resumable* job);
void internal_enqueue(Worker* self, resumable* job);
void resume_job_later(Worker* self, resumable* job);
resumable* dequeue(Worker* self);
void before_resume(Worker* self, resumable* job);
void after_resume(Worker* self, resumable* job);
void after_completion(Worker* self, resumable* job);
};
Whenever a new work item is scheduled—usually by sending a message to an idle
actor—, one of the functions central_enqueue,
external_enqueue, and internal_enqueue is called. The
first function is called whenever non-actor code interacts with the actor
system. For example when spawning an actor from main. Its first
argument is a pointer to the coordinator singleton and the second argument is
the new work item—usually an actor that became ready. The function
external_enqueue is never called directly by CAF. It models the
transfer of a task to a worker by the coordinator or another worker. Its first
argument is the worker receiving the new task referenced in the second
argument. The third function, internal_enqueue, is called whenever
an actor interacts with other actors in the system. Its first argument is the
current worker and the second argument is the new work item.
Actors reaching the maximum number of messages per run are re-scheduled with
resume_job_later and workers acquire new work by calling
dequeue. The two functions before_resume and
after_resume allow programmers to measure individual actor runtime,
while after_completion allows to execute custom code whenever a
work item has finished execution by changing its state to done, but
before it is destroyed. In this way, the last three functions enable developers
to gain fine-grained insight into the scheduling order and individual execution
times.
Work Stealing¶
The default policy in CAF is work stealing. The key idea of this algorithm is
to remove the bottleneck of a single, global work queue. The original
algorithm was developed for fully strict computations by Blumofe et al in 1994.
It schedules any number of tasks to P workers, where P
is the number of processors available.
Each worker dequeues work items from an individual queue until it is drained. Once this happens, the worker becomes a thief. It picks one of the other workers—usually at random—as a victim and tries to steal a work item. As a consequence, tasks (actors) are bound to workers by default and only migrate between threads as a result of stealing. This strategy minimizes communication between threads and maximizes cache locality. Work stealing has become the algorithm of choice for many frameworks. For example, Java’s Fork-Join (which is used by Akka), Intel’s Threading Building Blocks, several OpenMP implementations, etc.
CAF uses a double-ended queue for its workers, which is synchronized with two spinlocks. One downside of a decentralized algorithm such as work stealing is, that idle states are hard to detect. Did only one worker run out of work items or all? Since each worker has only local knowledge, it cannot decide when it could safely suspend itself. Likewise, workers cannot resume if new job items arrived at one or more workers. For this reason, CAF uses three polling intervals. Once a worker runs out of work items, it tries to steal items from others. First, it uses the aggressive polling interval. It falls back to a moderate interval after a predefined number of trials. After another predefined number of trials, it will finally use a relaxed interval.
Per default, the aggressive strategy performs 100 steal attempts with no sleep interval in between. The moderate strategy tries to steal 500 times with 50 microseconds sleep between two steal attempts. Finally, the relaxed strategy runs indefinitely but sleeps for 10 milliseconds between two attempts. These defaults can be overridden via system config at startup (see Configuring Actor Applications).
Work Sharing¶
Work sharing is an alternative scheduler policy in CAF that uses a single, global work queue. This policy uses a mutex and a condition variable on the central queue. Thus, the policy supports only limited concurrency but does not need to poll. Using this policy can be a good fit for low-end devices where power consumption is an important metric.
Registry¶
The actor registry in CAF keeps track of the number of running actors and allows
to map actors to their ID or a custom atom (see Atoms) representing a
name. The registry does not contain all actors. Actors have to be stored in
the registry explicitly. Users can access the registry through an actor system
by calling system.registry(). The registry stores actors using
strong_actor_ptr (see Pointer).
Users can use the registry to make actors system-wide available by name. The Middleman uses the registry to keep track of all actors known to remote nodes in order to serialize and deserialize them. Actors are removed automatically when they terminate.
It is worth mentioning that the registry is not synchronized between connected actor system. Each actor system has its own, local registry in a distributed setting.
| Types | |
name_map |
unordered_map<atom_value, strong_actor_ptr> |
| Observers | |
strong_actor_ptr get(actor_id) |
Returns the actor associated to given ID. |
strong_actor_ptr get(atom_value) |
Returns the actor associated to given name. |
name_map named_actors() |
Returns all name mappings. |
size_t running() |
Returns the number of currently running actors. |
| Modifiers | |
put(actor_id, strong_actor_ptr) |
Maps an actor to its ID. |
erase(actor_id) |
Removes an ID mapping from the registry. |
put(atom_value, strong_actor_ptr) |
Maps an actor to a name. |
erase(atom_value) |
Removes a name mapping from the registry. |
Reference Counting¶
Actors systems can span complex communication graphs that make it hard to decide when actors are no longer needed. As a result, manually managing lifetime of actors is merely impossible. For this reason, CAF implements a garbage collection strategy for actors based on weak and strong reference counts.
Shared Ownership in C++¶
The C++ standard library already offers shared_ptr and
weak_ptr to manage objects with complex shared ownership. The
standard implementation is a solid general purpose design that covers most use
cases. Weak and strong references to an object are stored in a control
block. However, CAF uses a slightly different design. The reason for this is
twofold. First, we need the control block to store the identity of an actor.
Second, we wanted a design that requires less indirections, because actor
handles are used extensively copied for messaging, and this overhead adds up.
Before discussing the approach to shared ownership in CAF, we look at the design of shared pointers in the C++ standard library.
The figure above depicts the default memory layout when using shared pointers.
The control block is allocated separately from the data and thus stores a
pointer to the data. This is when using manually-allocated objects, for example
shared_ptr<int> iptr{new int}. The benefit of this design is that
one can destroy T independently from its control block. While
irrelevant for small objects, it can become an issue for large objects.
Notably, the shared pointer stores two pointers internally. Otherwise,
dereferencing it would require to get the data location from the control block
first.
When using make_shared or allocate_shared, the standard
library can store reference count and data in a single memory block as shown
above. However, shared_ptr still has to store two pointers, because
it is unaware where the data is allocated.
Finally, the design of the standard library becomes convoluted when an object
should be able to hand out a shared_ptr to itself. Classes must
inherit from std::enable_shared_from_this to navigate from an
object to its control block. This additional navigation path is required,
because std::shared_ptr needs two pointers. One to the data and one
to the control block. Programmers can still use make_shared for
such objects, in which case the object is again stored along with the control
block.
Smart Pointers to Actors¶
In CAF, we use a different approach than the standard library because (1) we
always allocate actors along with their control block, (2) we need additional
information in the control block, and (3) we can store only a single raw
pointer internally instead of the two raw pointers std::shared_ptr
needs. The following figure summarizes the design of smart pointers to actors.
CAF uses strong_actor_ptr instead of
std::shared_ptr<...> and weak_actor_ptr instead of
std::weak_ptr<...>. Unlike the counterparts from the standard
library, both smart pointer types only store a single pointer.
Also, the control block in CAF is not a template and stores the identity of an
actor (actor_id plus node_id). This allows CAF to
access this information even after an actor died. The control block fits
exactly into a single cache line (64 Bytes). This makes sure no false
sharing occurs between an actor and other actors that have references to it.
Since the size of the control block is fixed and CAF guarantees the
memory layout enforced by actor_storage, CAF can compute the
address of an actor from the pointer to its control block by offsetting it by
64 Bytes. Likewise, an actor can compute the address of its control block.
The smart pointer design in CAF relies on a few assumptions about actor types.
Most notably, the actor object is placed 64 Bytes after the control block. This
starting address is cast to abstract_actor*. Hence, T*
must be convertible to abstract_actor* via
reinterpret_cast. In practice, this means actor subclasses must not
use virtual inheritance, which is enforced in CAF with a
static_assert.
Strong and Weak References¶
A strong reference manipulates the strong refs counter as shown above. An
actor is destroyed if there are zero strong references to it. If two actors
keep strong references to each other via member variable, neither actor can ever
be destroyed because they produce a cycle (see Breaking Cycles Manually). Strong
references are formed by strong_actor_ptr, actor, and
typed_actor<...> (see Actor References).
A weak reference manipulates the weak refs counter. This counter keeps
track of how many references to the control block exist. The control block is
destroyed if there are zero weak references to an actor (which cannot occur
before strong refs reached zero as well). No cycle occurs if two actors
keep weak references to each other, because the actor objects themselves can get
destroyed independently from their control block. A weak reference is only
formed by actor_addr (see Address).
Converting Actor References with actor_cast¶
The function actor_cast converts between actor pointers and
handles. The first common use case is to convert a strong_actor_ptr
to either actor or typed_actor<...> before being able
to send messages to an actor. The second common use case is to convert
actor_addr to strong_actor_ptr to upgrade a weak
reference to a strong reference. Note that casting actor_addr to a
strong actor pointer or handle can result in invalid handles. The syntax for
actor_cast resembles builtin C++ casts. For example,
actor_cast<actor>(x) converts x to an handle of type
actor.
Breaking Cycles Manually¶
Cycles can occur only when using class-based actors when storing references to
other actors via member variable. Stateful actors (see Stateful Actors)
break cycles by destroying the state when an actor terminates, before the
destructor of the actor itself runs. This means an actor releases all references
to others automatically after calling quit. However, class-based actors have
to break cycles manually, because references to others are not released until
the destructor of an actor runs. Two actors storing references to each other via
member variable produce a cycle and neither destructor can ever be called.
Class-based actors can break cycles manually by overriding on_exit() and
calling destroy(x) on each handle (see Handle). Using a handle
after destroying it is undefined behavior, but it is safe to assign a new value
to the handle.
Errors¶
Errors in CAF have a code and a category, similar to std::error_code and
std::error_condition. Unlike its counterparts from the C++ standard library,
error is plattform-neutral and serializable. Instead of using category
singletons, CAF stores categories as atoms (see Atoms). Errors can also
include a message to provide additional context information.
Class Interface¶
| Constructors | |
(Enum x) |
Construct error by calling make_error(x) |
(uint8_t x, atom_value y) |
Construct error with code x and category y |
(uint8_t x, atom_value y, message z) |
Construct error with code x, category y, and context z |
| Observers | |
uint8_t code() |
Returns the error code |
atom_value category() |
Returns the error category |
message context() |
Returns additional context information |
explicit operator bool() |
Returns code() != 0 |
Add Custom Error Categories¶
Adding custom error categories requires two steps: (1) declare an enum class of
type uint8_t with the first value starting at 1, (2) enable conversions from
the error code enum to error by using the macro CAF_ERROR_CODE_ENUM. The
following example illustrates these two steps with a math_error enum to
signal divisions by zero:
enum class math_error : uint8_t {
division_by_zero = 1,
};
CAF_ERROR_CODE_ENUM(math_error)
std::string to_string(math_error x) {
switch (x) {
case math_error::division_by_zero:
return "division_by_zero";
default:
return "-unknown-error-";
}
}
using divider = typed_actor<replies_to<div_atom, double, double>::with<double>>;
divider::behavior_type divider_impl() {
return {
[](div_atom, double x, double y) -> result<double> {
if (y == 0.0)
return math_error::division_by_zero;
return x / y;
},
};
}
Note
CAF stores the integer value with an atom that disambiguates error codes from
different categories. The macro CAF_ERROR_CODE_ENUM uses the given type
name as atom as well. This means the same limitations apply. In particular,
10 characters or less. If the type name exceeds this limit, users can pass
two arguments instead. In the example above, we could call
CAF_ERROR_CODE_ENUM(math_error, "math") instead.
System Error Codes¶
System Error Codes (SECs) represent errors in the actor system or one of its modules and are defined as follows:
/// SEC stands for "System Error Code". This enum contains error codes for
/// ::actor_system and its modules.
enum class sec : uint8_t {
/// No error.
none = 0,
/// Indicates that an actor dropped an unexpected message.
unexpected_message = 1,
/// Indicates that a response message did not match the provided handler.
unexpected_response,
/// Indicates that the receiver of a request is no longer alive.
request_receiver_down,
/// Indicates that a request message timed out.
request_timeout,
/// Indicates that requested group module does not exist.
no_such_group_module = 5,
/// Unpublishing or connecting failed: no actor bound to given port.
no_actor_published_at_port,
/// Connecting failed because a remote actor had an unexpected interface.
unexpected_actor_messaging_interface,
/// Migration failed because the state of an actor is not serializable.
state_not_serializable,
/// An actor received an unsupported key for `('sys', 'get', key)` messages.
unsupported_sys_key,
/// An actor received an unsupported system message.
unsupported_sys_message = 10,
/// A remote node disconnected during CAF handshake.
disconnect_during_handshake,
/// Tried to forward a message via BASP to an invalid actor handle.
cannot_forward_to_invalid_actor,
/// Tried to forward a message via BASP to an unknown node ID.
no_route_to_receiving_node,
/// Middleman could not assign a connection handle to a broker.
failed_to_assign_scribe_from_handle,
/// Middleman could not assign an acceptor handle to a broker.
failed_to_assign_doorman_from_handle = 15,
/// User requested to close port 0 or to close a port not managed by CAF.
cannot_close_invalid_port,
/// Middleman could not connect to a remote node.
cannot_connect_to_node,
/// Middleman could not open requested port.
cannot_open_port,
/// A C system call in the middleman failed.
network_syscall_failed,
/// A function received one or more invalid arguments.
invalid_argument = 20,
/// A network socket reported an invalid network protocol family.
invalid_protocol_family,
/// Middleman could not publish an actor because it was invalid.
cannot_publish_invalid_actor,
/// A remote spawn failed because the provided types did not match.
cannot_spawn_actor_from_arguments,
/// Serialization failed because there was not enough data to read.
end_of_stream,
/// Serialization failed because no CAF context is available.
no_context = 25,
/// Serialization failed because CAF misses run-time type information.
unknown_type,
/// Serialization of actors failed because no proxy registry is available.
no_proxy_registry,
/// An exception was thrown during message handling.
runtime_error,
/// Linking to a remote actor failed because actor no longer exists.
remote_linking_failed,
/// Adding an upstream to a stream failed.
cannot_add_upstream = 30,
/// Adding an upstream to a stream failed because it already exists.
upstream_already_exists,
/// Unable to process upstream messages because upstream is invalid.
invalid_upstream,
/// Adding a downstream to a stream failed.
cannot_add_downstream,
/// Adding a downstream to a stream failed because it already exists.
downstream_already_exists,
/// Unable to process downstream messages because downstream is invalid.
invalid_downstream = 35,
/// Cannot start streaming without next stage.
no_downstream_stages_defined,
/// Actor failed to initialize state after receiving a stream handshake.
stream_init_failed,
/// Unable to process a stream since due to missing state.
invalid_stream_state,
/// Stream aborted due to unexpected error.
unhandled_stream_error,
/// A function view was called without assigning an actor first.
bad_function_call = 40,
/// Feature is disabled in the actor system config.
feature_disabled,
/// Failed to open file.
cannot_open_file,
/// A socket descriptor argument is invalid.
socket_invalid,
/// A socket became disconnected from the remote host (hang up).
socket_disconnected,
/// An operation on a socket (e.g. `poll`) failed.
socket_operation_failed = 45,
/// A resource is temporarily unavailable or would block.
unavailable_or_would_block,
/// Connection refused because of incompatible CAF versions.
incompatible_versions,
/// Connection refused because of incompatible application IDs.
incompatible_application_ids,
/// The middleman received a malformed BASP message from another node.
malformed_basp_message,
/// The middleman closed a connection because it failed to serialize or
/// deserialize a payload.
serializing_basp_payload_failed = 50,
/// The middleman closed a connection to itself or an already connected node.
redundant_connection,
/// Resolving a path on a remote node failed.
remote_lookup_failed,
/// Disconnected from a BASP node after reaching the connection timeout.
connection_timeout,
};
Default Exit Reasons¶
CAF uses the error category "exit" for default exit reasons. These errors
are usually fail states set by the actor system itself. The two exceptions are
exit_reason::user_shutdown and exit_reason::kill. The former is used in
CAF to signalize orderly, user-requested shutdown and can be used by programmers
in the same way. The latter terminates an actor unconditionally when used in
send_exit, even if the default handler for exit messages (see
Exit Handler) is overridden.
/// This error category represents fail conditions for actors.
enum class exit_reason : uint8_t {
/// Indicates that an actor finished execution without error.
normal = 0,
/// Indicates that an actor died because of an unhandled exception.
unhandled_exception,
/// Indicates that the exit reason for this actor is unknown, i.e.,
/// the actor has been terminated and no longer exists.
unknown,
/// Indicates that an actor pool unexpectedly ran out of workers.
out_of_workers,
/// Indicates that an actor was forced to shutdown by a user-generated event.
user_shutdown,
/// Indicates that an actor was killed unconditionally.
kill,
/// Indicates that an actor finishied execution because a connection
/// to a remote link was closed unexpectedly.
remote_link_unreachable,
/// Indicates that an actor was killed because it became unreachable.
unreachable
};
Configuring Actor Applications¶
CAF configures applications at startup using an actor_system_config or a
user-defined subclass of that type. The config objects allow users to add custom
types, to load modules (optional components), and to fine-tune the behavior of
loaded modules with command line options or configuration files (see
Command Line Options and INI Configuration Files).
The following code example is a minimal CAF application with a Middleman but without any custom configuration options.
void caf_main(actor_system& system) {
// ...
}
CAF_MAIN(io::middleman)
The compiler expands this example code to the following.
void caf_main(actor_system& system) {
// ...
}
int main(int argc, char** argv) {
return exec_main<io::middleman>(caf_main, argc, argv);
}
The function exec_main performs several steps:
- Initialize all meta objects for the type ID blocks listed in
CAF_MAIN. - Create a config object. If
caf_mainhas two arguments, then CAF assumes that the second argument is the configuration and the type gets derived from that argument. Otherwise, CAF usesactor_system_config. - Parse command line arguments and configuration file (if present).
- Load all modules requested in
CAF_MAIN. - Create an actor system.
- Call
caf_mainwith the actor system and optionally withconfig.
When implementing the steps performed by CAF_MAIN by hand, the main
function would resemble the following (pseudo) code:
int main(int argc, char** argv) {
// Create the config.
actor_system_config cfg;
// Add runtime-type information for user-defined types.
cfg.add_message_types<...>();
// Read CLI options.
if (auto err = cfg.parse(argc, argv)) {
std::cerr << "error while parsing CLI and file options: "
<< to_string(err) << std::endl;
return EXIT_FAILURE;
}
// Return immediately if a help text was printed.
if (cfg.cli_helptext_printed)
return EXIT_SUCCESS;
// Load modules.
cfg.load<...>();
// Create the actor system.
actor_system sys{cfg};
// Run user-defined code.
caf_main(sys, cfg);
return EXIT_SUCCESS;
}
Using CAF_MAIN simply automates that boilerplate code. A minimal example
with a custom type ID block as well as a custom configuration class with the I/O
module loaded looks as follows:
CAF_BEGIN_TYPE_ID_BLOCK(my, first_custom_type_id)
// ...
CAF_END_TYPE_ID_BLOCK(my)
class my_config : public actor_system_config {
public:
my_config() {
// ...
}
};
void caf_main(actor_system& system, const my_config& cfg) {
// ...
}
CAF_MAIN(id_block::my, io::middleman)
Note
Using type ID blocks is optional. Users can also call add_message_type
for each user-defined type in the constructor of my_config.
Loading Modules¶
The simplest way to load modules is to use the macro CAF_MAIN and
to pass a list of all requested modules, as shown below.
void caf_main(actor_system& system) {
// ...
}
CAF_MAIN(mod1, mod2, ...)
Alternatively, users can load modules in user-defined config classes.
class my_config : public actor_system_config {
public:
my_config() {
load<mod1>();
load<mod2>();
// ...
}
};
The third option is to simply call x.load<mod1>() on a config
object before initializing an actor system with it.
Command Line Options and INI Configuration Files¶
CAF organizes program options in categories and parses CLI arguments as well as
INI files. CLI arguments override values in the INI file which override
hard-coded defaults. Users can add any number of custom program options by
implementing a subtype of actor_system_config. The example below
adds three options to the global category.
class config : public actor_system_config {
public:
uint16_t port = 0;
std::string host = "localhost";
bool server_mode = false;
config() {
opt_group{custom_options_, "global"}
.add(port, "port,p", "set port")
.add(host, "host,H", "set host (ignored in server mode)")
.add(server_mode, "server-mode,s", "enable server mode");
}
};
We create a new global category in custom_options_.
Each following call to add then appends individual options to the
category. The first argument to add is the associated variable. The
second argument is the name for the parameter, optionally suffixed with a
comma-separated single-character short name. The short name is only considered
for CLI parsing and allows users to abbreviate commonly used option names. The
third and final argument to add is a help text.
The custom config class allows end users to set the port for the application
to 42 with either -p 42 (short name) or --port=42 (long name). The long
option name is prefixed by the category when using a different category than
global. For example, adding the port option to the category foo means
end users have to type --foo.port=42 when using the long name. Short names
are unaffected by the category, but have to be unique.
Boolean options do not require arguments. The member variable
server_mode is set to true if the command line contains
either --server-mode or -s.
The example uses member variables for capturing user-provided settings for
simplicity. However, this is not required. For example,
add<bool>(...) allows omitting the first argument entirely. All
values of the configuration are accessible with get_or. Note that
all global options can omit the "global." prefix.
CAF adds the program options help (with short names -h
and -?) as well as long-help to the global
category.
The default name for the INI file is caf-application.ini. Users can
change the file name and path by passing --config-file=<path> on the
command line.
INI files are organized in categories. No value is allowed outside of a category
(no implicit global category). The parses uses the following syntax:
| Syntax | Type |
|---|---|
key=true |
Boolean |
key=1 |
Integer |
key=1.0 |
Floating point number |
key=1ms |
Timespan |
key='foo' |
Atom |
key="foo" |
String |
key=[0, 1, ...] |
List |
key={a=1, b=2, ...} |
Dictionary (map) |
The following example INI file lists all standard options in CAF and their
default value. Note that some options such as scheduler.max-threads
are usually detected at runtime and thus have no hard-coded default.
; This file shows all possible parameters with defaults.
; Values enclosed in <> are detected at runtime unless defined by the user.
; when using the default scheduler
[scheduler]
; accepted alternative: 'sharing'
policy='stealing'
; configures whether the scheduler generates profiling output
enable-profiling=false
; forces a fixed number of threads if set
max-threads=<number of cores>
; maximum number of messages actors can consume in one run
max-throughput=<infinite>
; measurement resolution in milliseconds (only if profiling is enabled)
profiling-resolution=100ms
; output file for profiler data (only if profiling is enabled)
profiling-output-file="/dev/null"
; when using 'stealing' as scheduler policy
[work-stealing]
; number of zero-sleep-interval polling attempts
aggressive-poll-attempts=100
; frequency of steal attempts during aggressive polling
aggressive-steal-interval=10
; number of moderately aggressive polling attempts
moderate-poll-attempts=500
; frequency of steal attempts during moderate polling
moderate-steal-interval=5
; sleep interval between poll attempts
moderate-sleep-duration=50us
; frequency of steal attempts during relaxed polling
relaxed-steal-interval=1
; sleep interval between poll attempts
relaxed-sleep-duration=10ms
; when loading io::middleman
[middleman]
; configures whether MMs try to span a full mesh
enable-automatic-connections=false
; application identifier of this node, prevents connection to other CAF
; instances with different identifier
app-identifier=""
; maximum number of consecutive I/O reads per broker
max-consecutive-reads=50
; heartbeat message interval in ms (0 disables heartbeating)
heartbeat-interval=0ms
; configures whether the MM attaches its internal utility actors to the
; scheduler instead of dedicating individual threads (needed only for
; deterministic testing)
attach-utility-actors=false
; configures whether the MM starts a background thread for I/O activity,
; setting this to true allows fully deterministic execution in unit test and
; requires the user to trigger I/O manually
manual-multiplexing=false
; disables communication via TCP
disable-tcp=false
; enable communication via UDP
enable-udp=false
; configures how many background workers are spawned for deserialization,
; by default CAF uses 1-4 workers depending on the number of cores
workers=<min(3, number of cores / 4) + 1>
; when compiling with logging enabled
[logger]
; file name template for output log file files (empty string disables logging)
file-name="actor_log_[PID]_[TIMESTAMP]_[NODE].log"
; format for rendering individual log file entries
file-format="%r %c %p %a %t %C %M %F:%L %m%n"
; configures the minimum severity of messages that are written to the log file
; (quiet|error|warning|info|debug|trace)
file-verbosity='trace'
; mode for console log output generation (none|colored|uncolored)
console='none'
; format for printing individual log entries to the console
console-format="%m"
; configures the minimum severity of messages that are written to the console
; (quiet|error|warning|info|debug|trace)
console-verbosity='trace'
; excludes listed components from logging (list of atoms)
component-blacklist=[]
Adding Custom Message Types¶
CAF requires serialization support for all of its message types (see Type Inspection). In addition, CAF also needs a mapping of unique type names to user-defined types at runtime. This is required to deserialize arbitrary messages from the network.
The function actor_system_config::add_message_type adds runtime-type
information for a single type. It takes a template parameter (the message type)
and one function argument (the type name). For example,
cfg.add_message_type<foo>("foo") would add runtime-type information for the
type foo. However, calling add_message_type for each type individually
is both verbose and prone to error.
For new code, we strongly recommend using the new type ID blocks. When setting
the CMake option CAF_ENABLE_TYPE_ID_CHECKS to ON (or calling the
configure script with --enable-type-id-checks), CAF raises a static
assertion that prohibits any message type that does not appear in such a type ID
block. When using this API, users can instead call add_message_types once
per message block. Combined with the type ID checks, this makes sure that the
runtime-type information never runs out of sync when adding new message types.
Passing the type ID blocks to CAF_MAIN also automates the setup steps for
adding new message types.
A type ID block in CAF starts by calling CAF_BEGIN_TYPE_ID_BLOCK. Inside the
block appear any number of CAF_ADD_TYPE_ID and CAF_ADD_ATOM statements.
The type ID block only requires forward declarations. The block ends at
CAF_END_TYPE_ID_BLOCK, as shown in the example below.
struct foo;
struct foo2;
CAF_BEGIN_TYPE_ID_BLOCK(custom_types_1, first_custom_type_id)
CAF_ADD_TYPE_ID(custom_types_1, (foo))
CAF_ADD_TYPE_ID(custom_types_1, (foo2))
CAF_ADD_TYPE_ID(custom_types_1, (std::pair<int32_t, int32_t>) )
CAF_END_TYPE_ID_BLOCK(custom_types_1)
Note
The second argument to CAF_ADD_TYPE_ID (the type) must appear in extra
parentheses. This unusual syntax is an artifact of argument handling in
macros. Without the extra set of parentheses, we would not be able to add
types with commas such as std::pair.
The first argument to all macros is the name of the type ID block. The macro
expands a name X to caf::id_block::X. In the example above, we can refer
to the custom type ID block with caf::id_block::custom_types_1. To add the
required runtime-type information to CAF, we can either call
cfg.add_message_types<caf::id_block::custom_types_1>() on a config object
pass the ID block to CAF_MAIN:
CAF_MAIN(caf::id_block::custom_types_1)
Note
At the point of calling CAF_MAIN or add_message_types, the compiler
must have the type declaration plus all inspect overloads available for
each type in the type ID block.
Adding Custom Actor Types experimental¶
Adding actor types to the configuration allows users to spawn actors by their
name. In particular, this enables spawning of actors on a different node (see
Remote Spawning of Actors experimental). For our example configuration, we consider the following
simple calculator actor.
using calculator = caf::typed_actor<
caf::replies_to<caf::add_atom, int32_t, int32_t>::with<int32_t>,
caf::replies_to<caf::sub_atom, int32_t, int32_t>::with<int32_t>>;
Adding the calculator actor type to our config is achieved by calling
add_actor_type. After calling this in our config, we can spawn the
calculator anywhere in the distributed actor system (assuming all nodes use
the same config). Note that the handle type still requires a type ID (see
Adding Custom Message Types).
After adding the actor type to the config, we can spawn our calculator by
name. Unlike the regular spawn overloads, this version requires wrapping the
constructor arguments into a message and the function might fail and thus
returns an expected:
if (auto x = system.spawn<calculator>("calculator", make_message())) {
// ... do something with *x ...
} else {
std::cerr << "*** unable to spawn a calculator: " << to_string(x.error())
<< std::endl;
// ...
}
Adding dynamically typed actors to the config is achieved in the same way. When
spawning a dynamically typed actor in this way, the template parameter is
simply actor. For example, spawning an actor “foo” which requires
one string is created with:
auto worker = system.spawn<actor>("foo", make_message("bar"));
Because constructor (or function) arguments for spawning the actor are stored in
a message, only actors with appropriate input types are allowed. Pointer
types, for example, are illegal in messages.
Log Output¶
CAF comes with a logger integrated into the actor system. By default, CAF itself
won’t emit any log messages. Developers can set the verbosity of CAF itself at
build time by setting the CMake option CAF_LOG_LEVEL manually or by passing
--with-log-level=... to the configure script. The available verbosity
levels are (from least to most output):
errorwarninginfodebugtrace
The logging infrastructure is always available to users, regardless of the verbosity level of CAF itself.
File Name¶
The output file is generated from the template configured by
logger-file-name. This template supports the following variables.
| Variable | Output |
|---|---|
[PID] |
The OS-specific process ID. |
[TIMESTAMP] |
The UNIX timestamp on startup. |
[NODE] |
The node ID of the CAF system. |
Console¶
Console output is disabled per default. Setting logger-console to either
uncolored or colored prints log events to std::clog. Using the
colored option will print the log events in different colors depending on
the severity level.
Format Strings¶
CAF uses log4j-like format strings for configuring printing of individual
events via logger-file-format and
logger-console-format. Note that format modifiers are not supported
at the moment. The recognized field identifiers are:
| Pattern | Output |
|---|---|
%c |
The category/component. |
%C |
The full qualifier of the current function. For example, the function void ns::foo::bar() would print ns.foo. |
%d |
The date in ISO 8601 format, i.e., "YYYY-MM-DDThh:mm:ss". |
%F |
The file name. |
%L |
The line number. |
%m |
The user-defined log message. |
%M |
The name of the current function. For example, the name of void ns::foo::bar() is printed as bar. |
%n |
A newline. |
%p |
The priority (severity level). |
%r |
Elapsed time since starting the application in milliseconds. |
%t |
ID of the current thread. |
%a |
ID of the current actor (or actor0 when not logging inside an actor). |
%% |
A single percent sign. |
Filtering¶
The two configuration options logger.component-blacklist and
logger.(file|console)-verbosity reduce the amount of generated log events.
The former is a list of excluded component names and the latter can increase the
reported severity level (but not decrease it beyond the level defined at compile
time).
Type-Erased Tuples, Messages and Message Views¶
Messages in CAF are stored in type-erased tuples. The actual message type
itself is usually hidden, as actors use pattern matching to decompose messages
automatically. However, the classes message and
message_builder allow more advanced use cases than only sending
data from one actor to another.
The interface type_erased_tuple encapsulates access to arbitrary
data. This data can be stored on the heap or on the stack. A
message is a type-erased tuple that is always heap-allocated and
uses copy-on-write semantics. When dealing with “plain” type-erased tuples,
users are required to check if a tuple is referenced by others via
type_erased_tuple::shared before modifying its content.
The convenience class message_view holds a reference to either a
stack-located type_erased_tuple or a message. The
content of the data can be access via message_view::content in both
cases, which returns a type_erased_tuple&. The content of the view
can be forced into a message object by calling
message_view::move_content_to_message. This member function either
returns the stored message object or moves the content of a stack-allocated
tuple into a new message.
RTTI and Type Numbers¶
All builtin types in CAF have a non-zero 6-bit type number. All
user-defined types are mapped to 0. When querying the run-time type information
(RTTI) for individual message or tuple elements, CAF returns a pair consisting
of an integer and a pointer to std::type_info. The first value is
the 6-bit type number. If the type number is non-zero, the second value is a
pointer to the C++ type info, otherwise the second value is null. Additionally,
CAF generates 32 bit type tokens. These tokens are type hints
that summarizes all types in a type-erased tuple. Two type-erased tuples are of
different type if they have different type tokens (the reverse is not true).
Class type_erased_tuple¶
Note: Calling modifiers on a shared type-erased tuple is undefined behavior.
| Observers | |
bool empty() |
Returns whether this message is empty. |
size_t size() |
Returns the size of this message. |
rtti_pair type(size_t pos) |
Returns run-time type information for the nth element. |
error save(serializer& x) |
Writes the tuple to x. |
error save(size_t n, serializer& x) |
Writes the nth element to x. |
const void* get(size_t n) |
Returns a const pointer to the nth element. |
std::string stringify() |
Returns a string representation of the tuple. |
std::string stringify(size_t n) |
Returns a string representation of the nth element. |
bool matches(size_t n, rtti_pair) |
Checks whether the nth element has given type. |
bool shared() |
Checks whether more than one reference to the data exists. |
bool match_element<T>(size_t n) |
Checks whether element n has type T. |
bool match_elements<Ts...>() |
Checks whether this message has the types Ts.... |
const T& get_as<T>(size_t n) |
Returns a const reference to the nth element. |
| Modifiers | |
void* get_mutable(size_t n) |
Returns a mutable pointer to the nth element. |
T& get_mutable_as<T>(size_t n) |
Returns a mutable reference to the nth element. |
void load(deserializer& x) |
Reads the tuple from x. |
Class message¶
The class message includes all member functions of
type_erased_tuple. However, calling modifiers is always guaranteed
to be safe. A message automatically detaches its content by copying
it from the shared data on mutable access. The class further adds the following
member functions over type_erased_tuple. Note that
apply only detaches the content if a callback takes mutable
references as arguments.
| Observers | |
message drop(size_t n) |
Creates a new message with all but the first n values. |
message drop_right(size_t n) |
Creates a new message with all but the last n values. |
message take(size_t n) |
Creates a new message from the first n values. |
message take_right(size_t n) |
Creates a new message from the last n values. |
message slice(size_t p, size_t n) |
Creates a new message from [p, p + n). |
message extract(message_handler) |
See extract. |
message extract_opts(...) |
See extract-opts. |
| Modifiers | |
optional<message> apply(message_handler f) |
Returns f(*this). |
| Operators | |
message operator+(message x, message y) |
Concatenates x and y. |
message& operator+=(message& x, message y) |
Concatenates x and y. |
Class message_builder¶
| Constructors | |
(void) |
Creates an empty message builder. |
(Iter first, Iter last) |
Adds all elements from range [first, last). |
| Observers | |
bool empty() |
Returns whether this message is empty. |
size_t size() |
Returns the size of this message. |
message to_message( ) |
Converts the buffer to an actual message object. |
append(T val) |
Adds val to the buffer. |
append(Iter first, Iter last) |
Adds all elements from range [first, last). |
message extract(message_handler) |
See extract. |
message extract_opts(...) |
See extract-opts. |
| Modifiers | |
optional<message> apply(message_handler f) |
Returns f(*this). |
message move_to_message() |
Transfers ownership of its data to the new message. |
Extracting¶
The member function message::extract removes matched elements from a
message. x Messages are filtered by repeatedly applying a message handler to the
greatest remaining slice, whereas slices are generated in the sequence
[0, size), [0, size-1), ..., [1, size-1), ...,
[size-1, size). Whenever a slice is matched, it is removed from the message
and the next slice starts at the same index on the reduced message.
For example:
auto msg = make_message(1, 2.f, 3.f, 4);
// remove float and integer pairs
auto msg2 = msg.extract({
[](float, float) { },
[](int, int) { }
});
assert(msg2 == make_message(1, 4));
Step-by-step explanation:
- Slice 1:
(1, 2.f, 3.f, 4), no match - Slice 2:
(1, 2.f, 3.f), no match - Slice 3:
(1, 2.f), no match - Slice 4:
(1), no match - Slice 5:
(2.f, 3.f, 4), no match - Slice 6:
(2.f, 3.f), match; new message is(1, 4) - Slice 7:
(4), no match
Slice 7 is (4), i.e., does not contain the first element, because
the match on slice 6 occurred at index position 1. The function
extract iterates a message only once, from left to right. The
returned message contains the remaining, i.e., unmatched, elements.
Extracting Command Line Options¶
The class message also contains a convenience interface to
extract for parsing command line options: the member function
extract_opts.
int main(int argc, char** argv) {
uint16_t port;
string host = "localhost";
auto res = message_builder(argv + 1, argv + argc).extract_opts({
{"port,p", "set port", port},
{"host,H", "set host (default: localhost)", host},
{"verbose,v", "enable verbose mode"}
});
if (! res.error.empty()) {
// read invalid CLI arguments
cerr << res.error << endl;
return 1;
}
if (res.opts.count("help") > 0) {
// CLI arguments contained "-h", "--help", or "-?" (builtin);
cout << res.helptext << endl;
return 0;
}
if (! res.remainder.empty()) {
// res.remainder stors all extra arguments that weren't consumed
}
if (res.opts.count("verbose") > 0) {
// enable verbose mode
}
// ...
}
/*
Output of ./program_name -h:
Allowed options:
-p [--port] arg : set port
-H [--host] arg : set host (default: localhost)
-v [--verbose] : enable verbose mode
*/
Group Communication¶
CAF supports publish/subscribe-based group communication. Dynamically typed actors can join and leave groups and send messages to groups. The following example showcases the basic API for retrieving a group from a module by its name, joining, and leaving.
std::string module = "local";
std::string id = "foo";
auto expected_grp = system.groups().get(module, id);
if (! expected_grp) {
std::cerr << "*** cannot load group: " << to_string(expected_grp.error())
<< std::endl;
return;
}
auto grp = std::move(*expected_grp);
scoped_actor self{system};
self->join(grp);
self->send(grp, "test");
self->receive(
[](const std::string& str) {
assert(str == "test");
}
);
self->leave(grp);
It is worth mentioning that the module "local" is guaranteed to
never return an error. The example above uses the general API for retrieving
the group. However, local modules can be easier accessed by calling
system.groups().get_local(id), which returns group
instead of expected<group>.
Anonymous Groups¶
Groups created on-the-fly with system.groups().anonymous() can be
used to coordinate a set of workers. Each call to this function returns a new,
unique group instance.
Local Groups¶
The "local" group module creates groups for in-process
communication. For example, a group for GUI related events could be identified
by system.groups().get_local("GUI events"). The group ID
"GUI events" uniquely identifies a singleton group instance of the
module "local".
Remote Groups¶
Calling``system.middleman().publish_local_groups(port, addr)`` makes all local groups available to other nodes in the network. The first argument denotes the port, while the second (optional) parameter can be used to whitelist IP addresses.
After publishing the group at one node (the server), other nodes (the clients)
can get a handle for that group by using the remote module:
system.groups().get("remote", "<group>@<host>:<port>"). This implementation
uses N-times unicast underneath and the group is only available as long as the
hosting server is alive.
Managing Groups of Workers experimental¶
When managing a set of workers, a central actor often dispatches requests to a
set of workers. For this purpose, the class actor_pool implements a
lightweight abstraction for managing a set of workers using a dispatching
policy. Unlike groups, pools usually own their workers.
Pools are created using the static member function make, which
takes either one argument (the policy) or three (number of workers, factory
function for workers, and dispatching policy). After construction, one can add
new workers via messages of the form ('SYS', 'PUT', worker), remove
workers with ('SYS', 'DELETE', worker), and retrieve the set of
workers as vector<actor> via ('SYS', 'GET').
An actor pool takes ownership of its workers. When forced to quit, it sends an exit messages to all of its workers, forcing them to quit as well. The pool also monitors all of its workers.
Pools do not cache messages, but enqueue them directly in a workers mailbox. Consequently, a terminating worker loses all unprocessed messages. For more advanced caching strategies, such as reliable message delivery, users can implement their own dispatching policies.
Dispatching Policies¶
A dispatching policy is a functor with the following signature:
using uplock = upgrade_lock<detail::shared_spinlock>;
using policy = std::function<void (actor_system& sys,
uplock& guard,
const actor_vec& workers,
mailbox_element_ptr& ptr,
execution_unit* host)>;
The argument guard is a shared lock that can be upgraded for unique
access if the policy includes a critical section. The second argument is a
vector containing all workers managed by the pool. The argument ptr
contains the full message as received by the pool. Finally, host is
the current scheduler context that can be used to enqueue workers into the
corresponding job queue.
The actor pool class comes with a set predefined policies, accessible via factory functions, for convenience.
actor_pool::policy actor_pool::round_robin();
This policy forwards incoming requests in a round-robin manner to workers. There is no guarantee that messages are consumed, i.e., work items are lost if the worker exits before processing all of its messages.
actor_pool::policy actor_pool::broadcast();
This policy forwards each message to all workers. Synchronous messages to the pool will be received by all workers, but the client will only recognize the first arriving response message—or error—and discard subsequent messages. Note that this is not caused by the policy itself, but a consequence of forwarding synchronous messages to more than one actor.
actor_pool::policy actor_pool::random();
This policy forwards incoming requests to one worker from the pool chosen
uniformly at random. Analogous to round_robin, this policy does not
cache or redispatch messages.
using join = function<void (T&, message&)>;
using split = function<void (vector<pair<actor, message>>&, message&)>;
template <class T>
static policy split_join(join jf, split sf = ..., T init = T());
This policy models split/join or scatter/gather work flows, where a work item is split into as many tasks as workers are available and then the individuals results are joined together before sending the full result back to the client.
The join function is responsible for glueing all result messages together to
create a single result. The function is called with the result object (initialed
using init) and the current result messages from a worker.
The first argument of a split function is a mapping from actors (workers) to tasks (messages). The second argument is the input message. The default split function is a broadcast dispatching, sending each worker the original request.
Streaming experimental¶
Streams in CAF describe data flow between actors. We are not aiming to provide functionality similar to Apache projects like Spark, Flink or Storm. Likewise, we have different goals than APIs such as RxJava, Reactive Streams, etc. Streams complement asynchronous messages, request/response communication and publish/subscribe in CAF. In a sense, actor streams in CAF are a building block that users could leverage for building feature-complete stream computation engines or reactive high-level Big Data APIs.
A stream establishes a logical channel between two or more actors for exchanging a potentially unbound sequence of values. This channel uses demand signaling to guarantee that senders cannot overload receivers.
Streams are directed and data flows only downstream, i.e., from sender (source) to receiver (sink). Establishing a stream requires a handshake in order to initialize required state and signal initial demand.
CAF distinguishes between three roles in a stream: (1) a source creates streams and generates data, (2) a stage transforms or filters data, and (3) a sink terminates streams by consuming data.
We usually draw streams as pipelines for simplicity. However, sources can have any number of outputs (downstream actors). Likewise, sinks can have any number of inputs (upstream actors) and stages can multiplex N inputs to M outputs. Hence, streaming topologies in CAF support arbitrary complexity with forks and joins.
Stream Managers¶
Streaming-related messages are handled separately. Under the hood, actors delegate to stream managers that in turn allow customization of their behavior with drivers and downstream managers.
Users usually can skip implementing driver classes and instead use the lambda-based interface showcased in the following sections. Drivers implement the streaming logic by taking inputs from upstream actors and pushing data to the downstream manager. A source has no input buffer. Hence, drivers only provide a generator function that downstream managers call according to demand.
A downstream manager is responsible for dispatching data to downstream actors. The default implementation broadcasts data, i.e., all downstream actors receive the same data. The downstream manager can also perform any sort multi- or anycast. For example, a load-balancer would use an anycast policy to dispatch data to the next available worker.
Defining Sources¶
// Simple source for generating a stream of integers from [0, n).
behavior int_source(event_based_actor* self) {
return {
[=](open_atom, int32_t n) {
// Produce at least one value.
if (n <= 0)
n = 1;
// Create a stream manager for implementing a stream source. The
// streaming logic requires three functions: initializer, generator, and
// predicate.
return attach_stream_source(
self,
// Initializer. The type of the first argument (state) is freely
// chosen. If no state is required, `caf::unit_t` can be used here.
[](int32_t& x) { x = 0; },
// Generator. This function is called by CAF to produce new stream
// elements for downstream actors. The `x` argument is our state again
// (with our freely chosen type). The second argument `out` points to
// the output buffer. The template argument (here: int) determines what
// elements downstream actors receive in this stream. Finally, `num` is
// a hint from CAF how many elements we should ideally insert into
// `out`. We can always insert fewer or more items.
[n](int32_t& x, downstream<int32_t>& out, size_t num) {
auto max_x = std::min(x + static_cast<int>(num), n);
for (; x < max_x; ++x)
out.push(x);
},
// Predicate. This function tells CAF when we reached the end.
[n](const int32_t& x) { return x == n; });
},
};
}
The simplest way to defining a source is to use the attach_stream_source
function and pass it four arguments: a pointer to self, initializer for the
state, generator for producing values, and predicate for signaling the end
of the stream.
Defining Stages¶
// Simple stage that only selects even numbers.
behavior int_selector(event_based_actor* self) {
return {
[=](stream<int32_t> in) {
// Create a stream manager for implementing a stream stage. Similar to
// `make_source`, we need three functions: initialzer, processor, and
// finalizer.
return attach_stream_stage(
self,
// Our input source.
in,
// Initializer. Here, we don't need any state and simply use unit_t.
[](unit_t&) {
// nop
},
// Processor. This function takes individual input elements as `val`
// and forwards even integers to `out`.
[](unit_t&, downstream<int32_t>& out, int32_t val) {
if (val % 2 == 0)
out.push(val);
},
// Finalizer. Allows us to run cleanup code once the stream terminates.
[=](unit_t&, const error& err) {
if (err) {
aout(self) << "int_selector aborted with error: " << err
<< std::endl;
} else {
aout(self) << "int_selector finalized" << std::endl;
}
// else: regular stream shutdown
});
},
};
}
The function make_stage also takes three lambdas but additionally the
received input stream handshake as first argument. Instead of a predicate,
make_stage only takes a finalizer, since the stage does not produce data on
its own and a stream terminates if no more sources exist.
Defining Sinks¶
behavior int_sink(event_based_actor* self) {
return {
[=](stream<int32_t> in) {
// Create a stream manager for implementing a stream sink. Once more, we
// have to provide three functions: Initializer, Consumer, Finalizer.
return attach_stream_sink(
self,
// Our input source.
in,
// Initializer. Here, we store all values we receive. Note that streams
// are potentially unbound, so this is usually a bad idea outside small
// examples like this one.
[](std::vector<int>&) {
// nop
},
// Consumer. Takes individual input elements as `val` and stores them
// in our history.
[](std::vector<int32_t>& xs, int32_t val) { xs.emplace_back(val); },
// Finalizer. Allows us to run cleanup code once the stream terminates.
[=](std::vector<int32_t>& xs, const error& err) {
if (err) {
aout(self) << "int_sink aborted with error: " << err << std::endl;
} else {
aout(self) << "int_sink finalized after receiving: " << xs
<< std::endl;
}
});
},
};
}
The function make_sink is similar to make_stage, except that is does not
produce outputs.
Initiating Streams¶
void caf_main(actor_system& sys, const config& cfg) {
auto src = sys.spawn(int_source);
auto snk = sys.spawn(int_sink);
auto pipeline = cfg.with_stage ? snk * sys.spawn(int_selector) * src
: snk * src;
anon_send(pipeline, open_atom_v, cfg.n);
}
In our example, we always have a source int_source and a sink int_sink
with an optional stage int_selector. Sending open_atom to the source
initiates the stream and the source will respond with a stream handshake.
Using the actor composition in CAF (snk * src reads sink after source)
allows us to redirect the stream handshake we send in caf_main to the sink
(or to the stage and then from the stage to the sink).
Testing¶
CAF comes with built-in support for writing unit tests in a domain-specific language (DSL). The API looks similar to well-known testing frameworks such as Boost.Test and Catch but adds CAF-specific macros for testing messaging between actors.
Our design leverages four main concepts:
- Checks represent single boolean expressions.
- Tests contain one or more checks.
- Fixtures equip tests with a fixed data environment.
- Suites group tests together.
The following code illustrates a very basic test case that captures the four main concepts described above.
// Adds all tests in this compilation unit to the suite "math".
#define CAF_SUITE math
// Pulls in all the necessary macros.
#include "caf/test/dsl.hpp"
namespace {
struct fixture {};
} // namespace
// Makes all members of `fixture` available to tests in the scope.
CAF_TEST_FIXTURE_SCOPE(math_tests, fixture)
// Implements our first test.
CAF_TEST(divide) {
CAF_CHECK(1 / 1 == 0); // fails
CAF_CHECK(2 / 2 == 1); // passes
CAF_REQUIRE(3 + 3 == 5); // fails and aborts test execution
CAF_CHECK(4 - 4 == 0); // unreachable due to previous requirement error
}
CAF_TEST_FIXTURE_SCOPE_END()
The code above highlights the two basic macros CAF_CHECK and
CAF_REQUIRE. The former reports failed checks, but allows the test
to continue on error. The latter stops test execution if the boolean expression
evaluates to false.
The third macro worth mentioning is CAF_FAIL. It unconditionally
stops test execution with an error message. This is particularly useful for
stopping program execution after receiving unexpected messages, as we will see
later.
Testing Actors¶
The following example illustrates how to add an actor system as well as a
scoped actor to fixtures. This allows spawning of and interacting with actors
in a similar way regular programs would. Except that we are using macros such
as CAF_CHECK and provide tests rather than implementing a
caf_main.
namespace {
struct fixture {
caf::actor_system_config cfg;
caf::actor_system sys;
caf::scoped_actor self;
fixture() : sys(cfg), self(sys) {
// nop
}
};
caf::behavior adder() {
return {
[=](int x, int y) {
return x + y;
}
};
}
} // namespace
CAF_TEST_FIXTURE_SCOPE(actor_tests, fixture)
CAF_TEST(simple actor test) {
// Our Actor-Under-Test.
auto aut = self->spawn(adder);
self->request(aut, caf::infinite, 3, 4).receive(
[=](int r) {
CAF_CHECK(r == 7);
},
[&](caf::error& err) {
// Must not happen, stop test.
CAF_FAIL(err);
});
}
CAF_TEST_FIXTURE_SCOPE_END()
The example above works, but suffers from several issues:
- Significant amount of boilerplate code.
- Using a scoped actor as illustrated above can only test one actor at a time. However, messages between other actors are invisible to us.
- CAF runs actors in a thread pool by default. The resulting nondeterminism makes triggering reliable ordering of messages near impossible. Further, forcing timeouts to test error handling code is even harder.
Deterministic Testing¶
CAF provides a scheduler implementation specifically tailored for writing unit
tests called test_coordinator. It does not start any threads and
instead gives unit tests full control over message dispatching and timeout
management.
To reduce boilerplate code, CAF also provides a fixture template called
test_coordinator_fixture that comes with ready-to-use actor system
(sys) and testing scheduler (sched). The optional
template parameter allows unit tests to plugin custom actor system
configuration classes.
Using this fixture unlocks three additional macros:
expectchecks for a single message. The macro verifies the content types of the message and invokes the necessary member functions on the test coordinator. Optionally, the macro checks the receiver of the message and its content. If the expected message does not exist, the test aborts.allowis similar toexpect, but it does not abort the test if the expected message is missing. This macro returnstrueif the allowed message was delivered,falseotherwise.disallowaborts the test if a particular message was delivered to an actor.
The following example implements two actors, ping and pong, that
exchange a configurable amount of messages. The test three pings then checks
the contents of each message with expect and verifies that no additional
messages exist using disallow.
namespace {
behavior ping(event_based_actor* self, actor pong_actor, int n) {
self->send(pong_actor, ping_atom_v, n);
return {
[=](pong_atom, int x) {
if (x > 1)
self->send(pong_actor, ping_atom_v, x - 1);
},
};
}
behavior pong() {
return {
[=](ping_atom, int x) { return make_result(pong_atom_v, x); },
};
}
struct ping_pong_fixture : test_coordinator_fixture<> {
actor pong_actor;
ping_pong_fixture() {
// Spawn the Pong actor.
pong_actor = sys.spawn(pong);
// Run initialization code for Pong.
run();
}
};
} // namespace
CAF_TEST_FIXTURE_SCOPE(ping_pong_tests, ping_pong_fixture)
CAF_TEST(three pings) {
// Spawn the Ping actor and run its initialization code.
auto ping_actor = sys.spawn(ping, pong_actor, 3);
sched.run_once();
// Test communication between Ping and Pong.
expect((ping_atom, int), from(ping_actor).to(pong_actor).with(_, 3));
expect((pong_atom, int), from(pong_actor).to(ping_actor).with(_, 3));
expect((ping_atom, int), from(ping_actor).to(pong_actor).with(_, 2));
expect((pong_atom, int), from(pong_actor).to(ping_actor).with(_, 2));
expect((ping_atom, int), from(ping_actor).to(pong_actor).with(_, 1));
expect((pong_atom, int), from(pong_actor).to(ping_actor).with(_, 1));
// No further messages allowed.
disallow((ping_atom, int), from(ping_actor).to(pong_actor).with(_, 1));
}
CAF_TEST_FIXTURE_SCOPE_END()
Middleman¶
The middleman is the main component of the I/O module and enables distribution.
It transparently manages proxy actor instances representing remote actors,
maintains connections to other nodes, and takes care of serialization of
messages. Applications install a middleman by loading caf::io::middleman as
module (see Configuring Actor Applications). Users can include "caf/io/all.hpp" to get
access to all public classes of the I/O module.
Class middleman¶
| Remoting | |
expected<uint16> open(uint16, const char*, bool) |
See Publishing and Connecting. |
expected<uint16> publish(T, uint16, const char*, bool) |
See Publishing and Connecting. |
expected<void> unpublish(T x, uint16) |
See Publishing and Connecting. |
expected<node_id> connect(std::string host, uint16_t port) |
See Publishing and Connecting. |
expected<T> remote_actor<T = actor>(string, uint16) |
See Publishing and Connecting. |
expected<T> spawn_broker(F fun, ...) |
See Network I/O with Brokers. |
expected<T> spawn_client(F, string, uint16, ...) |
See Network I/O with Brokers. |
expected<T> spawn_server(F, uint16, ...) |
See Network I/O with Brokers. |
Publishing and Connecting¶
The member function publish binds an actor to a given port, thereby
allowing other nodes to access it over the network.
template <class T>
expected<uint16_t> middleman::publish(T x, uint16_t port,
const char* in = nullptr,
bool reuse_addr = false);
The first argument is a handle of type actor or
typed_actor<...>. The second argument denotes the TCP port. The OS
will pick a random high-level port when passing 0. The third parameter
configures the listening address. Passing null will accept all incoming
connections (INADDR_ANY). Finally, the flag reuse_addr
controls the behavior when binding an IP address to a port, with the same
semantics as the BSD socket flag SO_REUSEADDR. For example, with
reuse_addr = false, binding two sockets to 0.0.0.0:42 and
10.0.0.1:42 will fail with EADDRINUSE since 0.0.0.0 includes 10.0.0.1.
With reuse_addr = true binding would succeed because 10.0.0.1 and
0.0.0.0 are not literally equal addresses.
The member function returns the bound port on success. Otherwise, an error
(see Errors) is returned.
template <class T>
expected<uint16_t> middleman::unpublish(T x, uint16_t port = 0);
The member function unpublish allows actors to close a port
manually. This is performed automatically if the published actor terminates.
Passing 0 as second argument closes all ports an actor is published to,
otherwise only one specific port is closed.
The function returns an error (see Errors) if the actor was not bound
to given port.
template<class T = actor>
expected<T> middleman::remote_actor(std::string host, uint16_t port);
After a server has published an actor with publish, clients can
connect to the published actor by calling remote_actor:
// node A
auto ping = spawn(ping);
system.middleman().publish(ping, 4242);
// node B
auto ping = system.middleman().remote_actor("node A", 4242);
if (!ping)
cerr << "unable to connect to node A: " << to_string(ping.error()) << '\n';
else
self->send(*ping, ping_atom::value);
There is no difference between server and client after the connection phase. Remote actors use the same handle types as local actors and are thus fully transparent.
The function pair open and connect allows users to connect CAF instances
without remote actor setup. The function connect returns a node_id that
can be used for remote spawning (see (see Remote Spawning of Actors experimental)).
Free Functions¶
The following free functions in the namespace caf::io avoid calling
the middleman directly. This enables users to easily switch between
communication backends as long as the interfaces have the same signatures. For
example, the (experimental) OpenSSL binding of CAF implements the same
functions in the namespace caf::openssl to easily switch between
encrypted and unencrypted communication.
expected<uint16> open(actor_system&, uint16, const char*, bool) |
See Publishing and Connecting. |
expected<uint16> publish(T, uint16, const char*, bool) |
See Publishing and Connecting. |
expected<void> unpublish(T x, uint16) |
See Publishing and Connecting. |
expected<node_id> connect(actor_system&, std::string host, uint16_t port) |
See Publishing and Connecting. |
expected<T> remote_actor<T = actor>(actor_system&, string, uint16) |
See Publishing and Connecting. |
Transport Protocols experimental¶
CAF communication uses TCP per default and thus the functions shown in the middleman API above are related to TCP. There are two alternatives to plain TCP: TLS via the OpenSSL module shortly discussed in (see Free Functions) and UDP.
UDP is integrated in the default multiplexer and BASP broker. Set the flag
middleman_enable_udp to true to enable it (see Configuring Actor Applications). This
does not require you to disable TCP. Use publish_udp and
remote_actor_udp to establish communication.
Communication via UDP is inherently unreliable and unordered. CAF reestablishes order and drops messages that arrive late. Messages that are sent via datagrams are limited to a maximum of 65.535 bytes which is used as a receive buffer size by CAF. Note that messages that exceed the MTU are fragmented by IP and are considered lost if a single fragment is lost. Optional reliability based on retransmissions and messages slicing on the application layer are planned for the future.
Network I/O with Brokers¶
When communicating to other services in the network, sometimes low-level socket I/O is inevitable. For this reason, CAF provides brokers. A broker is an event-based actor running in the middleman that multiplexes socket I/O. It can maintain any number of acceptors and connections. Since the broker runs in the middleman, implementations should be careful to consume as little time as possible in message handlers. Brokers should outsource any considerable amount of work by spawning new actors or maintaining worker actors.
Note that all UDP-related functionality is still experimental.
Spawning Brokers¶
Brokers are implemented as functions and are spawned by calling on of the three following member functions of the middleman.
template <spawn_options Os = no_spawn_options,
class F = std::function<void(broker*)>, class... Ts>
typename infer_handle_from_fun<F>::type
spawn_broker(F fun, Ts&&... xs);
template <spawn_options Os = no_spawn_options,
class F = std::function<void(broker*)>, class... Ts>
expected<typename infer_handle_from_fun<F>::type>
spawn_client(F fun, const std::string& host, uint16_t port, Ts&&... xs);
template <spawn_options Os = no_spawn_options,
class F = std::function<void(broker*)>, class... Ts>
expected<typename infer_handle_from_fun<F>::type>
spawn_server(F fun, uint16_t port, Ts&&... xs);
The function spawn_broker simply spawns a broker. The convenience
function spawn_client tries to connect to given host and port over
TCP and returns a broker managing this connection on success. Finally,
spawn_server opens a local TCP port and spawns a broker managing it
on success. There are no convenience functions spawn a UDP-based client or
server.
Class broker¶
void configure_read(connection_handle hdl, receive_policy::config config);
Modifies the receive policy for the connection identified by hdl.
This will cause the middleman to enqueue the next new_data_msg
according to the given config created by
receive_policy::exactly(x), receive_policy::at_most(x),
or receive_policy::at_least(x) (with x denoting the
number of bytes).
void write(connection_handle hdl, size_t num_bytes, const void* buf)
void write(datagram_handle hdl, size_t num_bytes, const void* buf)
Writes data to the output buffer.
void enqueue_datagram(datagram_handle hdl, std::vector<char> buf);
Enqueues a buffer to be sent as a datagram. Use of this function is encouraged
over write as it allows reuse of the buffer which can be returned to the broker
in a datagram_sent_msg.
void flush(connection_handle hdl);
void flush(datagram_handle hdl);
Sends the data from the output buffer.
template <class F, class... Ts>
actor fork(F fun, connection_handle hdl, Ts&&... xs);
Spawns a new broker that takes ownership of a given connection.
size_t num_connections();
Returns the number of open connections.
void close(connection_handle hdl);
void close(accept_handle hdl);
void close(datagram_handle hdl);
Closes the endpoint related to the handle.
expected<std::pair<accept_handle, uint16_t>>
add_tcp_doorman(uint16_t port = 0, const char* in = nullptr,
bool reuse_addr = false);
Creates new doorman that accepts incoming connections on a given port and returns the handle to the doorman and the port in use or an error.
expected<connection_handle>
add_tcp_scribe(const std::string& host, uint16_t port);
Creates a new scribe to connect to host:port and returns handle to it or an error.
expected<std::pair<datagram_handle, uint16_t>>
add_udp_datagram_servant(uint16_t port = 0, const char* in = nullptr,
bool reuse_addr = false);
Creates a datagram servant to handle incoming datagrams on a given port. Returns the handle to the servant and the port in use or an error.
expected<datagram_handle>
add_udp_datagram_servant(const std::string& host, uint16_t port);
Creates a datagram servant to send datagrams to host:port and returns a handle to it or an error.
Manually Triggering Events experimental¶
Brokers receive new events as new_connection_msg and
new_data_msg as soon and as often as they occur, per default. This
means a fast peer can overwhelm a broker by sending it data faster than the
broker can process it. In particular if the broker outsources work items to
other actors, because work items can accumulate in the mailboxes of the
workers.
Calling self->trigger(x,y), where x is a connection or
acceptor handle and y is a positive integer, allows brokers to halt
activities after y additional events. Once a connection or acceptor
stops accepting new data or connections, the broker receives a
connection_passivated_msg or acceptor_passivated_msg.
Brokers can stop activities unconditionally with self->halt(x) and
resume activities unconditionally with self->trigger(x).
Remote Spawning of Actors experimental¶
Remote spawning is an extension of the dynamic spawn using run-time type names
(see Adding Custom Actor Types experimental). The following example assumes a typed actor
handle named calculator with an actor implementing this messaging interface
named “calculator”.
void client(actor_system& system, const config& cfg) {
auto node = system.middleman().connect(cfg.host, cfg.port);
if (!node) {
cerr << "*** connect failed: " << to_string(node.error()) << endl;
return;
}
auto type = "calculator"; // type of the actor we wish to spawn
auto args = make_message(); // arguments to construct the actor
auto tout = std::chrono::seconds(30); // wait no longer than 30s
auto worker = system.middleman().remote_spawn<calculator>(*node, type, args,
tout);
if (!worker) {
cerr << "*** remote spawn failed: " << to_string(worker.error()) << endl;
return;
}
// start using worker in main loop
client_repl(make_function_view(*worker));
// be a good citizen and terminate remotely spawned actor before exiting
anon_send_exit(*worker, exit_reason::kill);
}
We first connect to a CAF node with middleman().connect(...). On success,
connect returns the node ID we need for remote_spawn. This requires the
server to open a port with middleman().open(...) or
middleman().publish(...). Alternatively, we can obtain the node ID from an
already existing remote actor handle—returned from remote_actor for
example—via hdl->node(). After connecting to the server, we can use
middleman().remote_spawn<...>(...) to create actors remotely.
Frequently Asked Questions¶
This Section is a compilation of the most common questions via GitHub, chat, and mailing list.
Can I Encrypt CAF Communication?¶
Yes, by using the OpenSSL module (see Free Functions).
Can I Create Messages Dynamically?¶
Yes.
Usually, messages are created implicitly when sending messages but can also be
created explicitly using make_message. In both cases, types and number of
elements are known at compile time. To allow for fully dynamic message
generation, CAF also offers message_builder:
message_builder mb;
// prefix message with some atom
mb.append(strings_atom::value);
// fill message with some strings
std::vector<std::string> strings{/*...*/};
for (auto& str : strings)
mb.append(str);
// create the message
message msg = mb.to_message();
What Debugging Tools Exist?¶
The scripts/ and tools/ directories contain some useful tools to aid in
development and debugging.
scripts/atom.py converts integer atom values back into strings.
scripts/demystify.py replaces cryptic typed_mpi<...>
templates with replies_to<...>::with<...> and
atom_constant<...> with a human-readable representation of the
actual atom.
scripts/caf-prof is an R script that generates plots from CAF
profiler output.
caf-vec is a (highly) experimental tool that annotates CAF logs
with vector timestamps. It gives you happens-before relations and a nice
visualization via ShiViz.
Utility¶
CAF includes a few utility classes that are likely to be part of C++ eventually (or already are in newer versions of the standard). However, until these classes are part of the standard library on all supported compilers, we unfortunately have to maintain our own implementations.
Class optional¶
Represents a value that may or may not exist.
| Constructors | |
(T value) |
Constructs an object with a value. |
(none_t = none) |
Constructs an object without a value. |
| Observers | |
explicit operator bool() |
Checks whether the object contains a value. |
T* operator->() |
Accesses the contained value. |
T& operator*() |
Accesses the contained value. |
Class expected¶
Represents the result of a computation that should return a value. If no value
could be produced, the expected<T> contains an error (see Errors).
| Constructors | |
(T value) |
Constructs an object with a value. |
(error err) |
Constructs an object with an error. |
| Observers | |
explicit operator bool() |
Checks whether the object contains a value. |
T* operator->() |
Accesses the contained value. |
T& operator*() |
Accesses the contained value. |
error& error() |
Accesses the contained error. |
Constant unit¶
The constant unit of type unit_t is the equivalent of
void and can be used to initialize optional<void> and
expected<void>.
Constant none¶
The constant none of type none_t can be used to
initialize an optional<T> to represent “nothing”.
Common Pitfalls¶
This Section highlights common mistakes or C++ subtleties that can show up when programming in CAF.
Defining Message Handlers¶
C++ evaluates comma-separated expressions from left-to-right, using only the last element as return type of the whole expression. This means that message handlers and behaviors must not be initialized like this:
message_handler wrong = (
[](int i) { /*...*/ },
[](float f) { /*...*/ }
);
The correct way to initialize message handlers and behaviors is to either
use the constructor or the member function assign:
message_handler ok1{
[](int i) { /*...*/ },
[](float f) { /*...*/ }
};
message_handler ok2;
// some place later
ok2.assign(
[](int i) { /*...*/ },
[](float f) { /*...*/ }
);
Event-Based API¶
The member function become does not block, i.e., always returns
immediately. Thus, lambda expressions should always capture by value.
Otherwise, all references on the stack will cause undefined behavior if the
lambda expression is executed.
Requests¶
A handle returned by request represents exactly one response
message. It is not possible to receive more than one response message.
The handle returned by request is bound to the calling actor. It is
not possible to transfer a handle to a response to another actor.
Sharing¶
It is strongly recommended to not share states between actors. In
particular, no actor shall ever access member variables or member functions of
another actor. Accessing shared memory segments concurrently can cause undefined
behavior that is incredibly hard to find and debug. However, sharing
data between actors is fine, as long as the data is immutable
and its lifetime is guaranteed to outlive all actors. The simplest way to meet
the lifetime guarantee is by storing the data in smart pointers such as
std::shared_ptr. Nevertheless, the recommended way of sharing
information is message passing. Sending the same message to multiple actors
does not result in copying the data several times.
Using aout – A Concurrency-safe Wrapper for cout¶
When using cout from multiple actors, output often appears
interleaved. Moreover, using cout from multiple actors – and thus
from multiple threads – in parallel should be avoided regardless, since the
standard does not guarantee a thread-safe implementation.
By replacing std::cout with caf::aout, actors can achieve a
concurrency-safe text output. The header caf/all.hpp also defines overloads
for std::endl and std::flush for aout, but does not support the full
range of ostream operations (yet). Each write operation to aout sends a
message to a “hidden” actor. This actor only prints lines, unless output is
forced using flush. The example below illustrates printing of lines of text
from multiple actors (in random order).
#include <random>
#include <chrono>
#include <cstdlib>
#include <iostream>
#include "caf/all.hpp"
using namespace caf;
using std::endl;
void caf_main(actor_system& system) {
for (int i = 1; i <= 50; ++i) {
system.spawn([i](blocking_actor* self) {
aout(self) << "Hi there! This is actor nr. "
<< i << "!" << endl;
std::random_device rd;
std::default_random_engine re(rd());
std::chrono::milliseconds tout{re() % 10};
self->delayed_send(self, tout, 42);
self->receive(
[i, self](int) {
aout(self) << "Actor nr. "
<< i << " says goodbye!" << endl;
}
);
});
}
}
CAF_MAIN()
Migration Guides¶
The guides in this section document all possibly breaking changes in the library for that last versions of CAF.
0.8 to 0.9¶
Version 0.9 included a lot of changes and improvements in its implementation, but it also made breaking changes to the API.
self has been removed¶
This is the biggest library change since the initial release. The major problem
with this keyword-like identifier is that it must have a single type as it’s
implemented as a thread-local variable. Since there are so many different kinds
of actors (event-based or blocking, untyped or typed), self needs
to perform type erasure at some point, rendering it ultimately useless. Instead
of a thread-local pointer, you can now use the first argument in functor-based
actors to “catch” the self pointer with proper type information.
actor_ptr has been replaced¶
CAF now distinguishes between handles to actors, i.e.,
typed_actor<...> or simply actor, and addresses
of actors, i.e., actor_addr. The reason for this change is that
each actor has a logical, (network-wide) unique address, which is used by the
networking layer of CAF. Furthermore, for monitoring or linking, the address
is all you need. However, the address is not sufficient for sending messages,
because it doesn’t have any type information. The function
current_sender() now returns the address of the sender. This
means that previously valid code such as send(current_sender(),...)
will cause a compiler error. However, the recommended way of replying to
messages is to return the result from the message handler.
The API for typed actors is now similar to the API for untyped actors¶
The APIs of typed and untyped actors have been harmonized. Typed actors can now be published in the network and also use all operations untyped actors can.
0.9 to 0.10 (libcppa to CAF)¶
The first release under the new name CAF is an overhaul of the entire library.
Some classes have been renamed or relocated, others have been removed. The
purpose of this refactoring was to make the library easier to grasp and to make
its API more consistent. All classes now live in the namespace caf and
all headers have the top level folder caf instead of cppa.
For example, cppa/actor.hpp becomes caf/actor.hpp. Further,
the convenience header to get all parts of the user API is now
"caf/all.hpp". The networking has been separated from the core
library. To get the networking components, simply include
caf/io/all.hpp and use the namespace caf::io.
Version 0.10 still includes the header cppa/cppa.hpp to make the
transition process for users easier and to not break existing code right away.
The header defines the namespace cppa as an alias for caf.
Furthermore, it provides implementations or type aliases for renamed or removed
classes such as cow_tuple. You won’t get any warning about deprecated
headers with 0.10. However, we will add this warnings in the next library
version and remove deprecated code eventually.
Even when using the backwards compatibility header, the new library has breaking changes. For instance, guard expressions have been removed entirely. The reasoning behind this decision is that we already have projections to modify the outcome of a match. Guard expressions add little expressive power to the library but a whole lot of code that is hard to maintain in the long run due to its complexity. Using projections to not only perform type conversions but also to restrict values is the more natural choice.
any_tuple => message
This type is only being used to pass a message from one actor to another.
Hence, message is the logical name.
partial_function => message_handler
Technically, it still is a partial function. However, we wanted to put emphasize on its use case.
cow_tuple => X
We want to provide a streamlined, simple API. Shipping a full tuple abstraction
with the library does not fit into this philosophy. The removal of
cow_tuple implies the removal of related functions such as
tuple_cast.
cow_ptr => X
This pointer class is an implementation detail of message and
should not live in the global namespace in the first place. It also had the
wrong name, because it is intrusive.
X => message_builder
This new class can be used to create messages dynamically. For example, the
content of a vector can be used to create a message using a series of
append calls.
accept_handle, connection_handle, publish, remote_actor,
max_msg_size, typed_publish, typed_remote_actor, publish_local_groups,
new_connection_msg, new_data_msg, connection_closed_msg, acceptor_closed_msg
These classes concern I/O functionality and have thus been moved to
caf::io
0.10 to 0.11¶
Version 0.11 introduced new, optional components. The core library itself,
however, mainly received optimizations and bugfixes with one exception: the
member function on_exit is no longer virtual. You can still provide
it to define a custom exit handler, but you must not use override.
0.11 to 0.12¶
Version 0.12 removed two features:
- Type names are no longer demangled automatically. Hence, users must explicitly pass the type name as first argument when using
announce, i.e.,announce<my_class>(...)becomesannounce<my_class>("my_class", ...). - Synchronous send blocks no longer support
continue_with. This feature has been removed without substitution.
0.12 to 0.13¶
This release removes the (since 0.9 deprecated) cppa headers and
deprecates all *_send_tuple versions (simply use the function
without _tuple suffix). local_actor::on_exit once again
became virtual.
In case you were using the old cppa::options_description API, you can
migrate to the new API based on extract (see Extracting Command Line Options).
Most importantly, version 0.13 slightly changes last_dequeued and
last_sender. Both functions will now cause undefined behavior (dereferencing
a nullptr) instead of returning dummy values when accessed from outside a
callback or after forwarding the current message. Besides, these function names
were not a good choice in the first place, since “last” implies accessing data
received in the past. As a result, both functions are now deprecated. Their
replacements are named current_message and current_sender (see
Messaging Interfaces).
0.13 to 0.14¶
The function timed_sync_send has been removed. It offered an
alternative way of defining message handlers, which is inconsistent with the
rest of the API.
The policy classes broadcast, random, and
round_robin in actor_pool were removed and replaced by
factory functions using the same name.
0.14 to 0.15¶
Version 0.15 replaces the singleton-based architecture with actor_system.
Most of the free functions in namespace caf are now member functions of
actor_system (see Environment / Actor Systems). Likewise, most functions in
namespace caf::io are now member functions of middleman (see
Middleman). The structure of CAF applications has changed fundamentally
with a focus on configurability. Setting and fine-tuning the scheduler, changing
parameters of the middleman, etc. is now bundled in the class
actor_system_config. The new configuration mechanism is also easily
extensible.
Patterns are now limited to the simple notation, because the advanced features (1) are not implementable for statically typed actors, (2) are not portable to Windows/MSVC, and (3) drastically impact compile times. Dropping this functionality also simplifies the implementation and improves performance.
The blocking_api flag has been removed. All variants of
spawn now auto-detect blocking actors.