Reading 23: Queues and Message-Passing

In our introduction to concurrency, we saw two models for concurrent programming: shared memory and message passing.

• 

  In the shared memory model, concurrent modules interact by reading and writing shared mutable objects in memory. Creating multiple threads inside a single Java process is our primary example of shared-memory concurrency.

• 

  In the message passing model, concurrent modules interact by sending immutable messages to one another over a communication channel. That communication channel might connect different computers over a network, as in some of our initial examples: web browsing, instant messaging, etc.

The message passing model has several advantages over the shared memory model, which boil down to greater safety from bugs. In message-passing, concurrent modules interact explicitly, by passing messages through the communication channel, rather than implicitly through mutation of shared data. The implicit interaction of shared memory can too easily lead to inadvertent interaction, sharing and manipulating data in parts of the program that don’t know they’re concurrent and aren’t cooperating properly in the thread safety strategy. Message passing also shares only immutable objects (the messages) between modules, whereas shared memory requires sharing mutable objects, which we have already seen can be a source of bugs.

We’ll discuss in this reading how to implement message passing within a single process. We’ll use blocking queues (an existing threadsafe type) to implement message passing between threads within a process. Some of the operations of a blocking queue are blocking in the sense that calling the operation blocks the progress of the thread until the operation can return a result. Blocking makes writing code easier, but it also means we must continue to contend with bugs that cause deadlock.

In the next concurrency reading we’ll see how to implement message passing between client/server processes over the network.

We saw in Locks and Synchronization that a thread blocks trying to acquire a lock until the lock has been released by its current owner. Blocking means that a thread waits (without doing further work) until an event occurs. We can use this term to describe methods and method calls: if a method is a blocking method, then a call to that method can block, waiting until some event occurs before it returns to the caller.

We can use a queue with blocking operations for message passing between threads — and the buffered network communication channel in client/server message passing will work the same way. Java provides the BlockingQueue interface for queues with blocking operations.

In an ordinary Queue:

• add(e) adds element e to the end of the queue.
• remove() removes and returns the element at the head of the queue, or throws an exception if the queue is empty.

A BlockingQueue extends this interface:

additionally supports operations that wait for the queue to become non-empty when retrieving an element, and wait for space to become available in the queue when storing an element.

• put(e) blocks until it can add element e to the end of the queue (if the queue does not have a size bound, put will not block).
• take() blocks until it can remove and return the element at the head of the queue, waiting until the queue is non-empty.

When you are using a BlockingQueue for message passing between threads, make sure to use the put() and take() operations, not add() and remove().

We will implement the producer-consumer design pattern for message passing between threads. Producer threads and consumer threads share a synchronized queue. Producers put data or requests onto the queue, and consumers remove and process them. One or more producers and one or more consumers might all be adding and removing items from the same queue. This queue must be safe for concurrency.

Java provides two implementations of BlockingQueue:

• ArrayBlockingQueue is a fixed-size queue that uses an array representation. Using put to add an item to the queue will block if the queue is full.
• LinkedBlockingQueue is a growable queue using a linked-list representation. If no maximum capacity is specified, the queue will never fill up, so put will never block.

Like other collections classes in Java, these synchronized queues can hold objects of an arbitrary type. We must choose or design a type for messages in the queue: we will choose an immutable type because our goal in using message passing is to avoid the problems of shared memory. Producers and consumers will communicate only by sending and receiving messages, and there will be no opportunity for (mis)communication by mutating an aliased message object.

And just as we designed the operations on a threadsafe ADT to prevent race conditions and enable clients to perform the atomic operations they need, we will design our message objects with those same requirements.

You can see all the code for this example. The parts relevant to discussion are excerpted below.

Here’s a message passing module that represents a refrigerator:

/**
 * A mutable type representing a refrigerator containing drinks.
 */
public class DrinksFridge {

    private int drinksInFridge;
    private final BlockingQueue<Integer> in;
    private final BlockingQueue<FridgeResult> out;

    // Abstraction function:
    //   AF(drinksInFridge, in, out) = a refrigerator containing `drinksInFridge` drinks
    //                                 that takes requests from `in` and sends replies to `out`
    // Rep invariant:
    //   drinksInFridge >= 0

    /**
     * Make a DrinksFridge that will listen for requests and generate replies.
     * 
     * @param requests queue to receive requests from
     * @param replies queue to send replies to
     */
    public DrinksFridge(BlockingQueue<Integer> requests, BlockingQueue<FridgeResult> replies) {
        this.drinksInFridge = 0;
        this.in = requests;
        this.out = replies;
        checkRep();
    }

    ...
}

The module has a start method that creates an internal thread to service requests on its input queue:

/**
 * Start handling drink requests.
 */
public void start() {
    new Thread(new Runnable() {
        public void run() {
            while (true) {
                try {
                    // block until a request arrives
                    int n = in.take();
                    FridgeResult result = handleDrinkRequest(n);
                    out.put(result);
                } catch (InterruptedException ie) {
                    ie.printStackTrace();
                }
            }
        }
    }).start();
}

Incoming messages to the DrinksFridge are integers, representing a number of drinks to take from (or add to) the fridge:

/**
 * Take (or add) drinks from the fridge.
 * @param n if >= 0, removes up to n drinks from the fridge;
 *          if < 0, adds -n drinks to the fridge.
 * @return FridgeResult reporting how many drinks were actually added or removed
 *      and how many drinks are left in the fridge. 
 */
private FridgeResult handleDrinkRequest(int n) {
    // adjust request to reflect actual drinks available
    int change = Math.min(n, drinksInFridge);
    drinksInFridge -= change;
    checkRep();
    return new FridgeResult(change, drinksInFridge);
}

Outgoing messages are instances of FridgeResult:

/**
 * A threadsafe immutable message describing the result of taking or adding drinks to a DrinksFridge.
 */
public class FridgeResult {
    private final int drinksTakenOrAdded;
    private final int drinksLeftInFridge;
    // Rep invariant: 
    //   TODO

    /**
     * Make a new result message.
     * @param drinksTakenOrAdded (precondition? TODO)
     * @param drinksLeftInFridge (precondition? TODO)
     */
    public FridgeResult(int drinksTakenOrAdded, int drinksLeftInFridge) {
        this.drinksTakenOrAdded = drinksTakenOrAdded;
        this.drinksLeftInFridge = drinksLeftInFridge;
    }

    // TODO: we will want more observers, but for now...

    @Override public String toString() {
        return (drinksTakenOrAdded >= 0 ? "you took " : "you put in ") 
                + Math.abs(drinksTakenOrAdded) + " drinks, fridge has " 
                + drinksLeftInFridge + " left";
    }
}

We would probably add additional observers to FridgeResult so clients can retrieve the values.

Finally, here’s a main method that loads the fridge:

public static void main(String[] args) {

    BlockingQueue<Integer> requests = new LinkedBlockingQueue<>();
    BlockingQueue<FridgeResult> replies = new LinkedBlockingQueue<>();

    // start an empty fridge
    DrinksFridge fridge = new DrinksFridge(0, requests, replies);
    fridge.start();

    try {
        // deliver some drinks to the fridge
        requests.put(-42);

        // maybe do something concurrently...

        // read the reply
        System.out.println(replies.take());
    } catch (InterruptedException ie) {
        ie.printStackTrace();
    }
    System.out.println("done");
    System.exit(0); // ends the program, including DrinksFridge
}

private void checkRep() {
  assert REP_INVARIANT;
}

What if we want to shut down the DrinksFridge so it is no longer waiting for new inputs? One strategy is a poison pill: a special message on the queue that signals the consumer of that message to end its work.

To shut down the fridge, since its input messages are merely integers, we would have to choose a magic poison integer (maybe nobody will ever ask for 0 drinks…? bad idea, don’t use magic numbers) or use null (bad idea, don’t use null). Instead, we might change the type of elements on the requests queue to an ADT:

FridgeRequest = DrinkRequest(n:int) + StopRequest

with operations:

drinksRequested : FridgeRequest → int
shouldStop : FridgeRequest → boolean

and when we want to stop the fridge, we enqueue a FridgeRequest where shouldStop returns true.

For example, in DrinksFridge.start(), fill in the blanks in the exercise below:

public void run() {
    while (true) {
        try {
            // block until a request arrives
            FridgeRequest req = in.take();
            // see if we should stop
            if (▶▶A◀◀) { ▶▶B◀◀; }
            // compute the answer and send it back
            int n = ▶▶C◀◀;
            FridgeResult result = handleDrinkRequest(n);
            out.put(result);
        } catch (InterruptedException ie) {
            ie.printStackTrace();
        }
    }
}

FridgeRequest = DrinkRequest(n:int) + StopRequest

interface FridgeRequest { ... }
class DrinkRequest implements FridgeRequest { ... }
class StopRequest implements FridgeRequest { ... }

class FridgeRequest { ... }
class DrinkRequest { ... }
class StopRequest { ... }

class FridgeRequest {
  private final String requestType;
  public static final String DRINK_REQUEST   = "drink";
  public static final String STOP_REQUEST    = "stop";
  ...
}

It is also possible to signal a thread that it should stop working by calling that thread’s interrupt() method. If the target thread is blocked waiting, the method it’s blocked in will throw an Interrupted­Exception. That’s why we have to try-catch that exception almost any time we call a blocking method. If the target thread was not blocked, an interrupted flag will be set. In order to use this approach to stop a thread, the target thread must both handle any Interrupted­Exceptions and check for the interrupted flag to see whether it should stop working. For example:

public void run() {
    // handle requests until we are interrupted
    while ( ! Thread.interrupted()) {
        try {
            // block until a request arrives
            int n = in.take();
            FridgeResult result = handleDrinkRequest(n);
            out.put(result);
        } catch (InterruptedException ie) {
            // stop
            break;
        }
    }
}

A thread safety argument with message passing might rely on:

• Existing threadsafe data types for the synchronized queue. This queue is definitely shared and definitely mutable, so we must ensure it is safe for concurrency.

• Immutability of messages or data that might be accessible to multiple threads at the same time.

• Confinement of data to individual producer/consumer threads. Local variables used by one producer or consumer are not visible to other threads, which only communicate with one another using messages in the queue.

• Confinement of mutable messages or data that are sent over the queue but will only be accessible to one thread at a time. This argument must be carefully articulated and implemented. Suppose one thread has some mutable data to send to another thread. If the first thread drops all references to the data like a hot potato as soon as it puts them onto a queue for delivery to the other thread, then only one thread will have access to those data at a time, precluding concurrent access.

In comparison to synchronization, message passing can make it easier for each module in a concurrent system to maintain its own thread safety invariants. We don’t have to reason about multiple threads accessing shared data if the data are instead transferred between modules using a threadsafe communication channel.

public interface Trade {
    public int numShares();
    public String stockName();
}

public class TradeWorker implements Runnable {
    private final Queue<Trade> tradesQueue;

    public TradeWorker(Queue<Trade> tradesQueue) {
        this.tradesQueue = tradesQueue;
    }

    public void run() {
        while (true) {
            Trade trade = tradesQueue.poll();
            TradeProcessor.handleTrade(trade.numShares(), trade.stockName());
        }
    }
}

public class TradeProcessor {
    public static void handleTrade(int numShares, String stockName) {
        /* ... process the trade ... takes a while ... */
    }
}

In a previous reading, we saw in the bank account example that message-passing doesn’t eliminate the possibility of race conditions. It’s still possible for concurrent message-passing processes to interleave their work in bad ways.

This particularly happens when a client must send multiple messages to the module to do what it needs, because those messages (and the client’s processing of their responses) may interleave with messages sent by other clients. The message protocol for DrinksFridge has been carefully designed to manage some of this interleaving, but there are still situations where a race condition can arise. The next exercises explore this problem.

The blocking behavior of blocking queues is very convenient for programming, but blocking also introduces the possibility of deadlock. In a deadlock, two (or more) concurrent modules are both blocked waiting for each other to do something. Since they’re blocked, no module will be able to make anything happen, and none of them will break the deadlock.

In general, in a system of multiple concurrent modules communicating with each other, we can imagine drawing a graph in which the nodes of the graph are modules, and there’s an edge from A to B if module A is blocked waiting for module B to do something. The system is deadlocked if at some point in time, there is a cycle in this graph. The simplest case is the two-node deadlock, A → B and B → A, but more complex systems can encounter larger deadlocks.

Deadlock is much more common with locks than with message-passing — but when the message-passing queues have limited capacity, and become filled up to that capacity with messages, then even a message-passing system can experience deadlock. A message passing system in deadlock appears to simply get stuck.

Let’s see an example of message-passing deadlock, using the same DrinksFridge we’ve been using so far. This time, instead of using LinkedBlockingQueues that can grow arbitrarily (limited only by the size of memory), we will use the ArrayBlockingQueue implementation that has a fixed capacity:

private static final int QUEUE_SIZE = 100;
...
    // make request and reply queues big enough to hold QUEUE_SIZE messages each
    BlockingQueue<Integer> requests = new ArrayBlockingQueue<>(QUEUE_SIZE);
    BlockingQueue<FridgeResult> replies = new ArrayBlockingQueue<>(QUEUE_SIZE);

Many message-passing systems use fixed-capacity queues for performance reasons, so this is a common situation.

Finally, to create the conditions needed for deadlock, the client code will make N requests, each asking for a drink, before checking for any of the replies from DrinksFridge. Here is the full code:

private static final int QUEUE_SIZE = 100;
private static final int N = 100;

/**  Send N thirsty people to the DrinksFridge. */
public static void main(String[] args) throws IOException {
    // make request and reply queues big enough to hold QUEUE_SIZE messages each
    BlockingQueue<Integer> requests = new ArrayBlockingQueue<>(QUEUE_SIZE);
    BlockingQueue<FridgeResult> replies = new ArrayBlockingQueue<>(QUEUE_SIZE);

    DrinksFridge fridge = new DrinksFridge(requests, replies);
    fridge.start();

    try {
        // put enough drinks in the fridge to start
        requests.put(-N);
        System.out.println(replies.take());

        // send the requests
        for (int x = 1; x <= N; ++x) {
            requests.put(1); // give me 1 drink!
            System.out.println("person #" + x + " is looking for a drink");
        }
        // collect the replies
        for (int x = 1; x <= N; ++x) {
            System.out.println("person #" + x + ": " + replies.take());
        }
    } catch (InterruptedException ie) {
        ie.printStackTrace();
    }
    System.out.println("done");
    System.exit(0); // ends the program, including DrinksFridge thread
}

It turns out with N=100 and QUEUE_SIZE=100, the code above works and doesn’t reach a deadlock. But notice that our client is making N requests before reading any replies. If N is larger than QUEUE_SIZE, the replies queue fills up with unread replies. Then DrinksFridge blocks trying to put one more reply into that queue, and it stops calling take on the requests queue. The client can continue putting more requests into the requests queue, but only up to the size of that queue. If there are more additional requests than can fit in that queue – i.e., when N is greater than 2×QUEUE_SIZE – then the client’s call to requests.put() will block too. And now we have our deadly embrace. DrinksFridge is waiting for the client to read some replies and free up space on the replies queue, but the client is waiting for DrinksFridge to accept some requests and free up space on the requests queue. Deadlock.

Final suggestions for preventing deadlock

• Rather than synchronize with locks, message passing systems synchronize on a shared communication channel, e.g. a stream or a queue.

• Threads communicating with blocking queues is a useful pattern for message passing within a single process.

If you would like to get more practice with the concepts covered in this reading, you can visit the question bank. The questions in this bank were written in previous semesters by students and staff, and are provided for review purposes only – doing them will not affect your classwork grades.