diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,15 @@ # Revision history for typed-protocols +## 1.2.2.0 -- 2026-08-21 + +### Non-breaking changes + +* Repository moved to https://github.com/IntersectMBO/typed-protocols +* Added `Lookahead`, a dual of `Pipelined`: a peer defers its sends to + background `Sender`s and receives ahead (`AwaitLookahead` / `FlushSender`), + with supporting proofs, drivers, and `Server` pattern synonyms. +* Support `QuickCheck >= 2.18`, `io-classes-1.11`. + ## 1.2.1.0 -- 2026-04-16 * Support GHC-9.14, io-classes >=1.8 && < 1.11 diff --git a/NOTICE b/NOTICE --- a/NOTICE +++ b/NOTICE @@ -1,4 +1,4 @@ -Copyright 2019-2025 Input Output Global Inc (IOG) +Copyright 2019-2026 Input Output Global Inc (IOG), 2026 Intersect Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License. diff --git a/README.md b/README.md --- a/README.md +++ b/README.md @@ -1,5 +1,5 @@ -[![Haskell CI](https://img.shields.io/github/actions/workflow/status/input-output-hk/typed-protocols/haskell.yml?branch=main&label=Build&style=for-the-badge)](https://github.com/input-output-hk/typed-protocols/actions/workflows/haskell.yml) -[![handbook](https://img.shields.io/badge/policy-Cardano%20Engineering%20Handbook-informational?style=for-the-badge)](https://input-output-hk.github.io/cardano-engineering-handbook) +[![Haskell CI](https://img.shields.io/github/actions/workflow/status/intersectmbo/typed-protocols/haskell.yml?branch=main&label=Build&style=for-the-badge)](https://github.com/intersectmbo/typed-protocols/actions/workflows/haskell.yml) +[![handbook](https://img.shields.io/badge/policy-Cardano%20Engineering%20Handbook-informational?style=for-the-badge)](https://intersectmbo.github.io/cardano-engineering-handbook) typed-protocols @@ -36,4 +36,4 @@ [typed-protocols-agda]: https://coot.me/agda/posts.agda.typed-protocols.html [coot]: https://github.com/coot [dcoutts]: https://github.com/dcoutts -[haddocks]: https://input-output-hk.github.io/typed-protocols +[haddocks]: https://intersectmbo.github.io/typed-protocols diff --git a/examples/Network/TypedProtocol/Driver/Simple.hs b/examples/Network/TypedProtocol/Driver/Simple.hs --- a/examples/Network/TypedProtocol/Driver/Simple.hs +++ b/examples/Network/TypedProtocol/Driver/Simple.hs @@ -15,9 +15,14 @@ , Role (..) -- * Pipelined peers , runPipelinedPeer + -- * Lookahead peers + , runLookaheadPeer + , runLookaheadFixedSenderPeer -- * Connected peers , runConnectedPeers , runConnectedPeersPipelined + , runConnectedPeersLookahead + , runConnectedPeersLookaheadFixedSender , runConnectedPeersAsymmetric -- * Driver utilities -- | This may be useful if you want to write your own driver. @@ -169,7 +174,54 @@ driver = driverSimple tracer codec channel +-- | Run a lookahead 'Peer' with the given 'Channel' and 'Codec'. -- +-- Like pipelined peers, lookahead peers rely on concurrency (the 'Sender's run +-- in parallel with the main peer), hence the 'MonadAsync' constraint. +-- +runLookaheadPeer + :: 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 + -> PeerLookahead ps pr st m a + -> m (a, Maybe bytes) +runLookaheadPeer tracer codec channel peer = + runLookaheadPeerWithDriver driver peer + where + driver = driverSimple tracer codec channel + + +-- | Run a fixed-'Sender' lookahead 'Peer' with the given 'Channel' and 'Codec'. +-- +runLookaheadFixedSenderPeer + :: 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 + -> PeerLookaheadFixedSender ps pr st m a + -> m (a, Maybe bytes) +runLookaheadFixedSenderPeer tracer codec channel peer = + runLookaheadFixedSenderPeerWithDriver driver peer + where + driver = driverSimple tracer codec channel + + +-- -- Utils -- @@ -250,6 +302,65 @@ (fst <$> runPipelinedPeer tracerClient codec clientChannel client) `concurrently` (fst <$> runPeer tracerServer codec serverChannel server) + where + tracerClient = contramap ((,) AsClient) tracer + tracerServer = contramap ((,) AsServer) tracer + + +-- | Run a pipelined client against a lookahead server over a pair of connected +-- 'Channel's. Both rely on concurrency — the client's receivers and the +-- server's 'Sender's each run in parallel with their main thread — so this is +-- where the interleavings lookahead exploits actually occur (unlike @connect@, +-- which forgets them). +-- +runConnectedPeersLookahead :: ( 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 + -> PeerLookahead ps (FlipAgency pr) st m b + -> m (a, b) +runConnectedPeersLookahead createChannels tracer codec client server = + createChannels >>= \(clientChannel, serverChannel) -> + + (fst <$> runPipelinedPeer tracerClient codec clientChannel client) + `concurrently` + (fst <$> runLookaheadPeer tracerServer codec serverChannel server) + where + tracerClient = contramap ((,) AsClient) tracer + tracerServer = contramap ((,) AsServer) tracer + + +-- | As 'runConnectedPeersLookahead', but the server is a fixed-'Sender' +-- lookahead peer run via 'runLookaheadFixedSenderPeer'. +-- +runConnectedPeersLookaheadFixedSender :: ( 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 + -> PeerLookaheadFixedSender ps (FlipAgency pr) st m b + -> m (a, b) +runConnectedPeersLookaheadFixedSender createChannels tracer codec client server = + createChannels >>= \(clientChannel, serverChannel) -> + + (fst <$> runPipelinedPeer tracerClient codec clientChannel client) + `concurrently` + (fst <$> runLookaheadFixedSenderPeer tracerServer codec serverChannel server) where tracerClient = contramap ((,) AsClient) tracer tracerServer = contramap ((,) AsServer) tracer diff --git a/examples/Network/TypedProtocol/ReqResp/Server.hs b/examples/Network/TypedProtocol/ReqResp/Server.hs --- a/examples/Network/TypedProtocol/ReqResp/Server.hs +++ b/examples/Network/TypedProtocol/ReqResp/Server.hs @@ -6,6 +6,7 @@ import Network.TypedProtocol.Core import Network.TypedProtocol.Peer.Server +import Network.TypedProtocol.Proofs (embedLookaheadUsingFixedSender) import Network.TypedProtocol.ReqResp.Type @@ -45,3 +46,58 @@ MsgReq req -> Effect $ do (resp, next) <- recvMsgReq req pure $ Yield (MsgResp resp) (reqRespServerPeer next) + + +-- | A lookahead 'ReqResp' server (the dual of a pipelined client). +-- +-- It is exactly 'reqRespServerPeerLookaheadFixedSender' with its single +-- 'Sender' plugged in at each 'AwaitLookahead' (via +-- 'embedLookaheadUsingFixedSender'), so the two never drift apart. +-- +reqRespServerPeerLookahead + :: forall resp m. Functor m + => m resp + -- ^ produce (and record) the next reply + -> ServerLookahead (ReqResp () resp) StIdle m () +reqRespServerPeerLookahead = + embedLookaheadUsingFixedSender . reqRespServerPeerLookaheadFixedSender + + +-- | A lookahead 'ReqResp' server that receives requests ahead of sending their +-- replies: each 'AwaitLookahead' hands the reply to the /previous/ request off +-- to the sender thread (via 'TheSender', the one 'Sender' carried by the +-- 'ServerLookaheadFixedSender' wrapper — it runs @nextResp@, which typically +-- reads and advances some state), while immediately awaiting the next request. +-- The request payload is ignored (hence @()@); replies come solely from +-- @nextResp@. This is the peer that +-- 'Network.TypedProtocol.Driver.runLookaheadFixedSenderPeerWithDriver' runs. +-- +reqRespServerPeerLookaheadFixedSender + :: forall resp m. Functor m + => m resp + -- ^ produce (and record) the next reply + -> ServerLookaheadFixedSender (ReqResp () resp) StIdle m () +reqRespServerPeerLookaheadFixedSender nextResp = + ServerLookaheadFixedSender sender start + where + sender :: Sender (ReqResp () resp) VariableSender StBusy StIdle m + sender = SenderEffect $ + (\resp -> SenderYield (MsgResp resp) SenderDone) <$> nextResp + + start :: Server (ReqResp () resp) (Lookahead Z (FixedSender StBusy StIdle)) StIdle m () + start = Await $ \msg -> case msg of + MsgReq _ -> busy Zero + MsgDone -> Done () + + busy :: forall n. + Nat n + -> Server (ReqResp () resp) (Lookahead n (FixedSender StBusy StIdle)) StBusy m () + busy n = AwaitLookahead TheSender $ \msg -> case msg of + MsgReq _ -> busy (Succ n) + MsgDone -> drain (Succ n) + + drain :: forall n. + Nat n + -> Server (ReqResp () resp) (Lookahead n (FixedSender StBusy StIdle)) StDone m () + drain Zero = Done () + drain (Succ n') = FlushSender Nothing (drain n') diff --git a/examples/Network/TypedProtocol/Stateful/ReqResp/Type.hs b/examples/Network/TypedProtocol/Stateful/ReqResp/Type.hs --- a/examples/Network/TypedProtocol/Stateful/ReqResp/Type.hs +++ b/examples/Network/TypedProtocol/Stateful/ReqResp/Type.hs @@ -88,6 +88,6 @@ WriteFile :: FilePath -> String -> FileAPI () -- write to a file --- TODO: input-output-hk/typed-protocols#57 +-- TODO: intersectmbo/typed-protocols#57 type FileRPC = ReqResp FileAPI diff --git a/src/Network/TypedProtocol/Core.hs b/src/Network/TypedProtocol/Core.hs --- a/src/Network/TypedProtocol/Core.hs +++ b/src/Network/TypedProtocol/Core.hs @@ -42,8 +42,10 @@ -- ** Pipelining -- *** IsPipelined , IsPipelined (..) + , SenderVariability (..) -- *** Outstanding , Outstanding + , OutstandingSenders -- *** N and Nat , N (..) , Nat (Succ, Zero) @@ -272,7 +274,7 @@ TheyHaveAgency :: RelativeAgency -- evidence of protocol termination NobodyHasAgency :: RelativeAgency --- TODO: input-output-hk/typed-protocols#57 +-- TODO: intersectmbo/typed-protocols#57 -- | Compute effective agency with respect to the peer role, for client role, @@ -491,25 +493,46 @@ -- | Promoted data type which indicates if 'Peer' is used in -- pipelined mode or not. -- -data IsPipelined where +type IsPipelined :: Type -> Type +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 + -- | The dual of 'Pipelined': defers sends to background 'Sender's while + -- receiving ahead. 'N' counts unflushed 'Sender's. + Lookahead :: N -> SenderVariability ps -> IsPipelined ps + +-- | Whether each lookahead 'Sender' is supplied per-step or reused with fixed +-- endpoints. +type SenderVariability :: Type -> Type +data SenderVariability ps where + VariableSender :: SenderVariability ps + FixedSender :: ps -> ps -> SenderVariability ps + -- | Type level count of the number of outstanding pipelined yields for which -- we have not yet collected a receiver result. Used to -- ensure that 'Collect' is only used when there are outstanding results to -- collect (e.g. after 'YieldPipeliend' was used); -- 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 ('Lookahead _ _) = Z + +-- | Dual of 'Outstanding': count of outstanding 'Sender's; blocks 'Yield' and +-- 'Done' (an inline send would race them). +-- +type OutstandingSenders :: IsPipelined ps -> N +type family OutstandingSenders pl where + OutstandingSenders 'NonPipelined = Z + OutstandingSenders ('Pipelined _ _) = Z + OutstandingSenders ('Lookahead n _) = n -- | A value level inductive natural number, indexed by the corresponding type -- level natural number 'N'. diff --git a/src/Network/TypedProtocol/Driver.hs b/src/Network/TypedProtocol/Driver.hs --- a/src/Network/TypedProtocol/Driver.hs +++ b/src/Network/TypedProtocol/Driver.hs @@ -13,14 +13,20 @@ , runPeerWithDriver -- * Pipelined peers , runPipelinedPeerWithDriver + -- * Lookahead peers + , runLookaheadPeerWithDriver + , runLookaheadFixedSenderPeerWithDriver ) 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 @@ -93,7 +99,7 @@ , -- | Initial state of the driver initialDState :: dstate } --- TODO: input-output-hk/typed-protocols#57 +-- TODO: intersectmbo/typed-protocols#57 -- | When decoding a 'Message' we only know the expected \"from\" state. We @@ -363,3 +369,194 @@ go dstate (ReceiverAwait refl k) = do (SomeMessage msg, dstate') <- recvMessage refl dstate go dstate' (k msg) + + +-- +-- Running lookahead peers +-- + +-- | A 'Sender' whose states are hidden, so it can be queued. +-- +data SomeSender ps pr m where + SomeSender :: Sender ps pr VariableSender st stdone m + -> SomeSender ps pr m + +-- | Run a lookahead peer with the given driver. +-- +-- Dual to 'runPipelinedPeerWithDriver': where a pipelined peer sends ahead and +-- defers its receives to a parallel receiver thread, a lookahead peer receives +-- ahead and defers its sends to a parallel sender thread. +-- +-- 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 'OutstandingSenders' index guarantees the peer thread's own +-- sends ('Yield'\/'Done') happen only when the sender thread is idle. +-- +runLookaheadPeerWithDriver + :: forall ps (st :: ps) pr dstate m a. + ( MonadAsync m + , MonadEvaluate m + , NFData a + ) + => Driver ps pr dstate m + -> PeerLookahead ps pr st m a + -> m (a, dstate) +runLookaheadPeerWithDriver driver@Driver{initialDState} (PeerLookahead peer) = do + senderQueue <- atomically newTQueue + doneVar <- newTVarIO 0 + r@(a, _dstate) <- runLookaheadPeerSender (readTQueue senderQueue) doneVar driver + `withAsyncLoop` + runLookaheadPeerMain + (\sender -> writeTQueue senderQueue (SomeSender sender)) + doneVar driver peer initialDState + + _ <- evaluate (force a) + return r + + where + withAsyncLoop :: m Void -> m x -> m x + withAsyncLoop left right = do + res <- race left right + case res of + Left v -> case v of {} + Right a -> return a + + +-- | The peer (main) thread shared by both lookahead drivers. It differs only +-- in how it hands off the 'Sender' deferred at each 'AwaitLookahead', which is +-- the @registerSender@ argument: the variable driver enqueues it, the fixed +-- driver bumps a counter (ignoring the abstract 'TheSender'). +-- +runLookaheadPeerMain + :: forall ps (sv :: SenderVariability ps) (st :: ps) pr dstate m a. + ( MonadSTM m + , MonadThread m + ) + => (forall stA stZ. Sender ps pr sv stA stZ m -> STM m ()) + -- ^ register the 'Sender' deferred by an 'AwaitLookahead' + -> TVar m Natural + -- ^ count of 'Sender's that have since finished, consumed by 'FlushSender'. + -- + -- The role of this 'TVar' is the same as the queue of results in the + -- pipelining case, but since a 'Sender' doesn't return any result we just + -- count how many 'Sender's have finished: the sender increments it when it + -- finishes running, and 'FlushSender' decrements it. + -> Driver ps pr dstate m + -> Peer ps pr ('Lookahead Z sv) st m a + -> dstate + -> m (a, dstate) +runLookaheadPeerMain registerSender doneVar + Driver{sendMessage, recvMessage} + peer0 dstate0 = do + threadId <- myThreadId + labelThread threadId "lookahead-peer-main" + go dstate0 peer0 + where + go :: forall st' n. + dstate + -> Peer ps pr ('Lookahead n sv) st' m a + -> m (a, dstate) + go dstate (Effect k) = k >>= go dstate + go dstate (Done _ x) = return (x, dstate) + + -- Only reachable at 'Lookahead Z' (the constructor demands + -- @OutstandingSenders ~ 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 'OutstandingSenders': receiving ahead is the whole point. + go dstate (Await refl k) = do + (SomeMessage msg, dstate') <- recvMessage refl dstate + go dstate' (k msg) + + go dstate (AwaitLookahead refl sender k) = do + atomically (registerSender sender) + (SomeMessage msg, dstate') <- recvMessage refl dstate + go dstate' (k msg) + + go dstate (FlushSender mbNonBlocking k) = + join $ atomically $ do + d <- readTVar doneVar + if d > 0 + then do writeTVar doneVar (d - 1); pure (go dstate k) + else case mbNonBlocking of + Nothing -> retry + Just k' -> pure (go dstate k') + + +-- | The sender thread shared by both lookahead drivers. It differs only in +-- how it obtains the next 'Sender' to run, which is the @nextSender@ argument: +-- the variable driver reads one off a queue, the fixed driver waits for its +-- counter and yields the one fixed 'Sender'. +-- +runLookaheadPeerSender + :: forall ps pr dstate m. + ( MonadSTM m + , MonadThread m + ) + => STM m (SomeSender ps pr m) + -- ^ obtain the next 'Sender' to run (blocking until one is available) + -> TVar m Natural + -> Driver ps pr dstate m + -> m Void +runLookaheadPeerSender nextSender doneVar + Driver{sendMessage} = do + threadId <- myThreadId + labelThread threadId "lookahead-sender" + forever $ do + SomeSender sender <- atomically nextSender + runSender sender + atomically $ modifyTVar' doneVar (+ 1) + where + runSender :: forall stA stZ. Sender ps pr VariableSender stA stZ m -> m () + runSender = \case + SenderEffect k -> k >>= runSender + SenderDone -> return () + SenderYield refl msg k -> do + sendMessage refl msg + runSender k + + +-- | Run a fixed-'Sender' lookahead peer with the given driver. +-- +-- Like 'runLookaheadPeerWithDriver', but every 'AwaitLookahead' reuses the one +-- 'Sender' supplied here, so the driver tracks outstanding sends with a plain +-- counter rather than a queue. ('embedLookaheadUsingFixedSender' could instead +-- reduce this to 'runLookaheadPeerWithDriver', but that would allocate and +-- queue a redundant 'Sender' per step.) +-- +runLookaheadFixedSenderPeerWithDriver + :: forall ps (st :: ps) pr dstate m a. + ( MonadAsync m + , MonadEvaluate m + , NFData a + ) + => Driver ps pr dstate m + -> PeerLookaheadFixedSender ps pr st m a + -> m (a, dstate) +runLookaheadFixedSenderPeerWithDriver driver@Driver{initialDState} (PeerLookaheadFixedSender sender peer) = do + sendVar <- newTVarIO (0 :: Natural) + doneVar <- newTVarIO 0 + let -- wait for the counter, then yield the one fixed 'Sender' + nextSender = do n <- readTVar sendVar + check (0 < n) + writeTVar sendVar $! n - 1 + return (SomeSender sender) + r@(a, _dstate) <- runLookaheadPeerSender nextSender doneVar driver + `withAsyncLoop` + runLookaheadPeerMain + (\TheSender -> modifyTVar' sendVar (+ 1)) + doneVar driver peer initialDState + + _ <- evaluate (force a) + return r + + where + withAsyncLoop :: m Void -> m x -> m x + withAsyncLoop left right = do + res <- race left right + case res of + Left v -> case v of {} + Right a -> return a diff --git a/src/Network/TypedProtocol/Peer.hs b/src/Network/TypedProtocol/Peer.hs --- a/src/Network/TypedProtocol/Peer.hs +++ b/src/Network/TypedProtocol/Peer.hs @@ -5,8 +5,12 @@ module Network.TypedProtocol.Peer ( Peer (..) , PeerPipelined (..) + , PeerLookahead (..) + , PeerLookaheadFixedSender (..) , Receiver (..) + , Sender (..) , Outstanding + , OutstandingSenders , N (..) , Nat (Zero, Succ) , natToInt @@ -79,7 +83,7 @@ -- type Peer :: forall ps -> PeerRole - -> IsPipelined + -> IsPipelined ps -> ps -> (Type -> Type) -- ^ monad's kind @@ -115,6 +119,7 @@ , StateTokenI st' , ActiveState st , Outstanding pl ~ Z + , OutstandingSenders pl ~ Z ) => WeHaveAgencyProof pr st -- ^ agency proof @@ -146,6 +151,7 @@ ( StateTokenI st , ActiveState st , Outstanding pl ~ Z + , OutstandingSenders pl ~ Z ) => TheyHaveAgencyProof pr st -- ^ agency proof @@ -169,6 +175,7 @@ ( StateTokenI st , StateAgency st ~ NobodyAgency , Outstanding pl ~ Z + , OutstandingSenders pl ~ Z ) => NobodyHasAgencyProof pr st -- ^ (no) agency proof @@ -214,6 +221,38 @@ -- ^ continuation -> Peer ps pr (Pipelined (S n) c) st m a + -- + -- Lookahead primitives + -- + + -- | The dual of 'YieldPipelined'. Defer the send @st -> st'@ to a background + -- 'Sender', then await ahead at @st'@. + -- + AwaitLookahead + :: forall ps pr (sv :: SenderVariability ps) (st :: ps) (st' :: ps) n m a. + ( StateTokenI st + , StateTokenI st' + , ActiveState st + , ActiveState st' + ) + => TheyHaveAgencyProof pr st' + -> Sender ps pr sv st st' m + -> (forall st''. Message ps st' st'' + -> Peer ps pr (Lookahead (S n) sv) st'' m a) + -> Peer ps pr (Lookahead n sv) st m a + + -- | The dual of 'Collect': await one deferred 'Sender' to finish. Unlike + -- 'Collect', this can also be used at a terminal state (to drain before + -- 'Done'), so it does not require 'ActiveState'. + -- + FlushSender + :: forall ps pr n (sv :: SenderVariability ps) st m a. + ( StateTokenI st + ) + => Maybe (Peer ps pr (Lookahead (S n) sv) st m a) + -> (Peer ps pr (Lookahead n sv) st m a) + -> Peer ps pr (Lookahead (S n) sv) st m a + deriving instance Functor m => Functor (Peer ps pr pl st m) @@ -260,6 +299,32 @@ deriving instance Functor m => Functor (Receiver ps pr st stdone m) +-- | The 'Lookahead' analog of 'Receiver' +type Sender :: forall ps + -> PeerRole + -> SenderVariability ps + -> ps + -> ps + -> (Type -> Type) + -> Type +data Sender ps pr sv st stdone m where + + TheSender :: Sender ps pr (FixedSender st stdone) st stdone m + + SenderEffect :: m (Sender ps pr VariableSender st stdone m) + -> Sender ps pr VariableSender st stdone m + + SenderDone :: Sender ps pr VariableSender stdone stdone m + + SenderYield :: ( StateTokenI st + , StateTokenI st' + , ActiveState st + ) + => !(WeHaveAgencyProof pr st) + -> Message ps st st' + -> Sender ps pr VariableSender st' stdone m + -> Sender ps pr VariableSender 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 +336,29 @@ -> PeerPipelined ps pr st m a deriving instance Functor m => Functor (PeerPipelined ps pr st m) + +-- | Wrapper for a lookahead peer that supplies its own 'Sender' at each +-- 'AwaitLookahead'. Expected by +-- 'Network.TypedProtocol.Driver.runLookaheadPeerWithDriver'. +-- +data PeerLookahead ps pr (st :: ps) m a where + PeerLookahead :: { runPeerLookahead :: Peer ps pr (Lookahead Z VariableSender) st m a } + -> PeerLookahead ps pr st m a + +deriving instance Functor m => Functor (PeerLookahead ps pr st m) + +-- | Wrapper for a lookahead peer whose 'AwaitLookahead's all use 'TheSender', +-- i.e. reuse the one 'Sender' carried here (with fixed endpoints @apst -> +-- apst'@). Expected by +-- 'Network.TypedProtocol.Driver.runLookaheadFixedSenderPeerWithDriver', which +-- can then track outstanding sends with a counter rather than a queue. +-- +-- The carried 'Sender' is a 'VariableSender' — the concrete one the driver +-- actually runs; the peer only refers to it abstractly via 'TheSender'. +-- +data PeerLookaheadFixedSender ps pr (st :: ps) m a where + PeerLookaheadFixedSender :: Sender ps pr VariableSender apst apst' m + -> Peer ps pr (Lookahead Z (FixedSender apst apst')) st m a + -> PeerLookaheadFixedSender ps pr st m a + +deriving instance Functor m => Functor (PeerLookaheadFixedSender ps pr st m) diff --git a/src/Network/TypedProtocol/Peer/Client.hs b/src/Network/TypedProtocol/Peer/Client.hs --- a/src/Network/TypedProtocol/Peer/Client.hs +++ b/src/Network/TypedProtocol/Peer/Client.hs @@ -36,7 +36,7 @@ type Client :: forall ps - -> IsPipelined + -> IsPipelined ps -> ps -> (Type -> Type) -> Type @@ -76,6 +76,7 @@ , StateTokenI st' , StateAgency st ~ ClientAgency , Outstanding pl ~ Z + , OutstandingSenders pl ~ Z ) => Message ps st st' -- ^ protocol message @@ -92,6 +93,7 @@ => ( StateTokenI st , StateAgency st ~ ServerAgency , Outstanding pl ~ Z + , OutstandingSenders pl ~ Z ) => (forall st'. Message ps st st' -> Client ps pl st' m a) @@ -107,6 +109,7 @@ => ( StateTokenI st , StateAgency st ~ NobodyAgency , Outstanding pl ~ Z + , OutstandingSenders pl ~ Z ) => a -- ^ protocol return value diff --git a/src/Network/TypedProtocol/Peer/Server.hs b/src/Network/TypedProtocol/Peer/Server.hs --- a/src/Network/TypedProtocol/Peer/Server.hs +++ b/src/Network/TypedProtocol/Peer/Server.hs @@ -14,17 +14,32 @@ , pattern Done , pattern YieldPipelined , pattern Collect + , pattern AwaitLookahead + , pattern FlushSender -- * Receiver type alias and its pattern synonyms , Receiver , pattern ReceiverEffect , pattern ReceiverAwait , pattern ReceiverDone + -- * Sender type alias and its pattern synonyms + , Sender + , pattern TheSender + , pattern SenderEffect + , pattern SenderYield + , pattern SenderDone -- * ServerPipelined type alias and its pattern synonym , ServerPipelined , TP.PeerPipelined (ServerPipelined, runServerPipelined) + -- * ServerLookahead type aliases and their pattern synonyms + , ServerLookahead + , TP.PeerLookahead (ServerLookahead, runServerLookahead) + , ServerLookaheadFixedSender + , pattern ServerLookaheadFixedSender -- * re-exports , IsPipelined (..) + , SenderVariability (..) , Outstanding + , OutstandingSenders , N (..) , Nat (..) ) where @@ -37,7 +52,7 @@ type Server :: forall ps - -> IsPipelined + -> IsPipelined ps -> ps -> (Type -> Type) -> Type @@ -60,6 +75,41 @@ {-# COMPLETE ServerPipelined #-} +-- TODO: mirror these lookahead pattern synonyms in +-- 'Network.TypedProtocol.Peer.Client' (a 'ClientLookahead' / +-- 'ClientLookaheadFixedSender'). They are not strictly needed since the API is +-- symmetric, but they would let a user work in terms of 'Client' as well as +-- 'Server'. + +-- | A lookahead server that supplies its own 'Sender' at each 'AwaitLookahead'. +-- +type ServerLookahead ps st m a = TP.PeerLookahead ps AsServer st m a + +pattern ServerLookahead :: forall ps st m a. + () + => Server ps (Lookahead Z VariableSender) st m a + -> ServerLookahead ps st m a +pattern ServerLookahead { runServerLookahead } = TP.PeerLookahead runServerLookahead + +{-# COMPLETE ServerLookahead #-} + + +-- | A lookahead server that reuses one 'Sender' at each 'AwaitLookaheadFixedSender'. +-- +type ServerLookaheadFixedSender ps st m a = TP.PeerLookaheadFixedSender ps AsServer st m a + +pattern ServerLookaheadFixedSender :: forall ps st m a. + () + => forall apst apst'. + () + => Sender ps VariableSender apst apst' m + -> Server ps (Lookahead Z (FixedSender apst apst')) st m a + -> ServerLookaheadFixedSender ps st m a +pattern ServerLookaheadFixedSender sender peer = TP.PeerLookaheadFixedSender sender peer + +{-# COMPLETE ServerLookaheadFixedSender #-} + + -- | Server role pattern for 'TP.Effect'. -- pattern Effect :: forall ps pl st m a. @@ -78,6 +128,7 @@ , StateTokenI st' , StateAgency st ~ ServerAgency , Outstanding pl ~ Z + , OutstandingSenders pl ~ Z ) => Message ps st st' -- ^ protocol message @@ -94,6 +145,7 @@ => ( StateTokenI st , StateAgency st ~ ClientAgency , Outstanding pl ~ Z + , OutstandingSenders pl ~ Z ) => (forall st'. Message ps st st' -> Server ps pl st' m a) @@ -109,6 +161,7 @@ => ( StateTokenI st , StateAgency st ~ NobodyAgency , Outstanding pl ~ Z + , OutstandingSenders pl ~ Z ) => a -- ^ protocol return value @@ -152,6 +205,45 @@ {-# COMPLETE Effect, Yield, Await, Done, YieldPipelined, Collect #-} +-- | Server role pattern for 'TP.AwaitLookahead' +-- +-- Use 'TheSender' as the first argument for a fixed-'Sender' peer, or a +-- concrete 'VariableSender' for a per-step one. +-- +pattern AwaitLookahead :: forall ps sv st n m a. + () + => forall st'. + ( StateTokenI st + , StateTokenI st' + , ActiveState st + , StateAgency st' ~ ClientAgency + ) + => Sender ps sv st st' m + -- ^ sender for the deferred send @st -> st'@ + -> (forall st''. Message ps st' st'' + -> Server ps (Lookahead (S n) sv) st'' m a) + -- ^ continuation, awaiting ahead at @st'@ + -> Server ps (Lookahead n sv) st m a +pattern AwaitLookahead sender k = TP.AwaitLookahead ReflClientAgency sender k + + +-- | Server role pattern for 'TP.FlushSender' +-- +pattern FlushSender :: forall ps st n sv m a. + () + => ( StateTokenI st + ) + => Maybe (Server ps (Lookahead (S n) sv) st m a) + -- ^ continuation if no 'Sender' has finished so far + -> (Server ps (Lookahead n sv) st m a) + -- ^ continuation once a 'Sender' has finished + -> Server ps (Lookahead (S n) sv) st m a +pattern FlushSender mk k = TP.FlushSender mk k + + +{-# COMPLETE Effect, Yield, Await, Done, AwaitLookahead, FlushSender #-} + + type Receiver ps st stdone m c = TP.Receiver ps AsServer st stdone m c pattern ReceiverEffect :: forall ps st stdone m c. @@ -177,3 +269,36 @@ pattern ReceiverDone c = TP.ReceiverDone c {-# COMPLETE ReceiverEffect, ReceiverAwait, ReceiverDone #-} + + +type Sender ps sv st stdone m = TP.Sender ps AsServer sv st stdone m + +pattern TheSender :: forall ps sv st stdone m. + () + => (sv ~ FixedSender st stdone) + => Sender ps sv st stdone m +pattern TheSender = TP.TheSender + +pattern SenderEffect :: forall ps st stdone m. + m (Sender ps VariableSender st stdone m) + -> Sender ps VariableSender st stdone m +pattern SenderEffect k = TP.SenderEffect k + +pattern SenderYield :: forall ps st stdone m. + () + => forall st'. + ( StateTokenI st + , StateTokenI st' + , StateAgency st ~ ServerAgency + ) + => Message ps st st' + -> Sender ps VariableSender st' stdone m + -> Sender ps VariableSender st stdone m +pattern SenderYield msg k = TP.SenderYield ReflServerAgency msg k + +pattern SenderDone :: forall ps stdone m. + Sender ps VariableSender stdone stdone m +pattern SenderDone = TP.SenderDone + +{-# COMPLETE TheSender #-} +{-# COMPLETE SenderEffect, SenderYield, SenderDone #-} diff --git a/src/Network/TypedProtocol/Proofs.hs b/src/Network/TypedProtocol/Proofs.hs --- a/src/Network/TypedProtocol/Proofs.hs +++ b/src/Network/TypedProtocol/Proofs.hs @@ -12,11 +12,17 @@ ( -- * Connect proofs connect , connectPipelined + , connectLookahead , TerminalStates (..) -- * Pipelining proofs -- | Additional proofs specific to the pipelining features , forgetPipelined , promoteToPipelined + -- * Lookahead proofs + -- | Additional proofs specific to the lookahead features + , forgetLookahead + , promoteToLookahead + , embedLookaheadUsingFixedSender -- ** Pipeline proof helpers , Queue (..) , enqueue @@ -224,6 +230,127 @@ -- ^ peers results and an evidence of their termination connectPipelined csA a b = connect (forgetPipelined csA a) b + + +-- +-- Remove Lookahead +-- + + +-- | Total conversion from lookahead peers to regular peers: the lookahead +-- analogue of 'forgetPipelined'. +-- +-- Where 'forgetPipelined' inlines each 'Receiver' at the point it is collected, +-- this inlines the 'Sender' supplied at each 'AwaitLookahead': its sends, which +-- the sender thread would have performed asynchronously, are performed +-- synchronously just before the fused receive. +-- +-- Dually to 'forgetPipelined', the @[Bool]@ chooses the interleaving at each +-- 'FlushSender': a @True@ pretends the 'Sender' has not finished yet, so the +-- peer takes its non-blocking continuation (when it has one) and leaves the +-- send outstanding; a @False@ (or @[]@) flushes it. +-- +forgetLookahead + :: forall ps (pr :: PeerRole) (st :: ps) m a. + Functor m + => [Bool] + -- ^ interleaving choices allowed by the 'FlushSender' primitive; @False@ + -- values or @[]@ leave nothing outstanding. + -> PeerLookahead ps pr st m a + -> Peer ps pr NonPipelined st m a +forgetLookahead cs0 (PeerLookahead peer0) = + goPeer cs0 peer0 + where + goPeer :: forall st' n. + [Bool] + -> Peer ps pr ('Lookahead n VariableSender) st' m a + -> Peer ps pr 'NonPipelined st' m a + goPeer cs (Effect k) = Effect (goPeer cs <$> k) + goPeer _ (Done refl k) = Done refl k + goPeer cs (Yield refl m k) = Yield refl m (goPeer cs k) + goPeer cs (Await refl k) = Await refl (goPeer cs . k) + goPeer cs (AwaitLookahead refl sender k) = + goSender sender (Await refl (goPeer cs . k)) + goPeer (True:cs') (FlushSender (Just k) _) = goPeer cs' k + goPeer (_:cs) (FlushSender _ k) = goPeer cs k + goPeer cs@[] (FlushSender _ k) = goPeer cs k + + goSender :: forall sst sst'. + Sender ps pr VariableSender sst sst' m + -> Peer ps pr 'NonPipelined sst' m a + -> Peer ps pr 'NonPipelined sst m a + goSender SenderDone k = k + goSender (SenderEffect ks) k = Effect ((`goSender` k) <$> ks) + goSender (SenderYield refl m ks) k = Yield refl m (goSender ks k) + + +-- | Promote a peer to a lookahead one, using an empty 'Sender'. +-- +-- This is a right inverse of 'forgetLookahead', e.g. +-- +-- >>> forgetLookahead . promoteToLookahead = id +-- +promoteToLookahead + :: forall ps (pr :: PeerRole) (st :: ps) m a. + Functor m + => Peer ps pr NonPipelined st m a + -- ^ a peer + -> PeerLookahead ps pr st m a + -- ^ a lookahead peer +promoteToLookahead p = PeerLookahead (go p) + where + go :: forall st'. + Peer ps pr 'NonPipelined st' m a + -> Peer ps pr ('Lookahead 'Z VariableSender) st' m a + go (Effect k) = Effect (go <$> k) + go (Yield refl m k) = Yield refl m (go k) + go (Await refl k) = Await refl (go . k) + go (Done refl k) = Done refl k + + +-- | Embed a fixed-'Sender' lookahead peer as a variable-'Sender' one, by +-- plugging the carried 'Sender' in for each 'TheSender'. This lets the +-- 'PeerLookahead' machinery (proofs, driver) subsume 'PeerLookaheadFixedSender' — +-- though a dedicated fixed driver is still preferable, as it can track +-- outstanding sends with a counter rather than a queue. +-- +embedLookaheadUsingFixedSender + :: forall ps (pr :: PeerRole) (st :: ps) m a. + Functor m + => PeerLookaheadFixedSender ps pr st m a + -> PeerLookahead ps pr st m a +embedLookaheadUsingFixedSender (PeerLookaheadFixedSender (sender :: Sender ps pr VariableSender apst apst' m) peer0) = + PeerLookahead (go peer0) + where + go :: forall st' n. + Peer ps pr ('Lookahead n (FixedSender apst apst')) st' m a + -> Peer ps pr ('Lookahead n VariableSender) st' m a + go (Effect k) = Effect (go <$> k) + go (Done refl k) = Done refl k + go (Yield refl m k) = Yield refl m (go k) + go (Await refl k) = Await refl (go . k) + go (AwaitLookahead refl TheSender k) = + AwaitLookahead refl sender (go . k) + go (FlushSender mk k) = FlushSender (go <$> mk) (go k) + + +-- | Analogous to 'connectPipelined' but for lookahead peers. +-- +connectLookahead + :: forall ps (pr :: PeerRole) + (st :: ps) m a b. + (Monad m, SingI pr) + => [Bool] + -- ^ an interleaving + -> PeerLookahead ps pr st m a + -- ^ a lookahead peer + -> Peer ps (FlipAgency pr) NonPipelined st m b + -- ^ a non-pipelined peer with flipped agency + -> m (a, b, TerminalStates ps) + -- ^ peers results and an evidence of their termination +connectLookahead cs a b = + connect (forgetLookahead cs a) b + -- | A reference specification for interleaving of requests and responses -- with pipelining, where the environment can choose whether a response is diff --git a/stateful/Network/TypedProtocol/Stateful/Codec.hs b/stateful/Network/TypedProtocol/Stateful/Codec.hs --- a/stateful/Network/TypedProtocol/Stateful/Codec.hs +++ b/stateful/Network/TypedProtocol/Stateful/Codec.hs @@ -58,7 +58,7 @@ -- local state, which contain extra context for the encoding -- process. -- - -- TODO: input-output-hk/typed-protocols#57 + -- TODO: intersectmbo/typed-protocols#57 -> Message ps st st' -- message to be encoded -> bytes, @@ -70,7 +70,7 @@ -- local state, which can contain extra context from the -- previous message. -- - -- TODO: input-output-hk/typed-protocols#57 + -- TODO: intersectmbo/typed-protocols#57 -> m (DecodeStep bytes failure m (SomeMessage st)) } diff --git a/stateful/Network/TypedProtocol/Stateful/Driver.hs b/stateful/Network/TypedProtocol/Stateful/Driver.hs --- a/stateful/Network/TypedProtocol/Stateful/Driver.hs +++ b/stateful/Network/TypedProtocol/Stateful/Driver.hs @@ -41,11 +41,11 @@ -- local state should not be sent to the remote side. -- However it provide extra context for the encoder. -- - -- TODO: input-output-hk/typed-protocols#57 + -- TODO: intersectmbo/typed-protocols#57 -> Message ps st st' -- message to send -- - -- TODO: input-output-hk/typed-protocols#57 + -- TODO: intersectmbo/typed-protocols#57 -> m () , -- | Receive a message, a blocking action which reads from the network @@ -61,12 +61,12 @@ -- local state which provides extra context for the -- decoder. -- - -- TODO: input-output-hk/typed-protocols#57 + -- TODO: intersectmbo/typed-protocols#57 -> dstate -- decoder state, e.g. bytes left from decoding of -- a previous message. -- - -- TODO: input-output-hk/typed-protocols#57 + -- TODO: intersectmbo/typed-protocols#57 -> m (SomeMessage st, dstate) , -- | Initial decoder state. diff --git a/stateful/Network/TypedProtocol/Stateful/Peer.hs b/stateful/Network/TypedProtocol/Stateful/Peer.hs --- a/stateful/Network/TypedProtocol/Stateful/Peer.hs +++ b/stateful/Network/TypedProtocol/Stateful/Peer.hs @@ -154,7 +154,7 @@ f st -- associated local state to the source protocol state 'st' -- - -- TODO: input-output-hk/typed-protocols#57 + -- TODO: intersectmbo/typed-protocols#57 -> Message ps st st' -> ( Peer ps pr st' f m a , f st' @@ -164,9 +164,9 @@ -- -- NOTE: the API is limited to pure transition of local state e.g. -- `f st -> Message ps st st' -> f st'`, - -- see https://github.com/input-output-hk/typed-protocols/discussions/63 + -- see https://github.com/intersectmbo/typed-protocols/discussions/63 -- - -- TODO: input-output-hk/typed-protocols#57 + -- TODO: intersectmbo/typed-protocols#57 ) -- ^ continuation -> Peer ps pr st f m a diff --git a/test/Network/TypedProtocol/PingPong/Tests.hs b/test/Network/TypedProtocol/PingPong/Tests.hs --- a/test/Network/TypedProtocol/PingPong/Tests.hs +++ b/test/Network/TypedProtocol/PingPong/Tests.hs @@ -47,6 +47,11 @@ import Test.Tasty.QuickCheck (testProperty) +#if !MIN_VERSION_QuickCheck(2, 18, 0) +withNumTests :: Testable prop => Int -> prop -> Property +withNumTests = withMaxSuccess +#endif + -- -- The list of all properties -- @@ -75,7 +80,7 @@ , testGroup "CBOR" [ testProperty "codec" prop_codec_cbor_PingPong , testProperty "codec 2-splits" prop_codec_cbor_splits2_PingPong - , testProperty "codec 3-splits" $ withMaxSuccess 30 prop_codec_cbor_splits3_PingPong + , testProperty "codec 3-splits" $ withNumTests 30 prop_codec_cbor_splits3_PingPong ] ] ] diff --git a/test/Network/TypedProtocol/ReqResp/Tests.hs b/test/Network/TypedProtocol/ReqResp/Tests.hs --- a/test/Network/TypedProtocol/ReqResp/Tests.hs +++ b/test/Network/TypedProtocol/ReqResp/Tests.hs @@ -17,13 +17,16 @@ import Network.TypedProtocol.ReqResp.Server import Network.TypedProtocol.ReqResp.Type +import Control.Concurrent.Class.MonadSTM.TVar (newTVarIO, readTVar, writeTVar) import Control.Exception (throw) import Control.Monad.Class.MonadAsync import Control.Monad.Class.MonadST import Control.Monad.Class.MonadSTM +import Control.Monad.Class.MonadTest (MonadTest (exploreRaces)) import Control.Monad.Class.MonadThrow import Control.Monad.IOSim import Control.Monad.ST (runST) +import Control.Monad.Trans.State.Strict (State, runState, state) import Control.Tracer (nullTracer) import Data.Functor.Identity (Identity (..)) @@ -45,6 +48,11 @@ import Text.Show.Functions () +#if !MIN_VERSION_QuickCheck(2, 18, 0) +withNumTests :: Testable prop => Int -> prop -> Property +withNumTests = withMaxSuccess +#endif + -- -- The list of all properties -- @@ -55,10 +63,16 @@ , testProperty "directPipelined" prop_directPipelined , testProperty "connect" prop_connect , testProperty "connectPipelined" prop_connectPipelined + , testProperty "connectLookahead" prop_connectLookahead , testProperty "channel ST" prop_channel_ST , testProperty "channel IO" prop_channel_IO , testProperty "channelPipelined ST" prop_channelPipelined_ST , testProperty "channelPipelined IO" prop_channelPipelined_IO + , testProperty "channelLookahead ST" prop_channelLookahead_ST + , testProperty "channelLookahead IO" prop_channelLookahead_IO + , testProperty "channelLookahead IOSimPOR" prop_channelLookahead_IOSimPOR + , testProperty "lookaheadFixedEquiv ST" prop_lookaheadFixedEquiv_ST + , testProperty "lookaheadFixedEquiv IO" prop_lookaheadFixedEquiv_IO #if !defined(mingw32_HOST_OS) , testProperty "namedPipePipelined" prop_namedPipePipelined_IO , testProperty "socketPipelined" prop_socketPipelined_IO @@ -66,11 +80,11 @@ , testGroup "Codec" [ testProperty "codec" prop_codec_ReqResp , testProperty "codec 2-splits" prop_codec_splits2_ReqResp - , testProperty "codec 3-splits" (withMaxSuccess 33 prop_codec_splits3_ReqResp) + , testProperty "codec 3-splits" (withNumTests 33 prop_codec_splits3_ReqResp) , testGroup "CBOR" [ testProperty "codec" prop_codec_cbor_ReqResp , testProperty "codec 2-splits" prop_codec_cbor_splits2_ReqResp - , testProperty "codec 3-splits" $ withMaxSuccess 30 prop_codec_cbor_splits3_ReqResp + , testProperty "codec 3-splits" $ withNumTests 30 prop_codec_cbor_splits3_ReqResp ] ] , testGroup "AnnotatedCodec" @@ -168,7 +182,26 @@ (s, c) == mapAccumL f 0 xs +-- | The dual of 'prop_connectPipelined': a lookahead server (which ignores the +-- request payload and reads each reply from the state) against a non-pipelined +-- client. The result must not depend on the interleaving choices. -- +prop_connectLookahead :: [Bool] -> (Int -> (Int, Int)) -> NonNegative Int -> Bool +prop_connectLookahead cs g (NonNegative n) = + case runState + (connectLookahead cs + (reqRespServerPeerLookahead nextResp) + (reqRespClientPeer (reqRespClientMap (replicate n ())))) + 0 + + of ((_, resps, TerminalStates SingDone SingDone), acc) -> + (acc, resps) == mapAccumL (\a () -> g a) 0 (replicate n ()) + where + nextResp :: State Int Int + nextResp = state (\a -> let (a', r) = g a in (r, a')) + + +-- -- Properties using channels, codecs and drivers. -- @@ -227,6 +260,129 @@ prop_channelPipelined_ST :: (Int -> Int -> (Int, Int)) -> [Int] -> Property prop_channelPipelined_ST f xs = let tr = runSimTrace (prop_channelPipelined f xs) in + counterexample (intercalate "\n" $ map show $ traceEvents tr) + $ case traceResult True tr of + Left err -> throw err + Right res -> res + + +-- | The channel/driver sibling of 'prop_connectLookahead': a pipelined client +-- against a lookahead server, run through the real 'runPipelinedPeer' / +-- 'runLookaheadPeer' drivers over a channel. Unlike @connect@, this genuinely +-- runs the 'Sender's concurrently with the main peer's receive-ahead, so the +-- lookahead machinery is actually exercised. The collected replies must equal +-- the reference sequence regardless of the interleaving the runtime picks. +-- +prop_channelLookahead :: ( MonadLabelledSTM m + , MonadAsync m + , MonadCatch m + , MonadEvaluate m + , MonadST m + , MonadTest m + ) + => (Int -> (Int, Int)) -> NonNegative Int + -> m Bool +prop_channelLookahead g (NonNegative n) = do + -- mark all threads forked from here on as system threads, so IOSimPOR will + -- reverse races between the client's receivers and the server's 'Sender's + -- (a no-op under IO / plain IOSim) + exploreRaces + accVar <- newTVarIO 0 + let nextResp = atomically $ do + a <- readTVar accVar + let (a', r) = g a + writeTVar accVar a' + return r + client = reqRespClientPeerPipelined (reqRespClientMapPipelined (replicate n ())) + server = reqRespServerPeerLookahead nextResp + (resps, ()) <- runConnectedPeersLookahead + (createPipelineTestChannels 100) + nullTracer + CBOR.codecReqResp + client server + return (resps == snd (mapAccumL (\a () -> g a) 0 (replicate n ()))) + +prop_channelLookahead_IO :: (Int -> (Int, Int)) -> NonNegative Int -> Property +prop_channelLookahead_IO g n = + ioProperty (prop_channelLookahead g n) + +prop_channelLookahead_ST :: (Int -> (Int, Int)) -> NonNegative Int -> Property +prop_channelLookahead_ST g n = + let tr = runSimTrace (prop_channelLookahead g n) in + counterexample (intercalate "\n" $ map show $ traceEvents tr) + $ case traceResult True tr of + Left err -> throw err + Right res -> res + +-- | Have IOSimPOR systematically explore the interleavings of the client's +-- receivers and the server's 'Sender's, checking that the lookahead driver +-- produces the reference result under every schedule. The request count is kept +-- small so the schedule space stays tractable. +-- +prop_channelLookahead_IOSimPOR :: (Int -> (Int, Int)) -> NonNegative Int -> Property +prop_channelLookahead_IOSimPOR g (NonNegative n) = + withNumTests 20 $ + exploreSimTrace id (prop_channelLookahead g (NonNegative (min n 6))) $ \_ tr -> + case traceResult False tr of + Left failure -> counterexample (show failure) (property False) + Right res -> property res + + +-- | 'runLookaheadFixedSenderPeerWithDriver' and +-- @'runLookaheadPeerWithDriver' . 'embedLookaheadUsingFixedSender'@ run the same +-- fixed-'Sender' peer two different ways — the former tracking outstanding sends +-- with a counter, the latter with a queue of (degenerate) 'Sender's. This runs +-- a pipelined client against the same fixed server via each and checks both +-- collect the same replies, matching the reference sequence. +-- +prop_lookaheadFixedEquiv :: ( MonadLabelledSTM m + , MonadAsync m + , MonadCatch m + , MonadEvaluate m + , MonadST m + ) + => (Int -> (Int, Int)) -> NonNegative Int + -> m Bool +prop_lookaheadFixedEquiv g (NonNegative n) = do + viaFixed <- runFixed + viaEmbed <- runEmbed + return (viaFixed == reference && viaEmbed == reference) + where + reqs = replicate n () + reference = snd (mapAccumL (\a () -> g a) 0 reqs) + + client = reqRespClientPeerPipelined (reqRespClientMapPipelined reqs) + + -- a fresh stateful reply source (so each run starts from the same state) + mkNextResp = do + accVar <- newTVarIO 0 + return $ atomically $ do + a <- readTVar accVar + let (a', r) = g a + writeTVar accVar a' + return r + + runFixed = do + nextResp <- mkNextResp + (resps, ()) <- runConnectedPeersLookaheadFixedSender + (createPipelineTestChannels 100) nullTracer CBOR.codecReqResp + client (reqRespServerPeerLookaheadFixedSender nextResp) + return resps + + runEmbed = do + nextResp <- mkNextResp + (resps, ()) <- runConnectedPeersLookahead + (createPipelineTestChannels 100) nullTracer CBOR.codecReqResp + client (embedLookaheadUsingFixedSender (reqRespServerPeerLookaheadFixedSender nextResp)) + return resps + +prop_lookaheadFixedEquiv_IO :: (Int -> (Int, Int)) -> NonNegative Int -> Property +prop_lookaheadFixedEquiv_IO g n = + ioProperty (prop_lookaheadFixedEquiv g n) + +prop_lookaheadFixedEquiv_ST :: (Int -> (Int, Int)) -> NonNegative Int -> Property +prop_lookaheadFixedEquiv_ST g n = + let tr = runSimTrace (prop_lookaheadFixedEquiv g n) in counterexample (intercalate "\n" $ map show $ traceEvents tr) $ case traceResult True tr of Left err -> throw err diff --git a/typed-protocols.cabal b/typed-protocols.cabal --- a/typed-protocols.cabal +++ b/typed-protocols.cabal @@ -1,12 +1,12 @@ cabal-version: 3.4 name: typed-protocols -version: 1.2.1.0 +version: 1.2.2.0 synopsis: A framework for strongly typed protocols description: A robust session type framework which supports protocol pipelining. - Haddocks are published [here](https://input-output-hk.github.io/typed-protocols/) + Haddocks are published [here](https://intersectmbo.github.io/typed-protocols/) license: Apache-2.0 license-files: LICENSE NOTICE -copyright: 2019-2025 Input Output Global Inc (IOG) +copyright: 2019-2026 Input Output Global Inc (IOG), 2026 Intersect author: Alexander Vieth, Duncan Coutts, Marcin Szamotulski maintainer: alex@well-typed.com, duncan@well-typed.com, marcin.szamotulski@iohk.io category: Control @@ -14,7 +14,13 @@ tested-with: GHC == {9.6, 9.8, 9.10, 9.12, 9.14} extra-doc-files: CHANGELOG.md README.md +bug-reports: https://github.com/intersectmbo/typed-protocols/issues +source-repository head + type: git + location: https://github.com/intersectmbo/typed-protocols + subdir: typed-protocols + -- Minimal GHC setup, additional extensions are enabled per package (e.g. -- pervasive type level extensions in the code base like `GADTs` or -- `DataKinds`, etc), or per module (e.g. `CPP` or other more exotic ones). @@ -49,7 +55,7 @@ other-modules: Network.TypedProtocol.Lemmas build-depends: base >=4.12 && <4.23, deepseq, - io-classes:io-classes ^>= 1.8 || ^>= 1.9 || ^>= 1.10, + io-classes:io-classes >=1.8 && <1.12 , singletons ^>= 3.0 hs-source-dirs: src default-extensions: DataKinds @@ -169,6 +175,7 @@ build-depends: base , bytestring , contra-tracer + , transformers , typed-protocols:{typed-protocols,codec-properties,examples} , io-classes:io-classes , io-sim