typed-session 0.2.0.1 → 0.3.0.0

raw patch · 6 files changed

+233/−102 lines, 6 filesPVP ok

version bump matches the API change (PVP)

API changes (from Hackage documentation)

Files

• src/TypedSession/Core.hs +2/−1
• src/TypedSession/Driver.hs +145/−53
• test/Book3/Main.hs +24/−29
• test/Book3/Peer.hs +61/−3
• test/Book3/Protocol.hs +0/−15
• typed-session.cabal +1/−1

@@ -14,6 +14,7 @@  module TypedSession.Core where +import Control.Concurrent.Class.MonadSTM (MonadSTM (..)) import Data.IFunctor import Data.IntMap (IntMap) import Data.Kind@@ -125,7 +126,7 @@ Messages received from the outside are placed in MsgCache. When interpreting Peer will use the Msg in MsgCache. -}-type MsgCache role' ps = IntMap (AnyMsg role' ps)+type MsgCache role' ps n = TVar n (IntMap (AnyMsg role' ps))  {- | Packaging of Msg, shielding part of the type information, mainly used for serialization.

@@ -25,18 +25,19 @@  3. Each role has one or more decode threads, and the decoded Msgs are placed in the MsgCache. -4. SendMap aggregates the send functions of multiple Channels together.-When sending a message, the send function of the receiver is searched from SendMap. -} module TypedSession.Driver where  import Control.Concurrent.Class.MonadSTM+import Control.Monad.Class.MonadFork (MonadFork (killThread), forkIO) import Control.Monad.Class.MonadThrow (MonadThrow, throwIO) import Control.Monad.Class.MonadTimer (MonadDelay, threadDelay)+import Data.Data (Typeable) import Data.IFunctor (At (..), Sing, SingI (sing)) import qualified Data.IFunctor as I import Data.IntMap (IntMap) import qualified Data.IntMap as IntMap+import Data.Traversable (for) import GHC.Exception (Exception) import TypedSession.Codec import TypedSession.Core@@ -63,6 +64,7 @@        . (SingToInt ps)       => Sing st'       -> m (AnyMsg role' ps)+  , terminalDecodeThreads :: [m ()]   }  {- |@@ -76,8 +78,10 @@   => Driver role' ps m   -> Peer role' ps r m (At a (Done r)) st   -> m a-runPeerWithDriver Driver{sendMsg, recvMsg} =-  go+runPeerWithDriver Driver{sendMsg, recvMsg, terminalDecodeThreads} peer = do+  a <- go peer+  sequence_ terminalDecodeThreads+  pure a  where   go     :: forall st'@@ -115,19 +119,26 @@ nullTracer _ = pure ()  {- |-SendMap aggregates the send functions of multiple Channels together.-When sending a message, the send function of the receiver is found from SendMap.+ConnChannels aggregates the multiple connect channels together. -}-type SendMap role' m bytes = IntMap (bytes -> m ())+type ConnChannels role' m bytes = [(SomeRole role', Channel m bytes)] +data NotConnect role' = NotConnect role'+  deriving (Show)++instance (Show role', Typeable role') => Exception (NotConnect role')++data SomeRole role' = forall (r :: role'). (SingToInt role') => SomeRole (Sing r)+ {- | -Build Driver through SendMap and MsgCache.+Build Driver through ConnChannels. Here we need some help from other functions:  1. `Tracer role' ps n` is similar to the log function, used to print received or sent messages. 2. `Encode role' ps` bytes encoding function, converts Msg into bytes.-3. `forall a. n a -> m a` This is a bit complicated, I will explain it in detail below.+3. `Decode role' ps failure bytes` bytes decode function, converts bytes into Msg.+4. `forall a. n a -> m a` This is a bit complicated, I will explain it in detail below.  I see Peer as three layers: @@ -136,53 +147,74 @@ 3. `n` bottom layer, responsible for receiving and sending bytes. It can have multiple options such as IO or IOSim. Using IOSim can easily test the code. -} driverSimple-  :: forall role' ps bytes m n+  :: forall role' ps failure bytes m n    . ( Monad m      , Monad n      , MonadSTM n+     , MonadFork n+     , MonadDelay n+     , MonadThrow n+     , SingToInt role'+     , Enum role'+     , Show role'+     , Typeable role'+     , Exception failure      )   => Tracer role' ps n   -> Encode role' ps bytes-  -> SendMap role' n bytes-  -> TVar n (MsgCache role' ps)+  -> Decode role' ps failure bytes+  -> ConnChannels role' n bytes   -> (forall a. n a -> m a)-  -> Driver role' ps m-driverSimple tracer Encode{encode} sendMap tvar liftFun =-  Driver{sendMsg, recvMsg}- where-  sendMsg-    :: forall (send :: role') (recv :: role') (from :: ps) (st :: ps) (st1 :: ps)-     . ( SingI recv-       , SingI from-       , SingToInt ps-       , SingToInt role'-       )-    => Sing recv-    -> Msg role' ps from send st recv st1-    -> m ()-  sendMsg role msg = liftFun $ do-    case IntMap.lookup (singToInt role) sendMap of-      Nothing -> error "np"-      Just sendFun -> sendFun (encode msg)-    tracer (TraceSendMsg (AnyMsg msg))+  -> n (Driver role' ps m)+driverSimple+  tracer+  Encode{encode}+  decodeV+  connChannels+  liftFun = do+    msgCache <- newTVarIO IntMap.empty+    ths <- for connChannels $ \(_, channel) -> forkIO $ decodeLoop Nothing decodeV channel msgCache+    let terminalDecodeThreads = map (\tid -> liftFun $ killThread tid) ths+    pure $ Driver{sendMsg, recvMsg = recvMsg' msgCache, terminalDecodeThreads}+   where+    sendMap = IntMap.fromList $ fmap (\(SomeRole r, c) -> (singToInt r, send c)) connChannels+    sendMsg+      :: forall (send :: role') (recv :: role') (from :: ps) (st :: ps) (st1 :: ps)+       . ( SingI recv+         , SingI from+         , SingToInt ps+         , SingToInt role'+         )+      => Sing recv+      -> Msg role' ps from send st recv st1+      -> m ()+    sendMsg role msg = liftFun $ do+      let recvKey = singToInt role+      case IntMap.lookup recvKey sendMap of+        Nothing -> do+          let recvRole = toEnum @role' recvKey+          throwIO (NotConnect recvRole)+        Just sendFun -> sendFun (encode msg)+      tracer (TraceSendMsg (AnyMsg msg)) -  recvMsg-    :: forall (st' :: ps)-     . (SingToInt ps)-    => Sing st'-    -> m (AnyMsg role' ps)-  recvMsg sst' = do-    let singInt = singToInt sst'-    liftFun $ do-      anyMsg <- atomically $ do-        agencyMsg <- readTVar tvar-        case IntMap.lookup singInt agencyMsg of-          Nothing -> retry-          Just v -> do-            writeTVar tvar (IntMap.delete singInt agencyMsg)-            pure v-      tracer (TraceRecvMsg (anyMsg))-      pure anyMsg+    recvMsg'+      :: forall (st' :: ps)+       . (SingToInt ps)+      => MsgCache role' ps n+      -> Sing st'+      -> m (AnyMsg role' ps)+    recvMsg' msgCache sst' = do+      let singInt = singToInt sst'+      liftFun $ do+        anyMsg <- atomically $ do+          agencyMsg <- readTVar msgCache+          case IntMap.lookup singInt agencyMsg of+            Nothing -> retry+            Just v -> do+              writeTVar msgCache (IntMap.delete singInt agencyMsg)+              pure v+        tracer (TraceRecvMsg (anyMsg))+        pure anyMsg  {- | decode loop, usually in a separate thread.@@ -199,13 +231,12 @@ -} decodeLoop   :: (Exception failure, MonadDelay n, MonadSTM n, MonadThrow n)-  => Tracer role' ps n-  -> Maybe bytes+  => Maybe bytes   -> Decode role' ps failure bytes   -> Channel n bytes-  -> TVar n (MsgCache role' ps)+  -> MsgCache role' ps n   -> n ()-decodeLoop tracer mbt d@Decode{decode} channel tvar = do+decodeLoop mbt d@Decode{decode} channel tvar = do   result <- runDecoderWithChannel channel mbt decode   case result of     Right (AnyMsg msg, mbt') -> do@@ -215,7 +246,68 @@         case IntMap.lookup agencyInt agencyMsg of           Nothing -> writeTVar tvar (IntMap.insert agencyInt (AnyMsg msg) agencyMsg)           Just _v -> retry-      decodeLoop tracer mbt' d channel tvar+      decodeLoop mbt' d channel tvar     Left failure -> do       threadDelay 1_000_000       throwIO failure++localDriverSimple+  :: forall role' ps m n+   . ( Monad m+     , Monad n+     , MonadSTM n+     , Enum role'+     , MonadThrow n+     , Show role'+     , Typeable role'+     )+  => Tracer role' ps n+  -> IntMap (MsgCache role' ps n)+  -> SomeRole role'+  -> (forall a. n a -> m a)+  -> Driver role' ps m+localDriverSimple tracer allMsgCache (SomeRole r) liftFun =+  Driver{sendMsg, recvMsg = recvMsg' (allMsgCache IntMap.! (singToInt r)), terminalDecodeThreads = []}+ where+  sendMsg+    :: forall (send :: role') (recv :: role') (from :: ps) (st :: ps) (st1 :: ps)+     . ( SingI recv+       , SingI from+       , SingToInt ps+       , SingToInt role'+       )+    => Sing recv+    -> Msg role' ps from send st recv st1+    -> m ()+  sendMsg role msg = liftFun $ do+    let recvKey = singToInt role+    case IntMap.lookup (singToInt role) allMsgCache of+      Nothing -> do+        let recvRole = toEnum @role' recvKey+        throwIO (NotConnect recvRole)+      Just ttvar -> atomically $ do+        agencyMsg <- readTVar ttvar+        let singInt = singToInt (sing @from)+        case IntMap.lookup singInt agencyMsg of+          Nothing -> writeTVar ttvar (IntMap.insert singInt (AnyMsg msg) agencyMsg)+          Just _v -> retry+    tracer (TraceSendMsg (AnyMsg msg))++  recvMsg'+    :: forall (st' :: ps)+     . (SingToInt ps)+    => MsgCache role' ps n+    -> Sing st'+    -> m (AnyMsg role' ps)+  recvMsg' msgCache sst' = do+    let singInt = singToInt sst'+    liftFun $ do+      anyMsg <- atomically $ do+        agencyMsg <- readTVar msgCache+        case IntMap.lookup singInt agencyMsg of+          Nothing -> retry+          Just v -> do+            writeTVar msgCache (IntMap.delete singInt agencyMsg)+            pure v+      tracer (TraceRecvMsg (anyMsg))+      pure anyMsg

@@ -25,13 +25,12 @@ import Control.Monad.Class.MonadFork (MonadFork, forkIO) import Control.Monad.Class.MonadSay import Control.Monad.Class.MonadThrow (MonadThrow)+import Control.Monad.Class.MonadTimer (MonadDelay) import Control.Monad.IOSim-import qualified Data.IntMap as IntMap import System.Random (StdGen, split) import TypedSession.Codec import TypedSession.Core import TypedSession.Driver-import Control.Monad.Class.MonadTimer (MonadDelay)  mvarsAsChannel   :: (MonadSTM m)@@ -62,36 +61,32 @@   buyerTMVar <- newEmptyTMVarIO @n @(AnyMsg BookRole Book)   buyer2TMVar <- newEmptyTMVarIO @n @(AnyMsg BookRole Book)   sellerTMVar <- newEmptyTMVarIO @n @(AnyMsg BookRole Book)-  let buyerSellerChannel = mvarsAsChannel @n buyerTMVar sellerTMVar-      buyerBuyer2Channel = mvarsAsChannel @n buyerTMVar buyer2TMVar -      sellerBuyerChannel = mvarsAsChannel @n sellerTMVar buyerTMVar--      buyer2BuyerChannel = mvarsAsChannel @n buyer2TMVar buyerTMVar--      sendFun bufferWrite x = atomically @n (putTMVar bufferWrite x)-      sendToRole =-        IntMap.fromList-          [ (singToInt SSeller, sendFun sellerTMVar)-          , (singToInt SBuyer, sendFun buyerTMVar)-          , (singToInt SBuyer2, sendFun buyer2TMVar)-          ]-  buyerTvar <- newTVarIO IntMap.empty-  buyer2Tvar <- newTVarIO IntMap.empty-  sellerTvar <- newTVarIO IntMap.empty-  let buyerDriver = driverSimple (myTracer "buyer :") encodeMsg sendToRole buyerTvar sendM-      buyer2Driver = driverSimple (myTracer "buyer2 :") encodeMsg sendToRole buyer2Tvar sendM-      sellerDriver = driverSimple (myTracer "seller :") encodeMsg sendToRole sellerTvar sendM-  -- fork buyer decode thread, seller -> buyer-  forkIO $ decodeLoop (myTracer "buyer :") Nothing (Decode decodeMsg) buyerSellerChannel buyerTvar-  -- fork buyer decode thread, buyer2 -> buyer-  forkIO $ decodeLoop (myTracer "buyer :") Nothing (Decode decodeMsg) buyerBuyer2Channel buyerTvar+  buyerDriver <-+    driverSimple+      (myTracer "buyer :")+      encodeMsg+      (Decode decodeMsg)+      [ (SomeRole SSeller, mvarsAsChannel buyerTMVar sellerTMVar)+      , (SomeRole SBuyer2, mvarsAsChannel buyerTMVar buyer2TMVar)+      ]+      sendM -  -- fork seller decode thread, buyer -> seller-  forkIO $ decodeLoop (myTracer "seller :") Nothing (Decode decodeMsg) sellerBuyerChannel sellerTvar+  buyer2Driver <-+    driverSimple+      (myTracer "buyer2 :")+      encodeMsg+      (Decode decodeMsg)+      [(SomeRole SBuyer, mvarsAsChannel buyer2TMVar buyerTMVar)]+      sendM -  -- fork buyer2 decode thread, buyer -> buyer2-  forkIO $ decodeLoop (myTracer "buyer2 :") Nothing (Decode decodeMsg) buyer2BuyerChannel buyer2Tvar+  sellerDriver <-+    driverSimple+      (myTracer "seller :")+      encodeMsg+      (Decode decodeMsg)+      [(SomeRole SBuyer, mvarsAsChannel sellerTMVar buyerTMVar)]+      sendM    let (g0, g1) = split g       (g2, g3) = split g0

@@ -12,16 +12,25 @@ {-# LANGUAGE ScopedTypeVariables #-} {-# LANGUAGE TypeApplications #-} {-# LANGUAGE TypeFamilies #-}+{-# LANGUAGE TypeOperators #-}+{-# OPTIONS_GHC -Wno-unused-do-bind #-}  module Book3.Peer where  import Book3.Protocol import Book3.Type-import Control.Algebra (Has)-import Control.Effect.Random (Random, uniform)-import Data.IFunctor (At (..), returnAt)+import Control.Carrier.Lift+import Control.Carrier.Random.Gen+import Control.Carrier.State.Strict (runState)+import Control.Effect.State+import Control.Monad+import Data.IFunctor (At (..), SingI (..), returnAt) import qualified Data.IFunctor as I+import Data.IntMap (IntMap)+import qualified Data.IntMap as IntMap+import System.Random (StdGen, mkStdGen, randomIO) import TypedSession.Core+import Unsafe.Coerce (unsafeCoerce)  budget :: Int budget = 16@@ -135,3 +144,52 @@         TwoNotBuy -> returnAt ()         TwoAccept -> yield (TwoDate 100)         TwoNotBuy1 -> returnAt ()++-- data AnyPeer role' ps m where+--   AnyPeer :: Peer role' ps r m (At () (Done r)) st -> AnyPeer role' ps m++-- runAnyPeers+--   :: forall n role' ps sig m+--    . ( SingToInt role'+--      , Has (State (IntMap (AnyPeer role' ps n))) sig m+--      )+--   => (forall a. n a -> m a) -> m ()+-- runAnyPeers liftFun = do+--   im <- get @(IntMap (AnyPeer role' ps n))+--   case IntMap.keys im of+--     [] -> pure ()+--     keys -> do+--       forM_ keys $ \key -> do+--         gets @(IntMap (AnyPeer role' ps n)) (IntMap.lookup key) >>= \case+--           Nothing -> error "np"+--           Just (AnyPeer peer) -> case peer of+--             IReturn (At ()) -> do+--               modify @(IntMap (AnyPeer role' ps n)) (IntMap.delete key)+--             LiftM fm -> do+--               np <- liftFun fm+--               modify @(IntMap (AnyPeer role' ps n)) (IntMap.insert key (AnyPeer np))+--             Yield (msg :: Msg role' ps st send sps recv rps) cont -> do+--               let recvKey = (singToInt $ sing @recv)+--               gets @(IntMap (AnyPeer role' ps n)) (IntMap.lookup recvKey) >>= \case+--                 Nothing -> error "np"+--                 Just (AnyPeer recvPeer) -> case recvPeer of+--                   Await scont -> do+--                     let nS = scont (unsafeCoerce msg)+--                     modify @(IntMap (AnyPeer role' ps n)) (IntMap.insert recvKey (AnyPeer nS))+--                   _ -> error "np"+--               modify @(IntMap (AnyPeer role' ps n)) (IntMap.insert key (AnyPeer cont))+--             Await{} -> pure ()+--       runAnyPeers @n @role' @ps liftFun++-- runAP = do+--   i <- randomIO @Int+--   runRandom (mkStdGen i)+--     $ runM @(RandomC StdGen IO)+--     $ runState+--       ( IntMap.fromList+--           [ (singToInt SBuyer, AnyPeer (buyerPeer @_ @(RandomC StdGen IO) I.>> returnAt ()))+--           , (singToInt SBuyer2, AnyPeer (buyer2Peer @_ @(RandomC StdGen IO) I.>> returnAt ()))+--           , (singToInt SSeller, AnyPeer (sellerPeer @_ @(RandomC StdGen IO) I.>> returnAt ()))+--           ]+--       )+--     $ (runAnyPeers @(RandomC StdGen IO) @BookRole @Book sendM)

@@ -112,18 +112,3 @@     TwoSuccess d -> "TwoSuccess " <> show d     TwoNotBuy1 -> "TwoNotBuy1"     TwoFailed -> "TwoFailed"--type Start s = Maybe s-type Start1  = Int--$( do - x <- reify ''Start - y <- reify ''Start1- runIO $ do -  putStrLn "--------------------------------------------"-  putStrLn $ show x-  putStrLn $ show y-  putStrLn "--------------------------------------------"- pure []--  )

@@ -20,7 +20,7 @@ -- PVP summary:     +-+------- breaking API changes --                  | | +----- non-breaking API additions --                  | | | +--- code changes with no API change-version:            0.2.0.1+version:            0.3.0.0  -- A short (one-line) description of the package. synopsis: typed session framework