Producer-consumer queue in multithreading, shutdown mechanism

Question

I wrote the following pice of pseudocode. It is about the classic producer-consumer queue in multithreading, but I am trying to add a graceful shutdown mechanism. I want consumeItem to keep working until all items in queue are consumed. Is this a good implementation? Or am I missing scenarios where it could get deadlocked or stale threads? Thank you.

thread addItems() {
while (!shutdown) {
    item = ItemFactory.produceItem();
    lock.acquire();
    try {
        while (self.queue.length() == MAX_QUEUE_SIZE) {
            conditionHasSpace.wait()
        }
        if (!shutdown) {
            self.queue.insert(order);
        }
        conditionHasItems.notify();
    }
    catch {
        handleException();
    }
    finally {
        lock.release()
    }
}

}

thread consumeItems() {
while (true) {
    lock.acquire();
    try { 
        while (self.queue.length() == 0) {
            if (shutdown) {
                exit;
            }
            conditionHasItems.wait();
        }
        self.queue.pop();
        conditionHasSpace.notify();
    }
    catch {
        handleException();
    }
    finally {
        lock.release();
    }
}

}

Answer 1

Consider using the "poison pill" method. It won't work any better than yours, but it's simpler.

The poison pill is a distinguished object or a distinguished value that you put into the queue in order to request that the consumer(s) shut down. If task objects are allocated from the heap, then the poison pill can be a "dummy" object that is recognized by pointer or referential equality, and its value can be ignored:

Assuming that your self.queue is a blocking queue with a thread-safe "pop" method that waits for the queue to become non-empty, then you can dispense with the lock, and your consumeItems method simplifies to this (Python version):

def consumeItems():
    while True:
        task = self.queue.get()
        if task is POISON_PILL:
            break
        task.perform()
    }
}

If there's more than one consumer thread, then a thread that wants to shut them all down must enqueue one reference to the POISON_PILL for each:

def shutDownAllConsumers():
    for _ in range(NUM_CONSUMERS):
        self.queue.put(POISON_PILL)
    self.queue.join()  # optional: wait for all poison pills to be swallowed.

Answer 2

Here is a solution using the Ada programming language. Ada provides tasks as active units of concurrency, often mapped to OS threads, and protected objects as passive units of concurrency. This example uses a protected object to implement the buffer shared by the producer and the consumer.

The protected object, the producer task and the consumer task are declared and defined in a package named producers_and_consumers. The package has two parts. The first part is the package specification, providing the API for the package. The second part is the package body, containing the implementation of the package logic.

The package specification is very simple. It simply declares the names of two tasks.

package producers_and_consumers is
   task producer;
   task consumer;
end producers_and_consumers;

The interesting part is in the package body where the protected buffer , the producer task and the consumer tasks are defined.

with Ada.Text_IO; use Ada.Text_IO;

package body producers_and_consumers is
   -- Protected object acting as the buffer
   -- for the producer and consumer tasks
   ------------------------------------------

   ------------
   -- buffer --
   ------------

   type Idx_type is mod 10;
   type circular_array is array (Idx_Type) of Integer;

   protected buffer is
      entry write (Item : in Integer);
      entry read (Item : out Integer);
      procedure set_done;
      function is_done return Boolean;
      function Buf_Empty return Boolean;
   private
      Buf           : circular_array;
      Write_Idx     : Idx_type := 0;
      Read_Idx      : Idx_type := 0;
      Producer_Done : Boolean  := False;
      Count         : Natural  := 0;
   end buffer;

   protected body buffer is
      entry write (Item : in Integer) when Count < Idx_type'Modulus is
      begin
         Buf (Write_Idx) := Item;
         Write_Idx       := Write_Idx + 1;
         Count           := Count + 1;
      end write;

      entry read (Item : out Integer) when Count > 0 is
      begin
         Item     := Buf (Read_Idx);
         Read_Idx := Read_Idx + 1;
         Count    := Count - 1;
      end read;

      procedure set_done is
      begin
         Producer_Done := True;
      end set_done;

      function Is_Done return Boolean is
      begin
         return Producer_Done;
      end Is_Done;

      function Buf_Empty return Boolean is
      begin
         return Count = 0;
      end Buf_Empty;

   end buffer;

   --------------
   -- producer --
   --------------

   task body producer is
   begin
      for I in 1 .. 30 loop
         buffer.write (I);
      end loop;
      buffer.set_done;
   end producer;

   --------------
   -- consumer --
   --------------

   task body consumer is
      Value : Integer;
   begin
      while not buffer.is_done loop
         buffer.read (Value);
         Put_Line ("Consumer read" & Value'Image);
      end loop;
      while not buffer.Buf_Empty loop
         buffer.read (Value);
         Put_Line ("Consumer read" & Value'Image);
      end loop;
   end consumer;

end producers_and_consumers;

The protected buffer is also written in two parts, a specification and a body. The specification declares the methods used to interface with the protected object in a public part and the data members of the protected object in a private part. Protected objects have three kinds of functions. A protected entry implicitly manipulates an exclusive read-write lock and executes only when its boundary condition evaluates to TRUE. A protected procedure manipulates an exclusive read-write lock and executes unconditionally. A protected function can only read values from the protected object. Protected functions implement a shared read lock.

This protected object implements two entries named write and read. It implements one procedure named set_done. It implements two functions named is_done and Buf_Empty.

The private part of the protected specification declares 5 data items.

• Buf is a circular array that will contain the data written to the protected object by the write entry and read from the protected object by the read entry.
• Write_Idx is a variable used to keep track of the index used by the write entry. Write_Idx is an instance of the type Idx_type defined in the package body. Idx_type is a modular type with valid values in the range of 0 through 9. All arithmetic applied to instances of a modular type are themselves modular. When an object of Idx_type containing a value of 9 is incremented the result will be 0.
• Read_Idx is a variable used to keep track of the index used by the read entry. Read_Idx, like Write_Idx, is an instance of Idx_type.
• Producer_Done is a Boolean variable initialized to False.
• Count is an instance of the subtype Natural, intialized to 0. The protected body defines all the entries, functions and the procedure.

The producer task body defines the logic for the producer task. The task simply writes the numbers 1 through 30 to the Buffer protected object then calls Buffer.set_done.

The consumer task body defines the logic for the consumer task. The task uses a while loop to iterate through the values in the Buffer protected object until Buffer.is_done returns TRUE. The consumer task then enters a second loop to iterate through the Buffer protected object until Buffer.Buf_Empty return True.

The Buffer protected object set_done, is_done and Buf_Empty methods provide the means to coordinate an orderly shut down of the producer and the consumer.

The main procedure, which is the program entry point in this example simply declares a dependency on the producers_and_consumers package, which automatically starts the producer and consumer tasks. The main procedure exits only after the producer and consumer tasks both complete.

The output of the program is:

C:\Users\jimma\Ada\Producer_Consumer\Auto_shutdown\obj\main.exe
Consumer read 1
Consumer read 2
Consumer read 3
Consumer read 4
Consumer read 5
Consumer read 6
Consumer read 7
Consumer read 8
Consumer read 9
Consumer read 10
Consumer read 11
Consumer read 12
Consumer read 13
Consumer read 14
Consumer read 15
Consumer read 16
Consumer read 17
Consumer read 18
Consumer read 19
Consumer read 20
Consumer read 21
Consumer read 22
Consumer read 23
Consumer read 24
Consumer read 25
Consumer read 26
Consumer read 27
Consumer read 28
Consumer read 29
Consumer read 30
[2024-10-04 20:07:51] process terminated successfully, elapsed time: 01.05s