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

typed-protocols/examples/Network/TypedProtocol/Driver/Simple.hs

@@ -15,9 +15,12 @@ module Network.TypedProtocol.Driver.Simple
   , Role (..)
     -- * Pipelined peers
   , runPipelinedPeer
+    -- * Anti-pipelined peers
+  , runAntiPipelinedPeer
     -- * Connected peers
   , runConnectedPeers
   , runConnectedPeersPipelined
+  , runConnectedPeersAntiPipelined
   , runConnectedPeersAsymmetric
     -- * Driver utilities
     -- | This may be useful if you want to write your own driver.
@@ -169,6 +172,33 @@ runPipelinedPeer tracer codec channel peer =
     driver = driverSimple tracer codec channel
 
 
+-- | Run an anti-pipelined peer with the given channel via the given codec.
+--
+-- This runs the peer to completion (if the protocol allows for termination).
+--
+-- Like pipelined peers, anti-pipelined peers rely on concurrency (the 'Sender'
+-- runs in parallel with the main peer), hence the 'MonadAsync' constraint.
+--
+runAntiPipelinedPeer
+  :: forall ps (st :: ps) pr failure bytes m a.
+     ( MonadAsync m
+     , MonadEvaluate m
+     , MonadThrow m
+     , Exception failure
+     , NFData failure
+     , NFData a
+     )
+  => Tracer m (TraceSendRecv ps)
+  -> Codec ps failure m bytes
+  -> Channel m bytes
+  -> PeerAntiPipelined ps pr st m a
+  -> m (a, Maybe bytes)
+runAntiPipelinedPeer tracer codec channel peer =
+    runAntiPipelinedPeerWithDriver driver peer
+  where
+    driver = driverSimple tracer codec channel
+
+
 --
 -- Utils
 --
@@ -255,6 +285,38 @@ runConnectedPeersPipelined createChannels tracer codec client server =
     tracerServer = contramap ((,) AsServer) tracer
 
 
+-- | Run a pipelined client against an anti-pipelined server via a pair of
+-- connected 'Channel's and a common 'Codec'. Both peers rely on concurrency:
+-- the client's receivers and the server's 'Sender' each run in parallel with
+-- their main thread, so this is where the interleavings anti-pipelining is
+-- meant to exploit actually occur (unlike @connect@, which forgets them).
+--
+runConnectedPeersAntiPipelined
+  :: ( MonadAsync m
+     , MonadCatch m
+     , MonadEvaluate m
+     , Exception failure
+     , NFData failure
+     , NFData a
+     , NFData b
+     )
+  => m (Channel m bytes, Channel m bytes)
+  -> Tracer m (PeerRole, TraceSendRecv ps)
+  -> Codec ps failure m bytes
+  -> PeerPipelined     ps             pr  st m a
+  -> PeerAntiPipelined ps (FlipAgency pr) st m b
+  -> m (a, b)
+runConnectedPeersAntiPipelined createChannels tracer codec client server =
+    createChannels >>= \(clientChannel, serverChannel) ->
+
+    (fst <$> runPipelinedPeer     tracerClient codec clientChannel client)
+      `concurrently`
+    (fst <$> runAntiPipelinedPeer tracerServer codec serverChannel server)
+  where
+    tracerClient = contramap ((,) AsClient) tracer
+    tracerServer = contramap ((,) AsServer) tracer
+
+
 -- Run the same protocol with different codes.  This is useful for testing
 -- 'Handshake' protocol which knows how to decode different versions.
 --

typed-protocols/examples/Network/TypedProtocol/ReqResp/Server.hs

@@ -45,3 +45,48 @@ reqRespServerPeer ReqRespServer{..} =
       MsgReq req -> Effect $ do
         (resp, next) <- recvMsgReq req
         pure $ Yield (MsgResp resp) (reqRespServerPeer next)
+
+
+-- | An anti-pipelined 'ReqResp' server.
+--
+-- The single 'Sender' takes no argument, so a reply can depend on its request
+-- only indirectly — by the peer stashing request-derived data into state (via
+-- an 'Effect') for the 'Sender' to read. This example doesn't do that: it
+-- ignores the request payload (hence @()@) and draws each reply from the
+-- supplied action, which typically reads and advances some state. It receives
+-- ahead, delegating every reply to the 'Sender', and is willing — at the
+-- environment's choice — either to collect an outstanding reply or keep
+-- receiving.
+--
+reqRespServerPeerAntiPipelined
+  :: forall resp m. Functor m
+  => m resp
+  -- ^ produce (and record) the next reply
+  -> ServerAntiPipelined (ReqResp () resp) StIdle m ()
+reqRespServerPeerAntiPipelined nextResp =
+    ServerAntiPipelined sender (go Zero)
+  where
+    sender :: Sender (ReqResp () resp) StBusy StIdle m
+    sender = SenderEffect $
+      (\resp -> SenderYield (MsgResp resp) SenderDone) <$> nextResp
+
+    -- with @n@ replies outstanding: collect one if the environment chooses,
+    -- but stay willing to receive ahead
+    go :: forall n.
+          Nat n
+       -> Server (ReqResp () resp) (AntiPipelined StBusy StIdle n) StIdle m ()
+    go  Zero     = await Zero
+    go (Succ n') = AntiCollect (await n') (Just (await (Succ n')))
+
+    await :: forall n.
+             Nat n
+          -> Server (ReqResp () resp) (AntiPipelined StBusy StIdle n) StIdle m ()
+    await n = Await $ \msg -> case msg of
+                MsgReq _ -> YieldAntiPipelined (go (Succ n))
+                MsgDone  -> drain n
+
+    drain :: forall n.
+             Nat n
+          -> Server (ReqResp () resp) (AntiPipelined StBusy StIdle n) StDone m ()
+    drain  Zero     = Done ()
+    drain (Succ n') = AntiCollect (drain n') Nothing

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
@@ -287,17 +293,16 @@ runPipelinedPeerSender receiveQueue collectQueue
     go n    dstate             (Effect k) = k >>= go n dstate
     go Zero (HasDState dstate) (Done _ x) = return (x, dstate)
 
-    go Zero dstate (Yield refl msg k) = do
+    go n dstate (Yield refl msg k) = do
       sendMessage refl msg
-      go Zero dstate k
+      go n dstate k
 
     go Zero (HasDState dstate) (Await stok k) = do
       (SomeMessage msg, dstate') <- recvMessage stok dstate
       go Zero (HasDState dstate') (k msg)
 
-    go n dstate (YieldPipelined refl msg receiver k) = do
+    go n dstate (AwaitPipelined receiver k) = do
       atomically (writeTQueue receiveQueue (ReceiveHandler dstate receiver))
-      sendMessage refl msg
       go (Succ n) NoDState k
 
     go (Succ n) NoDState (Collect Nothing k) = do
@@ -363,3 +368,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 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