---------------------- MODULE BlockingQueueFairMinimal ----------------------
EXTENDS Naturals, Sequences, FiniteSets, TLAPS

Range(f) == { f[x] : x \in DOMAIN f }

IsInjective(s) == \A i, j \in DOMAIN s: (s[i] = s[j]) => (i = j)

CONSTANTS Producers,   (* the (nonempty) set of producers                       *)
          Consumers,   (* the (nonempty) set of consumers                       *)
          BufCapacity  (* the maximum number of messages in the bounded buffer  *)

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 *)
       
-----------------------------------------------------------------------------

VARIABLES buffer, waitSeqC, waitSeqP
vars == <<buffer, waitSeqC, waitSeqP>>

WaitingThreads == Range(waitSeqC) \cup Range(waitSeqP)

RunningThreads == (Producers \cup Consumers) \ WaitingThreads

NotifyOther(ws) ==
            \/ /\ ws = <<>>
               /\ UNCHANGED ws
            \/ /\ ws # <<>>
               /\ ws' = Tail(ws)

(* @see java.lang.Object#wait *)
Wait(ws, t) == 
            /\ ws # <<>>
            /\ ws' = ws \o <<t>>
            /\ UNCHANGED <<buffer>>
           
-----------------------------------------------------------------------------

Put(t, d) ==
/\ t \notin Range(waitSeqP)
/\ \/ 
      /\ Len(buffer) < BufCapacity
      /\ buffer' = Append(buffer, d)
      /\ NotifyOther(waitSeqC)
      /\ UNCHANGED waitSeqP
   \/ /\ Len(buffer) = BufCapacity
      /\ Wait(waitSeqP, t)
      /\ UNCHANGED waitSeqC
      
Get(t) ==
/\ t \notin Range(waitSeqC)
/\ \/ /\ buffer # <<>>
      /\ buffer' = Tail(buffer)
      /\ NotifyOther(waitSeqP)
      /\ UNCHANGED waitSeqC
   \/ /\ buffer = <<>>
      /\ Wait(waitSeqC, t)
      /\ UNCHANGED waitSeqP

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

Init == /\ buffer = <<>>
        /\ waitSeqC = <<>>
        /\ waitSeqP = <<>>

Next == \/ \E p \in Producers: Put(p, p) \* Add some data to buffer
        \/ \E c \in Consumers: Get(c)

Spec == Init /\ [][Next]_vars 

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

BQS == INSTANCE BlockingQueueSplit WITH waitSetC <- Range(waitSeqC), 
                                        waitSetP <- Range(waitSeqP)

TypeInv == /\ Len(buffer) \in 0..BufCapacity
           /\ waitSeqP \in Seq(Producers)
           /\ IsInjective(waitSeqP)
           /\ waitSeqC \in Seq(Consumers)
           /\ IsInjective(waitSeqC)

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

THEOREM Spec => BQS!Spec
<1> USE Assumption, BQS!Assumption
<1>1. Init => BQS!Init 
<1>2. TypeInv /\ [Next]_vars => [BQS!Next]_BQS!vars
  <2> USE DEF BQS!Next, BQS!vars
  <2> SUFFICES ASSUME TypeInv,
                      [Next]_vars
               PROVE  [BQS!Next]_BQS!vars
    OBVIOUS
  <2>1. ASSUME NEW p \in Producers,
               Put(p, p)
        PROVE  [BQS!Next]_BQS!vars
    <3>1. CASE       /\ Len(buffer) < BufCapacity
                     /\ buffer' = Append(buffer, p)
                     /\ NotifyOther(waitSeqC)
                     /\ UNCHANGED waitSeqP
      <4>1. CASE /\ waitSeqC = <<>>
                 /\ UNCHANGED waitSeqC
      <4>2. CASE /\ waitSeqC # <<>>
                 /\ waitSeqC' = Tail(waitSeqC)
        BY SMT, <4>2, <2>1, <3>1, 
           BQS!ITypeInv \* This causes a "Queue.nth: internal error"
        DEF Put, NotifyOther,
            BQS!Put, BQS!NotifyOther, Range,
            BQS!TypeInv \* This causes a "Queue.nth: internal error"
      <4>3. QED
    <3>2. QED
  <2>2. ASSUME NEW c \in Consumers,
               Get(c)
        PROVE  [BQS!Next]_BQS!vars
  <2>3. CASE UNCHANGED vars
  <2>4. QED
<1>3. QED

===============

--------------------- MODULE BlockingQueueSplitMinimal ---------------------
EXTENDS Naturals, Sequences, FiniteSets, TLAPS

CONSTANTS Producers,   (* the (nonempty) set of producers                       *)
          Consumers,   (* the (nonempty) set of consumers                       *)
          BufCapacity  (* the maximum number of messages in the bounded buffer  *)

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 *)
       
VARIABLES buffer, waitSetC, waitSetP
vars == <<buffer, waitSetC, waitSetP>>

RunningThreads == (Producers \cup Consumers) \ (waitSetC \cup waitSetP)

NotifyOther(ws) == 
         \/ /\ ws = {}
            /\ UNCHANGED ws
         \/ /\ ws # {}
            /\ \E x \in ws: ws' = ws \ {x}

Wait(ws, t) == /\ ws' = ws \cup {t}
               /\ UNCHANGED <<buffer>>
           
Put(t, d) ==
/\ t \notin waitSetP
/\ \/ /\ Len(buffer) < BufCapacity
      /\ buffer' = Append(buffer, d)
      /\ NotifyOther(waitSetC)
      /\ UNCHANGED waitSetP
   \/ /\ Len(buffer) = BufCapacity
      /\ Wait(waitSetP, t)
      /\ UNCHANGED waitSetC
      
Get(t) ==
/\ t \notin waitSetC
/\ \/ /\ buffer # <<>>
      /\ buffer' = Tail(buffer)
      /\ NotifyOther(waitSetP)
      /\ UNCHANGED waitSetC
   \/ /\ buffer = <<>>
      /\ Wait(waitSetC, t)
      /\ UNCHANGED waitSetP


TypeInv == /\ Len(buffer) \in 0..BufCapacity
           /\ waitSetP \in SUBSET Producers
           /\ waitSetC \in SUBSET Consumers

Init == /\ buffer = <<>>
        /\ waitSetC = {}
        /\ waitSetP = {}

Next == \/ \E p \in Producers: Put(p, p)
        \/ \E c \in Consumers: Get(c)

Spec == Init /\ [][Next]_vars

LEMMA ITypeInv == Spec => []TypeInv
<1> USE Assumption DEF TypeInv
<1>1. Init => TypeInv
  BY SMT DEF Init
<1>2. TypeInv /\ [Next]_vars => TypeInv'
  BY SMT DEF Next, vars, Put, Get, Wait, NotifyOther
<1>3. QED
  BY <1>1, <1>2, PTL DEF Spec

=============================================================================