typed-protocols 1.2.1.0 → 1.2.2.0
raw patch · 18 files changed
+942/−31 lines, 18 filesdep +transformersdep ~io-classesnew-uploader
Dependencies added: transformers
Dependency ranges changed: io-classes
Files
- CHANGELOG.md +10/−0
- NOTICE +1/−1
- README.md +3/−3
- examples/Network/TypedProtocol/Driver/Simple.hs +111/−0
- examples/Network/TypedProtocol/ReqResp/Server.hs +56/−0
- examples/Network/TypedProtocol/Stateful/ReqResp/Type.hs +1/−1
- src/Network/TypedProtocol/Core.hs +29/−6
- src/Network/TypedProtocol/Driver.hs +198/−1
- src/Network/TypedProtocol/Peer.hs +92/−1
- src/Network/TypedProtocol/Peer/Client.hs +4/−1
- src/Network/TypedProtocol/Peer/Server.hs +126/−1
- src/Network/TypedProtocol/Proofs.hs +127/−0
- stateful/Network/TypedProtocol/Stateful/Codec.hs +2/−2
- stateful/Network/TypedProtocol/Stateful/Driver.hs +4/−4
- stateful/Network/TypedProtocol/Stateful/Peer.hs +3/−3
- test/Network/TypedProtocol/PingPong/Tests.hs +6/−1
- test/Network/TypedProtocol/ReqResp/Tests.hs +158/−2
- typed-protocols.cabal +11/−4
CHANGELOG.md view
@@ -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
NOTICE view
@@ -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.
README.md view
@@ -1,5 +1,5 @@-[](https://github.com/input-output-hk/typed-protocols/actions/workflows/haskell.yml)-[](https://input-output-hk.github.io/cardano-engineering-handbook)+[](https://github.com/intersectmbo/typed-protocols/actions/workflows/haskell.yml)+[](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
examples/Network/TypedProtocol/Driver/Simple.hs view
@@ -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
examples/Network/TypedProtocol/ReqResp/Server.hs view
@@ -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')
examples/Network/TypedProtocol/Stateful/ReqResp/Type.hs view
@@ -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
src/Network/TypedProtocol/Core.hs view
@@ -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'.
src/Network/TypedProtocol/Driver.hs view
@@ -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
src/Network/TypedProtocol/Peer.hs view
@@ -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)
src/Network/TypedProtocol/Peer/Client.hs view
@@ -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
src/Network/TypedProtocol/Peer/Server.hs view
@@ -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 #-}
src/Network/TypedProtocol/Proofs.hs view
@@ -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
stateful/Network/TypedProtocol/Stateful/Codec.hs view
@@ -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)) }
stateful/Network/TypedProtocol/Stateful/Driver.hs view
@@ -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.
stateful/Network/TypedProtocol/Stateful/Peer.hs view
@@ -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
test/Network/TypedProtocol/PingPong/Tests.hs view
@@ -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 ] ] ]
test/Network/TypedProtocol/ReqResp/Tests.hs view
@@ -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
typed-protocols.cabal view
@@ -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