<figure><img src="image_1.png" /></figure>

laboria

laboratoire de recherche
en informatique
et automatique (145)

SYNCHRONIZATION
FOR DISTRIBUTED SYSTEMS
USING A SINGLE
BROADCAST CHANNEL

Bibliothèque
IRIA

Jean-Serge BANINO

Claude KAISER

Hubert ZIMMERMANN

Rapport de Recherche N° 353

Mai 1979

25 p.

Institut de Recherche
d'Informatique
e+ d'Automatique

Domaine de Voluceau
Rocquencourt

B. P. 105 78150 - Le Chesnay
France

Tél.: 954 90 20
SYNCHRONIZATION FOR DISTRIBUTED SYSTEMS
USING A SINGLE BROADCAST CHANNEL

Jean Serge Banino*, Claude Kaiser**, Hubert Zimmermann*

Résumé :

Nous décrivons une technique de synchronisation pour des systèmes répartis utilisant un canal unique de communication par diffusion. Cette technique repose sur le type de transmission (diffusion) et sur l'unicité du canal. Comme exemple nous donnons d'abord un algorithme de maintien d'un nombre quelconque de copies d'un même objet. L'algorithme résiste à certaines erreurs ou pannes. Ensuite, nous montrons que cette technique permet l'implantation de sémaphores répartis, de compteurs d'événement répartis et de séquenceurs répartis. Enfin, nous décrivons comme exemple de solution simple à un problème général, la réalisation d'un schéma « producteurs multiples-consommateurs multiples » dans un système réparti.

Abstract :

We describe a synchronization technique for distributed systems using a single broadcast communication channel. This technique takes advantage of the broadcast type of transmission and uniqueness of the channel. As example we give first an algorithm which maintains an indifferent number of copies of a same object and which resists to some kinds of faults or failures. Then we show that this technique permits the implementation of distributed semaphores, distributed eventcounts and distributed sequencers. Finally, we give, as an example of a simple solution to a general problem, the implementation of a multiple-producers, multiple-consumers schema in a distributed system.

<figure><img src="image_2.png" /></figure>

<footnote>* IRIA BP 105 - 78150 Le Chesnay (France)</footnote>

<footnote>** CNAM 292, rue Saint-Martin, 75141 Paris Cedex 03 (France)</footnote>
# 1. INTRODUCTION

The usual definition of a distributed system is that processors do not access any shared memory and have no common clock (LE LANN 77). In systems (in general), the problem of synchronization can be expressed as the necessity of ordering some processes at a given point of their execution.

In a centralized system it is possible to realize this order by means of algorithms and shared memory (e.g. DEKKER's variables (DIJKSTRA 67) or eventcounts and sequencers (REED 79)) but more often the system offers to the programmer mechanisms (locks, test and set, semaphores, monitors, etc...) based in the last resort on technology and more precisely on a hardware mechanism offering access to a memory cell on the basis of mutual exclusion.

For distributed systems' designers, the same problems are to be solved. One way is to build algorithms in order to realize this ordering in some particular cases - e.g. to maintain in a consistent state multiple copies of information (MULLERY 75), (THOMAS 77), (ELLIS 77), (SEGUIN 77).

Another way is to imagine general techniques which should play, in the building of distributed systems, a basic role similar to the one played by the hardware mechanisms in centralized systems - e.g. token and virtual ring (LE LANN 77), logical time (LAMPORT 78), circulating sequencers (LE LANN 78).

In the first part of this paper we present an algorithm which maintains multiple copies of information in a consistent state, in the case of a distributed system communicating via a broadcast network with a single channel.

In the second part we isolate the basic mechanism used for synchronization in this algorithm, and show by means of examples that it offers a generality sufficient for solving most of the synchronization problems arising in such a distributed system.

.../...
# 2. AN ALGORITHM FOR MAINTAINING A MULTIPLE-COPIED INFORMATION IN A CONSISTENT STATE

## 2.1. Hypothesis

### 2.1.1. Hypothesis on the communication network

- The distributed system is made of computers without shared memory connected together by means of a communication network.

- The communication network is a broadcast network with a single channel (radio, bus, coax, etc...); the channel is allocated for the transmission of a single message at a time (e.g. by contention or by a controller).

- Every computer of the system is connected to the network via a "communicator" which sends the messages, recognizes addresses of messages passing through the channel, and catches messages sent to itself. If the channel is controlled in a decentralized way, the algorithm for its allocation is distributed in the communicators.

- Every message passing through the channel is headed by an address. An address is either the address of a communicator or a broadcast address.

- Every communicator recognizes its own address and at least one broadcast address. Therefore it catches messages headed by these addresses.

- Messages caught by a communicator are delivered to the computer connected to it (which we call the local host) in the FIFO order.

- When a communicator sends a message headed by a broadcast address, which it is programmed to recognize, it catches this message.

In the same way a communicator can send a message to itself through the channel.

### 2.1.2. Hypothesis for the algorithm

- On every host concerned, a server process handles the local copy of the duplicated object.

- Every server is addressed by a global and unique name in the system. The set of servers for all copies of the object is addressed by means of a generic address (i.e. a broadcast address).

.../...
- User processes access to the multiple-copied object by primitives (READ, WRITE,...)

Activation of such a primitive is realized by sending a message, through the network, to one of the servers or to the community of servers.

If the primitive needs a result, the process activating it is blocked until it receives a response message.

- Messages received by a server are put in a queue used in the FIFO order.

- Finally every server executes the same algorithm.

## 2.2. Principle of the algorithm

A simple way to maintain in a consistent state the multiple copies of one object is to integrate all the modifications on each copy in the same order. Modifications of the object are the fact of primitives. These primitives are activated by independant processes. Therefore we have :

- to synchronize these processes i.e. to realize an ordering of their actions related to the object.

- to respect this order when reporting modifications on each copy.

We meet these two goals by the use of peculiarities of the communication network.

First the unique channel is allocated to the transmission of one message at a time.

Therefore messages are ordered, and the order is the one of the channel allocation algorithm.

We can use this ordering, offered by the network, at the system level if every primitive modifying the duplicated object is expressed through the network in a single message (or in several messages chained to the first of them).

And then, we use for these primitives broadcast messages addressed to the community of servers. So, all the messages are transmitted to all the servers and the ordering of messages is the same in every server queue.

By working in this way we maintain the consistency of the copies. Each of them integrates the modifications at the speed of its server, but in the same order.

.../....
In addition we offer to the user the ability of making series of
operations on the object in an indivisible way - what we call a
transaction.

## 2.3. Description of the algorithm

### 2.3.1. Primitives offered to the user processes

#### Notation

Primitive activations are transmitted in messages. We suppose that user
processes and server processes are addressed by global names.

We note $M_i^j$ for a message M transmitted from a process named i to a process named j.

If b is the generic name of servers, a message M emitted from a process i to the set of servers is noted $M_i^b$.

For simplification we don't care about the mapping process name-host
address and speak only of process names.

#### Primitive types

**READ** A process i which wants to read the object sends, through the network, a message $READ_i^j$ (parameters) to one (j) among the servers.

**WRITE** A process i which wants to write on the object sends, through the network, a message $WRITE_i^b$ (parameters) to the set of servers.

The writing will be made on every copy.

**OPEN TRANSACTION** A process i which wants to make an indivisible sequence of readings and writings starts this sequence by sending, through the network, a message $OPEN TRANSACTION_i^b$ and it ends this sequence by sending a message **CLOSE TRANSACTION_i^b**.
So a transaction looks like the following sequence of messages :

OPEN TRANSACTION<sub>i</sub><sup>b</sup>
READ<sub>i</sub><sup>j</sup>
WRITE<sub>i</sub><sup>b</sup>
READ<sub>i</sub><sup>j</sup>
WRITE<sub>i</sub><sup>b</sup>
...
...
CLOSE TRANSACTION<sub>i</sub><sup>b</sup>

(In this example and in the others we don't take care of the parameters).

### 2.3.2. Activity of servers

- Every reading is made on a single copy. To ensure consistency of accesses, local reads must obey the same rules as distant reads. (e.g. during a transaction opened by process i, reads from process j must be delayed). This can be achieved by having local reads follow the same path as distant reads, including (local) transmission on the network.

- Every writing is requested by sending a broadcast message. Every server keeps a counter. The counter value is incremented by one each time the server realizes a writing (i.e. it executes a WRITE message taken in its queue). The counter value is called the "version number" of the copy.

All the version numbers are initialized at the same value when starting the system (§ 2.6).

- If the algorithm is respected and if the communication network is perfect all the version numbers of the copies are equal after all the requests have been treated.

A transaction, as a simple primitive, must be realized in parallelism on every copy. As soon as a server has treated an OPEN TRANSACTION<sub>i</sub><sup>b</sup> message it processes those messages following in the queue whose indices are i but puts in a secondary queue to be processed later all those whose indices are different from i.

- When a message CLOSE TRANSACTION<sub>i</sub><sup>b</sup> is treated, the server loads its secondary queue at the head of its principal queue.
## Comment

The opening of a transaction gives to the user the ability to work on the object on a mutual exclusion basis.

It doesn't forbid the other processes from emitting messages for requests. These messages will be put in a secondary queue for each server and their treatment will be delayed.

## 2.4. Fault-tolerance: mutual inspection principle

Every system can be suspected of failing. So we must, from the design, forcast and prevent the greatest number of faults. For this goal a distributed system can take advantage of the redundancy of processors.

It is what we call the mutual inspection principle : every server has the ability to compare, from time to time, the version number of its copy to the ones of others copies.

The comparison is realized by sending a message of type `VERSION NUMBER`. This message can be sent either by a user process to one or all of the servers, or by a server to one or all of the other servers. As a response the emitter of such a message receives one or several messages containing couples (process name, version number).

Of course, the sending of a message `VERSION NUMBER`<sub>i</sub><sup>j</sup> or `VERSION NUMBER`<sub>i</sub><sup>b</sup>, the sending of responses, and all the exchanges of messages related to this comparison phase, must hold in a transaction opened by process i.

If no message has been lost and if there was not a temporary failure of a processor, all the response messages will carry the same version number.

In the other cases some version numbers can be lower than the others.

This primitive `VERSION NUMBER` allows a server to compare its version number to the ones of others servers, by example. It can also detect which copies are "late".

Finally it allows a process to have an estimation of the number and of the identities of active servers at that time.

This kind of message can also be used by a user process which wants to choose a "good" copy and read in secure conditions.
When a process i, after sending a message VERSION NUMBER<sub>i</sub><sup>b</sup>, receives responses with different version numbers it executes a simple algorithm to determine which is the good copy (the greatest number or the majority number....).

The determination is as secure as the number of responses is large.

(We can remark that this number depends on the number of active servers, on the length of servers queues and on the time-out armed by the emitter of the VERSION NUMBER<sub>i</sub><sup>b</sup> message for the responses).

If a server has sent a message VERSION NUMBER and discovers that its copy is not "to-date" it must ask for a good copy from another server.

It sends it a message of type SEND COPY<sub>i</sub><sup>j</sup> (parameters) and receives as a response the good copy.

Finally, the initiator of such an inspection transaction may send, before closing the transaction, messages of type ALARM<sub>i</sub><sup>j</sup> to servers answering with bad version numbers.

A server receiving such a message must recover by itself a good copy.

It uses for that the self insertion procedure, as if it was a new server entering the game (§ 2.5).

To sum up an inspection transaction from a server i looks like the following sequence of messages :

OPEN TRANSACTION<sub>i</sub><sup>b</sup>
VERSION NUMBER<sub>i</sub><sup>b</sup>
{ Determination of the good version number }
SEND COPY<sub>i</sub><sup>j\*</sup>
{ updating the local version number }*
ALARM<sub>i</sub><sup>k\*</sup>
...
ALARM<sub>i</sub><sup>t\*</sup>
i
CLOSE TRANSACTION<sub>i</sub><sup>b</sup>

Notation : \* means eventual action

### Remark

When a server i sends a SEND-COPY<sub>i</sub><sup>j</sup> message a time out is armed in order to prevent the case of a failure from j in the course of transmission.
At the end of the time-out the same request is eventually sent
to another server k whose copy is also good.

To conclude with this mutual inspection principle, one can say that
on the one hand every process can verify the actuality and the
consistency of a copy before an important sequence of operations, by
example, and that on the other hand every server is periodically tested
and eventually signalled if it is "late".

This principle ensures an a-posteriori detection of some kinds of failures
(missing of messages by a server, temporary absence of a processor,
saturation of server queues etc...). But one cannot say that consistency of
copies is guaranteed at 100 % at any time.

The degree of security is as high as the number of inspection transactions
accomplished in the system is large. The user of the system must control
the use of these transactions and must evaluate the rate between security
and the cost to pay for it.

As an example at the user level we can imagine that in a seat reservation
system for airplanes the user wants to reserve three places for a flight
The server he addresses answers him : a hundred seats are free. The
transaction is quickly finished without further checks. But if the server
answers : four places are free it can be more satisfying to pay the price
for an inspection transaction to verify that the server has not, by chance,
missed the message related to the reservation of a numerous family !

2.5. Dynamic self-insertion procedure

A new server i which wants to dynamically enter the game must execute
the following sequence, after its own initialization on the host :

OPEN TRANSACTION b i
VERSION NUMBER b i
{ Determination of a server i having a good copy }
SEND COPY j i
{ Updating the local version number }
ALARM i to the servers "late"*
CLOSE TRANSACTION b i

*means eventual action

.../....
## Remarks

- In the course of a transaction opened by a process i, a server j which receives a message `SEND COPY<sup>j</sup><sub>i</sub>` will treat it and will answer though the transaction is opened; indeed this message is indexed with i, the name of the initiator of the transaction.

- In the course of an insertion transaction, all the servers are blocked, after having treated the `OPEN TRANSACTION<sup>b</sup><sub>i</sub>` messages, on an identical copy. They will send the same version number in response to the `VERSION NUMBER<sup>b</sup><sub>i</sub>` message.

So when the server which inserts itself asks to a server j for a copy, its secondary queue is going to fill up with the same requests as the secondary queues of other servers, and when treating the `CLOSE TRANSACTION<sup>b</sup><sub>i</sub>` message the same secondary queues will be integrated in the principal ones and all the copies will be "in phase".

- We must take care of the following case: a new server i wants to enter and opens a transaction by an `OPEN TRANSACTION<sup>b</sup><sub>i</sub>` message, but at this time an other transaction is in progress.

This message will be queued on the secondary queue of the servers, except for server i which entered the game after the beginning of the transaction. This server, when waiting for the responses to its message `VERSION NUMBER<sup>b</sup><sub>i</sub>` for example, will receive, in its secondary queue, messages related to the transaction not yet finished.

Because of this, at the time the message `CLOSE TRANSACTION<sup>b</sup><sub>i</sub>` is treated by server i, this one must "clean" its secondary queue and load in the principal one only the content of the secondary queue following the last `CLOSE TRANSACTION<sup>b</sup><sub>k</sub>` message it contains.

## 2.6. System initialization

The system initialization is very simple in its principle. These are two possible solutions :

- First solution (time costly)

.../...
```
{Initialization of servers}
{Initialization of the first copy on a site i}
{Initialization of its version number to zero}
ALARM<sub>i</sub><sup>b</sup>
```

**Comment :**

All the other servers than i will enter the game by themselves.

- **Second solution (more economical)**

```
{Initialization of servers}
{Initialization of the first copy on site i}
{Initialization of its version number}
OPEN TRANSACTION<sub>i</sub><sup>b</sup>
{Send the copy by broadcast messages}
CLOSE TRANSACTION<sub>i</sub><sup>b</sup>
```

## 2.7. The case of a failure of an emitter in the course of a transaction

The failure of a process during an emission is an inconvenience only in the case where this process has opened a transaction and has not yet closed it.

To take care of this kind of failure one can adopt the following technique :

In the course of a transaction, every server works on a secondary copy.

In the same way it increments a secondary counter (version number) during the transaction. At the time of treating the `CLOSE TRANSACTION` message the modifications of the secondary copy are integrated in the primary one and the value of the secondary counter is reported in the original one.

In order to detect a failure from the initiator of the transaction, every server arms a time-out when it treats an `OPEN TRANSACTION` message.

In the same way the time-out is re-armed at every message received during the transaction. At the end of a time-out in a server i, this one :

- destroy its secondary counter (version number)

- destroy its secondary copy

- destroy its primary queue and replace it by the secondary one.
2.8. "Resumé" of the algorithm

<table><thead><tr><th>Message received by server j</th><th>Activity of server j when treating the message</th></tr></thead><tbody><tr><td>READ<sup>j</sup><br>WRITE<sup>b</sup><br>VERSION NUMBER<sup>j</sup><br>VERSION NUMBER<sup>b</sup></td><td>Read; send result to process i;<br>Write; Increment counter by one;<br>Send to process i a (server name, version number) message</td></tr><tr><td>OPEN TRANSACTION<sup>b</sup></td><td>Open secondary queue; create secondary copy; create secondary counter; treat every message sent by process i; put the others in secondary queue;</td></tr><tr><td>CLOSE TRANSACTION<sup>b</sup></td><td>Replace primary queue by secondary one (from the last CLOSE TRANSACTION message in the case of a self insertion transaction); Integrate secondary copy in primary one;<br>Replace primary counter by secondary one;<br>Send to server i copy and the version number;</td></tr><tr><td>SEND COPY<sup>j</sup><br>ALARM<sup>j</sup></td><td>Send to server i copy and the version number;<br>Destroy queues; self-insertion procedure;</td></tr></tbody></table>

.../...
## Remarks

- ALARM is a priority message.

It is put at the head of the primary queue when it arrives at a server.

- In the course of a transaction initiated by a process i, response messages to a VERSION NUMBER<sup>b</sup><sub>i</sub> message (of type : server name, version number) are put in the primary queue of process i, if this one is a server. In the same way a server i which sends a SEND COPY<sup>j</sup><sub>i</sub> receives the response messages in its primary queue.

## 2.9. Some advantages of the algorithm

- The algorithm can work whatever the number of copies may be. This fact makes its implementation easy.

- The mutual inspection principle enforces reliability of the system.

- The algorithm resists to some kinds of errors (loss of a message by a receiver, for example) because of the mutual inspection.

- It also resists to the failure of one or several processors and allows their re-start during the system is working. This fact enforces the availability of the system.

- In the same way the system is dynamically extensible : a new copy can enter the game at every time.

- The traffic of control messages is minimized. Particularly the problem of acknowledgements, difficult in a broadcast network, is avoided.

- Servers work in an asynchronous way, though they respect the same ordering of operations.

- An opened transaction doesn't prevent other users from emitting requests. They will be memorized and treated after the end of the transaction.

- The algorithm is completely decentralized.

- User processes use mutual inspection messages as they want, according to the cost they are ready to pay for security !

.../....
### 3. SYNCHRONIZATION

The above example showed that, in order to solve the synchronization problem, two peculiarities of the communication network could be used:

1. the allocation of the channel on the basis of mutual exclusion carries out the ordering of messages as an indivisible instruction "test-and-set" can do it in a shared-memory system.

2. the broadcasting of messages transmits this order on each site.

The use of these characteristics simplifies in a large extent the solutions to synchronization problems in a distributed system: it is well known that a great difficulty in such system is to order events and to offer a uniform view of this order in the system (LAMPORT 78).

In this paragraph we are going to show, on some examples, how to use the peculiarities of the communication network to build general tools of synchronization and therefore solve most of problems arising this field.

#### 3.1. Implementation of a distributed semaphore

We have a mechanism which maintains in a consistent state several copies of an object. Let us apply this mechanism to the variable representing the value of the semaphore.

**Principle:**

A copy of the variable E is maintained on each host by a server process. The server realizes the activation of P and V primitives, on the invocation of user processes. These activations are treated one by one. Servers also send and receive messages through the network (conventions are the same than in § 2).

A user process invokes execution of a P or V primitive to the local server. The user processes are distributed in the system. Therefore the processes blocked by a P primitive are queued locally. Thus the semaphore queue is partitionned.

To avoid inspection of all the queues during a V primitive a user process takes a number B when it is blocked.

This number will be used to order the activation of blocked processes. The variable B is duplicated and handled by the servers. In the same way, a counter A is duplicated and maintained by the servers to determine which blocked process must be activate when a V primitive occurs. Then we can write the algorithm of P and V primitives when realized by a server i on the semaphore S: (E is initialized to the value of the semaphore; A and B are initialized to zero).
## Primitive P (S)

### begin

Send message OPEN TRANSACTION<sub>i</sub><sup>b</sup> ;

Send message DECREMENT<sub>i</sub><sup>b</sup> (E) ;

Wait for this message ;

Treat the message ;

If E < 0 then

### begin

Send message INCREMENT<sub>i</sub><sup>b</sup> (B)

Wait for this message ;

Treat the message ;

Give to the user process the number B ;

Block this process ;

Put it in FIFO on the local queue of blocked processes

### end

Send message CLOSE TRANSACTION<sub>i</sub><sup>b</sup>

### end

## Primitive V (S)

### begin

Send message OPEN TRANSACTION<sub>i</sub><sup>b</sup> ;

Send message INCREMENT<sub>i</sub><sup>b</sup> (E) ;

Wait for this message ;

Treat the message ;

If E < 0 then

### begin

Send message INCREMENT<sub>i</sub><sup>b</sup> (A) ;

Wait for this message ;

Treat the message ;

Send message ACTIVE<sub>i</sub><sup>b</sup> (A) ;

### end

Send message CLOSE TRANSACTION<sub>i</sub><sup>b</sup>

### end
## Activity of a server when it receives messages

<table><thead><tr><th>Message</th><th>Activity of the local server i</th></tr></thead><tbody><tr><td>INCREMENT<sup>b</sup><sub>j</sub> (X)</td><td>execute X:= X+1</td></tr><tr><td>DECREMENT<sup>b</sup><sub>j</sub> (X)</td><td>execute X:=X-1</td></tr><tr><td>ACTIVE<sup>b</sup><sub>j</sub> (X)</td><td>Activate every process of the local queue whose number is lower than or equal to X.</td></tr></tbody></table>

## Comments

- In a semaphore one must realize in an indivisible way the test of its value and the resulting decisions.
That is the reason of the transactions.

- For the same reason the server waits for broadcast messages its sends (and more it is a way to know if there is not another transaction in course).

- If an `ACTIVE<sup>b</sup><sub>j</sub>` (X) message is lost, the next `ACTIVE<sup>b</sup><sub>k</sub>` (X+1) message will activate two processes instead of one.

- Because the blocked process queue is partitioned it must be handled according to a simple policy (FIFO in this example).

If we want to have a more general strategy for handling the queue we must maintain a duplicated image of the order of blocked processes.
That is the following example.

### 3.2. An other implementation of a distributed semaphore (which allows various strategies for handling the queue)

#### Principle

The servers maintain a list of the blocked processes names. This list is duplicated.

Then we have (if u is the name of user process requiring the P primitive)
## Primitive P (S)

### begin

Send message OPEN TRANSACTION<sub>i</sub><sup>b</sup> ;

Send message DECREMENT<sub>i</sub><sup>b</sup> (E) ;

Wait for this message ;

Treat this message ;

If E < 0 then

### begin

Send message PUT<sub>i</sub><sup>b</sup> (u) ;

Block user process u

### end

Send message CLOSE TRANSACTION<sub>i</sub><sup>b</sup>

### end

## Primitive V (S)

### begin

Send message OPEN TRANSACTION<sub>i</sub><sup>b</sup> ;

Send message INCREMENT<sub>i</sub><sup>b</sup> (E) ;

Wait for this message ;

Treat the message ;

If E < 0 then

### begin

Determine which process w must be activated ;

Send message GET<sub>i</sub><sup>b</sup> (w) ;

Send message ACTIVE<sub>i</sub><sup>b</sup> (w)

### end

Send message CLOSE TRANSACTION<sub>i</sub><sup>b</sup>

### end

## Activity of a server when it receives messages

<table><thead><tr><td>Message</td><td>Activity of the local server i</td></tr></thead><tbody><tr><td>PUT<sub>j</sub><sup>b</sup> (X)</td><td>Put the name X in the list of blocked processes</td></tr><tr><td>GET<sub>j</sub><sup>b</sup> (X)</td><td>Remove the name X of the list of blocked processes</td></tr><tr><td>ACTIVE<sub>j</sub><sup>b</sup> (X)</td><td>If X is a local process, active it.</td></tr></tbody></table>
### 3.3. Implementation of a distributed eventcount and of a distributed sequencer

#### 3.3.1. Distributed eventcount

The distributed eventcount is a variable E which is duplicated on each host and managed on each host by a server process. A local blocked processes queue is attached to each server. User processes may invoke activations of the primitives: AWAIT (E,n), ADVANCE (E) and READ (E). We keep the same conventions as in the previous examples. Activations of primitives are realized by the local server. Then we can define the algorithms:

**Primitive READ (E)**

**begin**

read the value of E;

return the value to the user process

**end**

**Primitive ADVANCE (E)**

**begin**

send message ADVANCE<sub>i</sub><sup>b</sup> (E)

**end**

**Primitive AWAIT (E, n)**

begin

if n > E then

**begin** block user process;

put it on the local queue with number n

**end**

**end**

**Activity of a server receiving messages**

<table><thead><tr><th>Messages</th><th>Activity of local server i</th></tr></thead><tbody><tr><td>ADVANCE<sub>j</sub><sup>b</sup> (E)</td><td>execute E := E+1<br>Active all the processes in the queue<br>for which n &lt;= E</td></tr></tbody></table>
### Comment

We don't care at what time the server *i* which sends the message `ADVANCE_i^b` (E) will receive it and treat it. That's the reason why we don't need a transaction in this case.

### 3.3.2. Distributed sequencer

The distributed sequencer is a variable *T* copied on each host and managed by a server process on each host. User processes may invoke activations of the primitive `TICKET (T)` which are realized by the local server.

#### Primitive TICKET (T)

#### begin

send message `OPEN TRANSACTION_i^b` ;

send message `INCREMENT_i^b (T)` ;

wait for the message ;

treat it ;

Return value of *T* to the user process ;

Send message `CLOSE TRANSACTION_i^b`

#### end

### 3.4. Implementation of a distributed semaphore with a distributed eventcount and a distributed sequencer

Let *E* and *T* be respectively a distributed eventcount and a distributed sequencer implemented in the manner we saw in the previous examples.

It is easy to build a distributed semaphore *S* in the way used by Reed and Kanodia in a centralized system.

Then we have the primitives :

#### Primitive P (S)

#### begin

integer *t* ;

*t* := `TICKET (T)` ;

`AWAIT (E,t)`

#### end

#### Primitive V (S)

#### begin

`ADVANCE (E)`

#### end
## Comments

- The `P` primitive is not executed in mutual exclusion.

- We have an implementation of the same kind as the one of § 3.1., and with the same constraint : management of the queue in FIFO.

## 3.5. Implementation of a multiple-producers, multiple-consumers schema in a distributed environment

**Hypothesis** : In the distributed system we have two classes of processes : producers which work and produce results ; consumers which must wait for and take these results in order to make their own work. (e.g. consumers are a pool of printers). These processes are distributed at random in the system.

## Principle

We maintain in multiple copies two tables : the table of supplies and the table of demands. On each host the two tables are managed by a single server process.

When a producer produces a result it invokes the local server in order to get a rendez-vous with a consumer or, if it is not possible, to put its name and the number of its product in the supply table.

When a consumer wants to get a result it invokes the local server to get a rendez-vous with a producer or, if it is not possible, to put its name in the demand table.

Products of a producer process stay in its own host until a rendez-vous is realized with a consumer. So we suppose that each producer limit by itself its production, according with the place at its disposal. Producers may invoke the primitive `PUT` and consumers may invoke the primitive `GET`. These invocations are realized by the local server.

## Algorithms

**Notation** : 
* `p` : name of the producer process
* `n` : number of its product
* `i` : name of the local server
* `c` : name of the consumer process

.../...
Primitive PUT (p, n)

begin

send message OPEN TRANSACTION<sub>i</sub><sup>b</sup> ;

wait for this message ;

if demand table = empty then

begin

put (p, n) in the supply table ;

end

else

begin

remove a process name c of the demand table ;

return c to process p

send message PRODUCER NAME<sub>i</sub><sup>c</sup> (p, n)

end

send message CLOSE TRANSACTION<sub>i</sub><sup>b</sup>

end

Primitive GET (c)

begin

send message OPEN TRANSACTION<sub>i</sub><sup>b</sup> ;

wait for this message ;

if supply table = empty then

begin

put (c) in the demand table ;

block process c

end

else

begin

remove a couple (p, n) of the supply table ;

return (p, n) to the process c

send message CONSUMER NAME<sub>i</sub><sup>D</sup> (c, n)

end

send message CLOSE TRANSACTION<sub>i</sub><sup>b</sup>

end
Comments

- When a rendez-vous is successful, the both processes are alerted, one by the return of the primitive, and the other by a message which eventually activates it and gives it the name of its partner and the number of the product. They must then apply a protocol in order to communicate......

- We must wait for the message opening the transaction : if there is another transaction in course, we won't receive this message until the end of this transaction (it will be put in the secondary queue of the server).
## 4. CONCLUSION

In this paper we have exhibited a synchronization technique for a distributed system, using the characteristics of the communication network.

The first example (§2) is intended to offer the comprehension of the mechanism and to show how it can be used in realistic conditions.

The others examples are intended to demonstrate the potential generality of the technique but, of course, we don't pretend that it is the best way to implement it. Besides we didn't consider the eventual faults or failures in these last examples.

We have described a mechanism which plays, for this kind of distributed system, the same role as, in a centralized system, shared memory and access to a memory cell in mutual exclusion. Therefore it allows a large number of solutions to synchronization problems.

## ACKNOWLEDGEMENTS

We thank the other members of CHORUS project : A. CARISTAN, M. GUILLEMONT and M.L. LANGE for their suggestions and critics when we were writing this paper. And also thanks are due to professor G. WHITE who helped us to give an acceptable form to this paper.
BIBLIOGRAPHY

(DIJKSTRA 67) DIJKSTRA E.W. : Cooperating sequential processes. Programming languages. F. GENUYS Editor. Academic Press 1967.

(ELLIS 77) ELLIS C.A. : A robust algorithm for updating duplicate data bases. 2nd workshop "Distributed Data Management and Computer Networks" Berkeley, May 1977.

(LE LANN 77) LE LANN G. : Introduction à l'analyse des systèmes multiréférentiels. Thèse de Doctorat d'Etat, Université de Rennes, Mai 1977.

(LE LANN 78) LE LANN G. : Les problèmes de signalisation dans les systèmes informatiques à contrôle réparti. Décembre 1978.

(MULLERY 75) MULLERY A.P. : The distributed control of multiple copies of data. RR RC5782 IBM T.J. Watson, New York, 1975.

(LAMPORT 78) LAMPORT L. : Time, clocks and the ordering of events in a distributed system. Communications of the ACM, 1978.

(REED 79) REED R.P. and KANODIA R.K. : Synchronization with evencounts and sequencers. Communication of the ACM, 1979.

(SEGUIN 77) SEGUIN J., SERGEANT G., WILMS P. : Cohérence et gestion d'objets dupliqués dans les systèmes distribués. RR IMAG Mai 1977.

(THOMAS 77) THOMAS R.H. : A majority consensus approach to concurrency control for multiple copy data bases. BBN report n° 3733 December 1977.