Chapter 10: Message Passing

In previous chapters, we have seen that when threads interact through shared variables, these variables must be encapsulated in monitor objects to ensure correct behavior. An alternative way of organizing concurrent programs which does not require shared variables is to use message passing. In message-passing programs, processes interact by sending and receiving messages. Processes that interact solely by message exchange do not need to access shared memory and consequently can be located on different computers connected by a communication network. However, message passing is also frequently used when processes are intended to run within a single computer.

Fundamental to message passing are the operations to send and receive a message. There are a surprising number of different definitions for these operations in message-passing systems. We examine the two basic models for message passing: synchronous message passing, in which the sender of a message waits until it has been received; and asynchronous message passing, in which the sender does not wait and messages which have been sent but not yet received are buffered. These are both one-way forms of communication: the messages are transmitted in one direction only, from sender to receiver. In addition, we examine the rendezvous, a two-way message-passing protocol used for client – server interaction.

10.1 Synchronous Message Passing

An important design decision in a message-passing scheme is how to designate the sources and destinations of messages. Messages can be addressed directly to the destination process or indirectly to some intermediate entity. In discussing synchronous message passing, we adopt the scheme used in OCCAM (INMOS Ltd., 1988a), in which messages are sent to and received from channels. We will see that it is possible, in some message-passing schemes, for many senders to communicate with a receiver through a single communication entity. However, as in OCCAM, we specify that a channel connects two and only two processes. A single process can send to the channel and a single process can receive from the channel as shown in Figure 10.1. Communication is said to be one-to-one.

Figure 10.1: Synchronous message-passing channel.

The send and receive operations on channels are:

send(e, c) – send the value of the expression e to channel c. The process calling the send operation is blocked until the message is received from the channel.

v = receive(c) – receive a value into local variable v from channel c. The calling process is blocked waiting until a message is sent to the channel.

The first process, whether sender or receiver, to perform a channel operation is blocked until its partner performs the complementary action. After a communication occurs, sender and receiver can proceed independently. This form of message passing is termed synchronous, since the sender and receiver must be exactly synchronized for communication to occur. Another way of thinking of synchronous communication is that it implements a distributed assignment in which the sender’s expression is assigned to the receiver’s local variable (v = e). In the language OCCAM and the CSP formalism (Hoare, 1985) which inspired it, the notation for send is c!e and for receive is c?v.

Synchronous message operations do not require messages to be buffered. If the sender process is running on the same computer as the receiver then the message can be copied directly from the sender into the receiver’s local variable. This simplicity enabled the OCCAM send and receive operations to be implemented directly as Transputer (INMOS Ltd., 1988b) machine instructions.

10.1.1 Selective Receive

We have seen that choice between actions is important in both modeling and implementing concurrent programs. In Java, choice is implemented within a monitor. The monitor lock effectively selects a single synchronized method activation to execute from the set of possible activations. The synchronous, message receive operation blocks waiting on a single channel. How do we choose between receiving messages from a set of channels? The solution provided by languages such as OCCAM and Ada is to use a select statement. The general form of a select statement is as shown below:

 select    when G1 and v1 =receive(chan1) => S1; or    when G2 and v2 =receive(chan2) => S2; or    ... or    when Gn and vn =receive(chann) => Sn; end 

G₁ ..G_n are boolean expressions known as guards. A receive is eligible if the guard associated with it evaluates to true. The select statement chooses an eligible receive operation for which there is a sender waiting to send. The statement S_i is executed after a successful receive on a channel chan_i. The select statement then terminates. If none of the channels have waiting sends then the select statement blocks. The reader should immediately see the correspondence between this select construct and choice over guarded actions in FSP. Indeed, we will see later that this is exactly the way selective receive is modeled in FSP. A sin FSP, guards are usually optional in select statements such that an omitted guard is equivalent to when true and. Some select statements allow non-blocking semantics by providing an else alternative as follows:

 select    v=receive(chan)=> S; else    Selsepart; end 

If a sender is not waiting on the channel then the else part is immediately chosen. Another variation is to permit a timeout as an alternative. If no message arrives within the timeout period then the statements associated with the timeout part are executed and the select terminates.

The send and receive operations are symmetrical in synchronous message passing in that the send blocks if there is no corresponding receive and vice versa. Consequently, it is reasonable to suggest that send operations should be allowed as select alternatives in addition to receive operations. However, because of the resulting implementation complexity, only experimental message-passing languages have included both send and receive select alternatives.

10.1.2 Synchronous Message Passing in Java

We have seen in the preceding chapters that Java supports thread interaction through monitors. How do we write message-passing programs in Java? The answer is to use Java’s object-oriented programming facilities to implement and encapsulate message-passing abstractions. We can implement the synchronous message-passing channel as a class. The outline of the Java class that implements the channel abstraction is defined in Program 10.1.

Program 10.1: Channel class.

 public class Channel<T> extends Selectable{     public synchronized void send(T v)            throws InterruptedException {. . .}     public synchronized T receive()            throws InterruptedException {. . .} }

The implementation of Channel is a monitor that has synchronized access methods for send and receive. (The code may be found on the website that accompanies this book.) The class extends the Selectable base class to support the Java selective receive implementation, described later.

To demonstrate the operation of the Channel implemented in Java, we develop a simple program in which a sender thread communicates with a receiver thread using a single channel. The display for this program is depicted in Figure 10.2. The display depicts the situation where the sender thread is blocked waiting to send the value in e. The receiver has not yet executed the receive operation to copy the value into its local variable v. The sender simply transmits a sequence of integer values from 0 to 9 and then restarts at 0 again.

Figure 10.2: Synchronous message-passing applet display.

The code for both Sender and Receiver threads is given in Program 10.2. The threads use the display class that we defined in Chapter 9, SlotCanvas. The threads and channel are created by the following Java code:

Program 10.2: Sender and Receiver threads.

 class Sender implements Runnable {   private Channel<Integer> chan;   private SlotCanvas display;   Sender(Channel<Integer> c, SlotCanvas d)     {chan=c; display=d;}   public void run() {     try {       int ei = 0;       while(true) {         display.enter(String.valueOf(ei));         ThreadPanel.rotate(12);         chan.send(new Integer(ei));         display.leave(String.valueOf(ei));         ei=(ei+1)%10; ThreadPanel.rotate(348);       }     } catch (InterruptedException e){}   } } class Receiver implements Runnable {   private Channel<Integer> chan;   private SlotCanvas display;   Receiver(Channel<Integer> c, SlotCanvas d)     {chan=c; display=d;}   public void run() {     try {       Integer v=null;       while(true) {         ThreadPanel.rotate(180);         if (v!=null) display.leave(v.toString());         v = chan.receive();         display.enter(v.toString());         ThreadPanel.rotate(180);       }     } catch (InterruptedException e){}   } }

 Channel<Integer> chan = new Channel<Integer>(); tx.start(new Sender(chan,senddisp)); rx.start(new Receiver(chan,recvdisp));

where tx and rx are instances of the thread display class, ThreadPanel, and senddisp and recvdisp are instances of SlotCanvas.

While the code of Program 10.2 is straightforward, a subtlety can lead to problems if the programmer is not aware of it. Our implementation of channels in Java simply copies the reference to an object from sender to receiver, it does not make a copy of the referenced object. Consequently, it is possible for the receiver to modify an object held by the sender. When messages are passed by reference, the safest discipline to adopt is that a sender should not access an object if it has sent its reference to another thread. If a thread needs to reference the object in the future, it should copy it before sending it. With this discipline, a thread is guaranteed mutually exclusive access to the objects it has received but not sent. The sample program obeys this discipline even though the receiver does not modify the messages it receives.

10.1.3 Modeling Synchronous Message Passing

To illustrate how message-passing programs are modeled, we start with the simple Sender – Receiver example of the previous section. The model of the Sender thread is:

 range M = 0..9 SENDER = SENDER[0], SENDER[e:M] =(chan.send[e]->SENDER[(e+1)%10]).

where chan.send[e] models the action of sending a value e to a channel chan. The model of the Receiver thread is:

 RECEIVER = (chan.receive[v:M]->RECEIVER).

where chan.receive[v:M] models the action of receiving a value into a local variable v of type M from channel chan. The remaining question is how to model the channel entity. In fact, as sending and receiving are synchronous, they become the same action in the model. Consequently, we do not need a separate process to model a channel; we simply rename send and receive to be the same action. We can thus rename both the action chan.send and chan.receive to be the action chan shown in the structure diagram of Figure 10.3.

Figure 10.3: Modeling synchronous message passing.

To avoid the relabeling, we could have modeled the send action directly as chan[e] and the receive action as chan[v:M]. In the following, we model synchronous message passing by:

Message Operation	FSP Model
Send(e, chan)	chan[e]
v = receive(chan)	chan[e:M]

The only difference between the model for send and receive actions is that receive is modeled as a choice between a set of values M which can be sent over a channel whereas send specifies a specific value e. In the example, the values are in the range 0..9. The composite behavior for the example is given by the LTS of Figure 10.4.

Figure 10.4: SyncMsg labeled transition system.

10.1.4 Modeling and Implementing Selective Receive

To explain how to model and implement selective message reception, we use the car park example from Chapter 5. In this example, an arrivals gate signals the arrival of a car at the car park and the departures gate signals the departure of a car from the car park. The model for the car park is repeated in Figure 10.5.

Figure 10.5: Car park model.

Instead of implementing the CARPARKCONTROL process as a monitor with arrive and depart access methods, we implement the process as a thread which receives signals from channels called arrive and depart. The behaviors of the ARRIVALS and DEPARTURES processes are implemented by a common MsgGate class as shown in Program 10.3. A thread created from MsgGate sends messages to the channel with which it is initialized. Messages in this example contain no information, and are implemented by a Signal class. They signal the arrival or departure of a car from the car park.

Program 10.3: MsgGate class.

 class MsgGate implements Runnable {   private Channel<Signal> chan;   private Signal signal = new Signal();   public MsgGate (Channel<Signal> c) {chan=c;}   public void run() {     try {       while(true) {         ThreadPanel.rotate(12);         chan.send(signal);         ThreadPanel.rotate(348);       }     } catch (InterruptedException e){}   } }

The transformation of the CARPARKCONTROL process into a thread using message passing is straightforward since, as we noted earlier, choice and guarded actions express the behavior of selective receive. An outline implementation for the CARPARKCONTROL process using a selective receive is shown below:

 MsgCarPark::      while(true)         select             when spaces>0 and receive(arrive) => ++spaces;         or             when spaces< N and receive(depart) => --spaces;         end 

We use the object-oriented facilities of Java to implement the selective receive abstraction in the same way that we packaged channels as a class. Channels are derived from the Selectable base class, which provides the public method, guard. A selective receive is constructed using the Select class, which provides the add public method to include a Selectable object into a selective receive. Using these classes, the outline of the message version of car park control can be translated into Java, as shown in the MsgCarPark class of Program 10.4.

Program 10.4: MsgCarPark class.

 class MsgCarPark implements Runnable {   private Channel<Signal> arrive,depart;   private int spaces,N;   private StringCanvas disp;   public MsgCarPark(Channel<Signal> a, Channel<Signal> l,                  StringCanvas d,int capacity) {     depart=l; arrive=a; N=spaces=capacity; disp = d;     disp.setString("Cars: "+0+"   Spaces: "+spaces);   }   private void display(int s) throws InterruptedException {     disp.setString("Cars: "+(N-s)+"   Spaces: "+s);     ThreadPanel.rotate(348);   }   public void run() {     try {      Select sel = new Select();      sel.add(depart);      sel.add(arrive);      while(true) {        ThreadPanel.rotate(12);        arrive.guard(spaces>0);        depart.guard (spaces<N);        switch (sel.choose()) {        case 1:depart.receive();display(++spaces);               break;        case 2:arrive.receive();display(--spaces);               break;        }     }    } catch (InterruptedException e){}   } }

A selective receive is executed by invoking the choose() method on a Select object. This returns the index of a Selectable object which is ready. A selectable channel is ready if the sender has performed a send operation. The index is allocated to a selectable object based on the order that the object was added to the select object. Thus, in the example, the depart channel is allocated index 1 and the arrive channel index 2. When receive on the chosen channel is executed, it does not block since the channel is ready. If no selectable objects are ready then choose() blocks the calling thread waiting, in the example case, for a send on either the leave or the arrive channel.

This rather clumsy coding of a selective receive in Java would normally be done by a compiler if we had used a message-passing language. However, we have used Java so that readers can run message-passing programs in the same environment as the rest of the example programs in the book.