Section 15.7. ProducerConsumer Relationship with Thread Synchronization

15.7. Producer/Consumer Relationship with Thread Synchronization

Figures 15.915.10 demonstrate a producer and a consumer accessing a shared cell of memory with synchronization, so that the producer always goes first, the consumer consumes only after the producer produces a value and the producer produces a new value only after the consumer consumes the previous value produced. This examples reuses interface IBuffer (Fig. 15.4) and classes Producer (Fig. 15.5) and Consumer (Fig. 15.6) from the previous example. [Note: In this example, we demonstrate synchronization with class Monitor's Enter and Exit methods. In the next example, we demonstrate the same concepts via a lock block.]

Class SynchronizedBuffer (Figure 15.9) implements interface IBuffer and contains two instance variablesbufferValue (line 9) and occupiedBufferCount (line 12). Also, property Buffer's Get (lines 1657) and Set (lines 5888) accessors now use methods of class Monitor to synchronize access to instance variable bufferValue. Thus, each object of class SynchronizedBuffer has a SyncBlock to maintain synchronization. Instance variable occupiedBufferCount is known as a condition variableproperty Buffer's accessors use this int in conditions to determine whether it is the producer's turn to perform a task or the consumer's turn to perform a task. If occupiedBufferCount is 0, property Buffer's Set accessor can place a value into variable bufferValue, because the variable currently does not contain informationbut this means that property Buffer's Get accessor cannot read the value of bufferValue. If occupiedBufferCount is 1, the Buffer property's Get accessor can read a value from variable bufferValue, because the variable does contain information, but property Buffer's Set accessor cannot place a value into bufferValue.

As in the previous example, the producer thread performs the tasks specified in the producer object's Produce method. When line 23 of Fig. 15.5 sets the value of sharedLocation's property Buffer, the producer thread invokes the Set accessor at lines 5888 of Fig. 15.9. Line 60 invokes Monitor method Enter with the argument Me to acquire the lock on the SynchronizedBuffer object. The If statement (lines 6470) determines whether occupiedBufferCount is 1. If this condition is true, lines 6566 output a message indicating that the producer thread is trying to write a value, and lines 6768 invoke method DisplayState (lines 9295) to output another message indicating that the buffer is full and the producer thread is waiting. Line 69 invokes Monitor method Wait with the argument Me to place the calling thread (i.e., the producer) in the WaitSleepJoin state for the SynchronizedBuffer object. This also releases the lock on the SynchronizedBuffer object. The WaitSleepJoin state for an object is maintained by that object's SyncBlock. Now another thread can invoke an accessor method of the SynchronizedBuffer object's Buffer property.

The producer thread remains in the WaitSleepJoin state until the thread is notified by the consumer's call to Monitor method Pulse that it may proceedat which point the thread returns to the Running state and waits for the system to assign a processor to the thread. When the thread returns to the Running state, the thread implicitly reacquires the lock on the SynchronizedBuffer object, and the Set accessor continues executing with the next statement after Wait. Line 73 assigns value to bufferValue. Line 77 increments the occupiedBufferCount to indicate that the shared buffer now contains a value (i.e., a consumer can read the value, and a producer cannot yet put another value there). Lines 7980 invoke method DisplayState to output a line to the console window indicating that the producer is writing a new value into the bufferValue. Line 84 invokes Monitor method Pulse with the SynchronizedBuffer object (Me) as an argument. If there are any waiting threads in that object's SyncBlock, the first waiting thread enters the Running state, indicating that the thread can now attempt its task again (as soon as the thread is assigned a processor). The Pulse method returns immediately. Line 87 invokes Monitor method Exit to release the lock on the SynchronizedBuffer object, and the Set accessor returns to its caller (i.e., the Produce method of the Producer).

Common Programming Error 15.3

Forgetting to release the lock on an object when the lock is no longer needed is a logic error. This will prevent the other threads in your program from acquiring the lock to proceed with their tasks. These threads will be forced to wait (unnecessarily, because the lock is no longer needed). Such waiting can lead to deadlock and indefinite postponement.

The Get and Set accessors are implemented similarly. As in the previous example, the consumer thread performs the tasks specified in the consumer object's Consume method. The consumer thread gets the value of the SynchronizedBuffer object's Buffer property (line 25 of Fig. 15.6) by invoking the Get accessor in lines 1657 of Fig. 15.9. Line 18 invokes Monitor method Enter to acquire the lock on the SynchronizedBuffer object.

The If statement in lines 2228 determines whether occupiedBufferCount is 0. If this condition is TRue, lines 2324 output a message indicating that the consumer thread is trying to read a value, and lines 2526 invoke method DisplayState to output another message indicating that the buffer is empty and the consumer thread is waiting. Line 27 invokes Monitor method Wait with the argument Me to place the calling thread (i.e., the consumer) in the WaitSleepJoin state for the SynchronizedBuffer object and releases the lock on the object. Now another thread can invoke an accessor method of the SynchronizedBuffer object's Buffer property.

The consumer thread object remains in the WaitSleepJoin state until the thread is notified by the producer's call to Monitor method Pulse that it may proceedat which point the thread returns to the Running state and waits for the system to assign a processor to the thread. When the thread reenters the Running state, the thread implicitly reacquires the lock on the SynchronizedBuffer object, and the Get accessor continues executing with the next statement after Wait. Line 32 decrements occupiedBufferCount to indicate that the shared buffer is now empty (i.e., a consumer cannot read the value, but a producer can place another value into the shared buffer), line 34 outputs a line to the console window indicating the value the consumer is reading and line 38 invokes Monitor method Pulse with the SynchronizedBuffer object as an argument. If there are any waiting threads in that object's SyncBlock, the first waiting thread enters the Running state, indicating that the thread can now attempt its task again (as soon as the thread is assigned a processor). The Pulse method returns immediately. Line 51 gets a copy of bufferValue before releasing the lock. This is necessary because it is possible that the producer could be assigned the processor immediately after the lock is released (line 54) and before the Return statement executes (line 56). In this case, the producer would assign a new value to bufferValue before the Return statement returns the value to the consumer and the consumer would receive the new value. Making a copy of bufferValue and returning the copy ensures that the consumer receives the proper value. Line 54 invokes Monitor method Exit to release the lock on the SynchronizedBuffer object, and the Get accessor returns bufferCopy to its caller.

Class SynchronizedBufferTest (Fig. 15.10) is nearly identical to class UnsynchronizedBufferTest (Fig. 15.8). SynchronizedBufferTest's Main method declares sharedBuffer as an object of class SynchronizedBuffer (line 10) and also displays header information for the output (lines 1618).

Study the two sample outputs in Fig. 15.10. Observe that every integer produced is consumed exactly onceno values are lost and no values are consumed more than once. This occurs because the producer and consumer cannot perform tasks unless it is "their turn." The producer must go first; the consumer must wait if the producer has not produced since the consumer last consumed; and the producer must wait if the consumer has not yet consumed the value the producer most recently produced. Execute this program several times to confirm that every integer produced is consumed once. Notice the lines indicating when the producer and consumer must wait to perform their respective tasks.