Producer-Consumer Patterns

Producer-Consumer Patterns

The Producer-Consumer pattern is classically defined as two threads or tasks coordinating their behavior through a shared fixed length buffer. The producer writes data to the buffer. The consumer reads data from the buffer. The producer must stop writing to the buffer while the buffer is full. The consumer must stop reading from the buffer while the buffer is empty.

The size of the buffer is fixed and does not grow during the life of the program. The minimum size of the buffer must be one element. The maximum size of the buffer is limited only by memory constraints on the program.

The classic Producer-Consumer design uses one producer and one consumer.

Classic Producer-Consumer Pattern

In this pattern the producer and consumer have no direct interaction. Each interacts with the fixed length buffer, allowing the two tasks to execute asynchronously until the buffer is full or empty. It is important that the producer cannot write to the buffer while the consumer is reading from the buffer to avoid race conditions. Similarly, the consumer cannot read from the buffer while the producer is writing to the buffer. Coordination of writing and reading can be implemented through a simple binary semaphore when there is one producer and one consumer.

Coordination becomes more difficult when other configurations are used. Additional configurations include

• Single Producer with Multiple Consumers
• Multiple Producers with Single Consumer
• Multiple Producers with Multiple Consumers

Single Producer Multiple Consumer Pattern

Multiple Consumer Single Producer Pattern

Multiple Producer Multiple Consumer Pattern

The coordination problem becomes most complex when dealing with multiple producers writing to a single buffer and multiple consumers reading from the same single buffer.

Producer-Consumer Implemented Using Ada

Ada tasks are similar to threads in many other languages. Tasks are active objects. Ada protected objects are passive elements of concurrency. Protected objects are executed by tasks when a task calls one of the protected object's methods.

Ada Tasks

Tasks work independently unless they are suspended while waiting for a lock or in an entry queue. Tasks communicate synchronously with each other through task entries. They communicate asynchronously with each other through protected objects.

Task Entries

Task entries implement the Rendezvous model of communication. A task may declare an entry, which is a synchronization point during which data may be passed directly between tasks. A task may call another task's entry. The task declaring the entry can accept the entry call at the point in its logic best suited for handling the entry call.

A task calling the entry of another task will suspend until the called task reaches the accept statement for that entry. At this point the two tasks are synchronized to allow data to be passed from one task to the other. If the called task reaches its accept statement before another task calls the corresponding entry the called task will suspend, waiting for a calling task. Following completion of the task entry both tasks will continue executing independently of each other.

Select statements

Ada provides a method for conditionally accepting entry calls. The select clause provides the ability to handle one of many entries, or to poll an entry so that a rendezvous can be handled without incurring a protracted suspension of the task. An example from the Ada Language Reference Manual is

task body Server is

   Current_Work_Item : Work_Item;

begin

   loop

      select

         accept Next_Work_Item(WI : in Work_Item) do

            Current_Work_Item := WI;

         end;

         Process_Work_Item(Current_Work_Item);

      or

         accept Shut_Down;

         exit; -- Premature shut down requested

      or

         terminate; -- Normal shutdown at end of scope

      end select;

   end loop;

end Server;

Ada Protected Objects

Ada protected objects are protected against race conditions and deadlock. Protected objects are designed to be shared by multiple tasks. Protected objects implement sophisticated versions of Hoare monitors. Protected objects handle their own locking mechanisms, making their design highly Object Oriented. The calling tasks never directly manipulate locks.

Protected Methods

There are three kinds of protected methods:

• Procedures
  
• Entries
  
• Functions
  

Protected Procedures

Protected Procedures are allowed to read from or write to the protected object. Protected Procedures acquire exclusive unconditional access to the Protected Object through a read-write lock.

Protected Entries

Protected Entries are also allowed to read from or write to the protected object. Like Protected Procedures, Protected Entries acquire an exclusive read-write lock. The difference between a protected Procedure and a Protected Entry is the conditional nature of a Protected Entry call. Each Protected Entry is defined with a guard condition which must evaluate to True or False. When the guard condition is False execution of the Protected Entry is blocked and the calling task suspends until the guard condition evaluates to True. Each suspended task is placed in an Entry Queue so that calls to a Protected Entry are executed in the proper order. There are two common queuing policies; First In First Out (FIFO) and Priority. The default queuing policy is FIFO, so that the queued tasks execute the Protected Entry in the temporal order in which the Protected Entry was called.

Protected Functions

Protected functions are only allowed to read data from the Protected Object. They are not allowed to modify the Protected Object in any way. Protected functions acquire a shared read-only lock on the Protected Object. This lock prevents Protected Procedures and Protected Entries from executing while the lock is asserted, but permit any number of simultaneous function calls to execute on the Protected Object. Protected objects cannot execute while a Protected Entry or Protected Procedure read-write lock is asserted.

All these locking and queuing operations are performed implicitly. The programmer does not create any explicit lock or queue manipulation calls.

Ada Producer-Consumer Example

The following example uses Ada tasks, task entries, Ada Protected Objects and Protected Object entries.



with Ada.Strings.Bounded;
generic
   Buf_Size : Positive;
package Universal_PC is
   package Label_Strings is new Ada.Strings.Bounded.Generic_Bounded_Length(30);
   use Label_Strings;

   task type Producer is
      entry Set_Id(Label : Label_Strings.Bounded_String);
      entry Done;
   end Producer;


   task type Consumer is
      entry Set_Id(Label : Label_Strings.Bounded_String);
      entry Stop;
   end Consumer;


end Universal_PC;

The package specification defines a bounded string type named Label_Strings with a maximum length of 30 characters. It also defines two task types; Producer and Consumer. The Producer task type has two task entries; Set_Id, which passes in an instance of Bounded_String and Done which passes no information. Done only acts to synchronize the Producer and a calling task upon an event. In this case the event is when the Producer is done. The Consumer task type has two task entries; Set_Id, which passes an instance of Bounded_String to Consumer, and Stop, which commands the Consumer task instance to terminate.

with Ada.Text_IO.Bounded_IO;

package body Universal_PC is

   package Label_IO is new Ada.Text_IO.Bounded_IO (Label_Strings);

   use Label_IO;

   ------------

   -- Buffer --

   ------------

   type Internal_Buf is array (Natural range 0 .. Buf_Size - 1) of

       Label_Strings.Bounded_String;

   protected Buffer is

      entry Read (Item : out Label_Strings.Bounded_String);

      entry Write (Item : in Label_Strings.Bounded_String);

   private

      Count       : Natural := 0;

      Buf         : Internal_Buf;

      Write_Index : Natural := 0;

      Read_Index  : Natural := 0;

   end Buffer;

   protected body Buffer is

      entry Read (Item : out Label_Strings.Bounded_String)

         when Count > 0 is

      begin

         Item       := Buf (Read_Index);

         Count      := Count - 1;

         Read_Index := (Read_Index + 1) mod Buf_Size;

      end Read;

      entry Write (Item : in Label_Strings.Bounded_String)

         when Count < Buf_Size is

      begin

         Buf (Write_Index) := Item;

         Count             := Count + 1;

         Write_Index       := (Write_Index + 1) mod Buf_Size;

      end Write;

   end Buffer;

   --------------

   -- Producer --

   --------------

   task body Producer is

      Id : Label_Strings.Bounded_String;

   begin

      accept Set_Id (Label : Label_Strings.Bounded_String) do

         Id := Label;

      end Set_Id;

      for I in 1 .. 10 loop

         Buffer.Write (Id & ' ' & To_Bounded_String (I'Image));

         delay 0.01;

      end loop;

      accept Done;

   end Producer;

   --------------

   -- Consumer --

   --------------

   task body Consumer is

      Id : Label_Strings.Bounded_String;

      Value : Label_Strings.Bounded_String;

   begin

      accept Set_Id (Label : Label_Strings.Bounded_String) do

         Id := Label;

      end Set_Id;

      loop

         select

            accept Stop;

            exit;

         else

            select

               Buffer.Read (Value);

               Put_Line (Id & ':' & ' ' & Value);

            or

               delay 0.01;

            end select;

         end select;

      end loop;

   end Consumer;

end Universal_PC;

The package body provides the implementation of the tasks declared in the package specification. It also provides the implementation for the Protected Object named Buffer.
The Protected Object specification declares two Protected Entries for Buffer. The Write entry passes Bounded_String into Buffer and the Read entry passes a Bounded_String out of Buffer. The private part of the Protected Object specification defines the data contained in Buffer. Four data items are defined. Count maintains a count of the number of data elements currently in use in Buf array of Buffer. Buf is an array of Bounded_Strings. Write_Index contains the index for the next Write entry call. Read_Index contains the index for the next Read entry call.

The Protected Body contains the implementation of the two entry calls. The Read entry call has a condition stating that the call can only execute when Count > 0. This condition enforces the requirement that one cannot read from an empty Producer-Consumer queue. The Write entry call has a condition stating that the call can only execute when Count < Buf_Size. This condition enforces the requirement that one cannot write to a full Producer-Consumer queue. All the Protected Entry locking and queuing is written automatically by the compiler. The programmer is only concerned with writing the entry behaviors not associated with locking and queuing.

The package body also contains the implementation of the task bodies for Producer and Consumer.
Each Producer starts by accepting the Set_Id task entry, allowing the programmer to assign an ID label unique to each Producer task. The Producer task then iterates through the numbers in the range 1 through 10. Each iteration calls Buffer.Write, passing a bounded string containing the task ID concatenated with the iteration value. After writing all 10 values to Buffer the task accepts the Done entry then terminates.

Each Consumer task starts by accepting the Set_Id entry, just like the Producer tasks. The Consumer then enters a simple loop that iterates through a set of select calls polling the Stop task entry then polling the Buffer.Read protected entry. The loop exits when the Stop task entry is handled.

with universal_pc;

procedure Universal_pc_test is

   package Pc is new Universal_Pc(10);

   use PC;

   P1 : Producer;

   P2 : Producer;

   C1 : Consumer;

   C2 : Consumer;

begin

   P1.Set_Id(Label_Strings.to_Bounded_String("Producer 1"));

   P2.Set_Id(Label_Strings.To_Bounded_String("Producer 2"));

   C1.Set_Id(Label_Strings.To_Bounded_String("Consumer 1"));

   C2.Set_Id(Label_Strings.To_Bounded_String("Consumer 2"));

   P1.Done;

   P2.Done;

   C1.Stop;

   C2.Stop;

end Universal_pc_test;

This procedure is the “main” procedure or program entry point for the program. Every program entry point is also implicitly a task.

The generic package Universal_PC is instantiated with a Buf_Size of 10. Two producers, P1 and P2, are declared. Two Consumers, C1 and C2, are declared. The four tasks start as soon as execution of the program reaches the “begin” in the procedure Universal_pc_test. Initially all four tasks are suspended at their Set_Id task accept calls. The four task entries are called, passing appropriate labels to the four tasks.

P1.Done and P2.Done are called immediately. The calling task (Universal_pc_test) is suspended untils P1 and then P2 accept their Done entries. At that point both producers have completed. C1.Stop and C2.Stop are then called to terminate the two Consumer tasks.

The output of this program is

Consumer 1: Producer 1 1

Consumer 1: Producer 2 1

Consumer 1: Producer 2 2

Consumer 2: Producer 1 2

Consumer 1: Producer 1 3

Consumer 2: Producer 2 3

Consumer 1: Producer 2 4

Consumer 2: Producer 1 4

Consumer 1: Producer 2 5

Consumer 2: Producer 1 5

Consumer 1: Producer 2 6

Consumer 2: Producer 1 6

Consumer 2: Producer 2 7

Consumer 1: Producer 1 7

Consumer 2: Producer 1 8

Consumer 1: Producer 2 8

Consumer 2: Producer 2 9

Consumer 1: Producer 1 9

Consumer 2: Producer 1 10

Consumer 1: Producer 2 10

This solution will work with any Buf_Size greater than 0 and any number of producers and consumers.