Anti-pipelining (TODO better name) (AKA receiving despite having agency)

typed-protocols/src/Network/TypedProtocol/Core.hs

@@ -44,6 +44,7 @@ module Network.TypedProtocol.Core
   , IsPipelined (..)
     -- *** Outstanding
   , Outstanding
+  , AntiOutstanding
     -- *** N and Nat
   , N (..)
   , Nat (Succ, Zero)
@@ -491,13 +492,17 @@ data N = Z | S N
 -- | Promoted data type which indicates if 'Peer' is used in
 -- pipelined mode or not.
 --
-data IsPipelined where
+data IsPipelined ps where
     -- | Pipelined peer which is using `c :: Type` for collecting responses
     -- from a pipelined messages. 'N' indicates depth of pipelining.
-    Pipelined    :: N -> Type -> IsPipelined
+    Pipelined     :: N -> Type -> IsPipelined ps
 
     -- | Non-pipelined peer.
-    NonPipelined :: IsPipelined
+    NonPipelined  :: IsPipelined ps
+
+    -- | Pipelined peer for a /server/ that only ever uses one
+    -- 'Network.TypedProtocol.Peer.Sender'
+    AntiPipelined :: ps -> ps -> N -> IsPipelined ps
 
 -- | Type level count of the number of outstanding pipelined yields for which
 -- we have not yet collected a receiver result. Used to
@@ -506,10 +511,17 @@ data IsPipelined where
 -- and to ensure that the non-pipelined primitives 'Yield', 'Await' and 'Done'
 -- are only used when there are none unsatisfied pipelined requests.
 --
-type        Outstanding :: IsPipelined -> N
+type        Outstanding :: IsPipelined ps -> N
 type family Outstanding pl where
-  Outstanding 'NonPipelined    = Z
-  Outstanding ('Pipelined n _) = n
+  Outstanding 'NonPipelined          = Z
+  Outstanding ('Pipelined n _)       = n
+  Outstanding ('AntiPipelined _ _ _) = Z
+
+type        AntiOutstanding :: IsPipelined ps -> N
+type family AntiOutstanding pl where
+  AntiOutstanding 'NonPipelined          = Z
+  AntiOutstanding ('Pipelined _ _)       = Z
+  AntiOutstanding ('AntiPipelined _ _ n) = n
 
 -- | A value level inductive natural number, indexed by the corresponding type
 -- level natural number 'N'.

typed-protocols/src/Network/TypedProtocol/Driver.hs

@@ -1,5 +1,6 @@
-{-# LANGUAGE CPP          #-}
-{-# LANGUAGE TypeFamilies #-}
+{-# LANGUAGE CPP                 #-}
+{-# LANGUAGE ScopedTypeVariables #-}
+{-# LANGUAGE TypeFamilies        #-}
 
 -- | Actions for running 'Peer's with a 'Driver'
 --
@@ -13,14 +14,19 @@ module Network.TypedProtocol.Driver
   , runPeerWithDriver
     -- * Pipelined peers
   , runPipelinedPeerWithDriver
+    -- * Anti-pipelined peers
+  , runAntiPipelinedPeerWithDriver
   ) where
 
+import Control.Monad (forever, join)
 import Data.Void (Void)
+import Numeric.Natural (Natural)
 
 import Network.TypedProtocol.Core
 import Network.TypedProtocol.Peer
 
 import Control.Concurrent.Class.MonadSTM.TQueue
+import Control.Concurrent.Class.MonadSTM.TVar
 import Control.DeepSeq (NFData, force)
 import Control.Monad.Class.MonadAsync
 import Control.Monad.Class.MonadFork
@@ -363,3 +369,127 @@ runPipelinedPeerReceiver Driver{recvMessage} = go
     go dstate (ReceiverAwait refl k) = do
       (SomeMessage msg, dstate') <- recvMessage refl dstate
       go dstate' (k msg)
+
+
+--
+-- Running anti-pipelined peers
+--
+
+-- | Run an anti-pipelined peer with the given driver.
+--
+-- Dual to 'runPipelinedPeerWithDriver': where a pipelined peer sends ahead and
+-- defers its receives to a parallel receiver thread, an anti-pipelined peer
+-- receives ahead and defers its sends to a parallel sender thread.
+--
+-- Unlike the pipelined driver, there is no trailing-data handoff: the peer
+-- thread performs every 'recvMessage' (so it owns @dstate@ outright), and the
+-- sender thread performs every 'sendMessage'. The two only ever touch opposite
+-- directions of the channel, and the 'AntiOutstanding' index guarantees that
+-- the peer thread's own sends ('Yield'\/'Done') happen only when the sender
+-- thread is idle.
+--
+runAntiPipelinedPeerWithDriver
+  :: forall ps (st :: ps) pr dstate m a.
+     ( MonadAsync m
+     , MonadEvaluate m
+     , NFData a
+     )
+  => Driver ps pr dstate m
+  -> PeerAntiPipelined ps pr st m a
+  -> m (a, dstate)
+runAntiPipelinedPeerWithDriver driver@Driver{initialDState} (PeerAntiPipelined sender peer) = do
+    sendVar <- newTVarIO 0
+    doneVar <- newTVarIO 0
+    r@(a, _dstate) <- runAntiPipelinedPeerSender sender sendVar doneVar driver
+           `withAsyncLoop`
+         runAntiPipelinedPeerMain       sendVar doneVar driver peer initialDState
+
+    _ <- evaluate (force a)
+    return r
+
+  where
+    withAsyncLoop :: m Void -> m x -> m x
+    withAsyncLoop left right = do
+      -- race will throw if either of the threads throw
+      res <- race left right
+      case res of
+        Left v  -> case v of {}
+        Right a -> return a
+
+
+runAntiPipelinedPeerMain
+  :: forall ps (apst :: ps) (apst' :: ps) (st :: ps) pr dstate m a.
+     ( MonadSTM    m
+     , MonadThread m
+     )
+  => TVar m Natural
+  -> TVar m Natural
+  -> Driver ps pr dstate m
+  -> Peer ps pr ('AntiPipelined apst apst' Z) st m a
+  -> dstate
+  -> m (a, dstate)
+runAntiPipelinedPeerMain sendVar doneVar
+                         Driver{sendMessage, recvMessage}
+                         peer0 dstate0 = do
+    threadId <- myThreadId
+    labelThread threadId "antipipelined-peer-main"
+    go dstate0 peer0
+  where
+    go :: forall st' n.
+          dstate
+       -> Peer ps pr ('AntiPipelined apst apst' n) st' m a
+       -> m (a, dstate)
+    go dstate (Effect k) = k >>= go dstate
+    go dstate (Done _ x) = return (x, dstate)
+
+    -- Only reachable at 'AntiPipelined Z' (the constructor demands
+    -- @AntiOutstanding ~ Z@), i.e. when the sender thread is provably idle.
+    go dstate (Yield refl msg k) = do
+      sendMessage refl msg
+      go dstate k
+
+    -- Legal at any 'AntiOutstanding': receiving ahead is the whole point.
+    go dstate (Await refl k) = do
+      (SomeMessage msg, dstate') <- recvMessage refl dstate
+      go dstate' (k msg)
+
+    go dstate (YieldAntiPipelined _refl k) = do
+      atomically $ modifyTVar' sendVar (+ 1)
+      go dstate k
+
+    go dstate (AntiCollect k mbNonBlocking) = do
+      join $ atomically $ do
+        n <- readTVar doneVar
+        if n > 0
+          then do writeTVar doneVar (n - 1); pure $ go dstate k
+          else case mbNonBlocking of
+            Nothing -> retry
+            Just k' -> pure $ go dstate k'
+
+runAntiPipelinedPeerSender
+  :: forall ps pr apst apst' dstate m.
+     ( MonadSTM    m
+     , MonadThread m
+     )
+  => Sender ps pr apst apst' m
+  -> TVar m Natural
+  -> TVar m Natural
+  -> Driver ps pr dstate m
+  -> m Void
+runAntiPipelinedPeerSender sender sendVar doneVar
+                                Driver{sendMessage} = do
+
+    threadId <- myThreadId
+    labelThread threadId "antipipelined-sender"
+    forever $ do
+      atomically $ do n <- readTVar sendVar; check (0 < n); writeTVar sendVar $! n - 1
+      runSender sender
+      atomically $ modifyTVar' doneVar (+ 1)
+  where
+    runSender :: forall stA stZ. Sender ps pr stA stZ m -> m ()
+    runSender = \case
+      SenderEffect k         -> k >>= runSender
+      SenderDone             -> return ()
+      SenderYield refl msg k -> do
+        sendMessage refl msg
+        runSender k

typed-protocols/src/Network/TypedProtocol/Peer.hs

@@ -5,7 +5,9 @@
 module Network.TypedProtocol.Peer
   ( Peer (..)
   , PeerPipelined (..)
+  , PeerAntiPipelined (..)
   , Receiver (..)
+  , Sender (..)
   , Outstanding
   , N (..)
   , Nat (Zero, Succ)
@@ -79,7 +81,7 @@ import Network.TypedProtocol.Core as Core
 --
 type Peer :: forall ps
           -> PeerRole
-          -> IsPipelined
+          -> IsPipelined ps
           -> ps
           -> (Type -> Type)
           -- ^ monad's kind
@@ -115,6 +117,7 @@ data Peer ps pr pl st m a where
        , StateTokenI st'
        , ActiveState st
        , Outstanding pl ~ Z
+       , AntiOutstanding pl ~ Z
        )
     => WeHaveAgencyProof pr st
     -- ^ agency proof
@@ -169,6 +172,7 @@ data Peer ps pr pl st m a where
        ( StateTokenI st
        , StateAgency st ~ NobodyAgency
        , Outstanding pl ~ Z
+       , AntiOutstanding pl ~ Z
        )
     => NobodyHasAgencyProof pr st
     -- ^ (no) agency proof
@@ -214,8 +218,33 @@ data Peer ps pr pl st m a where
     -- ^ continuation
     -> Peer        ps pr (Pipelined (S n) c) st m a
 
-deriving instance Functor m => Functor (Peer ps pr pl st m)
+  -- | 'AntiPipelined' analog of 'YieldPipelined'
+  --
+  -- Always uses the 'Sender' that was provided to the driver
+  -- alongside this 'Peer'.
+  YieldAntiPipelined
+    :: forall ps pr (st :: ps) n (st' :: ps) m a.
+       ( StateTokenI st
+       , StateTokenI st'
+       , ActiveState st
+       )
+    => !(WeHaveAgencyProof pr st)
+    -> Peer ps pr (AntiPipelined st st' (S n)) st' m a
+       -- ^ continuation, before or after sending
+    -> Peer ps pr (AntiPipelined st st'  n ) st  m a
+
+  AntiCollect
+    :: forall ps pr n st apst apst' m a.
+       StateTokenI st
+    =>        Peer ps pr (AntiPipelined apst apst'   n ) st m a
+       -- ^ how to proceed if the @n+1@fst 'Sender' has already terminated
+    -> Maybe (Peer ps pr (AntiPipelined apst apst' (S n)) st m a)
+       -- ^ 'Just' if and only if the peer can proceed before the @n+1@st 'Sender' has terminated
+       --
+       -- This is ignored if a message has already been sent
+    -> Peer ps pr (AntiPipelined apst apst' (S n)) st m a
 
+deriving instance Functor m => Functor (Peer ps pr pl st m)
 
 -- | Receiver.  It is limited to only awaiting for messages and running monadic
 -- computations.  This means that one can only pipeline messages if they can be
@@ -260,6 +289,29 @@ data Receiver ps pr st stdone m c where
 
 deriving instance Functor m => Functor (Receiver ps pr st stdone m)
 
+-- | 'AntiPipelined' analog of 'Receiver'
+type Sender :: forall ps
+            -> PeerRole
+            -> ps
+            -> ps
+            -> (Type -> Type)
+            -> Type
+data Sender ps pr st stdone m where
+
+  SenderEffect :: m (Sender ps pr st stdone m)
+               ->    Sender ps pr st stdone m
+
+  SenderDone   :: Sender ps pr stdone stdone m
+
+  SenderYield  :: ( StateTokenI st
+                  , StateTokenI st'
+                  , ActiveState st
+                  )
+               => !(WeHaveAgencyProof pr st)
+               -> Message ps st st'
+               -> Sender ps pr st' stdone m
+               -> Sender ps pr st  stdone m
+
 -- | A description of a peer that engages in a protocol in a pipelined fashion.
 --
 -- This type is useful for wrapping pipelined peers to hide information which
@@ -271,3 +323,11 @@ data PeerPipelined ps pr (st :: ps) m a where
                   -> PeerPipelined ps pr st m a
 
 deriving instance Functor m => Functor (PeerPipelined ps pr st m)
+
+data PeerAntiPipelined ps pr (st :: ps) m a where
+    PeerAntiPipelined ::
+      Sender ps pr apst apst' m ->
+      Peer ps pr (AntiPipelined apst apst' Z) st m a ->
+      PeerAntiPipelined ps pr st m a
+
+deriving instance Functor m => Functor (PeerAntiPipelined ps pr st m)

typed-protocols/src/Network/TypedProtocol/Peer/Client.hs

@@ -36,7 +36,7 @@ import Network.TypedProtocol.Peer qualified as TP
 
 
 type Client :: forall ps
-            -> IsPipelined
+            -> IsPipelined ps
             -> ps
             -> (Type -> Type)
             -> Type
@@ -76,6 +76,7 @@ pattern Yield :: forall ps pl st m a.
                  , StateTokenI st'
                  , StateAgency st ~ ClientAgency
                  , Outstanding pl ~ Z
+                 , AntiOutstanding pl ~ Z
                  )
               => Message ps st st'
               -- ^ protocol message
@@ -107,6 +108,7 @@ pattern Done :: forall ps pl st m a.
              => ( StateTokenI st
                 , StateAgency st ~ NobodyAgency
                 , Outstanding pl ~ Z
+                , AntiOutstanding pl ~ Z
                 )
              => a
              -- ^ protocol return value

typed-protocols/src/Network/TypedProtocol/Peer/Server.hs

@@ -37,7 +37,7 @@ import Network.TypedProtocol.Peer qualified as TP
 
 
 type Server :: forall ps
-            -> IsPipelined
+            -> IsPipelined ps
             -> ps
             -> (Type -> Type)
             -> Type
@@ -78,6 +78,7 @@ pattern Yield :: forall ps pl st m a.
                  , StateTokenI st'
                  , StateAgency st ~ ServerAgency
                  , Outstanding pl ~ Z
+                 , AntiOutstanding pl ~ Z
                  )
               => Message ps st st'
               -- ^ protocol message
@@ -109,6 +110,7 @@ pattern Done :: forall ps pl st m a.
              => ( StateTokenI st
                 , StateAgency st ~ NobodyAgency
                 , Outstanding pl ~ Z
+                , AntiOutstanding pl ~ Z
                 )
              => a
              -- ^ protocol return value