- 13.1 Concurrentgate
- 13.2 A Brief History of Data Sharing
- 13.3 Look, Ma, No (Default) Sharing
- 13.4 Starting a Thread
- 13.5 Exchanging Messages between Threads
- 13.6 Pattern Matching with receive
- 13.7 File Copyingwith a Twist
- 13.8 Thread Termination
- 13.9 Out-of-Band Communication
- 13.10 Mailbox Crowding
- 13.11 The shared Type Qualifier
- 13.12 Operations with shared Data and Their Effects
- 13.13 Lock-Based Synchronization with synchronized classes
- 13.14 Field Typing in synchronized classes
- 13.15 Deadlocks and the synchronized Statement
- 13.16 Lock-Free Coding with shared classes
- 13.17 Summary
13.10 Mailbox Crowding
The producer-consumer file copy program works quite well but has an important shortcoming. Consider copying a large file between two devices of different speeds, for example, copying a legally acquired movie file from an internal drive (fast) to a network drive (possibly considerably slower). In that case, the producer (the main thread) issues buffers at considerable speed, much faster than the speed with which the consumer is able to unload them in the target file. The difference in the two speeds causes a net accumulation of buffers, which may cause the program to consume a lot of memory without achieving a boost in efficiency.
To avoid mailbox crowding, the concurrency API allows setting the maximum size of a thread's message queue, and also setting the action to take in case the maximum size has been reached. The signatures of relevance here are
// Inside std.concurrency void setMaxMailboxSize(Tid tid, size_t messages, bool(Tid) onCrowdingDoThis);
The call setMaxMailboxSize(tid, messages, onCrowdingDoThis) directs the concurrency API to call onCrowdingDoThis(tid) whenever a new message is to be passed but the queue already contains messages entries. If onCrowdingDoThis(tid) returns false or throws an exception, the new message is ignored. Otherwise, the size of the thread's queue is checked again, and if it is less than messages, the new message is posted to thread tid. Otherwise, the entire loop is resumed.
The call occurs in the caller thread, not the callee. In other words, the thread that initiates sending a message is also responsible for taking contingency action in case the maximum mailbox size of the recipient has been reached. It seems reasonable to ask why the call should not occur in the callee; that would, however, scale the wrong way in heavily threaded programs because threads with full mailboxes may become crippled by many calls from other threads attempting to send messages.
There are a few prepackaged actions to perform when the mailbox is full: block the caller until the queue becomes smaller, throw an exception, or ignore the new message. Such predefined actions are conveniently packaged as follows:
// Inside std.concurrency enum OnCrowding { block, throwException, ignore } void setMaxMailboxSize(Tid tid, size_t messages, OnCrowding doThis);
In our case, it's best to simply block the reader thread once the mailbox becomes too large, which we can effect by inserting the call
setMaxMailboxSize(tid, 1024, OnCrowding.block);
right after the call to spawn.
The following sections describe approaches to inter-thread communication that are alternative or complementary to message passing. Message passing is the recommended method of inter-thread communication; it is easy to understand, fast, well behaved, reliable, and scalable. You should descend to lower-level communication mechanisms only in special circumstances—and don't forget, "special" is not always as special as it seems.