RAMBO: A Reconfigurable Atomic Memory Service for

Dynamic Networks

Nancy Lynch1 and Alex A. Shvartsman2,1
1
2

Laboratory for Computer Science, Massachusetts Institute of Technology,

200 Technology Square, NE43-365, Cambridge, MA 02139, USA.
Department of Computer Science and Engineering, University of Connecticut,
191 Auditorium Road, Unit 3155, Storrs, CT 06269, USA.

Abstract. This paper presents an algorithm that emulates atomic read/write

shared objects in a dynamic network setting. To ensure availability and faulttolerance, the objects are replicated. To ensure atomicity, reads and writes are performed using quorum configurations, each of which consists of a set of members
plus sets of read-quorums and write-quorums. The algorithm is reconfigurable: the
quorum configurations may change during computation, and such changes do not
cause violations of atomicity. Any quorum configuration may be installed at any
time. The algorithm tolerates processor stopping failure and message loss. The algorithm performs three major tasks, all concurrently: reading and writing objects,
introducing new configurations, and garbage-collecting obsolete configurations.
The algorithm guarantees atomicity for arbitrary patterns of asynchrony and failure. The algorithm satisfies a variety of conditional performance properties, based
on timing and failure assumptions. In the "normal case", the latency of read and
write operations is at most 8d, where d is the maximum message delay.

Introduction

This paper presents an algorithm that can be used to implement atomic read/write shared
memory in a dynamic network setting, in which participants may join or fail during the
course of computation. Examples of such settings are mobile networks and peer-to-peer
networks. One use of this service might be to provide long-lived data in a dynamic and
volatile setting such as a military operation.
In order to achieve availability in the presence of failures, the objects are replicated.
In order to maintain memory consistency in the presence of small and transient changes,
the algorithm uses configurations, each of which consists of a set of members plus sets of
read-quorums and write-quorums. In order to accommodate larger and more permanent
changes, the algorithm supports reconfiguration, by which the set of members and the
sets of quorums are modified. Such changes do not cause violations of atomicity. Any
quorum configuration may be installed at any time.

This work was supported by the NSF ITR Grant 0121277. The work of the first author was also
supported by AFOSR under contract F49620-00-1-0097 and by NTT under contract MIT990412. The work of the second author was also supported by NSF CAREER Award 9984778 and
NSF Grant 9988304.

D. Malkhi (Ed.): DISC 2002, LNCS 2508, pp. 173190, 2002.

c Springer-Verlag Berlin Heidelberg 2002

174

N. Lynch and A.A. Shvartsman

We first provide a formal specification for reconfigurable atomic shared memory as

a global service. We call this service Rambo, which stands for Reconfigurable Atomic
Memory for Basic Objects (Basic" means Read/Write). The rest of the paper presents
our algorithm and its analysis. The algorithm carries out three major activities, all concurrently: reading and writing objects, introducing new configurations, and removing
(garbage-collecting) obsolete configurations. The algorithm is composed of a main
algorithm, which handles reading, writing, and garbage-collection, and a global reconfiguration service, Recon, which provides the main algorithm with a consistent sequence
of configurations. Reconfiguration is loosely coupled to the main algorithm, in particular, several configurations may be known to the algorithm at one time, and read and
write operations can use them all.
The main algorithm performs read and write operations using a two-phase strategy.
The first phase gathers information from read-quorums of active configurations and the
second phase propagates information to write-quorums of active configurations. This
communication is carried out using background gossiping, which allows the algorithm to
maintain only a small amount of protocol state information. Each phase is terminated by
a fixed point condition that involves a quorum from each active configuration. Different
read and write operations may execute concurrently: the restricted semantics of reads
and writes permit the effects of this concurrency to be sorted out afterwards.
The main algorithm also includes a facility for garbage-collecting old configurations
when their use is no longer necessary for maintaining consistency. Garbage-collection
also uses a two-phase strategy, where the first phase communicates with an old configuration and the second phase communicates with a new configuration. A garbage-collection
operation ensures that both a read-quorum and a write-quorum of the old configuration
learn about the new configuration, and that the latest value from the old configuration is
conveyed to a write-quorum of the new configuration.
The reconfiguration service is implemented by a distributed algorithm that uses
distributed consensus to agree on the successive configurations. Any member of the
latest configuration c may propose a new configuration at any time; different proposals
are reconciled by an execution of consensus among the members of c. Consensus is, in
turn, implemented using a version of the Paxos algorithm [17], as described formally
in [8]. Although such consensus executions may be slowin fact, in some situations,
they may not even terminatethey do not delay read and write operations.
We show atomicity for arbitrary patterns of asynchrony and failure. We analyze
performance conditionally, based on timing and failure assumptions. For example, assuming that gossip and garbage-collection occur periodically, that reconfiguration is
requested infrequently enough for garbage-collection to keep up, and that quorums of
active configurations do not fail, we show that read and write operations complete within
time 8d, where d is the maximum message latency.
Comparison with other approaches. Consensus algorithms can be used directly to
implement an atomic data service by allowing participants to agree on a global total
ordering of all operations [17]. In contrast, we use consensus to agree only on the
sequence of configurations and not on the individual operations. Also, in our algorithm,
the termination of consensus affects the termination of reconfiguration, but not of read
and write operations.

RAMBO: A Reconfigurable Atomic Memory Service for Dynamic Networks

175

Group communication services (GCSs) [1] can also be used to implement an atomic
data service, e.g., by implementing a global totally ordered broadcast service on top of a
view-synchronous GCS [11] using techniques of [16,2]. In most GCS implementations,
forming a new view takes a substantial amount of time, and client-level operations are
delayed during the view-formation period. In our algorithm, reads and writes can make
progress during reconfiguration. Also, in some standard GCS implementations, e.g., [5],
performance is degraded even if only one stopping failure occurs; our algorithm uses
quorums to tolerate small numbers of failures.
A dynamic primary configuration GCS was introduced in [7] and used to implement
atomic memory, using techniques of [3] within each configuration. That work restricts
the set of possible new configurations to those satisfying certain intersection properties
with respect to the previous configurations, whereas we impose no such restrictions.
Like other solutions based on GCSs, the algorithm of [7] delays reads and writes during
reconfiguration.
Single reconfigurer implementations for atomic memory are considered in [19,10].
In these approaches, the failure of the reconfigurer disables future reconfiguration. Also,
in [19,10], garbage-collection of an old configuration is tightly coupled to the introduction of a new configuration, whereas in our new algorithm, garbage-collection is carried
out in the background, concurrently with other processing.
Other related work. The first general scheme for emulating shared memory in messagepassing systems by using replication and accessing majorities of time-stamped replicas
was given in [21]. An algorithm for majority-based emulation of atomic read/write memory was presented in [3]. This algorithm introduced a two-phase paradigm in which the
first phase gathers information from a majority of participants, and the second phase
propagates information to a majority. A quorum system [13]a generalization of majority setsis a collection of sets such that any two sets, called quorums, intersect [12].
Quorum systems have been used to implement data replication protocols, e.g., [4,6,14].
Consensus algorithms have been used as building blocks in other work, e.g, [15].
Note. This paper is an extended abstract of a full report [20]. The full version includes
specifications of all components, complete proofs, and additional results.

Data Types

We assume distinguished elements and , which are not in any of the basic types. For
any type A, we define types A = A {}. and A = A {, }. If A is a poset,
we augment its ordering by assuming that < a < for every a A.
We assume the following data types and distinguished elements: I, the totally-ordered
set of locations. T , the set of tags, defined as N I. M , the set of messages. X, the set
of object identifiers, partitioned into subsets Xi , i I; Xi is the set of identifiers for
objects that may be created at location i. For any x X, (i0 )x denotes the unique i such
that x Xi . For each x X, we define Vx , the set of values that object x may take on,
and (v0 )x Vx , the initial value of x.
We also assume: C, the set of configuration ids; we use the trivial partial order
on C, in which all elements are incomparable. For each x X, (c0 )x C, the
initial configuration id for x. For each c C: members(c), a finite subset of I,

176

N. Lynch and A.A. Shvartsman

read-quorums(c) and write-quorums(c), two sets of finite subsets of members(c).

We assume: (1) members((c0 )x ) = {(i0 )x }, that is, the initial configuration for x has
one member, the creator of x. (2) For every c, every R read-quorums(c), and every
W write-quorums(c), R W = .
We define functions and sets for configurations: update, a binary function on C ,
defined by update(c, c ) = max(c, c ) if c and c are comparable, update(c, c ) = c
otherwise. extend , a binary function on C , defined by extend (c, c ) = c if c = and
c C, and extend (c, c ) = c otherwise. CMap, the set of configuration maps, defined
as N C . We extend the update and extend operators elementwise to binary operations on CMap. truncate, a unary function on CMap, defined by truncate(cm)(k) =
if there exists  k such that cm() = , truncate(cm)(k) = cm(k) otherwise. Truncation removes all the configuration ids that follow a . Truncated , the subset of CMap
such that cm Truncated if and only if truncate(cm) = cm. Usable, the subset of
CMap such that cm Usable iff the pattern occurring in cm consists of a prefix of
finitely many s, followed by an element of C, followed by an infinite sequence of
elements of C in which all but finitely many elements are .

Reconfigurable Atomic Memory Service Specification

Our specification for the Rambo service consists of an external signature plus a set of
traces that embody Rambos safety properties. No liveness properties are included; we
replace these with conditional latency bounds, which appear in Section 8. The external
signature appears in Figure 1. We use I/O automata notation for all our specifications.

Output:
Input:
join-ack(rambo)x,i , x X, i I
join(rambo, J)x,i , J a finite subset of I {i}, x X,
i I, such that if i = (i0 )x then J =
read-ack(v)x,i , v Vx , x X, i I
readx,i , x X, i I
write-ackx,i , x X, i I
write(v)x,i , v Vx , x X, i I
recon-ack(b)x,i , b {ok, nok}, x X, i I
recon(c, c
)x,i , c, c
C, i members(c), x X, i I
report(c)x,i , c C, c X, i I
faili , i I

Fig. 1. Rambo(x): External signature

The client at location i requests to join the system for a particular object x by
performing a join(rambo, J)x,i input action. The set J represents the clients guess at
a set of processes that have already joined the system for x. If i = (i0 )x , the set J
is empty, because (i0 )x is supposed to be the first process to join the system for x. If
the join attempt is successful, the Rambo service responds with a join-ack(rambo)x,i
output action. The client at i initiates a read (resp., write) operation using a readi (resp.,
writei ) input action, which the Rambo service acknowledges with a read-acki (resp.,
write-acki ) output. The client initiates a reconfiguration using a reconi input, which is
acknowledged with a recon-acki output. Rambo reports a new configuration to the client
using a reporti output. Finally, a stopping failure at location i is modelled using a faili
input action. We model process leaves as failures.

RAMBO: A Reconfigurable Atomic Memory Service for Dynamic Networks

177

The set of traces describing Rambos safety properties consists of those that satisfy
an implication of the form environment assumptions imply service guarantees. The
environment assumptions are simple well-formedness conditions:
Well-formedness: (1) For every x and i: (a) No join(rambo, )x,i , readx,i , write()x,i ,
or recon(, )x,i event is preceded by a faili event. (b) At most one join(rambo, )x,i
event occurs. (c) Any readx,i , write()x,i , or recon(, )x,i event is preceded by a
join-ack(rambo)x,i event. (d) Any readx,i , write()x,i , or recon(, )x,i event is preceded by an -ack event for any preceding event of any of these kinds. (2) For every x
and c, at most one recon(, c)x, event occurs. (Configuration ids that are proposed in
recon events are unique. This is not a serious restriction, because the same membership
and quorum sets may be associated with different configuration ids.) (3) For every c, c ,
x, and i, if a recon(c, c )x,i event occurs, then it is preceded by a report(c)x,i event and
by a join-ack(rambo)x,j event for every j members(c ).
The safety guarantees provided by the service are as follows:
Well-formedness: For every x and i: (a) No join-ack(rambo)x,i , read-ack()x,i ,
write-ackx,i , recon-ack()x,i , or report()x,i event is preceded by a faili event. (b)
Any join-ack(rambo)x,i (resp., read-ack()x,i , write-ackx,i , recon-ack()x,i ) event has
a preceding join(rambo, )x,i (resp., readx,i , write()x,i , recon(, )x,i ) event with no
intervening invocation or response action for x and i.
Atomicity:1 If all the read and write operations that are invoked complete, then the read
and write operations for object x can be partially ordered by an ordering , so that the
following conditions are satisfied: (1) No operation has infinitely many other operations
ordered before it. (2) The partial order is consistent with the external order of invocations
and responses, that is, there do not exist read or write operations 1 and 2 such that 1
completes before 2 starts, yet 2 1 . (3) All write operations are totally ordered and
every read operation is ordered with respect to all the writes. (4) Every read operation
ordered after any writes returns the value of the last write preceding it in the partial
order; any read operation ordered before all writes returns (v0 )x .
The rest of the paper presents our implementation of Rambo. The implementation is
a distributed algorithm in the asynchronous message-passing model. All processes may
communicate with each other. Processes may fail by stopping without warning.
Our implementation can be described formally as the composition of a separate
implementation for each x, so we describe the implementation for a generic x. We
suppress explicit mention of x, writing V , v0 , c0 , and i0 as shorthand for Vx , (v0 )x ,
(c0 )x , and (i0 )x , respectively.

Atomicity is often defined in terms of an equivalence with a serial memory. The definition
given here implies this equivalence, as shown, for example, in Lemma 13.16 in [18]. Although
Lemma 13.16 of [18] is presented for a setting with only finitely many locations, nothing in
Lemma 13.16 or its proof depends on the finiteness of the set of locations.

178

N. Lynch and A.A. Shvartsman

Reconfiguration Service Specification

Our Rambo implementation for object x consists of a main RW algorithm and a reconfiguration service, Recon. Here we present the specification for Recon. Our implementation
of Recon is described in Section 7.
The external signature appears in Figure 2. The client of Recon at location i requests
to join the reconfiguration service by performing a join(recon)i input action. The service
acknowledges this with a corresponding join-acki output action. The client requests
reconfiguration using a reconi input, which is acknowledged with a recon-acki output
action. The service reports a new configuration to the client using a reporti output
action. Outputs of the form new-cong(c, k)i announce at location i that c is the k th
configuration id. These outputs are used for communication with the portion of the
RW algorithm running at location i. Recon announces consistent information, only
one configuration id for each index in the configuration id sequence. Recon delivers
information about each configuration to members of the new configuration and members
of the immediately preceding configuration. Crashes are modeled using fail actions.

Input:
join(recon)i , i I
recon(c, c
)i , c, c
C, i members(c)
faili , i I

Output:
join-ack(recon)i , i I
recon-ack(b)i , b {ok, nok}, i I
report(c)i , c C, i I
new-cong(c, k)i , c C, k N+ , i I

Fig. 2. Recon: External signature

The set of traces describing Recons safety properties is defined by environment

assumptions and service guarantees. The environment assumptions are simple wellformedness conditions, consistent with those for Rambo:
Well-formedness: (1) For every i: (a) No join(recon)i or recon(, )i event is preceded
by a faili event. (b) At most one join(recon)i event occurs. (c) Any recon(, )i event
is preceded by a join-ack(recon)i event. (d) Any recon(, )i event is preceded by an
-ack for any preceding recon(, )i event. (2) For every c, at most one recon(, c) event
occurs. (3) For every c, c , x, and i, if a recon(c, c )i event occurs, then it is preceded
by: (a) A report(c)i event, and (b) A join-ack(recon)j for every j members(c ).
The safety guarantees are:
Well-formedness: For every i: (a) No join-ack(recon)i , recon-ack()i , report()i , or
new-cong(, )i event is preceded by a faili event. (b) Any join-ack(recon)i (resp.,
recon-ack(c)i ) event has a preceding join(recon)i (resp., reconi ) event with no intervening invocation or response action for x and i.
Agreement: If new-cong(c, k)i and new-cong(c , k)j both occur, then c = c .
Validity: If new-cong(c, k)i occurs, then it is preceded by a recon(, c)i
for some i
for which a matching recon-ack(nok)i
does not occur.
No duplication: If new-cong(c, k)i and new-cong(c, k  )i
both occur, then k = k  .

RAMBO: A Reconfigurable Atomic Memory Service for Dynamic Networks

179

Implementation of Rambo Using a Reconfiguration Service

Our Rambo implementation includes, for each i, a Joiner i automaton, which handles
joining, and a RWi automaton, which handles reading, writing, and installing new
configurations. These automata use asynchronous communication channels Channel i,j .
The RW automata also interact with an arbitrary implementation of Recon.
5.1

Joiner Automata

When Joiner i receives a join(rambo, J) request from its environment, it sends join
messages to the processes in J (with the hope that they are already participating, and so
can help in its attempt to join). Also, it submits join requests to the local RW and Recon
components and waits for acknowledgments. The join messages that are sent by Joiner
automata are handled by RW automata at other locations.
5.2

Reader-Writer Automata

RWi processes each read or write operation using one or more configurations, which
it learns about from the Recon service. It also handles the garbage-collection of older
configurations. The signature and state of RWi appear in Figure 3. Figure 4 presents
the transitions pertaining to joining the protocol and failing. Figure 5 presents those
pertaining to reading and writing, and Figure 6 presents those pertaining to garbagecollection.

Signature:
Input:
readi
write(v)i , v V
new-cong(c, k)i , c C, k N+
recv(join)j,i , j I {i}
recv(m)j,i , m M , j I
join(rw)i
faili

Output:
join-ack(rw)i
read-ack(v)i , v V
write-acki
send(m)i,j , m M , j I

Internal:
query-xi
prop-xi
gc(k)i , k N
gc-query-x(k)i , k N
gc-prop-x(k)i , k N
gc-ack(k)i , k N

State:
status {idle, joining, active},
initially idle
world, a finite subset of I, initially
value V , initially v0
tag T , initially (0, i0 )
cmap CMap, initially
cmap(0) = c0 ,
cmap(k) = for k 1
pnum1 N, initially 0
pnum2 I N,
initially everywhere 0
failed, a Boolean, initially false

op, a record with fields:

type {read, write}
phase {idle, query, prop,
done}, initially idle
pnum N
cmap CMap
acc, a finite subset of I
value V

gc, a record with fields:

phase {idle, query, prop},
initially idle
pnum N
acc, a finite subset of I
index N

Fig. 3. RWi : Signature and state

State variables. The status variable keeps track of the progress of the component as
it joins the protocol. When status = idle, RWi does not respond to any inputs (except

180

N. Lynch and A.A. Shvartsman

for join) and does not perform any locally controlled actions. When status = joining,
RWi responds to inputs but still does not perform any locally controlled actions. When
status = active, the automaton participates fully in the protocol.
The world variable is used to keep track of all processes that are known to have tried
to join the system. The value variable contains the current value of the local replica of
x, and tag holds the associated tag.
The cmap variable contains information about configurations. If cmap(k) = , it
means that RWi has not yet learned what the k th configuration id is. If cmap(k) =
c C, it means that RWi has learned that the k th configuration id is c, and has not
yet garbage-collected it. If cmap(k) = , it means that RWi has garbage-collected the
k th configuration id. RWi learns about configuration ids either directly from the Recon
service, or from other RW processes. The value of cmap is always in Usable, that is,
for some finite prefix of N, followed by an element of C, followed by elements of C ,
with only finitely many elements of C. When RWi processes a read or write operation,
it uses all the configurations whose ids appear in its cmap, up to the first .
The pnum1 variable and pnum2 array are used to implement a handshake that
identifies recent messages. RWi uses pnum1 to count the total number of operation
phases (either query or propagation phases) that it has initiated overall, including phases
occurring in read, write, and garbage-collection operations. For every j, including j = i,
RWi uses pnum2 (j) to record the largest number of a phase that i has learned that j has
started, via a message from j to i. Finally, two records, op and gc, are used to maintain
information about locally-initiated read, write, and garbage-collection operations.
Joining and failure transitions. When a join(rw)i input occurs when status = idle, if i
is the objects creator i0 , then status immediately becomes active, which means that RWi
is ready for full participation in the protocol. Otherwise, status becomes joining, which
means that RWi is receptive to inputs but not ready to perform any locally controlled
actions. In either case, RWi records itself as a member of its own world . From this point
on, RWi adds to its world any process from which it receives a join message. (Recall
that these join messages are sent by Joiner automata.)
If status = joining, then status becomes active when RWi receives a message
from another RW process. (The code for this appears in the recv transition definition in
Figure 5.) At this point, process i has acquired enough information to begin participating
fully. After status becomes active, process i can perform a join-ack(rw).
Information propagation transitions. Information is propagated between RW processes in the background, via point-to-point channels that are accessed using the send
and recv actions. The algorithm uses one kind of message, which contains a tuple consisting of the senders world , its latest known value and tag, its cmap, and two phase
numbersthe current phase number of the sender, pnum1 , and the latest known phase
number of the receiver, from the pnum2 array. These messages may be sent at any time,
to processes in the senders world .
When RWi receives a message, it sets its status to active, if it has not already done
so. It adds incoming world information, in W , to its world set. It compares the incoming
tag t to its own tag. If t is strictly greater, it represents a more recent version of the object;
in this case, RWi sets its tag to t and its value to the incoming value v. RWi also updates
its cmap with the information in the incoming CMap, cm, using the update operator

RAMBO: A Reconfigurable Atomic Memory Service for Dynamic Networks

Input join(rw)i
Effect:
if failed then
if status = idle then
if i = i0 then
status active
else
status joining
world world {i}
Input recv(join)j,i
Effect:
if failed then
if status = idle then
world world {j}

181

Output join-ack(rw)i
Precondition:
failed
status = active
Effect:
none
Input faili
Effect:
failed true

Fig. 4. RWi : Join-related and failure transitions

defined in Section 2. That is, for each k, if cmap(k) = and cm(k) C, process i
sets its cmap(k) to cm(k). Also, if cmap(k) C and cm(k) = , indicating that the
sender knows that configuration k has already been garbage-collected, then RWi sets its
cmap(k) to . RWi also updates its pnum2 (j) component for the sender j to reflect
new information about js phase number, which appears in the pns component of the
message.
While RWi is conducting a phase of a read, write, or garbage-collection operation, it
verifies that the incoming message is recent, in the sense that the sender j sent it after
j received a message from i that was sent after i began the current phase. RWi uses the
phase numbers to perform this check: it checks that the incoming phase number pnr is
at least as large as the current operation phase number ([Link] or [Link]). If the
message is recent, then RWi uses it to update the op or gc record.
Read and write operations. A read or write operation is performed in two phases: a
query phase and a propagation phase. In each phase, RWi obtains recent value, tag, and
cmap information from enough processes. This information is obtained by sending
and receiving messages, as described above.
When RWi starts a phase of a read or write, it sets [Link] to a CMap whose configurations are to be used to conduct the phase. Specifically, RWi uses truncate(cmap),
which is defined to include all the configuration ids in cmap up to the first . When
a new CMap cm is received during the phase, [Link] is extended by adding all
newly-discovered configuration ids, up to the first in cm. If adding these new configuration ids does not create a gap, that is, if the extended [Link] is in Truncated , then
the phase continues using the extended [Link]. On the other hand, if adding these new
configuration ids creates a gap, then RWi can infer that it has been using out-of-date
configuration ids. In this case, it restarts the phase using the best currently known CMap,
which is obtained by computing truncate(cmap) for the latest local cmap.
In between restarts, while RWi is engaged in a single attempt to complete a phase, it
never removes a configuration id from [Link]. In particular, if process i learns during
a phase that a configuration id in [Link](k) has been garbage-collected, it does not
remove it from [Link], but continues to include it in conducting the phase.

182

N. Lynch and A.A. Shvartsman

Output send(W, v, t, cm, pns, pnr)i,j

Precondition:
failed
status = active
j world
W, v, t, cm, pns, pnr =
world, value, tag, cmap, pnum1 , pnum2 (j)
Effect:
none
Input recv(W, v, t, cm, pns, pnr)j,i
Effect:
if failed then
if status = idle then
status active
world world W
if t > tag then (value, tag) (v, t)
cmap update(cmap, cm)
pnum2 (j) max(pnum2 (j), pns)
if [Link] {query, prop} and pnr [Link] then
[Link] extend([Link], truncate(cm))
if [Link] Truncated then
[Link] [Link] {j}
else
[Link]
[Link] truncate(cmap)
if [Link] {query, prop} and pnr [Link] then
[Link] [Link] {j}
Input new-cong(c, k)i
Effect:
if failed then
if status = idle then
cmap(k) update(cmap(k), c)

Internal query-xi
Precondition:
failed
status = active
[Link] {read, write}
[Link] = query
k N, c C : ([Link](k) = c)
(R read-quorums(c) :
R [Link])
Effect:
if [Link] = read then
[Link] value
else
value [Link]
tag [Link] + 1, i
pnum1 pnum1 + 1
[Link] pnum1
[Link] prop
[Link] truncate(cmap)
[Link]
Internal prop-xi
Precondition:
failed
status = active
[Link] {read, write}
[Link] = prop
k N, c C : ([Link](k) = c)
(W write-quorums(c) :
W [Link])
Effect:
[Link] = done

Input readi
Effect:
if failed then
if status = idle then
pnum1 pnum1 + 1
[Link], [Link], [Link], [Link], [Link]
pnum1 , read, query, truncate(cmap), 

Output read-ack(v)i
Precondition:
failed
status = active
[Link] = read
[Link] = done
v = [Link]
Effect:
[Link] = idle

Input write(v)i
Effect:
if failed then
if status = idle then
pnum1 pnum1 + 1
[Link], [Link], [Link], [Link], [Link],
[Link]
pnum1 , write, query, truncate(cmap), , v

Output write-acki
Precondition:
failed
status = active
[Link] = write
[Link] = done
Effect:
[Link] = idle

Fig. 5. RWi : Read/write transitions

The query phase of a read or write operation terminates when a query fixed point is
reached. This happens when RWi determines that it has received recent responses from
some read-quorum of each configuration in its current [Link]. Then t, defined to be
RWi s tag at the query fixed point, is at least as great as the tag value that each process
in each of these read-quorums had at the start of the query phase.
If the operation is a read, then RWi determines at this point that its current value
is the value to be returned to its client. However, before returning, RWi performs the

RAMBO: A Reconfigurable Atomic Memory Service for Dynamic Networks

183

propagation phase, whose purpose is to make sure that enough RW processes have
acquired tags that are at least t. Again, the information is propagated in the background,
and [Link] is managed as described above. The propagation phase ends once a propagation fixed point is reached, when RWi has received recent responses from some
write-quorum of each configuration in the current [Link]. When this occurs, the tag
of each process in each of these write-quorums is at least t.
Processing for a write operation starting with write(v)i is similar to that for a read.
The query phase is conducted exactly as for a read, but processing after the query fixed
point is different: Suppose t, process is tag at the query fixed point, is of the form (n, j).
Then RWi defines the tag for its write operation to be the pair (n + 1, i) and sets its tag
to (n + 1, i) and its value to the new value v. Then it performs its propagation phase.
Now the purpose of the propagation phase is to ensure that enough processes acquire
tags that are at least as great as (n + 1, i). The propagation phase is conducted exactly
as for a read.

Internal gc(k)i
Precondition:
failed
status = active
[Link] = idle
cmap(k) C
cmap(k + 1) C
k = 0 or cmap(k 1) =
Effect:
pnum1 pnum1 + 1
[Link], [Link], [Link], [Link] 
pnum1 , query, , k
Internal gc-query-x(k)i
Precondition:
failed
status = active
[Link] = query
[Link] = k
R read-quorums(cmap(k)) :
W write-quorums(cmap(k)) : R W [Link]
Effect:
pnum1 pnum1 + 1
[Link], [Link], [Link] pnum1 , prop, 

Internal gc-prop-x(k)i
Precondition:
failed
status = active
[Link] = prop
[Link] = k
W write-quorums(cmap(k + 1)) :
W [Link]
Effect:
cmap(k)
Internal gc-ack(k)i
Precondition:
failed
status = active
[Link] = k
cmap(k) =
Effect:
[Link] = idle

Fig. 6. RWi : Garbage-collection transitions

New configurations and garbage collection. When RWi learns about a new configuration id via a new-cong input action, it simply records it in its cmap. From time to
time, configuration ids get garbage-collected at i, in numerical order. The configuration
ids used in performing query and propagation phases of reads and writes are those in
truncate(cmap), that is, all configurations that have not been garbage-collected and
that appear before the first .
There are two situations in which RWi may garbage-collect the configuration id
in cmap(k). First, RWi can garbage-collect cmap(k) if it learns that another process
has already garbage-collected it. This happens when RWi receives a message in which
cm(k) = . Second, RWi may acquire enough information to garbage-collect con-

184

N. Lynch and A.A. Shvartsman

figuration k on its own. RWi accomplishes this by performing a a two-phase garbagecollection operation, with a structure similar to the read and write operations. RWi may
initiate garbage-collection of configuration k when its cmap(k) and cmap(k + 1) are
both in C, and when any configurations with indices smaller than k 1 have already
been garbage-collected. Garbage-collection may proceed concurrently with a read or
write operation at the same node.
In the query phase of a garbage-collection operation, RWi communicates with a
read-quorum and a write-quorum of configuration k. The query phase accomplishes two
tasks: First, RWi ensures that all the processes in the read-quorum and write-quorum
learn about configurations k and k + 1, and also learn that all configurations smaller than
k have been garbage-collected. If such a process, j, is contacted afterwards by someone
who is using configuration k, j can tell that process about configuration k + 1. Second,
in the query phase, RWi collects tag and value information from the read-quorum and
write-quorum. This ensures that, by the end of the query phase, RWi s tag, t, is at least as
great as the tag that each of the quorum members had when it sent a message to RWi for
the query phase. In the propagation phase, RWi ensures that all the processes in a writequorum of configuration k + 1 have acquired tags that are at least t. Note that, unlike a
read or write operation, a garbage-collection for k uses only two configurationsk in
the query phase and k + 1 in the propagation phase.
At any time while RWi is garbage-collecting configuration k, it may discover that
someone has already garbage-collected k; it discovers this by observing that cmap(k) =
. When this happens, RWi may simply terminate its garbage-collection.
5.3 The Complete Algorithm
We assume point to point channels Channel i,j , one for each i, j I (including i = j).
Channel i,j is accessed using send(m)i,j input actions, by which a sender at location i
submits message m to the channel, and recv(m)i,j output actions, by which a receiver at
location j receives m. Channels may lose and reorder messages, but cannot manufacture
new messages or duplicate messages. Formally, we model Channel i,j as a multiset,
where A send(m)i,j input action adds one copy of m to the multiset and A recv(m)i,j
output removes one copy of m. A lose input action allows any sub-multiset of messages
to be removed.
The complete implementation, which we call S, is the composition of the Joiner i ,
RWi , and Channel i,j automata, and any automaton whose traces satisfy the Recon
safety specification, with all actions that are not external actions of Rambo hidden.

Safety Proof

We show that S satisfies the safety guarantees of Rambo, as given in Section 3, assuming
the environment safety assumptions. An operation can be of type read, write, or garbagecollection. An operations is uniquely identified by its starting event: readi , write(v)i , or
gc()i event.
We introduce the following history variables: (1) For every k N: c(k) C. This
is set when the first new-cong(, k) occurs, to the configuration id that appears as the

RAMBO: A Reconfigurable Atomic Memory Service for Dynamic Networks

185

first parameter of this action. (2) For every operation : tag() T . This is set just after
s query-x or gc-query-x event, to the tag of the process running . (3) For every
read or write operation : (a) query-cmap(), a CMap. This is set in the query-x step
of , to the value of [Link] in the pre-state. (b) prop-cmap(), a CMap. This is set
in the prop-x step of , to the value of [Link] in the pre-state.
For any read or write operation , we designate the following events:
(1) query-phase-start(). This is defined in the query-x step of , to be the unique
earlier event at which the collection of query results was started and not subsequently
restarted (that is, [Link] is set to in the effects of the corresponding step, and [Link]
is not later reset to following that event and prior to the query-x step). This is either
a read, write, or recv event. (2) prop-phase-start(). This is defined in the prop-x step
of , to be the unique earlier event at which the collection of propagation results was
started and not subsequently restarted. This is either a query-x or recv event.
Now we present several lemmas describing information flow between operations.
All are stated for a generic execution satisfying the environment assumptions. The first
lemma describes information flow between garbage-collection operations. We say that a
gc-prop-x(k)i event is initial if it is the first gc-prop-x(k) event in , and a garbagecollection operation is initial if its gc-prop-x event is initial. The algorithm ensures
that garbage-collection of successive configurations is sequential, in fact, for each k, the
initial gc-prop-x(k) event precedes any attempt to garbage-collect k + 1. Sequential
garbage-collection implies that tags of garbage-collection operations are monotone with
respect to the configuration indices:
Lemma 1. Suppose k and  are garbage-collection operations for k and , respectively, where k  and k is initial. Suppose a gc-query-x() event for  occurs in .
Then tag(k ) tag( ).
Proof. By induction on , for fixed k. For the inductive step, assume that  k + 1 and
the result is true for  1. A write-quorum of c() is used in the propagation phase of
1 and a read-quorum of c() is used in the query phase of  . The quorum intersection
property for c() guarantees propagation of tag information.

The following lemma describes situations in which certain configurations must appear in
the query-cmap of a read or write operation . First, if no garbage-collection operation
for k completes before the query-phase-start event of , then some configuration with
index k must be included in query-cmap(). Second, if some garbage-collection for
k completes before the query-phase-start event of , then some configuration with index
k + 1 must be included in the query-cmap().
Lemma 2. Let be a read or write operation whose query-x event occurs
in . (1) If no gc-prop-x(k) event precedes the query-phase-start() event, then
query-cmap()() C for some  k. (2) If some gc-prop-x(k) event precedes
the query-phase-start() event, then query-cmap()() C for some  k + 1.
The next lemma describes propagation of tag information from a garbage-collection
operation to a following read or write operation.

186

N. Lynch and A.A. Shvartsman

Lemma 3. Let be an initial garbage-collection operation for k. Let be a read or

write operation whose query-x event occurs in . Suppose that the gc-prop-x(k) event
of precedes the query-phase-start() event. Then tag() tag(), and if is a write
operation then tag() < tag().
The next two lemmas describe relationships between reads and writes that execute sequentially. The first lemma says that the smallest configuration index used in the propagation phase of the first operation is less than or equal to the largest index used in
the query phase of the second operation. In other words, the second operations query
phase cannot use only configurations with indices that are less than any used in the first
operations propagation phase.
Lemma 4. Assume 1 and 2 are two read or write operations such that the prop-x
event of 1 precedes the query-phase-start(2 ) event in .
Then min({ : prop-cmap(1 )() C}) max({ : query-cmap(2 )() C}).
Proof. Suppose not. Let k = max({ : query-cmap(2 )() C}). Then
some gc-prop-x(k) event occurs before the prop-x of 1 , and so before the
query-phase-start(2 ) event. Lemma 2, Part 2, then implies that query-cmap(2 )()
C for some  k + 1, which contradicts the choice of k.

The second lemma describes propagation of tag information between sequential reads
and writes.
Lemma 5. Suppose 1 and 2 are two read or write operations, such that the prop-x
event of 1 precedes the query-phase-start(2 ) event in . Then tag(1 ) tag(2 ),
and if 2 is a write then tag(1 ) < tag(2 ).
Proof. Let i1 and i2 be the processes that run operations 1 and 2 , respectively. Let
cm 1 = prop-cmap(1 ) and cm 2 = query-cmap(2 ). If there exists k such that
cm 1 (k) C and cm 2 (k) C, then the quorum intersection property for configuration k implies the conclusions of the lemma. So we assume that no such k exists.
Lemma 4 implies that min({ : cm 1 () C}) max({ : cm 2 () C}). Since the
set of indices used in each phase consists of consecutive integers and the intervals have
no indices in common, it follows that k1 < k2 , where k1 = max({ : cm 1 () C})
and k2 = min({ : cm 2 () C}).
Since, for every k k2 1, [Link](2 )(k)
/ C, Lemma 2, Part 1, implies
that, for every such k, a gc-prop-x(k) event occurs before the query-phase-start(2 )
event. For each such k, define k to be the initial garbage-collection operation for k.
The propagation phase of 1 accesses a write-quorum of c(k1 ), and the query phase
of k1 accesses a read-quorum of c(k1 ). By the quorum intersection property, there is
some j in the intersection of these quorums. Let message m be the message sent from j
to i1 in the propagation phase of 1 , and let m be the message sent from j to the process
running k1 in its query phase. We claim that j sends m before it sends m . For if not,
then information about configuration k1 + 1 would be conveyed by j to i1 , who would
include it in cm 1 , contradicting the choice of k1 . Since j sends m before it sends m , j
conveys tag information from 1 to k1 , ensuring that tag(1 ) tag(k1 ).

RAMBO: A Reconfigurable Atomic Memory Service for Dynamic Networks

187

Since k1 k2 1, Lemma 1 implies that tag(k1 ) tag(k2 1 ). Lemma 3 implies

that tag(k2 1 ) tag(2 ), and if 2 is a write then tag(k2 1 ) < tag(2 ). Combining
all the inequalities then yields both conclusions.

Theorem 1. Let be a trace of S. If satisfy the Rambo environment assumptions,

then satisfies the Rambo service guarantees (well-formedness and atomicity).
Proof. Let be a trace of S that satisfies the Rambo environment assumptions. We
argue that satisfies the Rambo service guarantees. The proof that satisfies the Rambo
well-formedness guarantees is straightforward from the code. To show that satisfies
atomicity (as defined in Section 3), assume that all read and write operations complete
in . Let be an execution of S that satisfies the environment assumptions and whose
trace is . Define a partial order on read and write operations in : totally order the
writes in order of their tags, and order each read with respect to all the writes so that a
read with tag t is ordered after all writes with tags t and before all writes with tags
> t. Then we claim that satisfies the four conditions in the definition of atomicity.
The interesting condition is Condition 2; the other three are straightforward.
For Condition 2, suppose for the sake of contradiction that 1 and 2 are read or write
operations, completes before 2 starts, and 2 1 . If 2 is a write operation, then
since 1 completes before 2 starts, Lemma 5 implies that tag(2 ) > tag(1 ). But the
fact that 2 1 implies that tag(2 ) tag(1 ), yielding a contradiction. On the other
hand, if 2 is a read operation, then since 1 completes before 2 starts, Lemma 5 implies
that tag(2 ) tag(1 ). But the fact that 2 1 implies that tag(2 ) < tag(1 ), again
yielding a contradiction.

Implementation of the Reconfiguration Service

The Recon algorithm is considerably simpler than the RW algorithm. It consists of

a Recon i automaton for each location i, which interacts with a collection of global
consensus services Cons(k, c), one for each k 1 and each c C, and with a pointto-point communication service.
Cons(k, c) accepts inputs from members of configuration c, which it assumes to
be the k 1st configuration. These inputs are of the form init(c )k,c,i , where c is a
proposed new configuration. The configuration that Cons(k, c) decides upon (using
decide(c )k,c,i outputs) is deemed to be the k th configuration. The validity property of
consensus implies that this decision is one of the proposed configurations.
Recon i is activated by a join(recon)i action, which is an output of Joiner i . Recon i
accepts reconfiguration requests from clients, and initiates consensus to help determine new configurations. It records the new configurations that the consensus services
determine. Recon i also informs RWi about newly-determined configurations, and disseminates information about newly-determined configurations to the members of those
configurations. It returns acknowledgments and configuration reports to its client.
We implement Cons(k, c) using the Paxos consensus algorithm [17], as described
formally in [8]. Our complete implementation of Recon, Recon impl , consists of the

188

N. Lynch and A.A. Shvartsman

Recon i automata, channels connecting all the Recon i automata, and the implementations
of the Cons services. We use the same kinds of channels as for Rambo: point-to-point
channels that may lose and reorder messages, but not manufacture new messages or
duplicate messages. The complete Rambo system (for a particular object) consists of
Joiner , RW, and channel automata as described in Section 5, plus Recon impl . We call
the complete Rambo system S  .

Conditional Performance Analysis: Latency Bounds

We prove latency bounds for the full system S  . To handle timing, we convert all the
I/O automata to general timed automata (GTAs) as defined in [18], by allowing arbitrary
amounts of time to pass in any state. Fix d > 0, the normal message delay, and > 0.
Rambo allows sending of messages at arbitrary times. For the purpose of latency
analysis, we restrict Rambos sending pattern: We assume that each automaton has a local
real-valued clock, and sends messages at the first possible time and at regular intervals
of d thereafter, as measured on the local clock. Also, non-send locally controlled events
occur just once, within time 0 on the local clock.
Our results also require restrictions on timing and failure behavior: We define an
admissible timed execution to be normal provided that all local clocks progress at rate
exactly 1, all messages that are sent are delivered within time d, and timing and failure
behavior for all consensus services is normal, as defined in [8].2
Next, we define a reliability property for configurations. In general, in quorum-based
systems, operations that use quorums are guaranteed to terminate only if some quorums
do not fail. Because we use many configurations, we attempt to take into account which
configurations might be in use. We say that k is installed in a timed execution provided
that either k = 0 or there exists c C such that (1) some init()k,c, event occurs, and
(2) for every i members(c), either decide()k,c,i or faili occurs. (Thus, configuration
k 1 is c, and every non-failed member of c has learned about configuration k.) We
say that is e-configuration-viable, e 0, provided that for every c and k such that
some rec-cmap(k) = c in some state in , there exist R read-quorums(c) and
W write-quorums(c) such that either (1) no process in R W ever fails in , or
(2) k + 1 is installed in a finite prefix  of and no process in R W fails in
by time time( ) + e. (Quorums remain non-failed for at least time e after the next
configuration is installed.)
The e-configuration-viability assumption is useful in situations where a configuration
is no longer needed for performing operations after time e after the next configuration is
installed. This condition holds in Rambo executions in which certain timing assumptions
hold; the strength of those assumptions determines the value of e that must be considered.
We believe that such an assumption is reasonable for a reconfigurable system, because
it can be reconfigured when quorums appear to be in danger of failing.
We prove a bound of 2d on the time to join, and a bound of 11d + for the time
for reconfiguration, based on a bound of 10d + for consensus. We also establish a
2

This means that all messages are delivered within time d, local processing time is 0, and
information is gossiped at intervals of d.

RAMBO: A Reconfigurable Atomic Memory Service for Dynamic Networks

189

situation in which a system is guaranteed to produce a positive response to a reconfiguration request. We prove a bound of 4d on the time for garbage-collection, assuming
that enough of the relevant processes remain non-failed. We prove a bound of 4d on
the latency for read and write operations in a quiescent situation, in which all joins
and configuration management events have stopped, and the configuration map of the
operations initiator includes the latest configuration and has value for all earlier configurations. More generally, we show that this bound holds even if this map contains
more than one configuration: since the configurations are used concurrently, the use of
multiple configurations does not slow the operation down.
We show that all participants succeed in exchanging information about configurations, within a short time: if i and j have joined at least time e ago and do not fail, then
any information that i has about configurations is conveyed to j within time 2d. Using
this result, we show that, if reconfiguration requests are spaced sufficiently far apart, and
if quorums of configurations remain alive for sufficiently long, then garbage collection
keeps up with reconfiguration.
The main latency theorem bounds the time for read and write operations in the
steady-state case, where reconfigurations do not stop, but are spaced sufficiently far
apart. Fix e 0.
Theorem 2. Let be a normal admissible timed execution of S  such that:
(1) If a recon(, c)i event occurs at time t then for every j members(c),
join-ack(rambo)j occurs by time t e. (2) If join-ack(rambo)i and join-ack(rambo)j
both occur by time t, and neither i nor j fails by time t+e, then by time t+e, i world j .
(3) For any recon(c, )i that occurs in , the time since the corresponding report(c)i
event is 12d + . (4) satisfies 11d-configuration-viability. (5) contains decide
events for infinitely many configurations.
Suppose that a readi (resp., write()i ) event occurs at time t, and join-acki occurs strictly
before time t (e + 8d). Then the corresponding read-acki (resp., write-ack()i ) event
occurs by time t + 8d.
Proof. The various spacing properties and bounds on time to disseminate information
imply that each phase of the read or write completes with at most one restart for learning
about a new configuration. Therefore, each phase takes time at most 4d, for a total of 8d.

In the full paper we also present latency results analogous to those described above,
for executions that have normal timing and failure characteristics after some point in the
execution. These results are similar to the previous results, but include dependence on
the time when normal behavior begins.

Future Work

In future work, we plan to implement the complete Rambo algorithm in LAN, WAN, and
mobile settings. We will extend our performance analysis and compare it with empirical
results. We will investigate ways of increasing the concurrency of garbage-collection and
of reducing the amount of communication. Finally, this work leaves open the question
of how to choose good configurations, for various kinds of platforms.

190

N. Lynch and A.A. Shvartsman

References
1. Communications of the ACM, special section on group communications, vol. 39, no. 4, 1996.
2. Y. Amir, D. Dolev, P. Melliar-Smith and L. Moser, Robust and Efficient Replication Using
Group Communication Tech. Rep. 94-20, Dept. of Computer Science, Hebrew Univ., 1994.
3. H. Attiya, A. Bar-Noy and D. Dolev, Sharing Memory Robustly in Message Passing Systems", J. of the ACM, vol. 42, no. 1, pp. 124-142, 1996.
4. P.A. Bernstein, V. Hadzilacos and N. Goodman, Concurrency Control and Recovery in
Database Systems, Addison-Wesley, Reading, MA, 1987.
5. F. Cristian and F. Schmuck, Agreeing on Processor Group Membership in Asynchronous
Distributed Systems, TR. CSE95-428, Dept. of Comp. Sci., Univ. of California San Diego.
6. S.B. Davidson, H. Garcia-Molina and D. Skeen, Consistency in Partitioned Networks", ACM
Computing Surveys, vol. 15, no. 3, pp. 341-370, 1985.
7. R. De Prisco, A. Fekete, N. Lynch, A. Shvartsman, A Dynamic Primary Configuration Group
Communication Service, 13th Int-l Conference of Distributed Computing, 1999.
8. Roberto De Prisco, Nancy Lynch, Alex Shvartsman, Nicole Immorlica and Toh Ne Win A
Formal Treatment of Lamports Paxos Algorithm, manuscript, 2002.
9. C. Dwork, N. A. Lynch, L. J. Stockmeyer, Consensus in the presence of partial synchrony,
J. of ACM, 35(2), pp. 288-323, 1988.
10. B. Englert and A.A. Shvartsman, Graceful Quorum Reconfiguration in a Robust Emulation
of Shared Memory, in Proc. International Conference on Distributed Computer Systems
(ICDCS2000), pp. 454-463, 2000.
11. A. Fekete, N. Lynch and A. Shvartsman Specifying and using a partitionable group communication service, ACM Trans. on Computer Systems, vol. 19, no. 2, pp. 171216, 2001.
12. H. Garcia-Molina and D. Barbara, How to Assign Votes in a Distributed System," J. of the
ACM, vol. 32, no. 4, pp. 841-860, 1985.
13. D.K. Gifford, Weighted Voting for Replicated Data", in Proc. of 7th ACM Symp. on Oper.
Sys. Princ., pp. 150-162, 1979.
14. M.P. Herlihy, Dynamic Quorum Adjustment for Partitioned Data", ACM Trans. on Database
Systems, 12(2), pp. 170-194, 1987.
15. R. Guerraoui and A. Schiper, Consensus Service: A Modular Approach For Building FaultTolerant Agreement Protocols in Distributed Systems", Proc. of the 26th International Symposium on Fault-Tolerant Computing (FTCS-26), pp. 168-177, 1996.
16. I. Keidar and D. Dolev, Efficient Message Ordering in Dynamic Networks, in Proc. of 15th
Annual ACM Symp. on Princ. of Distr. Comput., pp. 68-76, 1996.
17. Leslie Lamport, "The Part-Time Parliament", ACM Transactions on Computer Systems, 16(2)
133-169, 1998.
18. N.A. Lynch, Distributed Algorithms, Morgan Kaufmann Publishers, San Mateo, CA, 1996.
19. N. Lynch and A. Shvartsman. Robust emulation of shared memory using dynamic quorumacknowledged broadcasts. 27th Int-l Symp. on Fault-Tolerant Comp., pp. 272-281, 1997.
20. Nancy Lynch and Alex Shvartsman. RAMBO: A Reconfigurable Atomic Memory Service
for Dynamic Networks. MIT-LCS-TR-856, 2002
21. E. Upfal and A. Wigderson, How to share memory in a distributed system, Journal of the
ACM, 34(1):116127, 1987.