ATC: Interrupting Blocking Calls

In this example we have a bounded buffer with operations to Put and Get
characters.  A producer thread puts characters into the buffer, and a
consumer thread gets characters.

If the buffer is full, we want the producer to block until there is more
room in the buffer.  If the buffer is empty, we want the consumer to
block until there's something in the buffer.

We also want to be able to tell the waiting consumer to stop getting
characters.  This requires some way of interrupting the blocked task.

Implementation

The bounded buffer is implemented as an array encapsulated by a protected
type, with a discriminant that indicates the size:

  generic
     type Item_Type is private;
  package Buffers is

     type Item_Array is array (Positive range <>) of Item_Type;

     protected type Buffer_Type (Size : Positive) is

        ...

     private

        Items     : Item_Array (1 .. Size);
        Get_Index : Positive := 1;
        Put_Index : Positive := Size;
        Count     : Natural := 0;

     end Buffer_Type;

  end Buffers;


We want the producer to block if the buffer is full, so we implement the
Put operation as a protected entry:

      entry Put (Item : in Item_Type);

The entry barrier tests whether the count of the number of items is less
then the buffer size, and if so, copies the item to the array:

      entry Put (Item : in Item_Type)
        when Count < Size is
      begin
         Put_Index := Put_Index mod Size + 1;
         Items (Put_Index) := Item;
         Count := Count + 1;
      end;

We implement Get using separate operations for waiting until there's
an item, and actually fetching the item:

      entry Wait_For_Item;

      procedure Get (Item : out Item_Type);


We've done it this way because we only want to interrupt waiting for an
item, not retrieving one.  If we were to interrupt a combined
wait-and-fetch operation, then an item could get lost.

The entry Wait_For_Item has only a barrier to test whether there's an
item in the buffer:

      entry Wait_For_Item when Count > 0 is
      begin
         null;
      end;

The protected procedure Get copies the first item in the array, and
increments the index:

      procedure Get (Item : out Item_Type) is
      begin
         Item := Items (Get_Index);
         Get_Index := Get_Index mod Size + 1;
         Count := Count - 1;
      end;


Our consumer abstraction binds to a character buffer via an access
discriminant, and provides an operation to shut down the thread:

  package Consumers is

     type Consumer_Type (Buffer : access Buffer_Type) is
        limited private;

     procedure Shutdown (Consumer : in out Consumer_Type);

  private

     ...

  end Consumers;


The Consumer_Type has an active thread of control, implemented as a task
component.  The type is not implemented as a task directly, because (as
we shall see) it needs another component.

  private

     task type Consumer_Task_Type (Consumer : access Consumer_Type) is
     end;

     ...

     type Consumer_Type (Buffer : access Buffer_Type) is
        limited record
           ...
           Consumer_Task : Consumer_Task_Type (Consumer_Type'Access);
        end record;

  end Consumers;


The Consumer_Task_Type has its own discriminant, to bind to the
enclosing Consumer_Type.  This gives the task visibility to the Buffer
and to the other components of the same record.

The consumer task is implemented as a loop that waits for a character,
fetches it, then displays it:

      Main:
      loop
         <wait for character>

         Buffer.Get (C);
         Put_Line ("Consumed '" & C & "'");
      end loop Main;


If we were to implement the wait as a straight call to the buffer's
protected entry:

         Buffer.Wait_For_Item;

then there would be no way to interrupt the entry call once in progress.
You can't use a task entry for this purpose, because the task is not
waiting to accept them -- it's blocked on the buffer call.

In order to interrupt the waiting task, we use Asynchronous Transfer of
Control (ATC).  By moving the wait to the abortable region,

         select
            <triggering alternative>
            exit Main;
         then abort
            Buffer.Wait_For_Item;
         end select;

then we can interrupt the call when a triggering alternative is
selected, which can be either a delay or an entry call.

The task itself can't handle the shutdown, because it's blocked waiting
for a character from the buffer.  So what we do is introduce an internal
protected type to receive the shutdown request from a client, and with
another entry to act as the triggering alternative for the task:

   protected type Trigger_Type is
      procedure Shutdown;
      entry Stop_Consuming;
   private
      Stop_Requested : Boolean := False;
   end Trigger_Type;


The protected procedure Shutdown is just sets a flag that indicates
we're done:

      procedure Shutdown is
      begin
         Stop_Requested := True;
      end;


The protected entry Stop_Consuming blocks until the done flag is true:

      entry Stop_Consuming when Stop_Requested is
      begin
         null;
      end;


The complete implementation of the Consumer_Type thus looks like this:

   type Consumer_Type (Buffer : access Buffer_Type) is
      limited record
         Trigger       : Trigger_Type;
         Consumer_Task : Consumer_Task_Type (Consumer_Type'Access);
      end record;


The consumer task calls the Stop_Consuming entry of the trigger:

         select
            Consumer.Trigger.Stop_Consuming;
            exit Main;
         then abort
            Buffer.Wait_For_Item;
         end select;


This (asynchronous) select statement will complete either because the
trigger accepts the Stop_Consuming entry, or because the Buffer accepts
the Wait_For_Item entry.

Shutdown is implemented by calling the trigger's protected Shutdown
procedure:

   procedure Shutdown (Consumer : in out Consumer_Type) is
   begin
      Consumer.Trigger.Shutdown;
   end;

This sets the done flag to True, which immediately forces the barrier
for Stop_Consuming to be reevaluated.  The trigger object accepts the
entry call, which triggers abortion of the Wait_For_Item call already in
progress.

There's one more issue to discuss.  Text_IO is not a concurrent
abstraction, and it's possible that simultaneous calls to Put_Line can
produce interleaved output.

The package Message_IO provides a thread-safe version of Put_Line that
synchronizes its callers:

  package Message_IO is

     procedure Put_Line (Message : in String);

  end Message_IO;


The body of Message_IO declares a semaphore that is seized and released
during calls to Put_Line:

  package body Message_IO is

     Semaphore : aliased Semaphore_Type;

     procedure Put_Line (Message : in String) is
        Control : Semaphore_Control (Semaphore'Access);
     begin
        Ada.Text_IO.Put_Line (Message);
     end;

  end Message_IO;


We use a controlled object to automatically seize and release the
semaphore, which guarantees that the semaphore will get released even if
there's an exception.

Note that we can't just wrap the Text_IO.Put_Line call in a protected
procedure:

  protected body Synchronization is

     procedure Do_Put_Line (S : String) is
     begin
        Text_IO.Put_Line (S);
     end;

  end Synchronization;

because Text_IO.Put_Line is a "potentially blocking call," and it is a
bounded error to invoke a blocking operation from within a protected
operation.  See RM95 9.5.1 (8 - 18).

For whatever reason, the test program dumps core as it terminates
because of a segmentation violation.  I don't know why...


--STX
package body Binary_Semaphores.Controls is

   procedure Initialize (Control : in out Semaphore_Control) is
   begin
      Control.Semaphore.Seize;
   end;

   procedure Finalize (Control : in out Semaphore_Control) is
   begin
      Control.Semaphore.Release;
   end;

end Binary_Semaphores.Controls;
with Ada.Finalization;

package Binary_Semaphores.Controls is

   pragma Preelaborate;

   type Semaphore_Control (Semaphore : access Semaphore_Type) is
      limited private;

private

   use Ada.Finalization;

   type Semaphore_Control (Semaphore : access Semaphore_Type) is
     new Limited_Controlled with null record;

   procedure Initialize (Control : in out Semaphore_Control);

   procedure Finalize (Control : in out Semaphore_Control);

end Binary_Semaphores.Controls;
package body Binary_Semaphores is

   protected body Semaphore_Type is

      procedure Release is
      begin
         In_Use := False;
      end;

      entry Seize when not In_Use is
      begin
         In_Use := True;
      end;

   end Semaphore_Type;

end Binary_Semaphores;
package Binary_Semaphores is

   pragma Pure;


   protected type Semaphore_Type is

      procedure Release;

      entry Seize;

   private

      In_Use : Boolean := False;

   end Semaphore_Type;

end Binary_Semaphores;
package body Buffers is

   protected body Buffer_Type is

      entry Put (Item : in Item_Type)
        when Count < Size is
      begin
         Put_Index := Put_Index mod Size + 1;
         Items (Put_Index) := Item;
         Count := Count + 1;
      end;

      entry Wait_For_Item when Count > 0 is
      begin
         null;
      end;

      procedure Get (Item : out Item_Type) is
      begin
         Item := Items (Get_Index);
         Get_Index := Get_Index mod Size + 1;
         Count := Count - 1;
      end;

      function Is_Full return Boolean is
      begin
         return Count = Size;
      end;

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

   end Buffer_Type;

end Buffers;



generic
   type Item_Type is private;
package Buffers is

   type Item_Array is array (Positive range <>) of Item_Type;

   protected type Buffer_Type (Size : Positive) is

      entry Put (Item : in Item_Type);

      entry Wait_For_Item;

      procedure Get (Item : out Item_Type);

      function Is_Full return Boolean;

      function Is_Empty return Boolean;

   private

      Items     : Item_Array (1 .. Size);
      Get_Index : Positive := 1;
      Put_Index : Positive := Size;
      Count     : Natural := 0;

   end Buffer_Type;

end Buffers;



with Buffers;

package Character_Buffers is new Buffers (Character);
with Message_IO; use Message_IO;

package body Consumers is

   procedure Shutdown (Consumer : in out Consumer_Type) is
   begin
      Consumer.Trigger.Shutdown;
   end;


   protected body Trigger_Type is

      procedure Shutdown is
      begin
         Stop_Requested := True;
      end;

      entry Stop_Consuming when Stop_Requested is
      begin
         null;
      end;

   end Trigger_Type;


   task body Consumer_Task_Type is

      C : Character;

      Buffer : Buffer_Type renames Consumer.Buffer.all;

   begin

      Main:
      loop

         select
            Consumer.Trigger.Stop_Consuming;
            exit Main;
         then abort
            Buffer.Wait_For_Item;
         end select;

         Buffer.Get (C);
         Put_Line ("Consumed '" & C & "'");

      end loop Main;

      while not Buffer.Is_Empty loop
         Buffer.Get (C);
         Put_Line ("Consumed '" & C & "'");
      end loop;

      Put_Line ("Consumer done.");

   end Consumer_Task_Type;

end Consumers;

with Character_Buffers;  use Character_Buffers;

package Consumers is

   type Consumer_Type (Buffer : access Buffer_Type) is
      limited private;

   procedure Shutdown (Consumer : in out Consumer_Type);

private

   task type Consumer_Task_Type (Consumer : access Consumer_Type) is
   end;

   protected type Trigger_Type is
      procedure Shutdown;
      entry Stop_Consuming;
   private
      Stop_Requested : Boolean := False;
   end Trigger_Type;

   type Consumer_Type (Buffer : access Buffer_Type) is
      limited record
         Trigger       : Trigger_Type;
         Consumer_Task : Consumer_Task_Type (Consumer_Type'Access);
      end record;

end Consumers;

with Binary_Semaphores.Controls;
with Ada.Text_IO;

package body Message_IO is

   use Binary_Semaphores;
   Semaphore : aliased Semaphore_Type;


   procedure Put_Line (Message : in String) is
      use Controls;
      Control : Semaphore_Control (Semaphore'Access);
   begin
      Ada.Text_IO.Put_Line (Message);
   end;

end Message_IO;

package Message_IO is

   procedure Put_Line (Message : in String);

end Message_IO;


with Ada.Text_IO;
with Message_IO;         use Message_IO;
with Character_Buffers;  use Character_Buffers;
with Consumers;          use Consumers;

procedure Test_Buffers is

   Buffer : aliased Buffer_Type (Size => 5);

   Consumer : Consumer_Type (Buffer'Access);

   C : Character;

begin

   Put_Line ("Enter characters, and use '0' to indicate you're done.");

   loop
      Ada.Text_IO.Get (C);
      Buffer.Put (C);
      exit when C = '0';
   end loop;

   Put_Line ("Shutting down consumer");
   Shutdown (Consumer);
   Put_Line ("Shutting down complete; exiting.");

end Test_Buffers;