Scalable Termination Detection for Distributed Actor Systems

A refob is a triple $(x,A,B)$, where $A$ is the owner actor’s address, $B$ is the target actor’s address, and $x$ is a globally unique token. An actor can cheaply generate such a token by combining its address with a local sequence number, since actor systems already guarantee that each address is unique. We will stylize a triple $(x,A,B)$ as ${x:A\multimap B}$. We will also sometimes refer to such a refob as simply $x$, since tokens act as unique identifiers.

When an actor $A$ spawns an actor $B$ (Fig. 2 (1, 2)) the DRL protocol creates a new refob ${x:A\multimap B}$ that is stored in both $A$ and $B$’s system-level state, and a refob ${y:B\multimap B}$ in $B$’s state. The refob $x$ allows $A$ to send application-level messages to $B$. These messages are denoted $\mathtt{app}(x,R)$, where $R$ is the sett of refobs contained in the message that $A$ has created for $B$. The refob $y$ corresponds to the self variable present in some actor languages.

If $A$ has active refobs ${x:A\multimap B}$ and ${y:A\multimap C}$, then it can create a new refob ${z:B\multimap C}$ by generating a token $z$. In addition to being sent to $B$, this refob must also temporarily be stored in $A$’s system-level state and marked as “created using $y$” (Fig. 2 (3)). Once $B$ receives $z$, it must add the refob to its system-level state and mark it as “active” (Fig. 2 (4)). Note that $B$ can have multiple distinct refobs that reference the same actor in its state; this can be the result of, for example, several actors concurrently sending refobs to $B$. Transition rules for spawning actors and sending messages are given in Section 4.3.

Actor $A$ may remove $z$ from its state once it has sent a (system-level) $\mathtt{info}$ message informing $C$ about $z$ (Fig. 2 (4)). Similarly, when $B$ no longer needs its refob for $C$, it can “deactivate” $z$ by removing it from local state and sending $C$ a (system-level) $\mathtt{release}$ message (Fig. 2 (5)). Note that if $B$ already has a refob ${z:B\multimap C}$ and then receives another ${z^{\prime}:B\multimap C}$, then it can be more efficient to defer deactivating the extraneous $z^{\prime}$ until $z$ is also no longer needed; this way, the $\mathtt{release}$ messages can be batched together.

When $C$ receives an $\mathtt{info}$ message, it records that the refob has been created, and when $C$ receives a $\mathtt{release}$ message, it records that the refob has been released (Fig. 2 (6)). Note that these messages may arrive in any order. Once $C$ has received both, it is permitted to remove all facts about the refob from its local state. Transition rules for these reference listing actions are given in Section 4.4.

Once a refob has been created, it cycles through four states: pending, active, inactive, or released. A refob ${z:B\multimap C}$ is said to be pending until it is received by its owner $B$. Once received, the refob is active until it is deactivated by its owner, at which point it becomes inactive. Finally, once $C$ learns that $z$ has been deactivated, the refob is said to be released. A refob that has not yet been released is unreleased.

Slightly amending the definition we gave in Section 2, we say that $B$ is a potential acquaintance of $A$ (and $A$ is a potential inverse acquaintance of $B$) when there exists an unreleased refob ${x:A\multimap B}$. Thus, $B$ becomes a potential acquaintance of $A$ as soon as $x$ is created, and only ceases to be an acquaintance once it has received a $\mathtt{release}$ message for every refob ${y:A\multimap B}$ that has been created so far.

For each refob ${x:A\multimap B}$, the owner $A$ counts the number of $\mathtt{app}$ and $\mathtt{info}$ messages sent along $x$; this count can be deleted when $A$ deactivates $x$. Each message is annotated with the refob used to send it. Whenever $B$ receives an $\mathtt{app}$ or $\mathtt{info}$ message along $x$, it correspondingly increments a receive count for $x$; this count can be deleted once $x$ has been released. Thus the memory overhead of message counts is linear in the number of unreleased refobs.

A snapshot is a copy of all the facts in an actor’s system-level state at some point in time. We will assume throughout the paper that in every set of snapshots $Q$, each snapshot was taken by a different actor. Such a set is also said to form a cut. Recall that a cut is consistent if no snapshot in the cut causally precedes any other [9]. Let us also say that $Q$ is a set of mutually quiescent snapshots if there are no undelivered messages between actors in the cut. That is, if $A\in Q$ sent a message to $B\in Q$ before taking a snapshot, then the message must have been delivered before $B$ took its snapshot. Notice that if all snapshots in $Q$ are mutually quiescent, then $Q$ is consistent.

Notice also that in Fig. 3, the snapshots of $B$ and $C$ are mutually quiescent when their send and receive counts agree. This is ensured in part because each refob has a unique token: If actors associated message counts with actor names instead of tokens, then $B$’s snapshots at $t_{0}$ and $t_{3}$ would both contain $\mathtt{Sent}(C,1)$. Thus, $B$’s snapshot at $t_{3}$ and $C$’s snapshot at $t_{0}$ would appear mutually quiescent, despite having undelivered messages in the cut.

We would like to conclude that snapshots from two actors $A,B$ are mutually quiescent if and only if their send and receive counts are agreed for every refob ${x:A\multimap B}$ or ${y:B\multimap A}$. Unfortunately, this fails to hold in general for systems with unordered message delivery. It also fails to hold when, for instance, the owner actor takes a snapshot before the refob is activated and the target actor takes a snapshot after the refob is released. In such a case, neither knowledge set includes a message count for the refob and they therefore appear to agree. However, we show that the message counts can nevertheless be used to bound the number of undelivered messages for purposes of our algorithm (Lemma A.5).

We use the capital letters $A,B,C,D,E$ to denote actor addresses. Tokens are denoted $x,y,z$, with a special reserved token $\mathtt{null}$ for messages from external actors.

A fact is a value that takes one of the following forms: $\mathtt{Created}(x)$, $\mathtt{Released}(x)$, $\mathtt{CreatedUsing}(x,y)$, $\mathtt{Active}(x)$, $\mathtt{Unreleased}(x)$, $\mathtt{Sent}(x,n)$, or $\mathtt{Received}(x,n)$ for some refobs $x,y$ and natural number $n$. Each actor’s state holds a set of facts about refobs and message counts called its knowledge set. We use $\phi,\psi$ to denote facts and $\Phi,\Psi$ to denote finite sets of facts. Each fact may be interpreted as a predicate that indicates the occurrence of some past event. Interpreting a set of facts $\Phi$ as a set of axioms, we write $\Phi\vdash\phi$ when $\phi$ is derivable by first-order logic from $\Phi$ with the following additional rules:

• •

  If $(\not\exists n\in\mathbb{N},\ \mathtt{Sent}(x,n)\in\Phi)$ then $\Phi\vdash\mathtt{Sent}(x,0)$

• •

  If $(\not\exists n\in\mathbb{N},\ \mathtt{Received}(x,n)\in\Phi)$ then $\Phi\vdash\mathtt{Received}(x,0)$

• •

  If $\Phi\vdash\mathtt{Created}(x)\land\lnot\mathtt{Released}(x)$ then $\Phi\vdash\mathtt{Unreleased}(x)$

• •

  If $\Phi\vdash\mathtt{CreatedUsing}(x,y)$ then $\Phi\vdash\mathtt{Created}(y)$

For convenience, we define a pair of functions $\emph{incSent}(x,\Phi),\emph{incRecv}(x,\Phi)$ for incrementing message send/receive counts, as follows: If $\mathtt{Sent}(x,n)\in\Phi$ for some $n$, then $\emph{incSent}(x,\Phi)=(\Phi\setminus\{\mathtt{Sent}(x,n)\})\cup\{\mathtt{Sent}(x,n+1)\}$; otherwise, $\emph{incSent}(x,\Phi)=\Phi\cup\{\mathtt{Sent}(x,1)\}$. Likewise for incRecv and $\mathtt{Received}$.

Recall that an actor is either busy (processing a message) or idle (waiting for a message). An actor with knowledge set $\Phi$ is denoted $[\Phi]$ if it is busy and $(\Phi)$ if it is idle.

Our specification includes both system messages (also called control messages) and application messages. The former are automatically generated by the DRL protocol and handled at the system level, whereas the latter are explicitly created and consumed by user-defined behaviors. Application-level messages are denoted $\mathtt{app}(x,R)$. The argument $x$ is the refob used to send the message. The second argument $R$ is a set of refobs created by the sender to be used by the destination actor. Any remaining application-specific data in the message is omitted in our notation.

The DRL communication protocol uses two kinds of system messages. $\mathtt{info}(y,z,B)$ is a message sent from an actor $A$ to an actor $C$, informing it that a new refob ${z:B\multimap C}$ was created using ${y:A\multimap C}$. $\mathtt{release}(x,n)$ is a message sent from an actor $A$ to an actor $B$, informing it that the refob ${x:A\multimap B}$ has been deactivated and should be released.

A configuration $\langle\!\langle\ \alpha\ |\ \mu\ \rangle\!\rangle^{\rho}_{\chi}$ is a quadruple $(\alpha,\mu,\rho,\chi)$ where: $\alpha$ is a mapping from actor addresses to knowledge sets; $\mu$ is a mapping from actor addresses to multisets of messages; and $\rho,\chi$ are sets of actor addresses. Actors in $\text{dom}(\alpha)$ are internal actors and actors in $\chi$ are external actors; the two sets may not intersect. The mapping $\mu$ associates each actor with undelivered messages to that actor. Actors in $\rho$ are receptionists. We will ensure $\rho\subseteq\text{dom}(\alpha)$ remains valid in any configuration that is derived from a configuration where the property holds (referred to as the locality laws in [12]).

Configurations are denoted by $\kappa$, $\kappa^{\prime}$, $\kappa_{0}$, etc. If an actor address $A$ (resp. a token $x$), does not occur in $\kappa$, then the address (resp. the token) is said to be fresh. We assume a facility for generating fresh addresses and tokens.

In order to express our transition rules in a pattern-matching style, we will employ the following shorthand. Let $\alpha,[\Phi]_{A}$ refer to a mapping $\alpha^{\prime}$ where $\alpha^{\prime}(A)=[\Phi]$ and $\alpha=\alpha^{\prime}|_{\text{dom}(\alpha^{\prime})\setminus\{A\}}$. Similarly, let $\mu,[A\triangleleft m]$ refer to a mapping $\mu^{\prime}$ where $m\in\mu^{\prime}(A)$ and $\mu=\mu^{\prime}|_{\text{dom}(\mu^{\prime})\setminus\{A\}}\cup\{A\mapsto\mu^{\prime}(A)\setminus\{m\}\}$. Informally, the expression $\alpha,[\Phi]_{A}$ refers to a set of actors containing both $\alpha$ and the busy actor $A$ (with knowledge set $\Phi$); the expression $\mu,[A\triangleleft m]$ refers to the set of messages containing both $\mu$ and the message $m$ (sent to actor $A$).

The rules of our transition system define atomic transitions from one configuration to another. Each transition rule has a label $l$, parameterized by some variables $\vec{x}$ that occur in the left- and right-hand configurations. Given a configuration $\kappa$, these parameters functionally determine the next configuration $\kappa^{\prime}$. Given arguments $\vec{v}$, we write $\kappa\xrightarrow{l(\vec{v})}\kappa^{\prime}$ to denote a semantic step from $\kappa$ to $\kappa^{\prime}$ using rule $l(\vec{v})$.

We refer to a label with arguments $l(\vec{v})$ as an event, denoted $e$. A sequence of events is denoted $\pi$. If $\pi=e_{1},\dots,e_{n}$ then we write $\kappa\xrightarrow{\pi}\kappa^{\prime}$ when $\kappa\xrightarrow{e_{1}}\kappa_{1}\xrightarrow{e_{2}}\dots\xrightarrow{e_{n}}\kappa^{\prime}$. If there exists $\pi$ such that $\kappa\xrightarrow{\pi}\kappa^{\prime}$, then $\kappa^{\prime}$ is derivable from $\kappa$. An execution is a sequence of events $e_{1},\dots,e_{n}$ such that $\kappa_{0}\xrightarrow{e_{1}}\kappa_{1}\xrightarrow{e_{2}}\dots\xrightarrow{e_{n}}\kappa_{n}$, where $\kappa_{0}$ is the initial configuration (Section 4.2). We say that a property holds at time $t$ if it holds in $\kappa_{t}$.