source: doc/theses/colby_parsons_MMAth/text/actors.tex @ 739bd7b

% ======================================================================
% ======================================================================
\chapter{Actors}\label{s:actors}
% ======================================================================
% ======================================================================

% C_TODO: add citations throughout chapter
Actors are an indirect concurrent feature that abstracts threading away from a programmer, and instead provides \gls{actor}s and messages as building blocks for concurrency, where message passing means there is no shared data to protect, making actors amenable in a distributed environment.
Actors are another message passing concurrency feature, similar to channels but with more abstraction, and are in the realm of \gls{impl_concurrency}, where programmers write concurrent code without dealing with explicit thread create or interaction.
The study of actors can be broken into two concepts, the \gls{actor_model}, which describes the model of computation and the \gls{actor_system}, which refers to the implementation of the model.
Before discussing \CFA's actor system in detail, it is important to first describe the actor model, and the classic approach to implementing an actor system.

\section{Actor Model}
The actor model is a concurrent paradigm where computation is broken into units of work called actors, and the data for computation is distributed to actors in the form of messages~\cite{Hewitt73}.
An actor is composed of a \Newterm{mailbox} (message queue) and a set of \Newterm{behaviours} that receive from the mailbox to perform work.
Actors execute asynchronously upon receiving a message and can modify their own state, make decisions, spawn more actors, and send messages to other actors.
Because the actor model is implicit concurrency, its strength is that it abstracts away many details and concerns needed in other concurrent paradigms.
For example, mutual exclusion and locking are rarely relevant concepts in an actor model, as actors typically only operate on local state.

An actor does not have a thread.
An actor is executed by an underlying \Newterm{executor} (kernel thread-pool) that fairly invokes each actor, where an actor invocation processes one or more messages from its mailbox.
The default number of executor threads is often proportional to the number of computer cores to achieve good performance.
An executor is often tunable with respect to the number of kernel threads and its scheduling algorithm, which optimize for specific actor applications and workloads \see{end of Section~\ref{s:CFAActor}}.

\subsection{Classic Actor System}
An implementation of the actor model with a community of actors is called an actor system.
Actor systems largely follow the actor model, but can differ in some ways.
While the semantics of message \emph{send} is asynchronous, the implementation may be synchronous or a combination.
The default semantics for message \emph{receive} is FIFO, so an actor receives messages from its mailbox in temporal (arrival) order;
however, messages sent among actors arrive in any order.
Some actor systems provide priority-based mailboxes and/or priority-based message-selection within a mailbox, where custom message dispatchers search among or within a mailbox(es) with a predicate for specific kinds of actors and/or messages.
Some actor systems provide a shared mailbox where multiple actors receive from a common mailbox~\cite{Akka}, which is contrary to the no-sharing design of the basic actor-model (and requires additional locking).
For non-FIFO service, some notion of fairness (eventual progress) must exist, otherwise messages have a high latency or starve, \ie never received.
Finally, some actor systems provide multiple typed-mailboxes, which then lose the actor-\lstinline{become} mechanism (see Section~\ref{s:SafetyProductivity}).
%While the definition of the actor model provides no restrictions on message ordering, actor systems tend to guarantee that messages sent from a given actor $i$ to actor $j$ will arrive at actor $j$ in the order they were sent.
Another way an actor system varies from the model is allowing access to shared global-state.
When this occurs, it complicates the implementation as this breaks any implicit mutual-exclusion guarantees when only accessing local-state.

\begin{figure}
\begin{tabular}{l|l}
\subfloat[Actor-centric system]{\label{f:standard_actor}\input{diagrams/standard_actor.tikz}} &
\subfloat[Message-centric system]{\label{f:inverted_actor}\raisebox{.1\height}{\input{diagrams/inverted_actor.tikz}}}
\end{tabular}
\caption{Classic and inverted actor implementation approaches with sharded queues.}
\end{figure}

\subsection{\CFA Actor System}
Figure~\ref{f:standard_actor} shows an actor system designed as \Newterm{actor-centric}, where a set of actors are scheduled and run on underlying executor threads~\cite{CAF,Akka,ProtoActor}.
The simplest design has a single global queue of actors accessed by the executor threads, but this approach results in high contention as both ends of the queue by the executor threads.
The more common design is to \Newterm{shard} the single queue among the executor threads, where actors are permanently assigned or can float among the queues.
Sharding significantly decreases contention among executor threads adding and removing actors to/from a queue.
Finally, each actor has a receive queue of messages (mailbox), which is a single consumer, multi-producer queue, \ie only the actor removes from the mailbox but multiple actors can attach messages.
When an actor receives a message in its mailbox, the actor is marked ready and scheduled by a thread to run the actor's current work unit on the message(s).

% cite parallel theatre and our paper
Figure \ref{f:inverted_actor} shows an actor system designed as \Newterm{message-centric}, where a set of messages are scheduled and run on underlying executor threads~\cite{uC++,Nigro21}.
Again, the simplest design has a single global queue of messages accessed by the executor threads, but this approach has the same contention problem by the executor threads.
Therefore, the messages (mailboxes) are sharded and executor threads schedule each message, which points to its corresponding actor.
Here, an actor's messages are permanently assigned to one queue to ensure FIFO receiving and/or reduce searching for specific actor/messages.
Since multiple actors belong to each message queue, actor messages are interleaved on a queue.
This design is \Newterm{inverted} because actors belong to a message queue, whereas in the classic approach a message queue belongs to each actor.
% In this inverted actor system instead of each executor threads owning a queue of actors, they each own a queue of messages.
% In this scheme work is consumed from their queue and executed by underlying threads.
The inverted model can be taken a step further by sharding the message queues for each executor threads, so each executor thread owns a set of queues and cycles through them.
Again, this extra level of sharding is to reduce queue contention.
% The arrows from the message queues to the actors in the diagram indicate interleaved messages addressed to each actor.

The actor system in \CFA uses a message-centric design, adopts several features from my prior actor work in \uC~\cite{}, and adds the following contributions related to \CFA:
\begin{enumerate}[topsep=5pt,itemsep=3pt,parsep=0pt]
\item
Provide insight into the impact of envelope allocation in actor systems.
In all actor systems, dynamic allocation is needed to ensure the lifetime of a unit of work persists from its creation until the unit of work is executed.
This allocation is often called an \Newterm{envelope} as it ``packages'' the information needed to run the unit of work, alongside any other information needed to send the unit of work, such as an actor's address or link fields.
This dynamic allocation occurs once per message sent.
Unfortunately, the high rate of message sends in an actor system results in significant contention on the memory allocator.
A novel data structure is introduced to consolidate allocations to improve performance by minimizing allocator contention.

\item
Improve performance of the inverted actor system using multiple approaches to minimize contention on queues, such as queue gulping and avoiding atomic operations.

\item
Introduce work stealing in the inverted actor system.
Work stealing in an actor-centric system involves stealing one or more actors among executor threads.
In the inverted system, the notion of stealing message queues is introduced.
The queue stealing is implemented such that the act of stealing work does not contend with non-stealing executor threads running actors.

\item
Introduce and evaluate a timestamp-based work-stealing heuristic with the goal of maintaining non-workstealing performance in work-saturated workloads and improving performance on unbalanced workloads.

\item
Provide a suite of safety and productivity features including static-typing, detection of erroneous message sends, statistics tracking, and more.
\end{enumerate}

\section{\CFA Actor}\label{s:CFAActor}
\CFA is not an object oriented language and it does not have \gls{rtti}.
As such, all message sends and receives among actors can only occur using static type-matching, as in Typed-Akka~\cite{AkkaTyped}.
Figure~\ref{f:BehaviourStyles} contrasts dynamic and static type-matching.
Figure~\ref{l:dynamic_style} shows the dynamic style with a heterogeneous message receive and an indirect dynamic type-discrimination for message processing.
Figure~\ref{l:static_style} shows the static style with a homogeneous message receive and a direct static type-discrimination for message processing.
The static-typing style is safer because of the static check and faster because there is no dynamic type-discrimination.
The dynamic-typing style is more flexible because multiple kinds of messages can be handled in a behaviour condensing the processing code.

\begin{figure}
\centering

\begin{lrbox}{\myboxA}
\begin{cfa}[morekeywords=case]
allocation receive( message & msg ) {
        case( @msg_type1@, msg ) {      // discriminate type
                ... msg_d-> ...;        // msg_type1 msg_d
        } else case( @msg_type2@, msg ) {
                ... msg_d-> ...;        // msg_type2 msg_d
        ...
}
\end{cfa}
\end{lrbox}

\begin{lrbox}{\myboxB}
\begin{cfa}
allocation receive( @msg_type1@ & msg ) {
        ... msg ...;
}
allocation receive( @msg_type2@ & msg ) {
        ... msg ...;
}
...
\end{cfa}
\end{lrbox}
\subfloat[dynamic typing]{\label{l:dynamic_style}\usebox\myboxA}
\hspace*{10pt}
\vrule
\hspace*{10pt}
\subfloat[static typing]{\label{l:static_style}\usebox\myboxB}
\caption{Behaviour Styles}
\label{f:BehaviourStyles}
\end{figure}

\begin{figure}
\centering

\begin{cfa}
// actor
struct my_actor {
        @inline actor;@                                                 $\C[3.25in]{// Plan-9 C inheritance}$
};
// messages
struct str_msg {
        char str[12];
        @inline message;@                                               $\C{// Plan-9 C inheritance}$
};
void ?{}( str_msg & this, char * str ) { strcpy( this.str, str ); }  $\C{// constructor}$
struct int_msg {
        int i;
        @inline message;@                                               $\C{// Plan-9 C inheritance}$
};
// behaviours
allocation receive( my_actor &, @str_msg & msg@ ) with(msg) {
        sout | "string message \"" | str | "\"";
        return Nodelete;                                                $\C{// actor not finished}$
}
allocation receive( my_actor &, @int_msg & msg@ ) with(msg) {
        sout | "integer message" | i;
        return Nodelete;                                                $\C{// actor not finished}$
}
int main() {
        str_msg str_msg{ "Hello World" };               $\C{// constructor call}$
        int_msg int_msg{ 42 };                                  $\C{// constructor call}$
        start_actor_system();                                   $\C{// sets up executor}$
        my_actor actor;                                                 $\C{// default constructor call}$
        @actor | str_msg | int_msg;@                    $\C{// cascade sends}$
        @actor | int_msg;@                                              $\C{// send}$
        @actor | finished_msg;@                                 $\C{// send => terminate actor (deallocation deferred)}$
        stop_actor_system();                                    $\C{// waits until actors finish}\CRT$
} // deallocate int_msg, str_msg, actor
\end{cfa}
\caption{\CFA Actor Syntax}
\label{f:CFAActor}
\end{figure}

Figure~\ref{f:CFAActor} shows a complete \CFA actor example starting with the actor type @my_actor@ created by defining a @struct@ that inherits from the base @actor@ @struct@ via the @inline@ keyword.
This inheritance style is the Plan-9 C-style inheritance discussed in Section~\ref{s:Inheritance}.
Similarly, the message types @str_msg@ and @int_msg@ are created by defining a @struct@ that inherits from the base @message@ @struct@ via the @inline@ keyword.
Only @str_msg@ needs a constructor to copy the C string;
@int_msg@ is initialized using its \CFA auto-generated constructors.
There are two matching @receive@ (behaviour) routines that process the corresponding typed messages.
Both @receive@ routines use a @with@ clause so message fields are not qualified and return @Nodelete@ indicating the actor is not finished.
Also, all messages are marked with @Nodelete@ as their default allocation state.
The program main begins by creating two messages on the stack.
Then the executor system is started by calling @start_actor_system@.
Now an actor is created on the stack and four messages are sent it using operator @?|?@.
The last message is the builtin @finish_msg@, which returns @Finished@ to an executor thread, causing it to removes the actor from the actor system \see{Section~\ref{s:ActorBehaviours}}.
The call to @stop_actor_system@ blocks the program main until all actors are finished and removed from the actor system.
The program main ends by deleting the actor and two messages from the stack.
The output for the program is:
\begin{cfa}
string message "Hello World"
integer message 42
integer message 42
\end{cfa}

\subsection{Actor Behaviours}\label{s:ActorBehaviours}
In general, a behaviour for some derived actor and derived message type is defined with following signature:
\begin{cfa}
allocation receive( my_actor & receiver, my_msg & msg )
\end{cfa}
where @my_actor@ and @my_msg@ inherit from types @actor@ and @message@, respectively.
The return value of @receive@ must be a value from enumerated type, @allocation@:
\begin{cfa}
enum allocation { Nodelete, Delete, Destroy, Finished };
\end{cfa}
The values represent a set of actions that dictate what the executor does with an actor or message after a given behaviour returns.
For actors, the @receive@ routine returns the @allocation@ status to the executor, which takes the appropriate action.
For messages, either the default allocation, @Nodelete@, or any changed value in the message is examined by the executor, which takes the appropriate action.
Message state is updated via a call to:
\begin{cfa}
void set_allocation( message & this, allocation state )
\end{cfa}

In detail, the actions taken by an executor for each of the @allocation@ values are:

\noindent@Nodelete@
tells the executor that no action is to be taken with regard to an actor or message.
This status is used when an actor continues receiving messages or a message may be reused.

\noindent@Delete@
tells the executor to call the object's destructor and deallocate (delete) the object.
This status is used with dynamically allocated actors and messages when they are not reused.

\noindent@Destroy@
tells the executor to call the object's destructor, but not deallocate the object.
This status is used with dynamically allocated actors and messages whose storage is reused.

\noindent@Finished@
tells the executor to mark the respective actor as finished executing, but not call the object's destructor nor deallocate the object.
This status is used when actors or messages are global or stack allocated, or a programmer wants to manage deallocation themselves.

For the actor system to terminate, all actors must have returned a status other than @Nodelete@.
After an actor is terminated, it is erroneous to send messages to it.
Similarly,  after a message is terminated, it cannot be sent to an actor.
Note, it is safe to construct an actor or message with a status other than @Nodelete@, since the executor only examines the allocation action after a behaviour returns.

\subsection{Actor Envelopes}\label{s:envelope}
As stated, each message, regardless of where it is allocated, can be sent to an arbitrary number of actors, and hence, appear on an arbitrary number of message queues.
Because a C program manages message lifetime, messages cannot be copied for each send, otherwise who manages the copies.
Therefore, it up to the actor program to manage message life-time across receives.
However, for a message to appear on multiple message queues, it needs an arbitrary number of link fields.
Hence, there is the concept of an envelop, which is dynamically allocated on each send, that wraps a message with any extra implementation fields needed to persist between send and receive.
Managing the envelop is straightforward because it is created at the send and deleted after the receive, \ie there is 1:1 relationship for an envelop and a many to one relationship for a message.

% In actor systems, messages are sent and received by actors.
% When a actor receives a message it executes its behaviour that is associated with that message type.
% However the unit of work that stores the message, the receiving actor's address, and other pertinent information needs to persist between send and the receive.
% Furthermore the unit of work needs to be able to be stored in some fashion, usually in a queue, until it is executed by an actor.
% All these requirements are fulfilled by a construct called an envelope.
% The envelope wraps up the unit of work and also stores any information needed by data structures such as link fields.

% One may ask, "Could the link fields and other information be stored in the message?".
% This is a good question to ask since messages also need to have a lifetime that persists beyond the work it delivers.
% However, if one were to use messages as envelopes then a message would not be able to be sent to multiple actors at a time.
% Therefore this approach would just push the allocation into another location, and require the user to dynamically allocate a message for every send, or require careful ordering to allow for message reuse.

\subsection{Actor System}\label{s:ActorSystem}
The calls to @start_actor_system@, and @stop_actor_system@ mark the start and end of a \CFA actor system.
The call to @start_actor_system@ sets up an executor and executor threads for the actor system.
It is possible to have multiple start/stop scenarios in a program.

@start_actor_system@ has three overloaded signatures that vary the executor's configuration:

\noindent@void start_actor_system()@
configures the executor to implicitly use all preallocated kernel-threads (processors), \ie the processors created by the program main prior to starting the actor system.
When the number of processors is greater than 1, each executor's message queue is sharded by a factor of 16 to reduce contention, \ie for 4 executor threads (processors), there is a total of 4 $\times$ 16 message queues evenly distributed across the executor threads.

\noindent@void start_actor_system( size_t num_thds )@
configures the number of executor threads to @num_thds@, with the same message queue sharding.

\noindent@void start_actor_system( executor & this )@
allows the programmer to explicitly create and configure an executor for use by the actor system.
Executor configuration options include are discussed in Section~\ref{s:executor}.

\noindent
All actors must be created \emph{after} calling @start_actor_system@ so the executor can keep track of the number of actors that have entered the system but not yet terminated.

% All message sends are done using the vertical-bar operator, @?|?@, similar to the syntax of \CC's stream output.
% \begin{cfa}
% allocation ?|?( my_actor & receiver, my_msg & msg )
% \end{cfa}
% Notice this signature is the same as the @receive@ routine, which is no coincidence.
% The \CFA compiler generates a @?|?@ routine definition and forward declaration for each @receive@ routine that has the appropriate signature.
% The generated routine packages the message and actor in an \hyperref[s:envelope]{envelope} and adds it to the executor's queues via an executor routine.
% As part of packaging the envelope, the @?|?@ routine sets a routine pointer in the envelope to point to the appropriate receive routine for given actor and message types.

\subsection{Actor Send}\label{s:ActorSend}
All message sends are done using the vertical-bar (bit-or) operator, @?|?@, similar to the syntax of the \CFA stream I/O.
As stated, \CFA does not have named inheritance with RTTI.
\CFA does have a preliminary form of virtual routines, but it is not mature enough for use in this work.
Therefore, there is no mechanism to write a generic @?|?@ routine taking a base actor and message type, and then dynamically selecting the @receive@ routine from the actor argument.
(For messages, the Plan-9 inheritance is sufficient because only the inherited fields are needed during the message send.)
Hence, programmers must write a matching @?|?@ routine for each @receive@ routine, which is awkward and generates a maintenance problem.
Therefore, I chose a template-like approach, where the compiler generates a matching @?|?@ routine for each @receive@ routine it finds with the correct actor/message type-signature.
(When the \CFA virtual routines mature, it should be possible to seamlessly transition to it from the template approach.)

% Funneling all message sends through a single @allocation ?|?(actor &, message &)@ routine is not feasible since the type of the actor and message would be erased, making it impossible to acquire a pointer to the correct @receive@.
% As such a @?|?@ routine per @receive@ provides type information needed to write the correct "address" on the envelope.

Figure~\ref{f:send_gen} shows the generated send routine for the @int_msg@ receive in Figure~\ref{f:CFAActor}.
Operator @?|?@ has the same parameter signature as the corresponding @receive@ routine and returns an @actor@ so the operator can be cascaded.
The routine sets @rec_fn@ to the matching @receive@ routine using the left-hand type to perform the selection.
Then the routine packages the base and derived actor and message and actor, along with the receive routine into an \hyperref[s:envelope]{envelope}.
Finally, the envelop is added to the executor queue designated by the actor using the executor routine @send@.

\begin{figure}
\begin{cfa}
$\LstCommentStyle{// from Figure~\ref{f:CFAActor}}$
struct my_actor { inline actor; };                                              $\C[3.75in]{// actor}$
struct int_msg { inline message; int i; };                              $\C{// message}$
allocation receive( @my_actor &, int_msg & msg@ ) {...} $\C{// receiver}$

// compiler generated send operator
typedef allocation (*receive_t)( actor &, message & );
actor & ?|?( @my_actor & receiver, int_msg & msg@ ) {
        allocation (*rec_fn)( my_actor &, int_msg & ) = @receive@; // deduce receive routine
        request req{ &receiver, (actor *)&receiver, &msg, (message *)&msg, (receive_t)rec_fn };
        send( receiver, req );                                                          $\C{// queue message for execution}\CRT$
        return receiver;
}
\end{cfa}
\caption{Generated Send Operator}
\label{f:send_gen}
\end{figure}

Figure~\ref{f:ConvenienceMessages} shows three builtin convenience messages and receive routines used to terminate actors, depending on how an actor is allocated: @Delete@, @Destroy@ or @Finished@.
For example, in Figure~\ref{f:CFAActor}, the builtin @finished_msg@ message and receive are used to terminate the actor because the actor is allocated on the stack, so no deallocation actions are performed by the executor.

\begin{figure}
\begin{cfa}
message __base_msg_finished $@$= { .allocation_ : Finished }; // no auto-gen constructors
struct __delete_msg_t { inline message; } delete_msg = __base_msg_finished;
struct __destroy_msg_t { inline message; } destroy_msg = __base_msg_finished;
struct __finished_msg_t { inline message; } finished_msg = __base_msg_finished;

allocation receive( actor & this, __delete_msg_t & msg ) { return Delete; }
allocation receive( actor & this, __destroy_msg_t & msg ) { return Destroy; }
allocation receive( actor & this, __finished_msg_t & msg ) { return Finished; }
\end{cfa}
\caption{Builtin Convenience Messages}
\label{f:ConvenienceMessages}
\end{figure}

\section{\CFA Executor}\label{s:executor}
This section describes the basic architecture of the \CFA executor.
An executor of an actor system is the scheduler that organizes where actor behaviours are run and how messages are sent and delivered.
In \CFA, the executor is message-centric \see{Figure~\ref{f:inverted_actor}}, but extended by over sharding of a message queue \see{left side of Figure~\ref{f:gulp}}, \ie there are $M$ message queues where $M$ is greater than the number of executor threads $N$ (usually a multiple of $N$).
This approach reduces contention by spreading message delivery among the $M$ queues rather than $N$, while still maintaining actor FIFO message-delivery semantics.
The only extra overhead is each executor cycling (usually round-robin) through its $M$/$N$ queues.
The goal is to achieve better performance and scalability for certain kinds of actor applications by reducing executor locking.
Note, lock-free queues do not help because busy waiting on any atomic instruction is the source of the slowdown whether it is a lock or lock-free.

\begin{figure}
\begin{center}
\input{diagrams/gulp.tikz}
\end{center}
\caption{Queue Gulping Mechanism}
\label{f:gulp}
\end{figure}

Each executor thread iterates over its own message queues until it finds one with messages.
At this point, the executor thread atomically \gls{gulp}s the queue, meaning it moves the contents of message queue to a local queue of the executor thread using a single atomic instruction.
An example of the queue gulping operation is shown in the right side of Figure \ref{f:gulp}, where a executor threads gulps queue 0 and begins to process it locally.
This step allows an executor thread to process the local queue without any atomics until the next gulp.
Other executor threads can continue adding to the ends of executor thread's message queues.
In detail, an executor thread performs a test-and-gulp, non-atomically checking if a queue is non-empty, before attempting to gulp it.
If an executor misses an non-empty queue due to a race, it eventually finds the queue after cycling through its message queues.
This approach minimizes costly lock acquisitions.

Processing a local queue involves: removing a unit of work from the queue, dereferencing the actor pointed-to by the work-unit, running the actor's behaviour on the work-unit message, examining the returned allocation status from the @receive@ routine for the actor and internal status in the delivered message, and taking the appropriate actions.
Since all messages to a given actor are in the same queue, this guarantees atomicity across behaviours of that actor since it can only execute on one thread at a time.
As each actor is created or terminated by an executor thread, it increments/decrements a global counter.
When an executor decrements the counter to zero, it sets a global boolean variable that is checked by each executor thread when it has no work.
Once a executor threads sees the flag is set it stops running.
After all executors stop, the actor system shutdown is complete.

\subsection{Copy Queue}\label{s:copyQueue}
Unfortunately, the frequent allocation of envelopes for each send results in heavy contention on the memory allocator.
This contention is reduced using a novel data structure, called a \Newterm{copy queue}.
The copy queue is a thin layer over a dynamically sized array that is designed with the envelope use case in mind.
A copy queue supports the typical queue operations of push/pop but in a different way from a typical array-based queue.

The copy queue is designed to take advantage of the \gls{gulp}ing pattern, giving an amortized runtime cost for each push/pop operation of $O(1)$.
In contrast, a na\"ive array-based queue often has either push or pop cost $O(n)$ and the other cost $O(1)$ since one of the operations requires shifting the elements of the queue.
Since the executor threads gulp a queue to operate on it locally, this creates a usage pattern where all elements are popped from the copy queue without any interleaved pushes.
As such, during pop operations there is no need to shift array elements.
Instead, an index is stored in the copy-queue data-structure that keeps track of which element to pop next allowing pop to be $O(1)$.
Push operations are amortized $O(1)$ since pushes may cause doubling reallocations of the underlying dynamic-sized array (like \CC @vector@).

% C_TODO: maybe make copy_queue diagram

Since the copy queue is an array, envelopes are allocated first on the stack and then copied into the copy queue to persist until they are no longer needed.
For many workload, the copy queues grow in size to facilitate the average number of messages in flight and there is no further dynamic allocations.
One downside of this approach that more storage is allocated than needed, \ie each copy queue is only partially full.
Comparatively, the individual envelope allocations of a list-based queue mean that the actor system always uses the minimum amount of heap space and cleans up eagerly.
Additionally, bursty workloads can cause the copy queues to allocate a large amounts of space to accommodate the peaks of the throughput, even if most of that storage is not needed for the rest of the workload's execution.

To mitigate memory wastage, a reclamation scheme is introduced.
Initially, the memory reclamation na\"ively reclaims one index of the array per \gls{gulp}, if the array size is above a low fixed threshold.
However, this approach has a problem.
The high memory watermark nearly doubled!
The issue is highlighted with an example.
Assume a fixed throughput workload, where a queue never has more than 19 messages at a time.
If the copy queue starts with a size of 10, it ends up doubling at some point to size 20 to accommodate 19 messages.
However, after 2 gulps and subsequent reclamations the array size is 18.
The next time 19 messages are enqueued, the array size is doubled to 36!
To avoid this issue, a second check is added.
% Each copy queue starts tracking the utilization of its array size.
Reclamation only occurs if less than half of the array is utilized.
This check achieves a lower total storage and overall memory utilization compared to the non-reclamation copy queues.
However, the use of copy queues still incurs a higher memory cost than list-based queueing, but the increase in memory usage is reasonable considering the performance gains \see{Section~\ref{s:actor_perf}}.

\section{Work Stealing}\label{s:steal}
Work stealing is a scheduling strategy to provide \Newterm{load balance}.
The goal is to increase resource utilization by having idle threads steal work from working threads.
While there are multiple parts in work-stealing scheduler, the two important components are victim selection and the stealing mechanism.

\subsection{Stealing Mechanism}
In work stealing, the stealing worker is called the \Newterm{thief} and the stolen-from worker is called the \Newterm{victim}.
The stealing mechanism presented here differs from existing work-stealing actor-systems because of the message-centric (inverted) actor-system.
Other actor systems, such as Akka~\cite{Akka} and CAF~\cite{CAF}, have work stealing, but use an actor-centric system where stealing is dequeuing from a non-empty ready-queue to an empty ready-queue.
As an example, in CAF, the sharded actor queue is a set of double-ended queues (dequeues).
When an actor has messages, it is inserted into a worker's dequeue (ready queue).
Workers then consume actors from the dequeue and execute their behaviours.
To steal work, thieves take one or more actors from a victim's dequeue.
By the pigeon hole principle, there are three dequeue operations (push/victim pop/thief pop) that can occur concurrently and only two ends to a dequeue, so work stealing in a dequeue-based system always results in a potential increase in contention on the dequeues.
This contention can slows down the victim's throughput.
Note, which end of the dequeue is used for stealing, consuming, and inserting is not discussed since the largest cost is the mutual exclusion and its duration for safely performing the queue operations.

Work steal now becomes queue stealing, where an entire actor/message queue is stolen, which trivially preserves message ordering in a queue \see{Section~\ref{s:steal}}.

% C_TODO: maybe insert stealing diagram

In \CFA, the actor work-stealing implementation is unique because of the message-centric system.
In this system, it is impractical to steal actors because an actor's messages are distributed in temporal order along the message queue.
To ensure sequential actor execution and FIFO message delivery, actor stealing requires finding and removing all of an actor's messages, and inserting them consecutively in another message queue.
This operation is $O(N)$ with a non-trivial constant.
The only way for work stealing to become practical is to shard the message queue, which also reduces contention, and steal queues to eliminate queue searching.

Given queue stealing, the goal is to have a zero-victim-cost stealing mechanism, which does not mean stealing has no cost.
It means work stealing does not affect the performance of the victim worker.
The implication is that thieves cannot contend with a victim, and that a victim should perform no stealing related work unless it becomes a thief.
In theory, this goal is not achievable, but results show the goal is achieved in practice.

In \CFA's actor system, workers own a set of sharded queues, which they iterate over and gulp.
If a worker has iterated over its message queues twice without finding any work, it tries to steal a queue from another worker.
Stealing a queue is done wait-free with a few atomic instructions that can only create contention with other stealing workers, not the victim.
To steal a queue, a worker does the following:
\begin{enumerate}[topsep=5pt,itemsep=3pt,parsep=0pt]
\item
The thief chooses a victim, which is trivial because all workers are stored in a shared array.

\item
The thief starts at a random index in the array of the victim's queues and searches for a candidate queue.
A candidate queue is any non-empty queue not being processed by the victim and not being stolen by another thief.
These rules are not strictly enforced.
A candidate is identified non-atomically, and as such, queues that do not satisfy these rules may be stolen.
However, steals not meeting the rules do not affect correctness and do not constitute failed steals as the queue is always swapped.

\item
Once a candidate queue is chosen, the thief attempts a wait-free swap of a victim's queue to a random empty thief queue.
If the swap successes, the steal is completed.
If the swap fails, the victim may have been gulping that message queue or another thief must have attempted to steal the victim's queue.
In either case, that message queue is highly likely to be empty.

\item
Once a thief fails or succeeds in stealing a queue, it iterates over its messages queues again because new messages may have arrived during stealing.
Stealing is only repeated after two consecutive iterations over its owned queues without finding work.
\end{enumerate}

The key to the stealing mechanism is that the queues can still be operated on while they are being swapped.
This functionality eliminates any contention among thieves and victims.

The first key to this is that actors and workers maintain two distinct arrays of references to queues.
Actors will always receive messages via the same queues.
Workers, on the other hand will swap the pointers to queues in their shared array and operate on queues in the range of that array that they own.
Swapping queues is a matter of atomically swapping two pointers in the worker array.
As such pushes to the queues can happen concurrently during the swap since pushes happen via the actor queue references.

Gulping can also occur during queue swapping, but the implementation requires more nuance than the pushes.
When a worker is not stealing it iterates across its own range of queues and gulps them one by one.
When a worker operates on a queue it first copies the current pointer from the worker array of references to a local variable.
It then uses that local variable for all queue operations until it moves to the next index of its range of the queue array.
This ensures that any swaps do not interrupt gulping operations, however this introduces a correctness issue.
If any behaviours from a queue are run by two workers at a time it violates both mutual exclusion and the actor ordering guarantees.
As such this must be avoided.
To avoid this each queue has a @being_processed@ flag that is atomically set to @true@ when a queue is gulped.
The flag indicates that a queue is being processed locally and is set back to @false@ once the local processing is finished.
If a worker attempts to gulp a queue and finds that the @being_processed@ flag is @true@, it does not gulp the queue and moves on to the next queue in its range.
This is a source of contention between victims and thieves since a thief may steal a queue and set @being_processed@ to @true@ between a victim saving a pointer to a queue and gulping it.
However, the window for this race is very small, making this contention rare.
This is why the claim is made that this mechanism is zero-victim-cost in practice but not in theory.
By collecting statistics on failed gulps due to the @being_processed@ flag, it is found that this contention occurs ~0.05\% of the time when a gulp occurs.
Hence, the claim is made that this stealing mechanism has zero-victim-cost in practice.


\subsection{Queue Swap Correctness}
Given the wait-free swap used is novel, it is important to show that it is correct.
Firstly, it is clear to show that the swap is wait-free since all workers will fail or succeed in swapping the queues in a finite number of steps since there are no locks or looping.
There is no retry mechanism in the case of a failed swap, since a failed swap either means the work was already stolen, or that work was stolen from the thief.
In both cases it is apropos for a thief to given up on stealing.
\CFA-style pseudocode for the queue swap is presented below.
The swap uses compare-and-swap (@CAS@) which is just pseudocode for C's @__atomic_compare_exchange_n@.
A pseudocode implementation of @CAS@ is also shown below.
The correctness of the wait-free swap will now be discussed in detail.
To first verify sequential correctness, consider the equivalent sequential swap below:

\begin{cfa}
void swap( uint victim_idx, uint my_idx ) {
        // Step 0:
        work_queue * my_queue = request_queues[my_idx];
        work_queue * vic_queue = request_queues[victim_idx];
        // Step 2:
        request_queues[my_idx] = 0p;
        // Step 3:
        request_queues[victim_idx] = my_queue;
        // Step 4:
        request_queues[my_idx] = vic_queue;
}
\end{cfa}

Step 1 is missing in the sequential example since in only matter in the concurrent context presented later.
By looking at the sequential swap it is easy to see that it is correct.
Temporary copies of each pointer being swapped are stored, and then the original values of each pointer are set using the copy of the other pointer.

\begin{cfa}
// This routine is atomic
bool CAS( work_queue ** ptr, work_queue ** old, work_queue * new ) {
        if ( *ptr != *old )
                return false;
        *ptr = new;
        return true;
}

bool try_swap_queues( worker & this, uint victim_idx, uint my_idx ) with(this) {
        // Step 0:
        // request_queues is the shared array of all sharded queues
        work_queue * my_queue = request_queues[my_idx];
        work_queue * vic_queue = request_queues[victim_idx];

        // Step 1:
        // If either queue is 0p then they are in the process of being stolen
        // 0p is Cforall's equivalent of C++'s nullptr
        if ( vic_queue == 0p ) return false;

        // Step 2:
        // Try to set thief's queue ptr to be 0p.
        // If this CAS fails someone stole thief's queue so return false
        if ( !CAS( &request_queues[my_idx], &my_queue, 0p ) )
                return false;

        // Step 3:
        // Try to set victim queue ptr to be thief's queue ptr.
        // If it fails someone stole the other queue, so fix up then return false
        if ( !CAS( &request_queues[victim_idx], &vic_queue, my_queue ) ) {
                request_queues[my_idx] = my_queue; // reset queue ptr back to prev val
                return false;
        }

        // Step 4:
        // Successfully swapped.
        // Thief's ptr is 0p so no one will touch it
        // Write back without CAS is safe
        request_queues[my_idx] = vic_queue;
        return true;
}
\end{cfa}\label{c:swap}

Now consider the concurrent implementation of the swap.
\begin{enumerate}[topsep=5pt,itemsep=3pt,parsep=0pt]
\item
Step 0 is the same as the sequential example, and the thief stores local copies of the two pointers to be swapped.
\item
Step 1 verifies that the stored copy of the victim queue pointer, @vic_queue@, is valid.
If @vic_queue@ is equal to @0p@, then the victim queue is part of another swap so the operation fails.
No state has changed at this point so no fixups are needed.
Note, @my_queue@ can never be equal to @0p@ at this point since thieves only set their own queues pointers to @0p@ when stealing.
At no other point will a queue pointer be set to @0p@.
Since each worker owns a disjoint range of the queue array, it is impossible for @my_queue@ to be @0p@.
\item
Step 2 attempts to set the thief's queue pointer to @0p@ via @CAS@.
The @CAS@ will only fail if the thief's queue pointer is no longer equal to @my_queue@, which implies that this thief has become a victim and its queue has been stolen.
At this point the thief-turned-victim will fail and since it has not changed any state it just fails and returns false.
If the @CAS@ succeeds then the thief's queue pointer will now be @0p@.
Nulling the pointer is safe since only thieves look at other worker's queue ranges, and whenever thieves need to dereference a queue pointer they check for @0p@.
\item
Step 3 attempts to set the victim's queue pointer to be @my_queue@ via @CAS@.
If the @CAS@ succeeds then the victim's queue pointer has been set and swap can no longer fail.
If the @CAS@ fails then the thief's queue pointer must be restored to its previous value before returning.
\item
Step 4 sets the thief's queue pointer to be @vic_queue@ completing the swap.
\end{enumerate}

\begin{theorem}
The presented swap is correct and concurrently safe in both the success and failure cases.
\end{theorem}

Correctness of the swap is shown through the existence of an invariant.
The invariant is that when a queue pointer is set to @0p@ by a thief, then the next write to the pointer can only be performed by the same thief.
To show that this invariant holds, it is shown that it is true at each step of the swap.
Step 0 and 1 do not write and as such they cannot invalidate the invariant of any other thieves.
In step 2 a thief attempts to write @0p@ to one of their queue pointers.
This queue pointer cannot be @0p@.
As stated above, @my_queue@ is never equal to @0p@ since thieves will only write @0p@ to queue pointers from their own queue range and all worker's queue ranges are disjoint.
As such step 2 upholds the invariant since in a failure case no write occurs, and in the success case, the value of the queue pointer is guaranteed to not be 0p.
In step 3 the thief attempts to write @my_queue@ to the victim's queue pointer.
If the current value of the victim's queue pointer is @0p@, then the CAS will fail since @vic_queue@ cannot be equal to @0p@ because of the check in step 1.
Therefore in the success case where the @CAS@ succeeds, the value of the victim's queue pointer must not be @0p@.
As such, the write will never overwrite a value of @0p@, hence the invariant is held in the @CAS@ of step 3.
The write back to the thief's queue pointer that happens in the failure case of step three and in step 4 hold the invariant since they are the subsequent write to a @0p@ queue pointer and they are being set by the same thief that set the pointer to @0p@.

Given this informal proof of invariance it can be shown that the successful swap is correct.
Once a thief atomically sets their queue pointer to be @0p@ in step 2, the invariant guarantees that pointer will not change.
As such, in the success case step 3 it is known that the value of the victim's queue pointer that was overwritten must be @vic_queue@ due to the use of @CAS@.
Given that pointers all have unique memory locations, this first write of the successful swap is correct since it can only occur when the pointer has not changed.
By the invariant the write back in the successful case is correct since no other worker can write to the @0p@ pointer.

In the failed case the outcome is correct in steps 1 and 2 since no writes have occurred so the program state is unchanged.
In the failed case of step 3 the program state is safely restored to its state it had prior to the @0p@ write in step 2, thanks to the invariant that makes the write back to the @0p@ pointer safe.

\subsection{Stealing Guarantees}

% C_TODO insert graphs for each proof
Given that the stealing operation can potentially fail, it is important to discuss the guarantees provided by the stealing implementation.
Given a set of $N$ swaps a set of connected directed graphs can be constructed where each vertex is a queue and each edge is a swap directed from a thief queue to a victim queue.
Since each thief can only steal from one victim at a time, each vertex can only have at most one outgoing edge.
A corollary that can be drawn from this, is that there are at most $V$ edges in this constructed set of connected directed graphs, where $V$ is the total number of vertices.

\begin{figure}
\begin{center}
\input{diagrams/M_to_one_swap.tikz}
\end{center}
\caption{Graph of $M$ thieves swapping with one victim.}
\label{f:M_one_swap}
\end{figure}

\begin{theorem}
Given $M$ thieves queues all attempting to swap with one victim queue, and no other swaps occurring that involve these queues, at least one swap is guaranteed to succeed.
\end{theorem}\label{t:one_vic}
A graph of the $M$ thieves swapping with one victim discussed in this theorem is presented in Figure~\ref{f:M_one_swap}.
\\
First it is important to state that a thief will not attempt to steal from themselves.
As such, the victim here is not also a thief.
Stepping through the code in \ref{c:swap}, for all thieves steps 0-1 succeed since the victim is not stealing and will have no queue pointers set to be @0p@.
Similarly for all thieves step 2 will succeed since no one is stealing from any of the thieves.
In step 3 the first thief to @CAS@ will win the race and successfully swap the queue pointer.
Since it is the first one to @CAS@ and @CAS@ is atomic, there is no way for the @CAS@ to fail since no other thief could have written to the victim's queue pointer and the victim did not write to the pointer since they aren't stealing.
Hence at least one swap is guaranteed to succeed in this case.

\begin{figure}
\begin{center}
\input{diagrams/chain_swap.tikz}
\end{center}
\caption{Graph of a chain of swaps.}
\label{f:chain_swap}
\end{figure}

\begin{theorem}
Given $M$ > 1, ordered queues pointers all attempting to swap with the queue in front of them in the ordering, except the first queue, and no other swaps occurring that involve these queues, at least one swap is guaranteed to succeed.
\end{theorem}\label{t:vic_chain}
A graph of the chain of swaps discussed in this theorem is presented in Figure~\ref{f:chain_swap}.
\\
This is a proof by contradiction.
Assume no swaps occur.
Then all thieves must have failed at step 1, step 2 or step 3.
For a given thief $b$ to fail at step 1, thief $b + 1$ must have succeeded at step 2 before $b$ executes step 0.
Hence, not all thieves can fail at step 1.
Furthermore if a thief $b$ fails at step 1 it logically splits the chain into two subchains $0 <- b$ and $b + 1 <- M - 1$, where $b$ has become solely a victim since its swap has failed and it did not modify any state.
There must exist at least one chain containing two or more queues after since it is impossible for a split to occur both before and after a thief, since that requires failing at step 1 and succeeding at step 2.
Hence, without loss of generality, whether thieves succeed or fail at step 1, this proof can proceed inductively.

For a given thief $i$ to fail at step 2, it means that another thief $j$ had to have written to $i$'s queue pointer between $i$'s step 0 and step 2.
The only way for $j$ to write to $i$'s queue pointer would be if $j$ was stealing from $i$ and had successfully finished step 3.
If $j$ finished step 3 then the at least one swap was successful.
Therefore all thieves did not fail at step 2.
Hence all thieves must successfully complete step 2 and fail at step 3.
However, since the first worker, thief $0$, is solely a victim and not a thief, it does not change the state of any of its queue pointers.
Hence, in this case thief $1$ will always succeed in step 3 if all thieves succeed in step 2.
Thus, by contradiction with the earlier assumption that no swaps occur, at least one swap must succeed.

% \raisebox{.1\height}{}
\begin{figure}
\centering
\begin{tabular}{l|l}
\subfloat[Cyclic Swap Graph]{\label{f:cyclic_swap}\input{diagrams/cyclic_swap.tikz}} &
\subfloat[Acyclic Swap Graph]{\label{f:acyclic_swap}\input{diagrams/acyclic_swap.tikz}}
\end{tabular}
\caption{Illustrations of cyclic and acyclic swap graphs.}
\end{figure}

\begin{theorem}
Given a set of $M > 1$ swaps occurring that form a single directed connected graph.
At least one swap is guaranteed to succeed if and only if the graph does not contain a cycle.
\end{theorem}\label{t:vic_cycle}
Representations of cyclic and acyclic swap graphs discussed in this theorem are presented in Figures~\ref{f:cyclic_swap} and \ref{f:acyclic_swap}.
\\
First the reverse direction is proven.
If the graph does not contain a cycle, then there must be at least one successful swap.
Since the graph contains no cycles and is finite in size, then there must be a vertex $A$ with no outgoing edges.
The graph can then be formulated as a tree with $A$ at the top since each node only has at most one outgoing edge and there are no cycles.
The forward direction is proven by contradiction in a similar fashion to \ref{t:vic_chain}.
Assume no swaps occur.
Similar to \ref{t:vic_chain}, this graph can be inductively split into subgraphs of the same type by failure at step 1, so the proof proceeds without loss of generality.
Similar to \ref{t:vic_chain} the conclusion is drawn that all thieves must successfully complete step 2 for no swaps to occur, since for step 2 to fail, a different thief has to successfully complete step 3, which would imply a successful swap.
Hence, the only way forward is to assume all thieves successfully complete step 2.
Hence for there to be no swaps all thieves must fail step 3.
However, since $A$ has no outgoing edges, since the graph is connected there must be some $K$ such that $K < M - 1$ thieves are attempting to swap with $A$.
Since all $K$ thieves have passed step 2, similar to \ref{t:one_vic} the first one of the $K$ thieves to attempt step 3 is guaranteed to succeed.
Thus, by contradiction with the earlier assumption that no swaps occur, if the graph does not contain a cycle, at least one swap must succeed.

The forward direction is proven by contrapositive.
If the graph contains a cycle then there exists a situation where no swaps occur.
This situation is constructed.
Since all vertices have at most one outgoing edge the cycle must be directed.
Furthermore, since the graph contains a cycle all vertices in the graph must have exactly one outgoing edge.
This is shown through construction of an arbitrary cyclic graph.
The graph contains a directed cycle by definition, so the construction starts with $T$ vertices in a directed cycle.
Since the graph is connected, and each vertex has at most one outgoing edge, none of the vertices in the cycle have available outgoing edges to accommodate new vertices with no outgoing edges.
Any vertices added to the graph must have an outgoing edge to connect, leaving the resulting graph with no available outgoing edges.
Thus, by induction all vertices in the graph must have exactly one outgoing edge.
Hence all vertices are thief queues.
Now consider the case where all thieves successfully complete step 0-1, and then they all complete step 2.
At this point all thieves are attempting to swap with a queue pointer whose value has changed to @0p@.
If all thieves attempt the @CAS@ before any write backs, then they will all fail.
Thus, by contrapositive, if the graph contains a cycle then there exists a situation where no swaps occur.
Hence, at least one swap is guaranteed to succeed if and only if the graph does not contain a cycle.

% C_TODO: go through and use \paragraph to format to make it look nicer
\subsection{Victim Selection}\label{s:victimSelect}
In any work stealing algorithm thieves have some heuristic to determine which victim to choose from.
Choosing this algorithm is difficult and can have implications on performance.
There is no one selection heuristic that is known to be the best on all workloads.
Recent work focuses on locality aware scheduling in actor systems\cite{barghi18}\cite{wolke17}.
However, while locality aware scheduling provides good performance on some workloads, something as simple as randomized selection performs better on other workloads\cite{barghi18}.
Since locality aware scheduling has been explored recently, this work introduces a heuristic called \textbf{longest victim} and compares it to randomized work stealing.
The longest victim heuristic maintains a timestamp per executor threads that is updated every time a worker attempts to steal work.
Thieves then attempt to steal from the thread with the oldest timestamp.
This means that if two thieves look to steal at the same time, they likely will attempt to steal from the same victim.
This does increase the chance at contention between thieves, however given that workers have multiple queues under them, often in the tens or hundreds of queues per worker it is rare for two queues to attempt so steal the same queue.
Furthermore in the case they attempt to steal the same queue at least one of them is guaranteed to successfully steal the queue as shown in Theorem \ref{t:one_vic}.
Additionally, the longest victim heuristic makes it very improbable that the no swap scenario presented in Theorem \ref{t:vic_cycle} manifests.
Given the longest victim heuristic, for a cycle to manifest it would require all workers to attempt to steal in a short timeframe.
This is the only way that more than one thief could choose another thief as a victim, since timestamps are only updated upon attempts to steal.
In this case, the probability of lack of any successful swaps is a non issue, since it is likely that these steals were not important if all workers are trying to steal.

\section{Safety and Productivity}\label{s:SafetyProductivity}
\CFA's actor system comes with a suite of safety and productivity features.
Most of these features are present in \CFA's debug mode, but are removed when code is compiled in nodebug mode.
The suit of features include the following.

\begin{itemize}
\item Static-typed message sends.
If an actor does not support receiving a given message type, the actor program is rejected at compile time, allowing unsupported messages to never be sent to actors.
\item Detection of message sends to Finished/Destroyed/Deleted actors.
All actors have a ticket that assigns them to a respective queue.
The maximum integer value of the ticket is reserved to indicate that an actor is dead, and subsequent message sends result in an error.
\item Actors made before the executor can result in undefined behaviour since an executor needs to be created beforehand so it can give out the tickets to actors.
As such, this is detected and an error is printed.
\item When an executor is created, the queues are handed out to executor threads in round robin order.
If there are fewer queues than executor threads, then some workers will spin and never do any work.
There is no reasonable use case for this behaviour so an error is printed if the number of queues is fewer than the number of executor threads.
\item A warning is printed when messages are deallocated without being sent.
Since the @Finished@ allocation status is unused for messages, it is used internally to detect if a message has been sent.
Deallocating a message without sending it could indicate to a user that they are touching freed memory later, or it could point out extra allocations that could be removed.
\end{itemize}

In addition to these features, \CFA's actor system comes with a suite of statistics that can be toggled on and off.
These statistics have minimal impact on the actor system's performance since they are counted on a per executor threads basis.
During shutdown of the actor system they are aggregated, ensuring that the only atomic instructions used by the statistics counting happen at shutdown.
The statistics measured are as follows.

\begin{description}
\item[\LstBasicStyle{\textbf{Actors Created}}]
Actors created.
Includes both actors made by the main and ones made by other actors.
\item[\LstBasicStyle{\textbf{Messages Sent}}]
Messages sent and received.
Includes termination messages send to the executor threads.
\item[\LstBasicStyle{\textbf{Gulps}}]
Gulps that occurred across the executor threads.
\item[\LstBasicStyle{\textbf{Average Gulp Size}}]
Average number of messages in a gulped queue.
\item[\LstBasicStyle{\textbf{Missed gulps}}]
Occurrences where a worker missed a gulp due to the concurrent queue processing by another worker.
\item[\LstBasicStyle{\textbf{Steal attempts}}]
Worker threads attempts to steal work.

\item[\LstBasicStyle{\textbf{Steal failures (no candidates)}}]
Work stealing failures due to selected victim not having any non empty or non-being-processed queues.
\item[\LstBasicStyle{\textbf{Steal failures (failed swaps)}}]
Work stealing failures due to the two stage atomic swap failing.
\item[\LstBasicStyle{\textbf{Messages stolen}}]
Aggregate of the number of messages in queues as they were stolen.
\item[\LstBasicStyle{\textbf{Average steal size}}]
Average number of messages in a stolen queue.
\end{description}

These statistics enable a user of \CFA's actor system to make informed choices about how to configure their executor, or how to structure their actor program.
For example, if there is a lot of messages being stolen relative to the number of messages sent, it could indicate to a user that their workload is heavily imbalanced across executor threads.
In another example, if the average gulp size is very high, it could indicate that the executor could use more queue sharding.

% C_TODO cite poison pill messages and add languages
Another productivity feature that is included is a group of poison-pill messages.
Poison-pill messages are common across actor systems, including Akka and ProtoActor \cite{}.
Poison-pill messages inform an actor to terminate.
In \CFA, due to the allocation of actors and lack of garbage collection, there needs to be a suite of poison-pills.
The messages that \CFA provides are @DeleteMsg@, @DestroyMsg@, and @FinishedMsg@.
These messages are supported on all actor types via inheritance and when sent to an actor, the actor takes the corresponding allocation action after receiving the message.
Note that any pending messages to the actor will still be sent.
It is still the user's responsibility to ensure that an actor does not receive any messages after termination.

\section{Performance}\label{s:actor_perf}
\CAP{I will update the figures to have the larger font size and different line markers once we start editing this chapter.}
The performance of \CFA's actor system is tested using a suite of microbenchmarks, and compared with other actor systems.
Most of the benchmarks are the same as those presented in \ref{}, with a few additions.
% C_TODO cite actor paper
At the time of this work the versions of the actor systems are as follows.
\CFA 1.0, \uC 7.0.0, Akka Typed 2.7.0, CAF 0.18.6, and ProtoActor-Go v0.0.0-20220528090104-f567b547ea07.
Akka Classic is omitted as Akka Typed is their newest version and seems to be the direction they are headed in.
The experiments are run on
\begin{list}{\arabic{enumi}.}{\usecounter{enumi}\topsep=5pt\parsep=5pt\itemsep=0pt}
\item
Supermicro SYS--6029U--TR4 Intel Xeon Gold 5220R 24--core socket, hyper-threading $\times$ 2 sockets (48 process\-ing units) 2.2GHz, running Linux v5.8.0--59--generic
\item
Supermicro AS--1123US--TR4 AMD EPYC 7662 64--core socket, hyper-threading $\times$ 2 sockets (256 processing units) 2.0 GHz, running Linux v5.8.0--55--generic
\end{list}

The benchmarks are run on up to 48 cores.
On the Intel, when going beyond 24 cores there is the choice to either hop sockets or to use hyperthreads.
Either choice will cause a blip in performance trends, which can be seen in the following performance figures.
On the Intel the choice was made to hyperthread instead of hopping sockets for experiments with more than 24 cores.

All benchmarks presented are run 5 times and the median is taken.
Error bars showing the 95\% confidence intervals are drawn on each point on the graphs.
If the confidence bars are small enough, they may be obscured by the point.
In this section \uC will be compared to \CFA frequently, as the actor system in \CFA was heavily based off \uC's actor system.
As such the performance differences that arise are largely due to the contributions of this work.

\begin{table}[t]
\centering
\setlength{\extrarowheight}{2pt}
\setlength{\tabcolsep}{5pt}

\caption{Static Actor/Message Performance: message send, program memory}
\label{t:StaticActorMessagePerformance}
\begin{tabular}{*{5}{r|}r}
        & \multicolumn{1}{c|}{\CFA (100M)} & \multicolumn{1}{c|}{CAF (10M)} & \multicolumn{1}{c|}{Akka (100M)} & \multicolumn{1}{c|}{\uC (100M)} & \multicolumn{1}{c@{}}{ProtoActor (100M)} \\
        \hline
        AMD             & \input{data/pykeSendStatic} \\
        \hline
        Intel   & \input{data/nasusSendStatic}
\end{tabular}

\bigskip

\caption{Dynamic Actor/Message Performance: message send, program memory}
\label{t:DynamicActorMessagePerformance}

\begin{tabular}{*{5}{r|}r}
        & \multicolumn{1}{c|}{\CFA (20M)} & \multicolumn{1}{c|}{CAF (2M)} & \multicolumn{1}{c|}{Akka (2M)} & \multicolumn{1}{c|}{\uC (20M)} & \multicolumn{1}{c@{}}{ProtoActor (2M)} \\
        \hline
        AMD             & \input{data/pykeSendDynamic} \\
        \hline
        Intel   & \input{data/nasusSendDynamic}
\end{tabular}
\end{table}

\subsection{Message Sends}
Message sending is the key component of actor communication.
As such latency of a single message send is the fundamental unit of fast-path performance for an actor system.
The following two microbenchmarks evaluate the average latency for a static actor/message send and a dynamic actor/message send.
Static and dynamic refer to the allocation of the message and actor.
In the static send benchmark a message and actor are allocated once and then the message is sent to the same actor repeatedly until it has been sent 100 million (100M) times.
The average latency per message send is then calculated by dividing the duration by the number of sends.
This benchmark evaluates the cost of message sends in the actor use case where all actors and messages are allocated ahead of time and do not need to be created dynamically during execution.
The CAF static send benchmark only sends a message 10M times to avoid extensively long run times.

In the dynamic send benchmark the same experiment is performed, with the change that with each send a new actor and message is allocated.
This evaluates the cost of message sends in the other common actor pattern where actors and message are created on the fly as the actor program tackles a workload of variable or unknown size.
Since dynamic sends are more expensive, this benchmark repeats the actor/message creation and send 20M times (\uC, \CFA), or 2M times (Akka, CAF, ProtoActor), to give an appropriate benchmark duration.

The results from the static/dynamic send benchmarks are shown in Figures~\ref{t:StaticActorMessagePerformance} and \ref{t:DynamicActorMessagePerformance} respectively.
\CFA leads the charts in both benchmarks, largely due to the copy queue removing the majority of the envelope allocations.
In the static send benchmark all systems except CAF have static send costs that are in the same ballpark, only varying by ~70ns.
In the dynamic send benchmark all systems experience slower message sends, as expected due to the extra allocations.
However, Akka and ProtoActor, slow down by a more significant margin than the \uC and \CFA.
This is likely a result of Akka and ProtoActor's garbage collection, which can suffer from hits in performance for allocation heavy workloads, whereas \uC and \CFA have explicit allocation/deallocation.

\subsection{Work Stealing}
\CFA's actor system has a work stealing mechanism which uses the longest victim heuristic, introduced in Section~ref{s:victimSelect}.
In this performance section, \CFA with the longest victim heuristic is compared with other actor systems on the benchmark suite, and is separately compared with vanilla non-stealing \CFA and \CFA with randomized work stealing.

\begin{figure}
        \centering
        \subfloat[AMD \CFA Balance-One Benchmark]{
                \resizebox{0.5\textwidth}{!}{\input{figures/nasusCFABalance-One.pgf}}
                \label{f:BalanceOneAMD}
        }
        \subfloat[Intel \CFA Balance-One Benchmark]{
                \resizebox{0.5\textwidth}{!}{\input{figures/pykeCFABalance-One.pgf}}
                \label{f:BalanceOneIntel}
        }
        \caption{The balance-one benchmark comparing stealing heuristics (lower is better).}
\end{figure}

\begin{figure}
        \centering
        \subfloat[AMD \CFA Balance-Multi Benchmark]{
                \resizebox{0.5\textwidth}{!}{\input{figures/nasusCFABalance-Multi.pgf}}
                \label{f:BalanceMultiAMD}
        }
        \subfloat[Intel \CFA Balance-Multi Benchmark]{
                \resizebox{0.5\textwidth}{!}{\input{figures/pykeCFABalance-Multi.pgf}}
                \label{f:BalanceMultiIntel}
        }
        \caption{The balance-multi benchmark comparing stealing heuristics (lower is better).}
\end{figure}

There are two benchmarks in which \CFA's work stealing is solely evaluated.
The main goal of introducing work stealing to \CFA's actor system is to eliminate the pathological unbalanced cases that can present themselves in a system without some form of load balancing.
The following two microbenchmarks construct two such pathological cases, and compare the work stealing variations of \CFA.
The balance benchmarks adversarially takes advantage of the round robin assignment of actors to load all actors that will do work on specific cores and create 'dummy' actors that terminate after a single message send on all other cores.
The workload on the loaded cores is the same as the executor benchmark described in \ref{s:executorPerf}, but with fewer rounds.
The balance-one benchmark loads all the work on a single core, whereas the balance-multi loads all the work on half the cores (every other core).
Given this layout, one expects the ideal speedup of work stealing in the balance-one case to be $N / N - 1$ where $N$ is the number of threads.
In the balance-multi case the ideal speedup is 0.5.
Note that in the balance-one benchmark the workload is fixed so decreasing runtime is expected.
In the balance-multi experiment, the workload increases with the number of cores so an increasing or constant runtime is expected.

On both balance microbenchmarks slightly less than ideal speedup compared to the non stealing variation is achieved by both the random and longest victim stealing heuristics.
On the balance-multi benchmark \ref{f:BalanceMultiAMD},\ref{f:BalanceMultiIntel} the random heuristic outperforms the longest victim.
This is likely a result of the longest victim heuristic having a higher stealing cost as it needs to maintain timestamps and look at all timestamps before stealing.
Additionally, a performance cost can be observed when hyperthreading kicks in in Figure~\ref{f:BalanceMultiIntel}.

In the balance-one benchmark on AMD \ref{f:BalanceOneAMD}, the performance bottoms out at 32 cores onwards likely due to the amount of work becoming less than the cost to steal it and move it across cores and cache.
On Intel \ref{f:BalanceOneIntel}, above 32 cores the performance gets worse for all variants due to hyperthreading.
Note that the non stealing variation of balance-one will slow down marginally as the cores increase due to having to create more dummy actors on the inactive cores during startup.

\subsection{Executor}\label{s:executorPerf}
The microbenchmarks in this section are designed to stress the executor.
The executor is the scheduler of an actor system and is responsible for organizing the interaction of executor threads to service the needs of a workload.
In the executor benchmark, 40'000 actors are created and assigned a group.
Each group of actors is a group of 100 actors who send and receive 100 messages from all other actors in their group.
Each time an actor completes all their sends and receives, they are done a round.
After all groups have completed 400 rounds the system terminates.
This microbenchmark is designed to flood the executor with a large number of messages flowing between actors.
Given there is no work associated with each message, other than sending more messages, the intended bottleneck of this experiment is the executor message send process.

\begin{figure}
        \centering
        \subfloat[AMD Executor Benchmark]{
                \resizebox{0.5\textwidth}{!}{\input{figures/nasusExecutor.pgf}}
                \label{f:ExecutorAMD}
        }
        \subfloat[Intel Executor Benchmark]{
                \resizebox{0.5\textwidth}{!}{\input{figures/pykeExecutor.pgf}}
                \label{f:ExecutorIntel}
        }
        \caption{The executor benchmark comparing actor systems (lower is better).}
\end{figure}

The results of the executor benchmark in Figures~\ref{f:ExecutorIntel} and \ref{f:ExecutorAMD} show \CFA with the lowest runtime relative to its peers.
The difference in runtime between \uC and \CFA is largely due to the usage of the copy queue described in Section~\ref{s:copyQueue}.
The copy queue both reduces and consolidates allocations, heavily reducing contention on the memory allocator.
Additionally, due to the static typing in \CFA's actor system, it is able to get rid of expensive dynamic casts that occur in \uC to discriminate messages by type.
Note that dynamic casts are usually not very expensive, but relative to the high performance of the rest of the implementation of the \uC actor system, the cost is significant.

\begin{figure}
        \centering
        \subfloat[AMD \CFA Executor Benchmark]{
                \resizebox{0.5\textwidth}{!}{\input{figures/nasusCFAExecutor.pgf}}
                \label{f:cfaExecutorAMD}
        }
        \subfloat[Intel \CFA Executor Benchmark]{
                \resizebox{0.5\textwidth}{!}{\input{figures/pykeCFAExecutor.pgf}}
                \label{f:cfaExecutorIntel}
        }
        \caption{Executor benchmark comparing \CFA stealing heuristics (lower is better).}
\end{figure}

When comparing the \CFA stealing heuristics in Figure~\ref{f:cfaExecutorAMD} it can be seen that the random heuristic falls slightly behind the other two, but in Figure~\ref{f:cfaExecutorIntel} the runtime of all heuristics are nearly identical to each other.

\begin{figure}
        \centering
        \subfloat[AMD Repeat Benchmark]{
                \resizebox{0.5\textwidth}{!}{\input{figures/nasusRepeat.pgf}}
                \label{f:RepeatAMD}
        }
        \subfloat[Intel Repeat Benchmark]{
                \resizebox{0.5\textwidth}{!}{\input{figures/pykeRepeat.pgf}}
                \label{f:RepeatIntel}
        }
        \caption{The repeat benchmark comparing actor systems (lower is better).}
\end{figure}

The repeat microbenchmark also evaluates the executor.
It stresses the executor's ability to withstand contention on queues, as it repeatedly fans out messages from a single client to 100000 servers who then all respond to the client.
After this scatter and gather repeats 200 times the benchmark terminates.
The messages from the servers to the client will likely all come in on the same queue, resulting in high contention.
As such this benchmark will not scale with the number of processors, since more processors will result in higher contention.
In Figure~\ref{f:RepeatAMD} we can see that \CFA performs well compared to \uC, however by less of a margin than the executor benchmark.
One factor in this result is that the contention on the queues poses a significant bottleneck.
As such the gains from using the copy queue are much less apparent.

\begin{figure}
        \centering
        \subfloat[AMD \CFA Repeat Benchmark]{
                \resizebox{0.5\textwidth}{!}{\input{figures/nasusCFARepeat.pgf}}
                \label{f:cfaRepeatAMD}
        }
        \subfloat[Intel \CFA Repeat Benchmark]{
                \resizebox{0.5\textwidth}{!}{\input{figures/pykeCFARepeat.pgf}}
                \label{f:cfaRepeatIntel}
        }
        \caption{The repeat benchmark comparing \CFA stealing heuristics (lower is better).}
\end{figure}

In Figure~\ref{f:RepeatIntel} \uC and \CFA are very comparable.
In comparison with the other systems \uC does well on the repeat benchmark since it does not have work stealing.
The client of this experiment is long running and maintains a lot of state, as it needs to know the handles of all the servers.
When stealing the client or its respective queue (in \CFA's inverted model), moving the client incurs a high cost due to cache invalidation.
As such stealing the client can result in a hit in performance.

This result is shown in Figure~\ref{f:cfaRepeatAMD} and \ref{f:cfaRepeatIntel} where the no-stealing version of \CFA performs better than both stealing variations.
In particular on the Intel machine in Figure~\ref{f:cfaRepeatIntel}, the cost of stealing is higher, which can be seen in the vertical shift of Akka, CAF and CFA results in Figure~\ref{f:RepeatIntel} (\uC and ProtoActor do not have work stealing).
The shift for CAF is particularly large, which further supports the hypothesis that CAF's work stealing is particularly eager.
In both the executor and the repeat benchmark CAF performs poorly.
It is hypothesized that CAF has an aggressive work stealing algorithm, that eagerly attempts to steal.
This results in poor performance in benchmarks with small messages containing little work per message.
On the other hand, in \ref{f:MatrixAMD} CAF performs much better since each message has a large amount of work, and few messages are sent, so the eager work stealing allows for the clean up of loose ends to occur faster.
This hypothesis stems from experimentation with \CFA.
CAF uses a randomized work stealing heuristic.
In \CFA if the system is tuned so that it steals work much more eagerly with a randomized it was able to replicate the results that CAF achieves in the matrix benchmark, but this tuning performed much worse on all other microbenchmarks that we present, since they all perform a small amount of work per message.

\begin{table}[t]
        \centering
        \setlength{\extrarowheight}{2pt}
        \setlength{\tabcolsep}{5pt}

        \caption{Executor Program Memory High Watermark}
        \label{t:ExecutorMemory}
        \begin{tabular}{*{5}{r|}r}
                & \multicolumn{1}{c|}{\CFA} & \multicolumn{1}{c|}{CAF} & \multicolumn{1}{c|}{Akka} & \multicolumn{1}{c|}{\uC} & \multicolumn{1}{c@{}}{ProtoActor} \\
                \hline
                AMD             & \input{data/pykeExecutorMem} \\
                \hline
                Intel   & \input{data/nasusExecutorMem}
        \end{tabular}
\end{table}

Figure~\ref{t:ExecutorMemory} shows the high memory watermark of the actor systems when running the executor benchmark on 48 cores.
\CFA has a high watermark relative to the other non-garbage collected systems \uC, and CAF.
This is a result of the copy queue data structure, as it will over-allocate storage and not clean up eagerly, whereas the per envelope allocations will always allocate exactly the amount of storage needed.

\subsection{Matrix Multiply}
The matrix benchmark evaluates the actor systems in a practical application, where actors concurrently multiplies two matrices.
The majority of the computation in this benchmark involves computing the final matrix, so this benchmark stresses the actor systems' ability to have actors run work, rather than stressing the executor or message sending system.

Given $Z_{m,r} = X_{m,n} \cdot Y_{n,r}$, the matrix multiply is defined as:
\begin{displaymath}
X_{i,j} \cdot Y_{j,k} = \left( \sum_{c=1}^{j} X_{row,c}Y_{c,column} \right)_{i,k}
\end{displaymath}

The benchmark uses input matrices $X$ and $Y$ that are both $3072$ by $3072$ in size.
An actor is made for each row of $X$ and is passed via message the information needed to calculate a row of the result matrix $Z$.


Given that the bottleneck of the benchmark is the computation of the result matrix, it follows that the results in Figures~\ref{f:MatrixAMD} and \ref{f:MatrixIntel} are clustered closer than other experiments.
In Figure~\ref{f:MatrixAMD} \uC and \CFA have identical performance and in Figure~\ref{f:MatrixIntel} \uC pulls ahead of \CFA after 24 cores likely due to costs associated with work stealing while hyperthreading.
As mentioned in \ref{s:executorPerf}, it is hypothesized that CAF performs better in this benchmark compared to others due to its eager work stealing implementation.
In Figures~\ref{f:cfaMatrixAMD} and \ref{f:cfaMatrixIntel} there is little negligible performance difference across \CFA stealing heuristics.

\begin{figure}
        \centering
        \subfloat[AMD Matrix Benchmark]{
                \resizebox{0.5\textwidth}{!}{\input{figures/nasusMatrix.pgf}}
                \label{f:MatrixAMD}
        }
        \subfloat[Intel Matrix Benchmark]{
                \resizebox{0.5\textwidth}{!}{\input{figures/pykeMatrix.pgf}}
                \label{f:MatrixIntel}
        }
        \caption{The matrix benchmark comparing actor systems (lower is better).}
\end{figure}

\begin{figure}
        \centering
        \subfloat[AMD \CFA Matrix Benchmark]{
                \resizebox{0.5\textwidth}{!}{\input{figures/nasusCFAMatrix.pgf}}
                \label{f:cfaMatrixAMD}
        }
        \subfloat[Intel \CFA Matrix Benchmark]{
                \resizebox{0.5\textwidth}{!}{\input{figures/pykeCFAMatrix.pgf}}
                \label{f:cfaMatrixIntel}
        }
        \caption{The matrix benchmark comparing \CFA stealing heuristics (lower is better).}
\end{figure}

% Local Variables: %
% tab-width: 4 %
% End: %