v37 (Termination): Gradually terminate Consumers when Producers shutd…

…own by sending Poison Apples.

BlockingQueuePoisonApple.cfg

@@ -0,0 +1,17 @@
+CONSTANTS
+    BufCapacity = 5
+    Producers = {p1,p2,p3,p4,p5}
+    Consumers = {c1,c2,c3,c4,c5}
+    Poison = Poison
+
+SPECIFICATION SSpec
+
+INVARIANT TypeInv
+
+INVARIANT NoDeadlock
+
+PROPERTIES Terminates
+
+PROPERTIES QueueEmpty
+
+PROPERTIES GlobalTermination

BlockingQueuePoisonApple.tla

@@ -0,0 +1,156 @@
+---------------------- MODULE BlockingQueuePoisonApple ----------------------
+EXTENDS Naturals, Sequences, FiniteSets
+
+CONSTANTS Producers,   (* the (nonempty) set of producers                       *)
+          Consumers,   (* the (nonempty) set of consumers                       *)
+          BufCapacity, (* the maximum number of messages in the bounded buffer  *)
+          Poison
+
+ASSUME Assumption ==
+       /\ Producers # {}                      (* at least one producer *)
+       /\ Consumers # {}                      (* at least one consumer *)
+       /\ Producers \intersect Consumers = {} (* no thread is both consumer and producer *)
+       /\ BufCapacity \in (Nat \ {0})         (* buffer capacity is at least 1 *)
+
+-----------------------------------------------------------------------------
+
+\* prod maps each producer to its apple slices that it will eventually send to
+\* consumers. cons maps each consumer to the apple slices it has eaten so far. 
+VARIABLES prod, cons
+
+-----------------------------------------------------------------------------
+
+ap ==
+    \* The set of active producers, i.e., producers that have yet to send out
+    \* some bites of their poison apple.
+    {p \in DOMAIN prod: prod[p] > 0} \* <=> prod \notin [P -> {0}]
+
+ac == 
+    \* The set of active consumers, i.e., consumers that have yet to take a
+    \* bite of poison apples before they die/can terminate.
+    {c \in DOMAIN cons: cons[c] > 0} \* <=> cons \notin [C -> {0}]
+
+-----------------------------------------------------------------------------
+
+VARIABLES buffer, waitSet
+vars == <<buffer, waitSet, prod, cons>>
+
+NotifyOther(Others) == 
+    IF waitSet \cap Others # {}
+    THEN \E t \in waitSet \cap Others : waitSet' = waitSet \ {t}
+    ELSE UNCHANGED waitSet
+
+(* @see java.lang.Object#wait *)
+Wait(t) == /\ waitSet' = waitSet \cup {t}
+           /\ UNCHANGED <<buffer>>
+
+-----------------------------------------------------------------------------
+
+(* 
+  Contrary to PoisonPill, there is no dedicated "janitor" process that requires
+  some form of *external* synchronization. Global termination is achieved without
+  synchronizing producers.  Instead, the tradeoff of this variant is a tighter
+  coupling between producers and consumers because producers *and* consumers
+  need to know the number (cardinality) of the other role (see the definition of
+  prod  and cons  in  Init  below to see why that is).
+  Why is this complexity necessary, why can't we just have each producer send
+  a poison pill to one consumer (and some producers send more than one pill iff
+  |C| > |P|)? While this would work iff |C| >= |P|, it fails if |C| < |P|. Let
+  P_p be the subset of P such that |P_p| = |C|, i.e. the producers that poison
+  consumers. If one or more of the producers not in P_p remain active after all
+  producers in P_p terminated, there are no (active) consumers left.
+*)
+Put(t, d) ==
+    /\ t \notin waitSet
+    \* The Producer t must not have initiated its termination.
+    /\ prod[t] > 0
+    /\ \/ /\ Len(buffer) < BufCapacity
+          /\ \/ /\ buffer' = Append(buffer, d)
+                /\ UNCHANGED prod
+             \/ /\ buffer' = Append(buffer, Poison)
+                \* Producer t messages one consumer.
+                /\ prod' = [ prod EXCEPT ![t] = @ - 1]
+          /\ NotifyOther(Consumers)
+       \/ /\ Len(buffer) = BufCapacity
+          /\ Wait(t)
+          /\ UNCHANGED prod
+    /\ UNCHANGED <<cons>>
+
+Get(t) ==
+/\ t \notin waitSet
+/\ cons[t] > 0
+/\ \/ /\ buffer # <<>>
+      /\ buffer' = Tail(buffer)
+      /\ NotifyOther(Producers)
+      /\ \/ /\ Head(buffer) # Poison
+            /\ UNCHANGED <<prod,cons>>
+         \/ /\ Head(buffer) = Poison
+            /\ cons' = [ cons EXCEPT ![t] = @ - 1]
+            /\ UNCHANGED <<prod>>
+   \/ /\ buffer = <<>>
+      /\ Wait(t)
+      /\ UNCHANGED <<prod,cons>>
+
+-----------------------------------------------------------------------------
+
+(* Initially, the buffer is empty and no thread is waiting. *)
+Init == 
+    /\ buffer = <<>>
+    /\ waitSet = {}
+    \* The system requires |C| * |P| poison pills to terminate reliably. Perhaps,
+    \* this is what Goetz et. al. mean by "unwiedly" on page 156 in their book
+    \* Java Concurrency in Practice. However, it is unclear why the authors write
+    \* on the same page that "Poison pills work reliably only with unbounded queues."
+    /\ cons = [ c \in Consumers |-> Cardinality(Producers) ]
+    /\ prod = [ p \in Producers |-> Cardinality(Consumers) ]
+
+(* Then, pick a thread out of all running threads and have it do its thing. *)
+Next == 
+    \/ \E p \in Producers: Put(p, p) \* Add some data to buffer
+    \/ \E c \in Consumers: Get(c)
+
+Spec ==
+    /\ Init 
+    /\ [][Next]_vars
+    /\ \A c \in Consumers : WF_vars(Get(c)) 
+    /\ \A p \in Producers : WF_vars(Put(p, p))
+
+-----------------------------------------------------------------------------
+
+(* TLA+ is untyped, thus lets verify the range of some values in each state. *)
+TypeInv == 
+    /\ buffer \in Seq(Producers \cup {Poison})
+    /\ Len(buffer) \in 0..BufCapacity
+    /\ waitSet \in SUBSET (Producers \cup Consumers)
+    /\ prod \in [ Producers -> 0..Cardinality(Consumers) ]
+    /\ cons \in [ Consumers -> 0..Cardinality(Producers) ]
+
+(* No Deadlock *)
+NoDeadlock == waitSet # (Producers \cup Consumers)
+
+THEOREM Spec => [](TypeInv /\ NoDeadlock)
+-----------------------------------------------------------------------------
+
+\* The queue is empty after (global) termination.
+QueueEmpty ==
+    []((ap \cup ac) = {}) => (buffer = <<>>)
+
+\* The system terminates iff all producers terminate.
+GlobalTermination ==
+    (ap = {}) ~> [](ac = {})
+
+THEOREM Spec => QueueEmpty /\ GlobalTermination
+-----------------------------------------------------------------------------
+
+SSpec ==
+    /\ Init 
+    /\ [][Next]_vars
+    /\ \A c \in Consumers : WF_vars(Get(c))
+    \* This fairness constraint causes all producers to eventually terminate.
+    /\ \A p \in Producers : SF_vars(Put(p, p) /\ prod' # prod)
+
+Terminates ==
+    <>[](ap = {})
+
+THEOREM SSpec => Terminates
+=============================================================================

README.md

@@ -13,6 +13,12 @@ This tutorial is work in progress. More chapters will be added in the future. In
 
 --------------------------------------------------------------------------
 
+### v37 (Termination): Gradually terminate Consumers when Producers shutdown by sending Poison Apples.
+
+Poison Pill variant where each producer sends `N` slices of a Poison Apple to `N` consumers. A consumer dies (terminates) when it has eaten `M` apple slices, where `M` is the number of producers.
+
+The advantage over the Poison Pill approach is that no extra synchronization primitive is required to detect global producer termination that triggers sending the poison pills. Also, consumers down-scale gradually following the producer termination.
+
 ### v36 (Termination): Check BlockingQueuePoisonPill for all subsets of the constants by mimicking Apalache's ConstInit feature.
 
 ### v35 (Termination): Check refinement of BlockingQueue by BlockingQueuePoisonPill.