extensible-effects-concurrent 0.24.0 → 0.24.1

raw patch · 24 files changed

+1155/−745 lines, 24 filesPVP: major bump suggested

API removals or changes: PVP suggests a major version bump

API changes (from Hackage documentation)

Files

• ChangeLog.md +17/−11
• examples/example-1/Main.hs +41/−41
• examples/example-2/Main.hs +40/−39
• examples/example-4/Main.hs +1/−1
• extensible-effects-concurrent.cabal +3/−2
• src/Control/Eff/Concurrent.hs +3/−0
• src/Control/Eff/Concurrent/Process.hs +126/−29
• src/Control/Eff/Concurrent/Process/ForkIOScheduler.hs +29/−17
• src/Control/Eff/Concurrent/Process/SingleThreadedScheduler.hs +87/−49
• src/Control/Eff/Concurrent/Process/Timer.hs +10/−8
• src/Control/Eff/Concurrent/Protocol.hs +85/−0
• src/Control/Eff/Concurrent/Protocol/EffectfulServer.hs +300/−0
• src/Control/Eff/Concurrent/Protocol/Observer/Queue.hs +5/−5
• src/Control/Eff/Concurrent/Protocol/Request.hs +20/−10
• src/Control/Eff/Concurrent/Protocol/Server.hs +0/−416
• src/Control/Eff/Concurrent/Protocol/StatefulServer.hs +206/−0
• src/Control/Eff/Concurrent/Protocol/Supervisor.hs +14/−14
• src/Control/Eff/Concurrent/Protocol/Supervisor/InternalState.hs +1/−1
• test/ForkIOScheduler.hs +10/−10
• test/GenServerTests.hs +27/−17
• test/LoopTests.hs +2/−2
• test/ProcessBehaviourTestCases.hs +117/−62
• test/SingleThreadedScheduler.hs +4/−4
• test/SupervisorTests.hs +7/−7

@@ -2,6 +2,7 @@  ## Plan for future Versions +- Put timing into pure processes! - Every `Api` type instance now **must** be an `NFData`    and a `ToLogText` instance - `call` will always require a `Timeout`@@ -14,20 +15,25 @@     - Remove `ToLogMessage`     - Remove `logXXX'` users have to use `logXXX` and `ToLogText`  +## 0.24.1++- Add more `EmbedProtocol` tuple instances (4-tuple, 5-tuple)+- Make `Effectful.Server` instances composable (See the `GenServerTests` for an example)+  [more details in a seperate file](./ChangeLog-Details-0.24.1.md)+- Add `ProcessTitle` - every process now must have a short title text for logging+- Add `ProcessDetails` - every process can call `UpdateProcessDetails` to update+  its infos about the current state of it for debugging and error tracing purposes.+- Add `GetProcessState` to retreive the `ProcessDetails` for some other process.+ ## 0.24.0 -- Add `GenServer` module for `Api` handling: +- Get rid of the `PrettyTypeShow` constraint in `Tangible`+- Get rid of `LogWriterEffects` and the necessity for some `UndecidableInstances` that came with it  +- Add `Server` module for `Api` handling via type classes+    - Add `Stateless` +    - Add `GenServer`+- Reimplement `Supervisor`            -    - Reduce the server to handle a single `Api` instance-        - Instances of `GenServer` must provide `lenses` and `prisms`-          for accessing the *other* Api instances like `Observer`-          and `ServerCallback` -    - Add `ServerCallback`, an internal API for interacting with these-      `GenServer` instances-        - `Api AnyMsg 'Asynchronous` -        - `Api GetInfo ('Synchronous Text)`  -         - ## 0.23.0  - Include the process id in the console and trace log renderer

@@ -6,7 +6,7 @@ import           Control.Monad import           Data.Dynamic import           Control.Eff.Concurrent-import           Control.Eff.Concurrent.Protocol.Server+import           Control.Eff.Concurrent.Protocol.EffectfulServer import qualified Control.Exception             as Exc import qualified Data.Text as T import           Control.DeepSeq@@ -42,7 +42,7 @@ main = defaultMain example  mainProcessSpawnsAChildAndReturns :: HasCallStack => Eff InterruptableProcEff ()-mainProcessSpawnsAChildAndReturns = void (spawn (void receiveAnyMessage))+mainProcessSpawnsAChildAndReturns = void (spawn "some child" (void receiveAnyMessage))  example:: HasCallStack => Eff InterruptableProcEff () example = do@@ -73,54 +73,54 @@             go   go -type instance GenServerProtocol TestProtocol = TestProtocol- testServerLoop :: Eff InterruptableProcEff (Endpoint TestProtocol)-testServerLoop = start (statelessGenServer handleReq "test-server-1")+testServerLoop = start (genServer (const id) handleReq "test-server-1")  where-  handleReq :: GenServerId TestProtocol -> Event TestProtocol -> Eff (ToStatelessEffects InterruptableProcEff) ()-  handleReq _myId (OnRequest (Call orig Terminate)) = do-    me <- self-    logInfo (T.pack (show me ++ " exiting"))-    sendReply orig ()-    interrupt NormalExitRequested--  handleReq _myId (OnRequest (Call orig (TerminateError e))) = do-    me <- self-    logInfo (T.pack (show me ++ " exiting with error: " ++ e))-    sendReply orig ()-    interrupt (ErrorInterrupt e)--  handleReq _myId (OnRequest (Call orig (SayHello mx))) =-    case mx of-      "e1" -> do+  handleReq :: GenServerId TestProtocol -> Event TestProtocol -> Eff InterruptableProcEff ()+  handleReq _myId (OnCall ser orig cm) =+    case cm of+      Terminate -> do         me <- self-        logInfo (T.pack (show me ++ " raising an error"))-        interrupt (ErrorInterrupt "No body loves me... :,(")+        logInfo (T.pack (show me ++ " exiting"))+        sendReply ser orig ()+        interrupt NormalExitRequested -      "e2" -> do+      TerminateError e -> do         me <- self-        logInfo (T.pack (show me ++ " throwing a MyException "))-        void (lift (Exc.throw MyException))+        logInfo (T.pack (show me ++ " exiting with error: " ++ e))+        sendReply ser orig ()+        interrupt (ErrorInterrupt e) -      "self" -> do-        me <- self-        logInfo (T.pack (show me ++ " casting to self"))-        cast (asEndpoint @TestProtocol me) (Shout "from me")-        sendReply orig False+      SayHello mx ->+        case mx of+          "e1" -> do+            me <- self+            logInfo (T.pack (show me ++ " raising an error"))+            interrupt (ErrorInterrupt "No body loves me... :,(") -      "stop" -> do-        me <- self-        logInfo (T.pack (show me ++ " stopping me"))-        sendReply orig False-        interrupt (ErrorInterrupt "test error")+          "e2" -> do+            me <- self+            logInfo (T.pack (show me ++ " throwing a MyException "))+            void (lift (Exc.throw MyException)) -      x -> do-        me <- self-        logInfo (T.pack (show me ++ " Got Hello: " ++ x))-        sendReply orig (length x > 3)+          "self" -> do+            me <- self+            logInfo (T.pack (show me ++ " casting to self"))+            cast (asEndpoint @TestProtocol me) (Shout "from me")+            sendReply ser orig False -  handleReq _myId (OnRequest (Cast (Shout x))) = do+          "stop" -> do+            me <- self+            logInfo (T.pack (show me ++ " stopping me"))+            sendReply ser orig False+            interrupt (ErrorInterrupt "test error")++          x -> do+            me <- self+            logInfo (T.pack (show me ++ " Got Hello: " ++ x))+            sendReply ser orig (length x > 3)++  handleReq _myId (OnCast (Shout x)) = do     me <- self     logInfo (T.pack (show me ++ " Shouting: " ++ x)) 

@@ -1,11 +1,11 @@-+{-# LANGUAGE UndecidableInstances #-} -- | Another complete example for the library module Main where  import           Data.Dynamic import           Control.Eff import           Control.Eff.Concurrent-import           Control.Eff.Concurrent.Protocol.Server+import           Control.Eff.Concurrent.Protocol.StatefulServer import           Control.Monad import           Data.Foldable import           Control.Lens@@ -87,36 +87,40 @@  type instance ToPretty (Counter, ObserverRegistry CounterChanged, SupiDupi) = PutStr "supi-counter" -instance (LogIo q) => Server SupiCounter (InterruptableProcess q) where+instance (LogIo q) => Server SupiCounter q where -  type instance Model SupiCounter = (Integer, Observers CounterChanged, Maybe (RequestOrigin SupiCounter (Maybe ())))+  type instance Model SupiCounter =+    ( Integer+    , Observers CounterChanged+    , Maybe ( Serializer (Reply SupiCounter (Maybe ()))+            , RequestOrigin SupiCounter (Maybe ())+            )+    ) -  data instance StartArgument SupiCounter (InterruptableProcess q) = MkEmptySupiCounter+  data instance StartArgument SupiCounter q = MkEmptySupiCounter    setup _ = return ((0, emptyObservers, Nothing), ())    update _ = \case-    OnRequest req ->-      case req of-        Call orig callReq ->-          case callReq of-            ToPdu1 Cnt ->-              sendReply orig =<< useModel @SupiCounter _1-            ToPdu2 _ -> error "unreachable"-            ToPdu3 (Whoopediedoo c) ->-              modifyModel @SupiCounter (_3 .~ if c then Just orig else Nothing)+    OnCall ser orig callReq ->+      case callReq of+        ToPdu1 Cnt ->+          sendReply ser orig =<< useModel @SupiCounter _1+        ToPdu2 _ -> error "unreachable"+        ToPdu3 (Whoopediedoo c) ->+          modifyModel @SupiCounter (_3 .~ if c then Just (ser, orig) else Nothing) -        Cast castReq ->-          case castReq of-            ToPdu1 Inc -> do-              val' <- view _1 <$> modifyAndGetModel @SupiCounter (_1 %~ (+ 1))-              zoomModel @SupiCounter _2 (observed (CounterChanged val'))-              when (val' > 5) $-                getAndModifyModel @SupiCounter (_3 .~ Nothing)-                >>= traverse_ (flip sendReply (Just ())) . view _3-            ToPdu2 x ->-              zoomModel @SupiCounter _2 (handleObserverRegistration x)-            ToPdu3 _ -> error "unreachable"+    OnCast castReq ->+      case castReq of+        ToPdu1 Inc -> do+          val' <- view _1 <$> modifyAndGetModel @SupiCounter (_1 %~ (+ 1))+          zoomModel @SupiCounter _2 (observed (CounterChanged val'))+          when (val' > 5) $+            getAndModifyModel @SupiCounter (_3 .~ Nothing)+            >>= traverse_ (\(ser, orig) -> sendReply ser orig (Just ())) . view _3+        ToPdu2 x ->+          zoomModel @SupiCounter _2 (handleObserverRegistration x)+        ToPdu3 _ -> error "unreachable"      other -> logWarning (T.pack (show other)) @@ -130,19 +134,16 @@   :: (LogsTo IO q, Lifted IO q, Typeable q)   => Eff (InterruptableProcess q) (Observer CounterChanged) logCounterObservations = do-  svr <- start-          $ genServer @(Observer CounterChanged)-            (\_me -> pure (emptyObservers, ()))-            (\_me ->-                \case-                  OnRequest (Cast r) ->-                    handleObservations (\msg -> logInfo' ("observed: " ++ show msg)) r-                  wtf -> logNotice (T.pack (show wtf))-            )-            "counter logger"-+  svr <- start OCCStart   pure (toObserver svr) -type instance GenServerModel (Observer CounterChanged) = Observers CounterChanged-type instance GenServerSettings (Observer CounterChanged) = ()-type instance GenServerProtocol (Observer CounterChanged) = Observer CounterChanged+instance Member Logs q => Server (Observer CounterChanged) q where+  data StartArgument (Observer CounterChanged) q = OCCStart+  type Model (Observer CounterChanged) = Observers CounterChanged+  type Settings (Observer CounterChanged) = ()+  type Protocol (Observer CounterChanged) = Observer CounterChanged+  setup _ = pure (emptyObservers, ())+  update _ =+    \case+      OnCast r -> handleObservations (\msg -> logInfo' ("observed: " ++ show msg)) r+      wtf -> logNotice (T.pack (show wtf))

@@ -16,7 +16,7 @@  firstExample :: (HasCallStack, Member Logs q) => Eff (InterruptableProcess q) () firstExample = do-  person <- spawn+  person <- spawn "first-example"     (do       logInfo "I am waiting for someone to ask me..."       WhoAreYou replyPid <- receiveMessage

@@ -1,6 +1,6 @@ cabal-version:  2.0 name:           extensible-effects-concurrent-version:        0.24.0+version:        0.24.1 description:    Please see the README on GitHub at <https://github.com/sheyll/extensible-effects-concurrent#readme> synopsis:       Message passing concurrency as extensible-effect homepage:       https://github.com/sheyll/extensible-effects-concurrent#readme@@ -83,7 +83,8 @@                   Control.Eff.Concurrent,                   Control.Eff.Concurrent.Protocol,                   Control.Eff.Concurrent.Protocol.Client,-                  Control.Eff.Concurrent.Protocol.Server,+                  Control.Eff.Concurrent.Protocol.EffectfulServer,+                  Control.Eff.Concurrent.Protocol.StatefulServer,                   Control.Eff.Concurrent.Protocol.Request,                   Control.Eff.Concurrent.Protocol.Supervisor,                   Control.Eff.Concurrent.Process,

@@ -66,6 +66,7 @@                                                 , toStrictDynamic                                                 , fromStrictDynamic                                                 , unwrapStrictDynamic+                                                , Serializer(..)                                                 , ProcessId(..)                                                 , fromProcessId                                                 , ConsProcess@@ -77,6 +78,8 @@                                                 , sendShutdown                                                 , sendInterrupt                                                 , makeReference+                                                , getProcessState+                                                , updateProcessDetails                                                 , receiveAnyMessage                                                 , receiveMessage                                                 , receiveSelectedMessage

@@ -15,11 +15,19 @@   ( -- * Process Effect     -- ** Effect Type Handling     Process(..)+    -- ** Process Info+  , ProcessTitle(..)+  , fromProcessTitle+  , ProcessDetails(..)+  , fromProcessDetails   , -- ** Message Data     StrictDynamic()   , toStrictDynamic   , fromStrictDynamic   , unwrapStrictDynamic+  , Serializer(..)++     -- ** ProcessId Type   , ProcessId(..)   , fromProcessId@@ -54,7 +62,9 @@   -- ** Process Life Cycle Management   , self   , isProcessAlive-  -- ** Spawning+  , getProcessState+  , updateProcessDetails+    -- ** Spawning   , spawn   , spawn_   , spawnLink@@ -109,29 +119,30 @@   ) where -import           GHC.Generics                   ( Generic-                                                , Generic1-                                                )+import           Control.Applicative import           Control.DeepSeq import           Control.Eff import           Control.Eff.Exception import           Control.Eff.Extend import           Control.Eff.Log.Handler+import qualified Control.Exception             as Exc import           Control.Lens import           Control.Monad                  ( void                                                 , (>=>)                                                 ) import           Data.Default import           Data.Dynamic+import           Data.Functor.Contravariant     () import           Data.Kind-import           GHC.Stack import           Data.Function-import           Control.Applicative import           Data.Maybe-import           Data.String                    (fromString)-import           Data.Text                     (Text, pack, unpack)+import           Data.String                    ( IsString, fromString )+import           Data.Text                      ( Text, pack, unpack) import qualified Data.Text                     as T-import qualified Control.Exception             as Exc+import           GHC.Stack+import           GHC.Generics                    ( Generic+                                                 , Generic1+                                                 )   -- | The process effect is the basis for message passing concurrency. This@@ -169,13 +180,11 @@   SelfPid :: Process r (ResumeProcess ProcessId)   -- | Start a new process, the new process will execute an effect, the function   -- will return immediately with a 'ProcessId'.-  Spawn :: Eff (Process r ': r) () -> Process r (ResumeProcess ProcessId)+  Spawn :: ProcessTitle -> Eff (Process r ': r) () -> Process r (ResumeProcess ProcessId)   -- | Start a new process, and 'Link' to it .   --   -- @since 0.12.0-  SpawnLink :: Eff (Process r ': r) () -> Process r (ResumeProcess ProcessId)-  -- | Get the process state (or 'Nothing' if the process is dead)-  GetProcessState :: ProcessId -> Process r (ResumeProcess (Maybe ProcessState))+  SpawnLink :: ProcessTitle -> Eff (Process r ': r) () -> Process r (ResumeProcess ProcessId)   -- | Shutdown the process; irregardless of the exit reason, this function never   -- returns,   Shutdown :: Interrupt 'NoRecovery -> Process r a@@ -222,13 +231,20 @@   -- @since 0.12.0   Unlink :: ProcessId -> Process r (ResumeProcess ()) +  -- | Update the 'ProcessDetails' of a process+  UpdateProcessDetails :: ProcessDetails -> Process r (ResumeProcess ())++  -- | Get the 'ProcessState' (or 'Nothing' if the process is dead)+  GetProcessState :: ProcessId -> Process r (ResumeProcess (Maybe (ProcessTitle, ProcessDetails, ProcessState)))++ instance Show (Process r b) where   showsPrec d = \case     FlushMessages -> showString "flush messages"     YieldProcess  -> showString "yield process"     SelfPid       -> showString "lookup the current process id"-    Spawn     _   -> showString "spawn a new process"-    SpawnLink _   -> showString "spawn a new process and link to it"+    Spawn   t _   -> showString "spawn a new process: " . shows t+    SpawnLink t _ -> showString "spawn a new process and link to it" . shows t     Shutdown sr ->       showParen (d >= 10) (showString "shutdown " . showsPrec 10 sr)     SendShutdown toPid sr -> showParen@@ -253,13 +269,48 @@       . showsPrec 10 sr       )     ReceiveSelectedMessage _   -> showString "receive a message"-    GetProcessState pid -> showString "get the process state of " . shows pid     MakeReference              -> showString "generate a unique reference"     Monitor   pid              -> showString "monitor " . shows pid     Demonitor i                -> showString "demonitor " . shows i     Link      l                -> showString "link " . shows l     Unlink    l                -> showString "unlink " . shows l+    GetProcessState pid        -> showString "get the process state of " . shows pid+    UpdateProcessDetails l     -> showString "update the process details to: " . shows l +-- | A short title for a 'Process' for logging purposes.+--+-- @since 0.24.1+newtype ProcessTitle =+  MkProcessTitle { _fromProcessTitle :: Text }+  deriving (Eq, Ord, NFData, Generic, IsString, Typeable, Semigroup, Monoid)++-- | An isomorphism lens for the 'ProcessTitle'+--+-- @since 0.24.1+fromProcessTitle :: Lens' ProcessTitle Text+fromProcessTitle = iso _fromProcessTitle MkProcessTitle++instance Show ProcessTitle where+  showsPrec _ (MkProcessTitle t) = showString (T.unpack t)++-- | A multi-line text describing the __current__+-- state of a process for debugging purposes.+--+-- @since 0.24.1+newtype ProcessDetails =+  MkProcessDetails { _fromProcessDetails :: Text }+  deriving (Eq, Ord, NFData, Generic, IsString, Typeable, Semigroup, Monoid)++-- | An isomorphism lens for the 'ProcessDetails'+--+-- @since 0.24.1+fromProcessDetails :: Lens' ProcessDetails Text+fromProcessDetails = iso _fromProcessDetails MkProcessDetails++instance Show ProcessDetails where+  showsPrec _ (MkProcessDetails t) = showString (T.unpack t)++ -- | Data flows between 'Process'es via these messages. -- -- This is just a newtype wrapper around 'Dynamic'.@@ -272,9 +323,27 @@   MkDynamicMessage :: Dynamic -> StrictDynamic   deriving Typeable +instance NFData StrictDynamic where+  rnf (MkDynamicMessage d) = d `seq` ()+ instance Show StrictDynamic where   show (MkDynamicMessage d) = show d ++-- | Serialize a @message@ into a 'StrictDynamic' value to be sent via 'sendAnyMessage'.+--+-- This indirection allows, among other things, the composition of+-- 'Control.Eff.Concurrent.Protocol.Effectful.Server's.+--+-- @since 0.24.1+newtype Serializer message =+  MkSerializer+    { runSerializer :: message -> StrictDynamic+    } deriving (Typeable)++instance Contravariant Serializer where+  contramap f (MkSerializer b) = MkSerializer (b . f)+ -- | Deeply evaluate the given value and wrap it into a 'StrictDynamic'. -- -- @since 0.22.0@@ -382,6 +451,7 @@                                 --   scheduled yet.   | ProcessIdle                 -- ^ The process yielded it's time slice   | ProcessBusy                 -- ^ The process is busy with non-blocking+  | ProcessBusyUpdatingDetails  -- ^ The process is busy with 'UpdateProcessDetails'   | ProcessBusySending          -- ^ The process is busy with sending a message   | ProcessBusySendingShutdown  -- ^ The process is busy with killing   | ProcessBusySendingInterrupt -- ^ The process is busy with killing@@ -481,7 +551,7 @@     -- | An error causes the process to exit immediately.     -- For example an unexpected runtime exception was thrown, i.e. an exception     -- derived from 'Control.Exception.Safe.SomeException'-    -- Or a 'Recoverable' Interrupt was not recoverd.+    -- Or a 'Recoverable' Interrupt was not recovered.     ExitUnhandledError       :: Text -> Interrupt 'NoRecovery     -- | A process shall exit immediately, without any cleanup was cancelled (e.g. killed, in 'Async.cancel')@@ -831,18 +901,20 @@ spawn   :: forall r q    . (HasCallStack, SetMember Process (Process q) r, Member Interrupts r)-  => Eff (InterruptableProcess q) ()+  => ProcessTitle+  -> Eff (InterruptableProcess q) ()   -> Eff r ProcessId-spawn child =-  executeAndResumeOrThrow (Spawn @q (provideInterruptsShutdown child))+spawn t child =+  executeAndResumeOrThrow (Spawn @q t (provideInterruptsShutdown child))  -- | Like 'spawn' but return @()@. spawn_   :: forall r q    . (HasCallStack, SetMember Process (Process q) r, Member Interrupts r)-  => Eff (InterruptableProcess q) ()+  => ProcessTitle+  -> Eff (InterruptableProcess q) ()   -> Eff r ()-spawn_ child = void (spawn child)+spawn_ t child = void (spawn t child)  -- | Start a new process, and immediately link to it. --@@ -850,10 +922,11 @@ spawnLink   :: forall r q    . (HasCallStack, SetMember Process (Process q) r, Member Interrupts r)-  => Eff (InterruptableProcess q) ()+  => ProcessTitle+  -> Eff (InterruptableProcess q) ()   -> Eff r ProcessId-spawnLink child =-  executeAndResumeOrThrow (SpawnLink @q (provideInterruptsShutdown child))+spawnLink t child =+  executeAndResumeOrThrow (SpawnLink @q t (provideInterruptsShutdown child))  -- | Start a new process, the new process will execute an effect, the function -- will return immediately with a 'ProcessId'. The spawned process has only the@@ -862,17 +935,19 @@ spawnRaw   :: forall r q    . (HasCallStack, SetMember Process (Process q) r, Member Interrupts r)-  => Eff (ConsProcess q) ()+  => ProcessTitle+  -> Eff (ConsProcess q) ()   -> Eff r ProcessId-spawnRaw child = executeAndResumeOrThrow (Spawn @q child)+spawnRaw t child = executeAndResumeOrThrow (Spawn @q t child)  -- | Like 'spawnRaw' but return @()@. spawnRaw_   :: forall r q    . (HasCallStack, SetMember Process (Process q) r, Member Interrupts r)-  => Eff (ConsProcess q) ()+  => ProcessTitle+  -> Eff (ConsProcess q) ()   -> Eff r ()-spawnRaw_ = void . spawnRaw+spawnRaw_ t = void . spawnRaw t  -- | Return 'True' if the process is alive. --@@ -883,6 +958,28 @@   => ProcessId   -> Eff r Bool isProcessAlive pid = isJust <$> executeAndResumeOrThrow (GetProcessState pid)+++-- | Return the 'ProcessTitle', 'ProcessDetails'  and 'ProcessState',+-- for the given process, if the process is alive.+--+-- @since 0.24.1+getProcessState+  :: forall r q+   . (HasCallStack, SetMember Process (Process q) r, Member Interrupts r)+  => ProcessId+  -> Eff r (Maybe (ProcessTitle, ProcessDetails, ProcessState))+getProcessState pid = executeAndResumeOrThrow (GetProcessState pid)++-- | Replace the 'ProcessDetails' of the process.+--+-- @since 0.24.1+updateProcessDetails+  :: forall r q+   . (HasCallStack, SetMember Process (Process q) r, Member Interrupts r)+  => ProcessDetails+  -> Eff r ()+updateProcessDetails pd = executeAndResumeOrThrow (UpdateProcessDetails pd)  -- | Block until a message was received. -- See 'ReceiveSelectedMessage' for more documentation.

@@ -87,7 +87,8 @@ -- a 'MessageQ'. data ProcessInfo = ProcessInfo   { _processId :: ProcessId-  , _processState :: TVar ProcessState+  , _processTitle :: ProcessTitle+  , _processState :: TVar (ProcessDetails, ProcessState)   , _messageQ :: TVar MessageQ   , _processLinks :: TVar (Set ProcessId)   }@@ -172,9 +173,9 @@   show p = "process info: " ++ show (p ^. processId)  -- | Create a new 'ProcessInfo'-newProcessInfo :: HasCallStack => ProcessId -> STM ProcessInfo-newProcessInfo a =-  ProcessInfo a <$> newTVar ProcessBooting <*> newTVar def <*> newTVar def+newProcessInfo :: HasCallStack => ProcessId -> ProcessTitle ->  STM ProcessInfo+newProcessInfo a t =+  ProcessInfo a t  <$> newTVar (mempty, ProcessBooting) <*> newTVar def <*> newTVar def  -- * Scheduler Implementation @@ -296,7 +297,8 @@ -- setMyProcessState st = do --  oldSt <- lift (atomically (readTVar myProcessStateVar <* setMyProcessStateSTM st)) --  logDebug ("state change: "<> show oldSt <> " -> " <> show st)-  setMyProcessStateSTM = writeTVar myProcessStateVar+  setMyProcessStateSTM = modifyTVar myProcessStateVar . set _2+  setMyProcessDetailsSTM = modifyTVar myProcessStateVar . set _1   myMessageQVar        = myProcessInfo ^. messageQ   kontinueWith     :: forall s v a@@ -368,8 +370,8 @@     SendInterrupt _ _ -> diskontinue shutdownRequest     SendShutdown toPid r ->       if toPid == myPid then diskontinue r else diskontinue shutdownRequest-    Spawn                  _ -> diskontinue shutdownRequest-    SpawnLink              _ -> diskontinue shutdownRequest+    Spawn                _ _ -> diskontinue shutdownRequest+    SpawnLink            _ _ -> diskontinue shutdownRequest     ReceiveSelectedMessage _ -> diskontinue shutdownRequest     FlushMessages            -> diskontinue shutdownRequest     SelfPid                  -> diskontinue shutdownRequest@@ -377,6 +379,7 @@     YieldProcess             -> diskontinue shutdownRequest     Shutdown        r        -> diskontinue r     GetProcessState _        -> diskontinue shutdownRequest+    UpdateProcessDetails _   -> diskontinue shutdownRequest     Monitor         _        -> diskontinue shutdownRequest     Demonitor       _        -> diskontinue shutdownRequest     Link            _        -> diskontinue shutdownRequest@@ -396,8 +399,8 @@       SendShutdown  toPid r -> if toPid == myPid         then diskontinue r         else kontinue (Interrupted interruptRequest)-      Spawn                  _ -> kontinue (Interrupted interruptRequest)-      SpawnLink              _ -> kontinue (Interrupted interruptRequest)+      Spawn                _ _ -> kontinue (Interrupted interruptRequest)+      SpawnLink            _ _ -> kontinue (Interrupted interruptRequest)       ReceiveSelectedMessage _ -> kontinue (Interrupted interruptRequest)       FlushMessages            -> kontinue (Interrupted interruptRequest)       SelfPid                  -> kontinue (Interrupted interruptRequest)@@ -405,6 +408,7 @@       YieldProcess             -> kontinue (Interrupted interruptRequest)       Shutdown        r        -> diskontinue r       GetProcessState _        -> kontinue (Interrupted interruptRequest)+      UpdateProcessDetails _   -> kontinue (Interrupted interruptRequest)       Monitor         _        -> kontinue (Interrupted interruptRequest)       Demonitor       _        -> kontinue (Interrupted interruptRequest)       Link            _        -> kontinue (Interrupted interruptRequest)@@ -432,10 +436,10 @@         interpretSendShutdownOrInterrupt toPid (SomeExitReason msg)         >>= kontinue nextRef         .   ResumeWith-    Spawn child ->-      spawnNewProcess Nothing child >>= kontinue nextRef . ResumeWith . fst-    SpawnLink child ->-      spawnNewProcess (Just myProcessInfo) child+    Spawn title child ->+      spawnNewProcess Nothing title child >>= kontinue nextRef . ResumeWith . fst+    SpawnLink title child ->+      spawnNewProcess (Just myProcessInfo) title child         >>= kontinue nextRef         .   ResumeWith         .   fst@@ -449,6 +453,8 @@     Shutdown r    -> diskontinue r     GetProcessState toPid ->       interpretGetProcessState toPid >>= kontinue nextRef . ResumeWith+    UpdateProcessDetails d ->+      interpretUpdateDetails d >>= kontinue nextRef . ResumeWith     Monitor target ->       interpretMonitor target >>= kontinue nextRef . ResumeWith     Demonitor ref -> interpretDemonitor ref >>= kontinue nextRef . ResumeWith@@ -484,8 +490,13 @@       let procInfoVar = schedulerState ^. processTable       lift $ atomically $ do         procInfoTable <- readTVar procInfoVar-        traverse (\toProcInfo -> readTVar (toProcInfo ^. processState))+        traverse (\toProcInfo -> do+                      (pDetails, pState) <- readTVar (toProcInfo ^. processState)+                      return (toProcInfo ^. processTitle, pDetails, pState))                  (procInfoTable ^. at toPid)+    interpretUpdateDetails !td = do+      setMyProcessState ProcessBusyUpdatingDetails+      lift $ atomically $ setMyProcessDetailsSTM td     interpretLink !toPid = do       setMyProcessState ProcessBusyLinking       schedulerState <- getSchedulerState@@ -557,7 +568,7 @@   liftBaseWith       (\runS -> Async.withAsync         (runS $ withNewSchedulerState $ do-          (_, mainProcAsync) <- spawnNewProcess Nothing $ do+          (_, mainProcAsync) <- spawnNewProcess Nothing "init" $ do             logNotice "++++++++ main process started ++++++++"             provideInterruptsShutdown procEff             logNotice "++++++++ main process returned ++++++++"@@ -573,9 +584,10 @@ spawnNewProcess   :: (HasCallStack)   => Maybe ProcessInfo+  -> ProcessTitle   -> Eff ProcEff ()   -> Eff SchedulerIO (ProcessId, Async (Interrupt 'NoRecovery))-spawnNewProcess mLinkedParent mfa = do+spawnNewProcess mLinkedParent title mfa = do   schedulerState <- getSchedulerState   procInfo       <- allocateProcInfo schedulerState   traverse_ (linkToParent procInfo) mLinkedParent@@ -589,7 +601,7 @@             processInfoVar = schedulerState ^. processTable         pid <- readTVar nextPidVar         modifyTVar' nextPidVar (+ 1)-        procInfo <- newProcessInfo pid+        procInfo <- newProcessInfo pid title         modifyTVar' processInfoVar (at pid ?~ procInfo)         return procInfo       )

@@ -37,13 +37,34 @@ import Data.Dynamic (dynTypeRep)  -- --------------------------------------------------------------------------------  STS+--  STS and ProcessInfo -- ----------------------------------------------------------------------------- +data ProcessInfo =+  MkProcessInfo+  { _processInfoTitle :: !ProcessTitle+  , _processInfoDetails :: !ProcessDetails+  , _processInfoMessageQ :: !(Seq StrictDynamic)+  }++instance Show ProcessInfo where+  showsPrec d (MkProcessInfo pTitle pDetails pQ) =+    showParen+      (d >= 10)+      (appEndo+         (Endo (showChar ' ' . shows pTitle . showString ": ") <>+          foldMap (Endo . shows . dynTypeRep . unwrapStrictDynamic) (toList pQ) <>+          Endo (shows pDetails)))++makeLenses ''ProcessInfo++newProcessInfo :: ProcessTitle -> ProcessInfo+newProcessInfo t = MkProcessInfo t "" Seq.empty+ data STS r m = STS   { _nextPid :: !ProcessId   , _nextRef :: !Int-  , _msgQs :: !(Map.Map ProcessId (Seq StrictDynamic))+  , _msgQs :: !(Map.Map ProcessId ProcessInfo)   , _monitors :: !(Set.Set (MonitorReference, ProcessId))   , _processLinks :: !(Set.Set (ProcessId, ProcessId))   , _runEff :: forall a . Eff r a -> m a@@ -51,7 +72,7 @@   }  initStsMainProcess :: (forall a . Eff r a -> m a) -> m () -> STS r m-initStsMainProcess = STS 1 0 (Map.singleton 0 Seq.empty) Set.empty Set.empty+initStsMainProcess = STS 1 0 (Map.singleton 0 (newProcessInfo "init" )) Set.empty Set.empty  makeLenses ''STS @@ -64,9 +85,8 @@     . showString " msgQs: "     . appEndo         (foldMap-          (\(pid, msgs) ->-            Endo (showString "  " . shows pid . showString ": ")-              <> foldMap (Endo . shows . dynTypeRep . unwrapStrictDynamic) (toList msgs)+          (\(pid, p) ->+            Endo (showChar ' ' . shows pid . showChar ' ' . shows p)           )           (sts ^.. msgQs . itraversed . withIndex)         )@@ -75,44 +95,48 @@ dropMsgQ :: ProcessId -> STS r m -> STS r m dropMsgQ pid = msgQs . at pid .~ Nothing -getProcessState :: ProcessId -> STS r m -> Maybe ProcessState-getProcessState pid sts = toPS <$> sts ^. msgQs . at pid- where-  toPS Empty     = ProcessIdle-  toPS (_ :<| _) = ProcessBusy -- TODO get more detailed state+getProcessStateFromScheduler :: ProcessId -> STS r m -> Maybe (ProcessTitle, ProcessDetails, ProcessState)+getProcessStateFromScheduler pid sts = toPS <$> sts ^. msgQs . at pid+  where+    toPS p =+      ( p ^. processInfoTitle+      , p ^. processInfoDetails+      , case (p ^. processInfoMessageQ) -- TODO get more detailed state+              of+          _ :<| _ -> ProcessBusy+          _ -> ProcessIdle)  incRef :: STS r m -> (Int, STS r m) incRef sts = (sts ^. nextRef, sts & nextRef %~ (+ 1))  enqueueMsg :: ProcessId -> StrictDynamic -> STS r m -> STS r m-enqueueMsg toPid msg = msgQs . ix toPid %~ (:|> msg)+enqueueMsg toPid msg = msgQs . ix toPid . processInfoMessageQ %~ (:|> msg) -newProcessQ :: Maybe ProcessId -> STS r m -> (ProcessId, STS r m)-newProcessQ parentLink sts =+newProcessQ :: Maybe ProcessId -> ProcessTitle -> STS r m -> (ProcessId, STS r m)+newProcessQ parentLink title sts =   ( sts ^. nextPid-  , let stsQ =-          sts & nextPid %~ (+ 1) & msgQs . at (sts ^. nextPid) ?~ Seq.empty-    in  case parentLink of+  , let stsQ = sts & nextPid %~ (+ 1) & msgQs . at (sts ^. nextPid) ?~ newProcessInfo title+     in case parentLink of           Nothing -> stsQ           Just pid ->-            let (Nothing, stsQL) = addLink pid (sts ^. nextPid) stsQ in stsQL-  )+            let (Nothing, stsQL) = addLink pid (sts ^. nextPid) stsQ+             in stsQL)  flushMsgs :: ProcessId -> STS m r -> ([StrictDynamic], STS m r) flushMsgs pid = State.runState $ do-  msgs <- msgQs . ix pid <<.= Empty+  msgs <- msgQs . ix pid . processInfoMessageQ <<.= Empty   return (toList msgs)  receiveMsg   :: ProcessId -> MessageSelector a -> STS m r -> Maybe (Maybe (a, STS m r)) receiveMsg pid messageSelector sts =-  case sts ^. msgQs . at pid of+  case sts ^? msgQs . at pid . _Just . processInfoMessageQ of     Nothing -> Nothing     Just msgQ ->       Just $       case partitionMessages msgQ Empty of         Nothing -> Nothing-        Just (result, otherMessages) -> Just (result, sts & msgQs . ix pid .~ otherMessages)+        Just (result, otherMessages) -> Just (result, sts & msgQs . ix pid . processInfoMessageQ .~ otherMessages)   where     partitionMessages Empty _acc = Nothing     partitionMessages (m :<| msgRest) acc =@@ -265,6 +289,7 @@   OnSelf :: (ResumeProcess ProcessId -> Eff r (OnYield r a))          -> OnYield r a   OnSpawn :: Bool+          -> ProcessTitle           -> Eff (Process r ': r) ()           -> (ResumeProcess ProcessId -> Eff r (OnYield r a))           -> OnYield r a@@ -280,12 +305,16 @@          -> OnYield r a   OnGetProcessState          :: ProcessId-         -> (ResumeProcess (Maybe ProcessState) -> Eff r (OnYield r a))+         -> (ResumeProcess (Maybe (ProcessTitle, ProcessDetails, ProcessState)) -> Eff r (OnYield r a))          -> OnYield r a+  OnUpdateProcessDetails+         :: ProcessDetails+         -> (ResumeProcess () -> Eff r (OnYield r a))+         -> OnYield r a   OnSendShutdown :: !ProcessId -> Interrupt 'NoRecovery-                    -> (ResumeProcess () -> Eff r (OnYield r a)) -> OnYield r a+                 -> (ResumeProcess () -> Eff r (OnYield r a)) -> OnYield r a   OnSendInterrupt :: !ProcessId -> Interrupt 'Recoverable-                    -> (ResumeProcess () -> Eff r (OnYield r a)) -> OnYield r a+                  -> (ResumeProcess () -> Eff r (OnYield r a)) -> OnYield r a   OnMakeReference :: (ResumeProcess Int -> Eff r (OnYield r a)) -> OnYield r a   OnMonitor     :: ProcessId@@ -306,24 +335,25 @@  instance Show (OnYield r a) where   show = \case-    OnFlushMessages _     -> "OnFlushMessages"-    OnYield         _     -> "OnYield"-    OnSelf          _     -> "OnSelf"-    OnSpawn False _ _     -> "OnSpawn"-    OnSpawn True  _ _     -> "OnSpawn (link)"-    OnDone     _          -> "OnDone"-    OnShutdown e          -> "OnShutdown " ++ show e-    OnInterrupt e _       -> "OnInterrupt " ++ show e-    OnSend toP _ _        -> "OnSend " ++ show toP-    OnRecv            _ _ -> "OnRecv"-    OnGetProcessState p _ -> "OnGetProcessState " ++ show p-    OnSendShutdown  p e _ -> "OnSendShutdow " ++ show p ++ " " ++ show e-    OnSendInterrupt p e _ -> "OnSendInterrupt " ++ show p ++ " " ++ show e-    OnMakeReference _     -> "OnMakeReference"-    OnMonitor   p _       -> "OnMonitor " ++ show p-    OnDemonitor p _       -> "OnDemonitor " ++ show p-    OnLink      p _       -> "OnLink " ++ show p-    OnUnlink    p _       -> "OnUnlink " ++ show p+    OnFlushMessages _          -> "OnFlushMessages"+    OnYield         _          -> "OnYield"+    OnSelf          _          -> "OnSelf"+    OnSpawn False t _ _        -> "OnSpawn" ++ show t+    OnSpawn True  t _ _        -> "OnSpawn (link)" ++ show t+    OnDone     _               -> "OnDone"+    OnShutdown e               -> "OnShutdown " ++ show e+    OnInterrupt e _            -> "OnInterrupt " ++ show e+    OnSend toP _ _             -> "OnSend " ++ show toP+    OnRecv            _ _      -> "OnRecv"+    OnGetProcessState p _      -> "OnGetProcessState " ++ show p+    OnUpdateProcessDetails p _ -> "OnUpdateProcessDetails " ++ show p+    OnSendShutdown  p e _      -> "OnSendShutdow " ++ show p ++ " " ++ show e+    OnSendInterrupt p e _      -> "OnSendInterrupt " ++ show p ++ " " ++ show e+    OnMakeReference _          -> "OnMakeReference"+    OnMonitor   p _            -> "OnMonitor " ++ show p+    OnDemonitor p _            -> "OnDemonitor " ++ show p+    OnLink      p _            -> "OnLink " ++ show p+    OnUnlink    p _            -> "OnUnlink " ++ show p  runAsCoroutinePure   :: forall v r m@@ -340,12 +370,13 @@   cont k q FlushMessages                = return (OnFlushMessages (k . qApp q))   cont k q YieldProcess                 = return (OnYield (k . qApp q))   cont k q SelfPid                      = return (OnSelf (k . qApp q))-  cont k q (Spawn     e               ) = return (OnSpawn False e (k . qApp q))-  cont k q (SpawnLink e               ) = return (OnSpawn True e (k . qApp q))+  cont k q (Spawn     t e             ) = return (OnSpawn False t e (k . qApp q))+  cont k q (SpawnLink t e             ) = return (OnSpawn True t e (k . qApp q))   cont _ _ (Shutdown  !sr             ) = return (OnShutdown sr)   cont k q (SendMessage !tp !msg      ) = return (OnSend tp msg (k . qApp q))   cont k q (ReceiveSelectedMessage f  ) = return (OnRecv f (k . qApp q))   cont k q (GetProcessState        !tp) = return (OnGetProcessState tp (k . qApp q))+  cont k q (UpdateProcessDetails   !td) = return (OnUpdateProcessDetails td (k . qApp q))   cont k q (SendInterrupt !tp  !er    ) = return (OnSendInterrupt tp er (k . qApp q))   cont k q (SendShutdown  !pid !sr    ) = return (OnSendShutdown pid sr (k . qApp q))   cont k q MakeReference                = return (OnMakeReference (k . qApp q))@@ -412,9 +443,10 @@                         OnSelf _tk -> return (OnShutdown sr)                         OnSend _ _ _tk -> return (OnShutdown sr)                         OnRecv _ _tk -> return (OnShutdown sr)-                        OnSpawn _ _ _tk -> return (OnShutdown sr)+                        OnSpawn _ _ _ _tk -> return (OnShutdown sr)                         OnDone x -> return (OnDone x)                         OnGetProcessState _ _tk -> return (OnShutdown sr)+                        OnUpdateProcessDetails _ _tk -> return (OnShutdown sr)                         OnShutdown sr' -> return (OnShutdown sr')                         OnInterrupt _er _tk -> return (OnShutdown sr)                         OnMakeReference _tk -> return (OnShutdown sr)@@ -441,14 +473,19 @@           nextK <- kontinue sts k ()           handleProcess (enqueueMsg toPid msg sts) (rest :|> (nextK, pid))         OnGetProcessState toPid k -> do-          nextK <- kontinue sts k (getProcessState toPid sts)+          nextK <- kontinue sts k (getProcessStateFromScheduler toPid sts)           handleProcess sts (rest :|> (nextK, pid))-        OnSpawn link f k -> do+        OnUpdateProcessDetails pd k -> do+          let newSts = sts & msgQs . ix pid . processInfoDetails .~ pd+          nextK <- kontinue newSts k ()+          handleProcess newSts (rest :|> (nextK, pid))+        OnSpawn link title f k -> do           let (newPid, newSts) =                 newProcessQ                   (if link                      then Just pid                      else Nothing)+                  title                   sts           fk <- runAsCoroutinePure (newSts ^. runEff) (f >> exitNormally)           nextK <- kontinue newSts k newPid@@ -514,9 +551,10 @@                 OnSelf tk -> tk (Interrupted sr)                 OnSend _ _ tk -> tk (Interrupted sr)                 OnRecv _ tk -> tk (Interrupted sr)-                OnSpawn _ _ tk -> tk (Interrupted sr)+                OnSpawn _ _ _ tk -> tk (Interrupted sr)                 OnDone x -> return (OnDone x)                 OnGetProcessState _ tk -> tk (Interrupted sr)+                OnUpdateProcessDetails _ tk -> tk (Interrupted sr)                 OnShutdown sr' -> return (OnShutdown sr')                 OnInterrupt er tk -> tk (Interrupted er)                 OnMakeReference tk -> tk (Interrupted sr)

@@ -25,6 +25,7 @@ import           Control.DeepSeq import           Control.Monad.IO.Class import           Data.Typeable+import           Data.Text as T import           Control.Applicative import           GHC.Stack @@ -158,14 +159,15 @@   -> Timeout   -> (TimerReference -> message)   -> Eff r TimerReference-sendAfter pid (TimeoutMicros 0) mkMsg = TimerReference <$> spawn-  (yieldProcess >> self >>= (sendMessage pid . force . mkMsg . TimerReference))-sendAfter pid (TimeoutMicros t) mkMsg = TimerReference <$> spawn-  (   liftIO (threadDelay t)-  >>  self-  >>= (sendMessage pid . force . mkMsg . TimerReference)-  )-+sendAfter pid (TimeoutMicros us) mkMsg =+  TimerReference <$>+  spawn+    (MkProcessTitle ("after_" <> T.pack (show us) <> "us_to_" <> T.pack (show pid)))+    ((if us == 0+        then yieldProcess+        else liftIO (threadDelay us)) >>+     self >>=+     (sendMessage pid . force . mkMsg . TimerReference)) -- | Start a new timer, after the time has elapsed, 'TimerElapsed' is sent to -- calling process. The message also contains the 'TimerReference' returned by -- this function. Use 'cancelTimer' to cancel the timer. Use

@@ -236,3 +236,88 @@   fromPdu (ToPdu3 l) = Just l   fromPdu _ = Nothing +data instance Pdu (a1, a2, a3, a4) x where+  ToPdu1Of4 :: Pdu a1 r -> Pdu (a1, a2, a3, a4) r+  ToPdu2Of4 :: Pdu a2 r -> Pdu (a1, a2, a3, a4) r+  ToPdu3Of4 :: Pdu a3 r -> Pdu (a1, a2, a3, a4) r+  ToPdu4Of4 :: Pdu a4 r -> Pdu (a1, a2, a3, a4) r+  deriving Typeable++instance (NFData (Pdu a1 r), NFData (Pdu a2 r), NFData (Pdu a3 r), NFData (Pdu a4 r)) => NFData (Pdu (a1, a2, a3, a4) r) where+  rnf (ToPdu1Of4 x) = rnf x+  rnf (ToPdu2Of4 y) = rnf y+  rnf (ToPdu3Of4 z) = rnf z+  rnf (ToPdu4Of4 w) = rnf w++instance (Show (Pdu a1 r), Show (Pdu a2 r), Show (Pdu a3 r), Show (Pdu a4 r)) => Show (Pdu (a1, a2, a3, a4) r) where+  showsPrec d (ToPdu1Of4 x) = showsPrec d x+  showsPrec d (ToPdu2Of4 y) = showsPrec d y+  showsPrec d (ToPdu3Of4 z) = showsPrec d z+  showsPrec d (ToPdu4Of4 w) = showsPrec d w++instance EmbedProtocol (a1, a2, a3, a4) a1 where+  embedPdu = ToPdu1Of4+  fromPdu (ToPdu1Of4 l) = Just l+  fromPdu _ = Nothing++instance EmbedProtocol (a1, a2, a3, a4) a2 where+  embedPdu = ToPdu2Of4+  fromPdu (ToPdu2Of4 l) = Just l+  fromPdu _ = Nothing++instance EmbedProtocol (a1, a2, a3, a4) a3 where+  embedPdu = ToPdu3Of4+  fromPdu (ToPdu3Of4 l) = Just l+  fromPdu _ = Nothing++instance EmbedProtocol (a1, a2, a3, a4) a4 where+  embedPdu = ToPdu4Of4+  fromPdu (ToPdu4Of4 l) = Just l+  fromPdu _ = Nothing++data instance Pdu (a1, a2, a3, a4, a5) x where+  ToPdu1Of5 :: Pdu a1 r -> Pdu (a1, a2, a3, a4, a5) r+  ToPdu2Of5 :: Pdu a2 r -> Pdu (a1, a2, a3, a4, a5) r+  ToPdu3Of5 :: Pdu a3 r -> Pdu (a1, a2, a3, a4, a5) r+  ToPdu4Of5 :: Pdu a4 r -> Pdu (a1, a2, a3, a4, a5) r+  ToPdu5Of5 :: Pdu a5 r -> Pdu (a1, a2, a3, a4, a5) r+  deriving Typeable++instance (NFData (Pdu a1 r), NFData (Pdu a2 r), NFData (Pdu a3 r), NFData (Pdu a4 r), NFData (Pdu a5 r)) => NFData (Pdu (a1, a2, a3, a4, a5) r) where+  rnf (ToPdu1Of5 x) = rnf x+  rnf (ToPdu2Of5 y) = rnf y+  rnf (ToPdu3Of5 z) = rnf z+  rnf (ToPdu4Of5 w) = rnf w+  rnf (ToPdu5Of5 w) = rnf w++instance (Show (Pdu a1 r), Show (Pdu a2 r), Show (Pdu a3 r), Show (Pdu a4 r), Show (Pdu a5 r)) => Show (Pdu (a1, a2, a3, a4, a5) r) where+  showsPrec d (ToPdu1Of5 x) = showsPrec d x+  showsPrec d (ToPdu2Of5 y) = showsPrec d y+  showsPrec d (ToPdu3Of5 z) = showsPrec d z+  showsPrec d (ToPdu4Of5 w) = showsPrec d w+  showsPrec d (ToPdu5Of5 v) = showsPrec d v++instance EmbedProtocol (a1, a2, a3, a4, a5) a1 where+  embedPdu = ToPdu1Of5+  fromPdu (ToPdu1Of5 l) = Just l+  fromPdu _ = Nothing++instance EmbedProtocol (a1, a2, a3, a4, a5) a2 where+  embedPdu = ToPdu2Of5+  fromPdu (ToPdu2Of5 l) = Just l+  fromPdu _ = Nothing++instance EmbedProtocol (a1, a2, a3, a4, a5) a3 where+  embedPdu = ToPdu3Of5+  fromPdu (ToPdu3Of5 l) = Just l+  fromPdu _ = Nothing++instance EmbedProtocol (a1, a2, a3, a4, a5) a4 where+  embedPdu = ToPdu4Of5+  fromPdu (ToPdu4Of5 l) = Just l+  fromPdu _ = Nothing++instance EmbedProtocol (a1, a2, a3, a4, a5) a5 where+  embedPdu = ToPdu5Of5+  fromPdu (ToPdu5Of5 l) = Just l+  fromPdu _ = Nothing

@@ -0,0 +1,300 @@+-- | A better, more safe implementation of the Erlang/OTP gen_server behaviour.+--+-- @since 0.24.0+module Control.Eff.Concurrent.Protocol.EffectfulServer+  ( Server(..)+  , Event(..)+  , start+  , startLink+  , protocolServerLoop+  -- * GenServer+  , TangibleGenServer+  , GenServer+  , GenServerId(..)+  , genServer+  -- * Re-exports+  , RequestOrigin(..)+  , Reply(..)+  , sendReply+  )+  where++import Control.Applicative+import Control.DeepSeq+import Control.Eff+import Control.Eff.Extend ()+import Control.Eff.Concurrent.Process+import Control.Eff.Concurrent.Process.Timer+import Control.Eff.Concurrent.Protocol+import Control.Eff.Concurrent.Protocol.Request+import Control.Eff.Log+import Control.Lens+import Data.Kind+import Data.String+import Data.Typeable+import Data.Type.Pretty+import qualified Data.Text as T+import GHC.Stack (HasCallStack)++-- | A type class for building supervised processes, that handle 'Event's+-- with 'Request's for 'Pdu' instance.+--+-- @since 0.24.1+class Server (a :: Type) (e :: [Type -> Type])+  where+  -- | The value that defines what is required to initiate a 'Server'+  -- loop.+  data Init a e++  -- | The index type of the 'Event's that this server processes.+  -- This is the first parameter to the 'Request' and therefore of+  -- the 'Pdu' family.+  type ServerPdu a :: Type+  type ServerPdu a = a++  -- | Effects of the implementation+  --+  -- @since 0.24.1+  type Effects a e :: [Type -> Type]+  type Effects a e = e++  -- | Return the 'ProcessTitle'.+  --+  -- Usually you should rely on the default implementation+  serverTitle :: Init a e -> ProcessTitle++  default serverTitle :: Typeable (ServerPdu a) => Init a e -> ProcessTitle+  serverTitle _ = fromString $ prettyTypeableShows (typeRep (Proxy @(ServerPdu a))) "-server"++  -- | Process the effects of the implementation+  runEffects :: Init a e -> Eff (Effects a e) x -> Eff e x++  default runEffects :: Effects a e ~ e => Init a e -> Eff (Effects a e) x -> Eff e x+  runEffects = const id++  -- | Update the 'Model' based on the 'Event'.+  onEvent :: Init a e -> Event (ServerPdu a) -> Eff (Effects a e) ()++  default onEvent :: (Show (Init a e),  Member Logs (Effects a e)) => Init a e -> Event (ServerPdu a) -> Eff (Effects a e) ()+  onEvent i e = logInfo ("unhandled: " <> T.pack (show i) <> " " <> T.pack (show e))+++-- | Execute the server loop.+--+-- @since 0.24.0+start+  :: forall a q h+  . ( Server a (InterruptableProcess q)+    , Typeable a+    , Typeable (ServerPdu a)+    , LogsTo h (InterruptableProcess q)+    , SetMember Process (Process q) (Effects a (InterruptableProcess q))+    , Member Interrupts             (Effects a (InterruptableProcess q))+    , HasCallStack)+  => Init a (InterruptableProcess q)+  -> Eff (InterruptableProcess q) (Endpoint (ServerPdu a))+start a = asEndpoint <$> spawn (serverTitle a) (protocolServerLoop a)++-- | Execute the server loop.+--+-- @since 0.24.0+startLink+  :: forall a q h+  . ( Typeable a+    , Typeable (ServerPdu a)+    , Server a (InterruptableProcess q)+    , LogsTo h (InterruptableProcess q)+    , SetMember Process (Process q) (Effects a (InterruptableProcess q))+    , Member Interrupts (Effects a (InterruptableProcess q))+    , HasCallStack)+  => Init a (InterruptableProcess q)+  -> Eff (InterruptableProcess q) (Endpoint (ServerPdu a))+startLink a = asEndpoint <$> spawnLink (serverTitle a) (protocolServerLoop a)++-- | Execute the server loop.+--+-- @since 0.24.0+protocolServerLoop+     :: forall q h a+     . ( Server a (InterruptableProcess q)+       , LogsTo h (InterruptableProcess q)+       , SetMember Process (Process q) (Effects a (InterruptableProcess q))+       , Member Interrupts (Effects a (InterruptableProcess q))+       , Typeable a+       , Typeable (ServerPdu a)+       )+  => Init a (InterruptableProcess q) -> Eff (InterruptableProcess q) ()+protocolServerLoop a = do+  myEp <- T.pack . show . asEndpoint @(ServerPdu a) <$> self+  censorLogs (lmAddEp myEp) $ do+    logDebug ("starting")+    runEffects a (receiveSelectedLoop sel mainLoop)+    return ()+  where+    lmAddEp myEp = lmProcessId ?~ myEp+    sel :: MessageSelector (Event (ServerPdu a))+    sel = onRequest <$> selectMessage @(Request (ServerPdu a))+      <|> OnDown    <$> selectMessage @ProcessDown+      <|> OnTimeOut <$> selectMessage @TimerElapsed+      <|> OnMessage <$> selectAnyMessage+      where+        onRequest :: Request (ServerPdu a) -> Event (ServerPdu a)+        onRequest (Call o m) = OnCall (MkSerializer toStrictDynamic) o m+        onRequest (Cast m) = OnCast m+    handleInt i = onEvent a (OnInterrupt i) *> pure Nothing+    mainLoop :: (Typeable a)+      => Either (Interrupt 'Recoverable) (Event (ServerPdu a))+      -> Eff (Effects a (InterruptableProcess q)) (Maybe ())+    mainLoop (Left i) = handleInt i+    mainLoop (Right i) = onEvent a i *> pure Nothing++-- | Internal protocol to communicate incoming messages and other events to the+-- instances of 'Server'.+--+-- Note that this is required to receive any kind of messages in 'protocolServerLoop'.+--+-- @since 0.24.0+data Event a where+  -- | A 'Synchronous' message was received. If an implementation wants to delegate nested 'Pdu's, it can+  -- 'contramap' the reply 'Serializer' such that the 'Reply' received by the caller has the correct type.+  --+  -- @since 0.24.1+  OnCall :: forall a r. (Tangible r, TangiblePdu a ('Synchronous r)) => Serializer (Reply a r) -> RequestOrigin a r -> Pdu a ('Synchronous r) -> Event a+  OnCast :: forall a. TangiblePdu a 'Asynchronous => Pdu a 'Asynchronous -> Event a+  OnInterrupt  :: (Interrupt 'Recoverable) -> Event a+  OnDown  :: ProcessDown -> Event a+  OnTimeOut  :: TimerElapsed -> Event a+  OnMessage  :: StrictDynamic -> Event a+  deriving Typeable++instance Show (Event a) where+  showsPrec d e =+    showParen (d>=10) $+      showString "event: "+      . case e of+          OnCall _ o p -> shows (Call o p)+          OnCast p -> shows (Cast p)+          OnInterrupt r -> shows r+          OnDown r -> shows r+          OnTimeOut r -> shows r+          OnMessage r -> shows r++instance NFData a => NFData (Event a) where+   rnf = \case+       OnCall _ o p -> rnf o `seq` rnf p+       OnCast p -> rnf p+       OnInterrupt r -> rnf r+       OnDown r  -> rnf r+       OnTimeOut r -> rnf r+       OnMessage r -> r `seq` ()++type instance ToPretty (Event a) = ToPretty a <+> PutStr "event"++-- * GenServer++-- | Make a 'Server' from a /data record/ instead of type-class instance.+--+-- Sometimes it is much more concise to create an inline server-loop. In those cases+-- it might not be practical to go through all this type class boilerplate.+--+-- In these cases specifying a server by from a set of callback functions seems+-- much more appropriate.+--+-- This is such a helper. The @GenServer@ is a record with to callbacks,+-- and a 'Server' instance that simply invokes the given callbacks.+--+-- 'Server's that are directly based on 'LogIo' and 'InterruptableProcess' effects.+--+-- The name prefix @Gen@ indicates the inspiration from Erlang's @gen_server@ module.+--+-- @since 0.24.1+data GenServer tag eLoop e where+  MkGenServer+    :: (TangibleGenServer tag eLoop e, HasCallStack) =>+    { genServerRunEffects :: forall x . (Eff eLoop x -> Eff (InterruptableProcess e) x)+    , genServerOnEvent :: Event tag -> Eff eLoop ()+    } -> GenServer tag eLoop e+  deriving Typeable++-- | The constraints for a /tangible/ 'GenServer' instance.+--+-- @since 0.24.1+type TangibleGenServer tag eLoop e =+       ( LogIo e+       , SetMember Process (Process e) eLoop+       , Member Interrupts eLoop+       , Typeable e+       , Typeable eLoop+       , Typeable tag+       )++-- | The name/id of a 'GenServer' for logging purposes.+--+-- @since 0.24.0+newtype GenServerId tag =+  MkGenServerId { _fromGenServerId :: T.Text }+  deriving (Typeable, NFData, Ord, Eq, IsString)++instance (Typeable k, Typeable (tag :: k)) => Show (GenServerId tag) where+  showsPrec d px@(MkGenServerId x) =+    showParen+      (d >= 10)+      (showString (T.unpack x)+      . showString " :: "+      . prettyTypeableShows (typeOf px)+      )++instance (TangibleGenServer tag eLoop e) => Server (GenServer (tag :: Type) eLoop e) (InterruptableProcess e) where+  type ServerPdu (GenServer tag eLoop e) = tag+  type Effects (GenServer tag eLoop e) (InterruptableProcess e) = eLoop+  data instance Init (GenServer tag eLoop e) (InterruptableProcess e) =+        GenServerInit+         { genServerCallbacks :: GenServer tag eLoop e+         , genServerId :: GenServerId tag+         } deriving Typeable+  runEffects (GenServerInit cb cId) m =+    censorLogs+      (lmMessage <>~ (" | " <> _fromGenServerId cId))+      (genServerRunEffects cb m)+  onEvent (GenServerInit cb _cId) req = genServerOnEvent cb req++instance NFData (Init (GenServer tag eLoop e) (InterruptableProcess e)) where+  rnf (GenServerInit _ x) = rnf x++instance Typeable tag => Show (Init (GenServer tag eLoop e) (InterruptableProcess e)) where+  showsPrec d (GenServerInit _ x) =+    showParen (d>=10)+      ( showsPrec 11 x+      . showChar ' ' . prettyTypeableShows (typeRep (Proxy @tag))+      . showString " gen-server"+      )++-- | Create a 'GenServer'.+--+-- This requires the callback for 'Event's, a initial 'Model' and a 'GenServerId'.+--+-- There must be a 'GenServerProtocol' instance.+--+-- This is Haskell, so if this functions is partially applied+-- to some 'Event' callback, you get a function back,+-- that generates 'Init's from 'GenServerId's, like a /factory/+--+-- @since 0.24.0+genServer+  :: forall tag eLoop e .+     ( HasCallStack+     , TangibleGenServer tag eLoop e+     , Server (GenServer tag eLoop e) (InterruptableProcess e)+     )+  => (forall x . GenServerId tag -> Eff eLoop x -> Eff (InterruptableProcess e) x)+  -> (GenServerId tag -> Event tag -> Eff eLoop ())+  -> GenServerId tag+  -> Init (GenServer tag eLoop e) (InterruptableProcess e)+genServer initCb stepCb i =+  GenServerInit+    { genServerId = i+    , genServerCallbacks =+        MkGenServer { genServerRunEffects = initCb i+                    , genServerOnEvent = stepCb i+                    }+    }

@@ -1,3 +1,4 @@+{-# LANGUAGE UndecidableInstances #-} -- | A small process to capture and _share_ observation's by enqueueing them into an STM 'TBQueue'. module Control.Eff.Concurrent.Protocol.Observer.Queue   ( ObservationQueue(..)@@ -14,8 +15,7 @@ import           Control.Eff import           Control.Eff.Concurrent.Protocol import           Control.Eff.Concurrent.Protocol.Observer-import           Control.Eff.Concurrent.Protocol.Request-import           Control.Eff.Concurrent.Protocol.Server+import           Control.Eff.Concurrent.Protocol.StatefulServer import           Control.Eff.Concurrent.Process import           Control.Eff.ExceptionExtra     ( ) import           Control.Eff.Log@@ -148,14 +148,14 @@   cbo <- startLink (MkObservationQueue q)   pure (toObserver cbo) -instance (TangibleObserver o, TangiblePdu (Observer o) 'Asynchronous, Lifted IO q, Member Logs q) => Server (ObservationQueue o) (InterruptableProcess q) where+instance (TangibleObserver o, TangiblePdu (Observer o) 'Asynchronous, Lifted IO q, Member Logs q) => Server (ObservationQueue o) q where   type Protocol (ObservationQueue o) = Observer o -  data instance StartArgument (ObservationQueue o) (InterruptableProcess q) =+  data instance StartArgument (ObservationQueue o) q =      MkObservationQueue (ObservationQueue o)    update (MkObservationQueue (ObservationQueue q)) =     \case-      OnRequest (Cast r) ->+      OnCast r ->         handleObservations (lift . atomically . writeTBQueue q) r       otherMsg -> logError ("unexpected: " <> T.pack (show otherMsg))

@@ -31,7 +31,7 @@     -> Pdu protocol ('Synchronous reply)     -> Request protocol   Cast-    :: forall protocol. (TangiblePdu protocol 'Asynchronous)+    :: forall protocol. (TangiblePdu protocol 'Asynchronous, NFData (Pdu protocol 'Asynchronous))     => Pdu protocol 'Asynchronous     -> Request protocol   deriving (Typeable)@@ -46,11 +46,15 @@   rnf (Call o req) = rnf o `seq` rnf req   rnf (Cast req) = rnf req + -- | The wrapper around replies to 'Call's. -- -- @since 0.15.0 data Reply protocol reply where-  Reply :: (Tangible reply) => RequestOrigin protocol reply -> reply -> Reply protocol reply+  Reply :: (Tangible reply) =>+    { _replyTo :: RequestOrigin protocol reply+    , _replyValue :: reply+    } -> Reply protocol reply   deriving (Typeable)  instance NFData (Reply p r) where@@ -64,22 +68,28 @@ -- -- @since 0.15.0 data RequestOrigin (proto :: Type) reply = RequestOrigin-  { _requestOriginPid :: !ProcessId+  { _requestOriginPid     :: !ProcessId   , _requestOriginCallRef :: !Int-  } deriving (Eq, Ord, Typeable, Generic)+  } deriving (Typeable, Generic, Eq, Ord)  instance Show (RequestOrigin p r) where   showsPrec d (RequestOrigin o r) =-    showParen (d >= 10) (showString "origin: " . showsPrec 11 o . showChar ' ' . showsPrec 11 r)+    showParen (d >= 10) (showString "origin: " . showsPrec 10 o . showChar ' ' . showsPrec 10 r)  -- | Create a new, unique 'RequestOrigin' value for the current process. -- -- @since 0.24.0-makeRequestOrigin :: (SetMember Process (Process q0) e, '[Interrupts] <:: e) => Eff e (RequestOrigin p r)+makeRequestOrigin+  :: ( Typeable r+     , NFData r+     , SetMember Process (Process q0) e+     , '[Interrupts] <:: e)+  => Eff e (RequestOrigin p r) makeRequestOrigin = RequestOrigin <$> self <*> makeReference  instance NFData (RequestOrigin p r) + -- | Send a 'Reply' to a 'Call'. -- -- The reply will be deeply evaluated to 'rnf'.@@ -87,7 +97,7 @@ -- @since 0.15.0 sendReply ::      (SetMember Process (Process q) eff, Member Interrupts eff, Tangible reply, Typeable protocol)-  => RequestOrigin protocol reply-  -> reply-  -> Eff eff ()-sendReply origin reply = sendMessage (_requestOriginPid origin) (Reply origin $! force reply)+  => Serializer (Reply protocol reply) -> RequestOrigin protocol reply -> reply -> Eff eff ()+sendReply ser o r = sendAnyMessage (_requestOriginPid o) $! runSerializer ser $! Reply o r++

@@ -1,416 +0,0 @@--- | A better, more safe implementation of the Erlang/OTP gen_server behaviour.------ @since 0.24.0-module Control.Eff.Concurrent.Protocol.Server-  ( Server(..)-  , StartArgument(..)-  , ToServerEffects-  , ModelState-  , modifyModel-  , getAndModifyModel-  , modifyAndGetModel-  , getModel-  , putModel-  , getAndPutModel-  , useModel-  , zoomModel-  , SettingsReader-  , askSettings-  , viewSettings-  , Event(..)-  , start-  , startLink-  , protocolServerLoop-  , GenServer(..)-  , ToGenServerEffects-  , GenServerId(..)-  , GenServerProtocol-  , GenServerModel-  , GenServerSettings-  , genServer-  , Stateless-  , ToStatelessEffects-  , statelessGenServer-  -- * Re-exports-  , Request(..)-  , sendReply-  , RequestOrigin(..)-  )-  where--import Control.Applicative-import Control.DeepSeq-import Control.Eff-import Control.Eff.Extend ()-import Control.Eff.Concurrent.Process-import Control.Eff.Concurrent.Process.Timer-import Control.Eff.Concurrent.Protocol-import Control.Eff.Concurrent.Protocol.Request-import Control.Eff.Log-import Control.Eff.Reader.Strict-import Control.Eff.State.Strict-import Control.Lens-import Data.Coerce-import Data.Default-import Data.Kind-import Data.String-import Data.Typeable-import Data.Type.Pretty-import qualified Data.Text as T-import GHC.Stack (HasCallStack)-import GHC.Generics---- | A type class for 'Pdu' values that have an implementation--- which handles the corresponding protocol.------ @since 0.24.0-class-  (Typeable (Protocol a)) =>-      Server (a :: Type) e-  where-  -- | The value that defines what is required to initiate a 'Server'-  -- loop.-  data StartArgument a e-  -- | The index type of the 'Event's that this server processes.-  -- This is the first parameter to the 'Request' and therefore of-  -- the 'Pdu' family.-  type Protocol a :: Type-  type Protocol a = a-  -- | Type of the /model/ data, given to every invocation of 'update'-  -- via the 'ModelState' effect.-  -- The /model/ of a server loop is changed through incoming 'Event's.-  -- It is initially calculated by 'setup'.-  type Model a :: Type-  type Model a = ()-  -- | Type of read-only state.-  type Settings a :: Type-  type Settings a = ()--  setup ::-       StartArgument a e-    -> Eff e (Model a, Settings a)--  default setup ::-       (Default (Model a), Default (Settings a))-    => StartArgument a e-    -> Eff e (Model a, Settings a)-  setup _ = pure (def, def)--  update ::-       StartArgument a e-    -> Event (Protocol a)-    -> Eff (ToServerEffects a e) ()---- | /Cons/ (i.e. prepend) 'ModelState' and 'SettingsReader' to an--- effect list.------ @since 0.24.0-type ToServerEffects a e =-  ModelState a ': SettingsReader a ': e---- | The 'Eff'ect type of mutable 'Model' in a 'Server' instance.------ @since 0.24.0-type ModelState a = State (Model a)---- | Modify the 'Model' of a 'Server'.------ @since 0.24.0-modifyModel :: forall a e . Member (ModelState a) e => (Model a -> Model a) -> Eff e ()-modifyModel f = getModel @a >>= putModel @a . f---- | Modify the 'Model' of a 'Server' and return the old value.------ @since 0.24.0-getAndModifyModel :: forall a e . Member (ModelState a) e => (Model a -> Model a) -> Eff e (Model a)-getAndModifyModel f = getModel @a <* modify f---- | Modify the 'Model' of a 'Server' and return the new value.------ @since 0.24.0-modifyAndGetModel :: forall a e . Member (ModelState a) e => (Model a -> Model a) -> Eff e (Model a)-modifyAndGetModel f = modifyModel @a f *> getModel @a---- | Return the 'Model' of a 'Server'.------ @since 0.24.0-getModel :: forall a e . Member (ModelState a) e => Eff e (Model a)-getModel = get---- | Return a element selected by a 'Lens' of the 'Model' of a 'Server'.------ @since 0.24.0-useModel :: forall a b e . Member (ModelState a) e => Getting b (Model a) b -> Eff e b-useModel l = view l <$> getModel @a---- | Overwrite the 'Model' of a 'Server'.------ @since 0.24.0-putModel :: forall a e . Member (ModelState a) e => Model a -> Eff e ()-putModel = put---- | Overwrite the 'Model' of a 'Server', return the old value.------ @since 0.24.0-getAndPutModel :: forall a e . Member (ModelState a) e => Model a -> Eff e (Model a)-getAndPutModel m = getModel @a <* putModel @a m---- | Run an action that modifies portions of the 'Model' of a 'Server' defined by the given 'Lens'.------ @since 0.24.0-zoomModel :: forall a b c e. Member (ModelState a) e => Lens' (Model a) b -> Eff (State b ': e) c -> Eff e c-zoomModel l a = do-  m0 <- getModel @a-  (c, m1) <- runState (view l m0) a-  modifyModel @a (l .~ m1)-  return c---- | The 'Eff'ect type of readonly 'Settings' in a 'Server' instance.------ @since 0.24.0-type SettingsReader a = Reader (Settings a)---- | Return the read-only 'Settings' of a 'Server'------ @since 0.24.0-askSettings :: forall a e . Member (SettingsReader a) e => Eff e (Settings a)-askSettings = ask---- | Return the read-only 'Settings' of a 'Server' as viewed through a 'Lens'------ @since 0.24.0-viewSettings :: forall a b e . Member (SettingsReader a) e =>  Getting b (Settings a) b -> Eff e b-viewSettings l = view l <$> askSettings @a---- | Execute the server loop.------ @since 0.24.0-start-  :: forall a q h-  . ( Server a (InterruptableProcess q)-    , Typeable a-    , LogsTo h (InterruptableProcess q)-    , HasCallStack)-  => StartArgument a (InterruptableProcess q) -> Eff (InterruptableProcess q) (Endpoint (Protocol a))-start a = asEndpoint <$> spawn (protocolServerLoop a)---- | Execute the server loop.------ @since 0.24.0-startLink-  :: forall a q h . (Typeable a, Server a (InterruptableProcess q), LogsTo h (InterruptableProcess q), HasCallStack)-  => StartArgument a (InterruptableProcess q) -> Eff (InterruptableProcess q) (Endpoint (Protocol a))-startLink a = asEndpoint <$> spawnLink (protocolServerLoop a)---- | Execute the server loop.------ @since 0.24.0-protocolServerLoop-     :: forall q e h a-     . ( Server a e-       , SetMember Process (Process q) e-       , Member Interrupts e-       , LogsTo h e-       , Typeable a-       )-  => StartArgument a e -> Eff e ()-protocolServerLoop a = do-  (st, env) <- setup a-  _ <- runReader env (runState st (receiveSelectedLoop sel mainLoop))-  return ()-  where-    sel :: MessageSelector (Event (Protocol a))-    sel =-          OnRequest <$> selectMessage @(Request (Protocol a))-      <|> OnDown    <$> selectMessage @ProcessDown-      <|> OnTimeOut <$> selectMessage @TimerElapsed-      <|> OnMessage <$> selectAnyMessage-    handleInt i = update a (OnInterrupt i) *> pure Nothing-    mainLoop ::-         (Typeable a)-      => Either (Interrupt 'Recoverable) (Event (Protocol a))-      -> Eff (ToServerEffects a e) (Maybe ())-    mainLoop (Left i) = handleInt i-    mainLoop (Right i) = update a i *> pure Nothing---- | Internal protocol to communicate incoming messages and other events to the--- instances of 'Server'.------ Note that this is required to receive any kind of messages in 'protocolServerLoop'.------ @since 0.24.0-data Event a =-    OnRequest (Request a)-  | OnInterrupt (Interrupt 'Recoverable)-  | OnDown ProcessDown-  | OnTimeOut TimerElapsed-  | OnMessage StrictDynamic-  deriving (Show,Generic,Typeable)--instance NFData a => NFData (Event a) where-   rnf = \case-       OnRequest r      -> rnf r-       OnInterrupt r    -> rnf r-       OnDown r  -> rnf r-       OnTimeOut r -> rnf r-       OnMessage r -> r `seq` ()--type instance ToPretty (Event a) = ToPretty a <+> PutStr "event"---- * GenServer---- | A helper for 'Server's that are directly based on logging and 'IO': 'GenIO'------ A record that contains callbacks to provide a 'Server' instance for the--- @tag@ parameter, .------ The name prefix @Gen@ indicates the inspiration from Erlang's @gen_server@ module.------ @since 0.24.0-data GenServer tag e where-  MkGenServer :: LogIo e =>-      { _setupCallback :: Eff (InterruptableProcess e) (Model (GenServer tag e), Settings (GenServer tag e))-      , _updateCallback-          :: Event (GenServerProtocol tag)-          -> Eff (ToGenServerEffects tag e) ()-      } -> GenServer tag e---- | Prepend the 'ModelState' for 'GenServerModel' and 'SettingsReader' for 'GenServerSettings' of a 'GenServer'--- 'Server' to an effect list.------ @since 0.24.0-type ToGenServerEffects tag e = ToServerEffects (GenServer tag e) (InterruptableProcess e)---- | The name/id of a 'GenServer' for logging purposes.------ @since 0.24.0-newtype GenServerId tag =-  MkGenServerId { _fromGenServerId :: T.Text }-  deriving (Typeable, NFData, Ord, Eq, IsString)--instance Show (GenServerId tag) where-  showsPrec _d (MkGenServerId x) = showString (T.unpack x)---- | The 'Protocol' un-wrapper type function.------ @since 0.24.0-type family GenServerProtocol tag---- | Type of state for 'GenServer' based 'Server's------ @since 0.24.0-type family GenServerModel tag---- | Type of the environment for 'GenServer' based 'Server's------ @since 0.24.0-type family GenServerSettings tag--instance ( Typeable (GenServerProtocol tag)-         , LogIo e )-         => Server (GenServer (tag :: Type) e) (InterruptableProcess e) where-  type Protocol (GenServer tag e) = GenServerProtocol tag-  type Model (GenServer tag e) = GenServerModel tag-  type Settings (GenServer tag e) = GenServerSettings tag-  data instance StartArgument (GenServer tag e) (InterruptableProcess e) =-        MkGenStartArgument-         { _genServerId :: GenServerId tag-         , _genServerCallbacks :: GenServer tag e-         } deriving Typeable-  setup (MkGenStartArgument _ cb) = _setupCallback cb-  update (MkGenStartArgument _ cb) req = _updateCallback cb req--instance NFData (StartArgument (GenServer tag e) (InterruptableProcess e)) where-  rnf (MkGenStartArgument x _) = rnf x--instance Typeable tag => Show (StartArgument (GenServer tag e) (InterruptableProcess e)) where-  showsPrec d (MkGenStartArgument x _) =-    showParen (d>=10)-      ( showsPrec 11 x-      . showChar ' ' . showsTypeRep (typeRep (Proxy @tag))-      . showString " gen-server"-      )---- ** 'GenServer' based Server constructors----- | Create a 'GenServer'.------ This requires the callback for 'Event's, a initial 'Model' and a 'GenServerId'.------ There must be a 'GenServerModel' instance.--- There must be a 'GenServerSettings' instance.--- There must be a 'GenServerProtocol' instance.------ This is Haskell, so if this functions is partially applied--- to some 'Event' callback, you get a function back,--- that generates 'StartArgument's from 'GenServerId's, like a /factory/------ @since 0.24.0-genServer-  :: forall tag e  .-     ( Typeable tag-     , HasCallStack-     , LogIo e-     , Server (GenServer tag e) (InterruptableProcess e)-     )-  => (GenServerId tag -> Eff (InterruptableProcess e) (GenServerModel tag, GenServerSettings tag))-  -> (GenServerId tag -> Event (GenServerProtocol tag) -> Eff (ToGenServerEffects tag e) ())-  -> GenServerId tag-  -> StartArgument (GenServer tag e) (InterruptableProcess e)-genServer initCb stepCb i =-  MkGenStartArgument-    { _genServerId = i-    , _genServerCallbacks =-        MkGenServer { _setupCallback = initCb i-                    , _updateCallback = stepCb i . coerce-                    }-    }---- | The type-level tag indicating a stateless 'Server' instance.------ There are 'GenServerModel', 'GenServerSettings' and 'GenServerProtocol' as well as--- 'ToPretty' instances for this type.------ See also 'ToStatelessEffects'.------ @since 0.24.0-data Stateless tag deriving Typeable---- | Prepend the 'ModelState' and 'SettingsReader' of a 'Stateless'--- 'Server' to an effect list. The 'Model' and 'Settings' of a 'Stateless'--- 'Server' are just @()@ /unit/.------ @since 0.24.0-type ToStatelessEffects e = State () ': Reader () ': e--type instance GenServerModel (Stateless tag) = ()-type instance GenServerSettings (Stateless tag) = ()-type instance GenServerProtocol (Stateless tag) = GenServerProtocol tag-type instance ToPretty (Stateless t) = ToPretty t---- | Create a 'Stateless' 'GenServer'.------ This requires only the callback for 'Event's--- and a 'GenServerId'.------ This is Haskell, so if this functions is partially applied--- to some 'Event' callback, you get a function back,--- that generates 'StartArgument's from 'GenServerId's, like a /factory/------ @since 0.24.0-statelessGenServer-  :: forall tag e . ( Typeable tag-     , HasCallStack-     , LogIo e-     , Typeable tag-     , Server (GenServer (Stateless tag) e) (InterruptableProcess e)-     )-  => (GenServerId tag -> Event (GenServerProtocol tag) -> Eff (ToStatelessEffects (InterruptableProcess e)) ())-  -> GenServerId tag-  -> StartArgument (GenServer (Stateless tag) e) (InterruptableProcess e)-statelessGenServer stepCb (MkGenServerId i) =-  genServer (const (pure ((), ()))) runStep (MkGenServerId i)-   where-    runStep :: GenServerId (Stateless tag) -> Event (GenServerProtocol (Stateless tag)) -> Eff (ToStatelessEffects (InterruptableProcess e)) ()-    runStep (MkGenServerId i') loopEvent = stepCb (MkGenServerId i') (coerce loopEvent)

@@ -0,0 +1,206 @@+-- | A better, more safe implementation of the Erlang/OTP gen_server behaviour.+--+-- @since 0.24.0+module Control.Eff.Concurrent.Protocol.StatefulServer+  ( Server(..)+  , start+  , startLink+  , ModelState+  , modifyModel+  , getAndModifyModel+  , modifyAndGetModel+  , getModel+  , putModel+  , getAndPutModel+  , useModel+  , zoomModel+  , SettingsReader+  , askSettings+  , viewSettings+  -- * Re-exports+  , Effectful.Event(..)+  , RequestOrigin(..)+  , Reply(..)+  , sendReply+  )+  where++import Control.Eff+import Control.Eff.Extend ()+import Control.Eff.Concurrent.Process+import Control.Eff.Concurrent.Protocol+import qualified Control.Eff.Concurrent.Protocol.EffectfulServer as Effectful+import Control.Eff.Concurrent.Protocol.Request+import Control.Eff.Log+import Control.Eff.Reader.Strict+import Control.Eff.State.Strict+import Control.Lens+import Data.Default+import Data.Kind+import Data.Typeable+import GHC.Stack (HasCallStack)++-- | A class for 'Stateful' server processes.+--+-- This is inspired by The Elm Architecture, without the @view@ callback.+--+-- This can be used for a variety of typical server loop implementations.+--+-- @since 0.24.0+class (Typeable (Protocol a)) => Server (a :: Type) q where+  -- | The value that defines what is required to initiate a 'Server'+  -- loop.+  data StartArgument a q+  -- | The index type of the 'Event's that this server processes.+  -- This is the first parameter to the 'Request' and therefore of+  -- the 'Pdu' family.+  type Protocol a :: Type+  type Protocol a = a+  -- | Type of the /model/ data, given to every invocation of 'update'+  -- via the 'ModelState' effect.+  -- The /model/ of a server loop is changed through incoming 'Event's.+  -- It is initially calculated by 'setup'.+  type Model a :: Type+  type Model a = ()+  -- | Type of read-only state.+  type Settings a :: Type+  type Settings a = ()++  -- | Return an initial 'Model' and 'Settings'+  setup ::+       StartArgument a q+    -> Eff (InterruptableProcess q) (Model a, Settings a)++  default setup ::+       (Default (Model a), Default (Settings a))+    => StartArgument a q+    -> Eff (InterruptableProcess q) (Model a, Settings a)+  setup _ = pure (def, def)++  -- | Update the 'Model' based on the 'Event'.+  update ::+       StartArgument a q+    -> Effectful.Event (Protocol a)+    -> Eff (ModelState a ': SettingsReader a ': InterruptableProcess q) ()++-- | This type is used to build stateful 'EffectfulServer' instances.+--+-- The types that are suitable to be have to be instances of 'Server'+--+-- @since 0.24.0+data Stateful a deriving Typeable++instance Server a q => Effectful.Server (Stateful a) (InterruptableProcess q) where+  data Init (Stateful a) (InterruptableProcess q) = Init (StartArgument a q)+  type ServerPdu (Stateful a) = Protocol a+  type Effects (Stateful a) (InterruptableProcess q) = ModelState a ': SettingsReader a ': InterruptableProcess q++  runEffects (Init sa) m = do+    (st, env) <- setup sa+    runReader env (evalState st m)++  onEvent (Init sa) = update sa+++-- | Execute the server loop.+--+-- @since 0.24.0+start+  :: forall a q h+  . ( HasCallStack+    , Typeable a+    , LogsTo h (InterruptableProcess q)+    , Effectful.Server (Stateful a) (InterruptableProcess q)+    , Server a q+    )+  => StartArgument a q -> Eff (InterruptableProcess q) (Endpoint (Protocol a))+start = Effectful.start . Init++-- | Execute the server loop.+--+-- @since 0.24.0+startLink+  :: forall a q h+  . ( HasCallStack+    , Typeable a+    , LogsTo h (InterruptableProcess q)+    , Effectful.Server (Stateful a) (InterruptableProcess q)+    , Server a q+    )+  => StartArgument a q -> Eff (InterruptableProcess q) (Endpoint (Protocol a))+startLink = Effectful.startLink . Init+++-- | The 'Eff'ect type of mutable 'Model' in a 'Server' instance.+--+-- @since 0.24.0+type ModelState a = State (Model a)++-- | Modify the 'Model' of a 'Server'.+--+-- @since 0.24.0+modifyModel :: forall a e . Member (ModelState a) e => (Model a -> Model a) -> Eff e ()+modifyModel f = getModel @a >>= putModel @a . f++-- | Modify the 'Model' of a 'Server' and return the old value.+--+-- @since 0.24.0+getAndModifyModel :: forall a e . Member (ModelState a) e => (Model a -> Model a) -> Eff e (Model a)+getAndModifyModel f = getModel @a <* modify f++-- | Modify the 'Model' of a 'Server' and return the new value.+--+-- @since 0.24.0+modifyAndGetModel :: forall a e . Member (ModelState a) e => (Model a -> Model a) -> Eff e (Model a)+modifyAndGetModel f = modifyModel @a f *> getModel @a++-- | Return the 'Model' of a 'Server'.+--+-- @since 0.24.0+getModel :: forall a e . Member (ModelState a) e => Eff e (Model a)+getModel = get++-- | Return a element selected by a 'Lens' of the 'Model' of a 'Server'.+--+-- @since 0.24.0+useModel :: forall a b e . Member (ModelState a) e => Getting b (Model a) b -> Eff e b+useModel l = view l <$> getModel @a++-- | Overwrite the 'Model' of a 'Server'.+--+-- @since 0.24.0+putModel :: forall a e . Member (ModelState a) e => Model a -> Eff e ()+putModel = put++-- | Overwrite the 'Model' of a 'Server', return the old value.+--+-- @since 0.24.0+getAndPutModel :: forall a e . Member (ModelState a) e => Model a -> Eff e (Model a)+getAndPutModel m = getModel @a <* putModel @a m++-- | Run an action that modifies portions of the 'Model' of a 'Server' defined by the given 'Lens'.+--+-- @since 0.24.0+zoomModel :: forall a b c e. Member (ModelState a) e => Lens' (Model a) b -> Eff (State b ': e) c -> Eff e c+zoomModel l a = do+  m0 <- getModel @a+  (c, m1) <- runState (view l m0) a+  modifyModel @a (l .~ m1)+  return c++-- | The 'Eff'ect type of readonly 'Settings' in a 'Server' instance.+--+-- @since 0.24.0+type SettingsReader a = Reader (Settings a)++-- | Return the read-only 'Settings' of a 'Server'+--+-- @since 0.24.0+askSettings :: forall a e . Member (SettingsReader a) e => Eff e (Settings a)+askSettings = ask++-- | Return the read-only 'Settings' of a 'Server' as viewed through a 'Lens'+--+-- @since 0.24.0+viewSettings :: forall a b e . Member (SettingsReader a) e =>  Getting b (Settings a) b -> Eff e b+viewSettings l = view l <$> askSettings @a

@@ -68,7 +68,7 @@ import Control.Eff.Concurrent.Protocol import Control.Eff.Concurrent.Protocol.Client import Control.Eff.Concurrent.Protocol.Request-import Control.Eff.Concurrent.Protocol.Server as Server+import Control.Eff.Concurrent.Protocol.StatefulServer as Server import Control.Eff.Concurrent.Protocol.Supervisor.InternalState import Control.Eff.Concurrent.Process import Control.Eff.Concurrent.Process.Timer@@ -145,8 +145,8 @@   ( Lifted IO q, LogsTo IO q   , TangibleSup p   , Tangible (ChildId p)-  , Server p (InterruptableProcess q)-  ) => Server (Sup p) (InterruptableProcess q) where+  , Server p q+  ) => Server (Sup p) q where    -- | Options that control the 'Sup p' process.   --@@ -156,34 +156,34 @@   -- * the 'Timeout' after requesting a normal child exit before brutally killing the child.   --   -- @since 0.24.0-  data StartArgument (Sup p) (InterruptableProcess q) = MkSupConfig+  data StartArgument (Sup p) q = MkSupConfig     {       -- , supConfigChildRestartPolicy :: ChildRestartPolicy       -- , supConfigResilience :: Resilience       supConfigChildStopTimeout :: Timeout-    , supConfigStartFun :: ChildId p -> Server.StartArgument p (InterruptableProcess q)+    , supConfigStartFun :: ChildId p -> Server.StartArgument p q     }    type Model (Sup p) = Children (ChildId p) p    setup _cfg = pure (def, ())-  update supConfig (OnRequest (Call orig req)) =+  update supConfig (OnCall ser orig req) =     case req of       GetDiagnosticInfo ->  do         p <- (pack . show <$> getChildren @(ChildId p) @p)-        sendReply orig p+        sendReply ser orig p        LookupC i -> do         p <- fmap _childEndpoint <$> lookupChildById @(ChildId p) @p i-        sendReply orig p+        sendReply ser orig p        StopC i t -> do         mExisting <- lookupAndRemoveChildById @(ChildId p) @p i         case mExisting of-          Nothing -> sendReply orig False+          Nothing -> sendReply ser orig False           Just existingChild -> do             stopOrKillChild i existingChild t-            sendReply orig True+            sendReply ser orig True        StartC i -> do         childEp <- raise (raise (Server.start (supConfigStartFun supConfig i)))@@ -193,9 +193,9 @@         case mExisting of           Nothing -> do             putChild i (MkChild @p childEp cMon)-            sendReply orig (Right childEp)+            sendReply ser orig (Right childEp)           Just existingChild ->-            sendReply orig (Left (AlreadyStarted i (existingChild ^. childEndpoint)))+            sendReply ser orig (Left (AlreadyStarted i (existingChild ^. childEndpoint)))    update _supConfig (OnDown (ProcessDown mrChild reason)) = do       oldEntry <- lookupAndRemoveChildByMonitor @(ChildId p) @p mrChild@@ -255,9 +255,9 @@     , LogsTo IO (InterruptableProcess e)     , Lifted IO e     , TangibleSup p-    , Server (Sup p) (InterruptableProcess e)+    , Server (Sup p) e     )-  => StartArgument (Sup p) (InterruptableProcess e)+  => StartArgument (Sup p) e   -> Eff (InterruptableProcess e) (Endpoint (Sup p)) startSupervisor = Server.start 

@@ -4,7 +4,7 @@ import Control.Eff as Eff import Control.Eff.Concurrent.Process import Control.Eff.Concurrent.Protocol-import Control.Eff.Concurrent.Protocol.Server+import Control.Eff.Concurrent.Protocol.StatefulServer import Control.Eff.State.Strict as Eff import Control.Lens hiding ((.=), use) import Data.Default

@@ -30,15 +30,15 @@           aVar <- newEmptyTMVarIO           (Scheduler.defaultMain               (do-                p1 <- spawn $ foreverCheap busyEffect+                p1 <- spawn "test" $ foreverCheap busyEffect                 lift (threadDelay 1000)-                void $ spawn $ do+                void $ spawn "test" $ do                   lift (threadDelay 1000)                   doExit                 lift (threadDelay 100000)                  me <- self-                spawn_ (lift (threadDelay 10000) >> sendMessage me ())+                spawn_  "test" (lift (threadDelay 10000) >> sendMessage me ())                 resultOrError <- receiveWithMonitor p1 (selectMessage @())                 case resultOrError of                   Left  _down -> lift (atomically (putTMVar aVar False))@@ -65,7 +65,7 @@     , ( "spawn-ing"       , void         (send-          (Spawn @SchedulerIO+          (Spawn @SchedulerIO  "test"             (void               (send (ReceiveSelectedMessage @SchedulerIO selectAnyMessage))             )@@ -87,7 +87,7 @@   (testCase     "spawn a child and return"       (Scheduler.defaultMain-        (void (spawn (void receiveAnyMessage)))+        (void (spawn "test" (void receiveAnyMessage)))       )   ) @@ -97,7 +97,7 @@     "spawn a child and exit normally"       (Scheduler.defaultMain         (do-          void (spawn (void receiveAnyMessage))+          void (spawn "test" (void receiveAnyMessage))           void exitNormally           fail "This should not happen!!"         )@@ -112,7 +112,7 @@       "spawn a child with a busy send loop and exit normally"         (Scheduler.defaultMain           (do-            void (spawn (foreverCheap (void (sendMessage 1000 ("test" :: String)))))+            void (spawn "test" (foreverCheap (void (sendMessage 1000 ("test" :: String)))))             void exitNormally             fail "This should not happen!!"           )@@ -126,7 +126,7 @@     "spawn a child and let it return and return"     (Scheduler.defaultMain         (do-          child <- spawn (void (receiveMessage @String))+          child <- spawn "test" (void (receiveMessage @String))           sendMessage child ("test" :: String)           return ()         )@@ -138,7 +138,7 @@     "spawn a child and let it exit and exit"       (Scheduler.defaultMain            (do-              child <- spawn $ void $ provideInterrupts $ exitOnInterrupt+              child <- spawn "test" $ void $ provideInterrupts $ exitOnInterrupt                 (do                   void (receiveMessage @String)                   void exitNormally@@ -166,7 +166,7 @@                 Left  _to -> return ()                 Right _   -> flushMessagesLoop         me <- self-        spawn_+        spawn_  "test-worker"           (do             replicateM_ n $ sendMessage me ("bad message" :: String)             replicateM_ n $ sendMessage me (3123 :: Integer)

@@ -1,4 +1,4 @@-+{-# LANGUAGE UndecidableInstances #-} module GenServerTests   ( test_genServer   ) where@@ -8,9 +8,12 @@ import Control.Eff import Control.Eff.Concurrent import Control.Eff.Concurrent.Protocol.Supervisor as Sup-import Control.Eff.Concurrent.Protocol.Server as Server+import qualified Control.Eff.Concurrent.Protocol.EffectfulServer as E+import qualified Control.Eff.Concurrent.Protocol.StatefulServer as S+import Control.Eff.Concurrent.Protocol.Request import Control.Lens import Data.Coerce+import Data.Functor.Contravariant ((>$<)) import Data.Text as T import Data.Type.Pretty import Data.Typeable (Typeable)@@ -39,12 +42,14 @@  -- ---------------------------------------------------------------------------- -instance LogIo e => Server Small (InterruptableProcess e) where-  data StartArgument Small (InterruptableProcess e) = MkSmall+instance LogIo e => S.Server Small e where+  data StartArgument Small e = MkSmall   type Model Small = String   update MkSmall x =     case x of-      OnRequest msg ->+      E.OnCall ser o (SmallCall f) ->+       sendReply ser o f+      E.OnCast msg ->        logInfo' (show msg)       other ->         interrupt (ErrorInterrupt (show other))@@ -81,22 +86,20 @@  -- ---------------------------------------------------------------------------- -instance LogIo e => Server Big (InterruptableProcess e) where-  data instance StartArgument Big (InterruptableProcess e) = MkBig+instance LogIo e => S.Server Big e where+  data instance StartArgument Big e = MkBig   type Model Big = String   update MkBig = \case-    OnRequest msg ->-      case msg of-        Call orig req ->+    E.OnCall ser orig req ->           case req of             BigCall o -> do               logNotice ("BigCall " <> pack (show o))-              sendReply orig o-            BigSmall x -> update MkSmall (OnRequest (Call (coerce orig) x))-        Cast req ->-          case req of-            BigCast o -> putModel @Big o-            BigSmall x -> update MkSmall (OnRequest (Cast x))+              sendReply ser orig o+            BigSmall x -> S.update MkSmall (S.OnCall (coerce >$< ser) (coerce orig) x)+    E.OnCast req ->+        case req of+          BigCast o -> S.putModel @Big o+          BigSmall x -> S.update MkSmall (S.OnCast x)     other ->       interrupt (ErrorInterrupt (show other)) -- ----------------------------------------------------------------------------@@ -104,8 +107,15 @@ test_genServer :: HasCallStack => TestTree test_genServer = setTravisTestOptions $ testGroup "Server" [   runTestCase "When a server is started it handles call Pdus without dieing" $ do-    big <- start MkBig+    big <- S.start MkBig     call big (BigCall True) >>=  lift . assertBool "invalid result 1"     isProcessAlive (_fromEndpoint big) >>= lift . assertBool "process dead"     call big (BigCall False) >>=  lift . assertBool "invalid result 2" . not+    isProcessAlive (_fromEndpoint big) >>= lift . assertBool "process dead"+    cast big (BigCast "rezo")+    isProcessAlive (_fromEndpoint big) >>= lift . assertBool "process dead"+    cast big (BigSmall (SmallCast "yo diggi"))+    isProcessAlive (_fromEndpoint big) >>= lift . assertBool "process dead"+    call big (BigSmall (SmallCall False )) >>=  lift . assertBool "invalid result 3" . not+    isProcessAlive (_fromEndpoint big) >>= lift . assertBool "process dead"   ]

@@ -47,7 +47,7 @@                       res <- Scheduler.scheduleIOWithLogging  (mkLogWriterIO (T.putStrLn . (">>> " <>) . renderLogMessageConsoleLog))                               $ do                                     me <- self-                                    spawn_ (foreverCheap $ sendMessage me ())+                                    spawn_ "test" (foreverCheap $ sendMessage me ())                                     replicateCheapM_                                         soMany                                         (void (receiveMessage @()))@@ -88,7 +88,7 @@                           Scheduler.scheduleIOWithLogging  (mkLogWriterIO (T.putStrLn . (">>> " <>) . renderLogMessageConsoleLog))                               $ do                                     me <- self-                                    spawn_ (forever $ sendMessage me ())+                                    spawn_ "test" (forever $ sendMessage me ())                                     replicateM_ soMany                                                 (void (receiveMessage @())) 

@@ -8,7 +8,7 @@ import           Control.Eff.Concurrent.Process.Timer import           Control.Eff.Concurrent.Protocol import           Control.Eff.Concurrent.Protocol.Client-import           Control.Eff.Concurrent.Protocol.Server+import           Control.Eff.Concurrent.Protocol.EffectfulServer import qualified Control.Eff.Concurrent.Process.ForkIOScheduler                                                as ForkIO import qualified Control.Eff.Concurrent.Process.SingleThreadedScheduler@@ -26,6 +26,7 @@ import           Control.DeepSeq import           Data.List                      ( sort ) import           Data.Foldable                  ( traverse_ )+import           Data.Maybe import           Data.Typeable import           Data.Type.Pretty import           Data.Void@@ -80,6 +81,7 @@     , linkingTests schedulerFactory     , monitoringTests schedulerFactory     , timerTests schedulerFactory+    , processDetailsTests schedulerFactory     ]   ) @@ -119,23 +121,22 @@   -> Eff r () stopReturnToSender toP = call toP StopReturnToSender -type instance GenServerProtocol ReturnToSender = ReturnToSender- returnToSenderServer   :: forall q    . (HasCallStack, Lifted IO q, LogsTo IO q, Member Logs q, Typeable q)   => Eff (InterruptableProcess q) (Endpoint ReturnToSender) returnToSenderServer = start-  (statelessGenServer @ReturnToSender+  (genServer @ReturnToSender+    (const id)     (\_me evt ->       case evt of-        OnRequest (Call orig msg) ->+        OnCall ser orig msg ->           case msg of             StopReturnToSender -> interrupt testInterruptReason             ReturnToSender fromP echoMsg -> do               sendMessage fromP echoMsg               yieldProcess-              sendReply orig True+              sendReply ser orig True         OnInterrupt i ->           interrupt i         other -> interrupt (ErrorInterrupt (show other))@@ -169,8 +170,8 @@             traverse_ (sendMessage destination) [1 .. nMax]          me          <- self-        receiverPid <- spawn (receiverLoop me)-        spawn_ (senderLoop receiverPid)+        receiverPid <- spawn "reciever loop" (receiverLoop me)+        spawn_ "sender loop" (senderLoop receiverPid)         ok <- receiveMessage @Bool         lift (ok @? "selective receive failed")     , testCase "receive a message while waiting for a call reply"@@ -182,11 +183,11 @@         lift (ok @? "selective receive failed")     , testCase "flush messages" $ applySchedulerFactory schedulerFactory $ do       me <- self-      spawn_ $ replicateM_ 10 (sendMessage me True) >> sendMessage me ()-      spawn_+      spawn_ "sender-bool" $ replicateM_ 10 (sendMessage me True) >> sendMessage me ()+      spawn_ "sender-float"         $  replicateM_ 10 (sendMessage me (123.23411 :: Float))         >> sendMessage me ()-      spawn_ $ replicateM_ 10 (sendMessage me ("123"::String)) >> sendMessage me ()+      spawn_ "sender-string" $ replicateM_ 10 (sendMessage me ("123"::String)) >> sendMessage me ()       ()   <- receiveMessage       ()   <- receiveMessage       ()   <- receiveMessage@@ -257,9 +258,9 @@             sendMessage ponger (Ping me)             Pong <- receiveMessage             sendMessage parent True-      pongPid <- spawn pongProc+      pongPid <- spawn "pong" pongProc       me      <- self-      spawn_ (pingProc pongPid me)+      spawn_ "ping" (pingProc pongPid me)       ok <- receiveMessage @Bool       lift (ok @? "ping pong failed")   , testCase "ping pong a message between two processes, with massive yielding"@@ -283,11 +284,11 @@             sendMessage parent True             yieldProcess       yieldProcess-      pongPid <- spawn pongProc+      pongPid <- spawn "pong" pongProc       yieldProcess       me <- self       yieldProcess-      spawn_ (pingProc pongPid me)+      spawn_ "ping" (pingProc pongPid me)       yieldProcess       ok <- receiveMessage @Bool       yieldProcess@@ -301,7 +302,7 @@       let pongProc = foreverCheap $ do             Pong <- receiveMessage             lift (putMVar pongVar Pong)-      ponger <- spawn pongProc+      ponger <- spawn "pong" pongProc       sendMessage ponger Pong       let waitLoop = do             p <- lift (tryTakeMVar pongVar)@@ -339,7 +340,7 @@           me <- self           let n = 15           traverse_-            (\(i :: Int) -> spawn $ foreverCheap+            (\(i :: Int) -> spawn "test" $ foreverCheap               (void                 (  sendMessage (888888 + fromIntegral i) ("test message" :: String)                 >> yieldProcess@@ -348,7 +349,7 @@             )             [0 .. n]           traverse_-            (\(i :: Int) -> spawn $ do+            (\(i :: Int) -> spawn "test" $ do               void $ sendMessage me i               void (exitWithError (show i ++ " died"))               error "this should not be reached"@@ -376,7 +377,7 @@           me <- self           traverse_             (const-              (spawn+              (spawn "reciever"                 (do                   m <- receiveAnyMessage                   void (sendAnyMessage me m)@@ -390,18 +391,18 @@       $ \assertEff -> do -- start massive amount of children that exit as soon as they are              -- executed, this will only work smoothly when scheduler schedules              -- the new child before the parent-          traverse_ (const (spawn exitNormally)) [1 .. n]+          traverse_ (const (spawn "lemming" exitNormally)) [1 .. n]           assertEff "" True       , testCase "two concurrent processes"       $ scheduleAndAssert schedulerFactory       $ \assertEff -> do           me     <- self-          child1 <- spawn+          child1 <- spawn "reciever"             (do               m <- receiveAnyMessage               void (sendAnyMessage me m)             )-          child2 <- spawn (foreverCheap (void (sendMessage 888888 (""::String))))+          child2 <- spawn "sender" (foreverCheap (void (sendMessage 888888 (""::String))))           sendMessage child1 ("test" :: String)           i <- receiveMessage           sendInterrupt child2 testInterruptReason@@ -411,7 +412,7 @@       $ \assertEff -> do           me <- self           traverse_-            (\(i :: Int) -> spawn $ do+            (\(i :: Int) -> spawn "sender"$ do               when (i `rem` 5 == 0) $ void $ sendMessage me i               foreverCheap $ void (sendMessage 888 ("test message to 888" :: String))             )@@ -423,7 +424,7 @@       $ \assertEff -> do           me <- self           traverse_-            (\(i :: Int) -> spawn $ do+            (\(i :: Int) -> spawn "sender"$ do               when (i `rem` 5 == 0) $ void $ sendMessage me i               foreverCheap $ void self             )@@ -435,7 +436,7 @@       $ \assertEff -> do           me <- self           traverse_-            (\(i :: Int) -> spawn $ do+            (\(i :: Int) -> spawn "killer" $ do               when (i `rem` 5 == 0) $ void $ sendMessage me i               foreverCheap $ void (sendShutdown 999 ExitNormally)             )@@ -447,11 +448,11 @@       $ \assertEff -> do           me <- self           traverse_-            (\(i :: Int) -> spawn $ do+            (\(i :: Int) -> spawn "sender 0" $ do               when (i `rem` 5 == 0) $ void $ sendMessage me i               parent <- self               foreverCheap $ void-                (spawn (void (sendMessage parent ("test msg from child"::String))))+                (spawn  "sender" (void (sendMessage parent ("test msg from child"::String))))             )             [0 .. n]           oks <- replicateM (length [0, 5 .. n]) receiveMessage@@ -461,7 +462,7 @@       $ \assertEff -> do           me <- self           traverse_-            (\(i :: Int) -> spawn $ do+            (\(i :: Int) -> spawn  "sender" $ do               when (i `rem` 5 == 0) $ void $ sendMessage me i               foreverCheap $ void receiveAnyMessage             )@@ -520,7 +521,7 @@           , ( "spawn-ing"             , void               (send-                (Spawn @r+                (Spawn @r  "reciever"                   (void (send (ReceiveSelectedMessage @r selectAnyMessage)))                 )               )@@ -533,7 +534,7 @@         [ testCase ("a child process, busy with " ++ busyWith)           $ applySchedulerFactory schedulerFactory           $ do-              void $ spawn $ foreverCheap busyEffect+              void $ spawn "busyloop" $ foreverCheap busyEffect               lift (threadDelay 10000)         | (busyWith, busyEffect) <-           [ ( "receiving"@@ -553,7 +554,7 @@             , void               (send                 (Spawn @r-                  (void (send (ReceiveSelectedMessage @r selectAnyMessage)))+                  "reciever"  (void (send (ReceiveSelectedMessage @r selectAnyMessage)))                 )               )             )@@ -569,9 +570,9 @@             )           $ scheduleAndAssert schedulerFactory           $ \assertEff -> do-              p1 <- spawn $ foreverCheap busyEffect+              p1 <- spawn "busyloop" $ foreverCheap busyEffect               lift (threadDelay 1000)-              void $ spawn $ do+              void $ spawn "sleep loop"  $ do                 lift (threadDelay 1000)                 doExit               lift (threadDelay 100000)@@ -599,6 +600,7 @@             , void               (send                 (Spawn @r+                  "receiver"                   (void (send (ReceiveSelectedMessage @r selectAnyMessage)))                 )               )@@ -647,7 +649,7 @@     $ scheduleAndAssert schedulerFactory     $ \assertEff -> do         me    <- self-        other <- spawn+        other <- spawn "interrupted"           (do             untilInterrupted (SendMessage @r 666666 (toStrictDynamic ("test"::String)))             void (sendMessage me ("OK"::String))@@ -659,7 +661,7 @@     $ scheduleAndAssert schedulerFactory     $ \assertEff -> do         me    <- self-        other <- spawn+        other <- spawn "interrupted"           (do             untilInterrupted (ReceiveSelectedMessage @r selectAnyMessage)             void (sendMessage me ("OK" :: String))@@ -671,7 +673,7 @@     $ scheduleAndAssert schedulerFactory     $ \assertEff -> do         me    <- self-        other <- spawn+        other <- spawn "interrupted"           (do             untilInterrupted (SelfPid @r)             void (sendMessage me ("OK" :: String))@@ -683,9 +685,9 @@     $ scheduleAndAssert schedulerFactory     $ \assertEff -> do         me    <- self-        other <- spawn+        other <- spawn "interrupted"           (do-            untilInterrupted (Spawn @r (return ()))+            untilInterrupted (Spawn @r "returner" (return ()))             void (sendMessage me ("OK"::String))           )         void (sendInterrupt other testInterruptReason)@@ -695,7 +697,7 @@     $ scheduleAndAssert schedulerFactory     $ \assertEff -> do         me    <- self-        other <- spawn+        other <- spawn "interrupted"           (do             untilInterrupted (SendShutdown @r 777 ExitNormally)             void (sendMessage me ("OK"::String))@@ -743,7 +745,7 @@     , testCase "linked process exit message is NormalExit"     $ applySchedulerFactory schedulerFactory     $ do-        foo <- spawn (void (receiveMessage @Void))+        foo <- spawn "reciever" (void (receiveMessage @Void))         handleInterrupts           (lift . (\e -> e /= LinkedProcessCrashed foo @? show e))           (do@@ -754,7 +756,7 @@     , testCase "linked process exit message is Crash"     $ applySchedulerFactory schedulerFactory     $ do-        foo <- spawn (void (receiveMessage @Void))+        foo <- spawn "reciever" (void (receiveMessage @Void))         handleInterrupts           (lift . (@?= LinkedProcessCrashed foo))           (do@@ -765,7 +767,7 @@     , testCase "link multiple times"     $ applySchedulerFactory schedulerFactory     $ do-        foo <- spawn (void (receiveMessage @Void))+        foo <- spawn "reciever" (void (receiveMessage @Void))         handleInterrupts           (lift . (@?= LinkedProcessCrashed foo))           (do@@ -782,11 +784,11 @@     , testCase "unlink multiple times"     $ applySchedulerFactory schedulerFactory     $ do-        foo <- spawn (void (receiveMessage @Void))+        foo <- spawn "reciever" (void (receiveMessage @Void))         handleInterrupts           (lift . (False @?) . show)           (do-            spawn_ (void receiveAnyMessage)+            spawn_ "reciever" (void receiveAnyMessage)             linkProcess foo             linkProcess foo             linkProcess foo@@ -801,14 +803,14 @@     , testCase "spawnLink" $ applySchedulerFactory schedulerFactory $ do       let foo = void (receiveMessage @Void)       handleInterrupts (\er -> lift (isProcessDownInterrupt Nothing er @? show er)) $ do-        x <- spawnLink foo+        x <- spawnLink "foo" foo         sendShutdown x ExitProcessCancelled         void (receiveMessage @Void)     , testCase "spawnLink and child exits by returning from spawn" $ applySchedulerFactory schedulerFactory $ do         me <- self-        u <- spawn $ do+        u <- spawn "unlinker" $ do           logCritical "unlinked child started"-          l <- spawnLink $ do+          l <- spawnLink "linked" $ do             logCritical "linked child started"             () <- receiveMessage             logCritical "linked child done"@@ -829,9 +831,9 @@      , testCase "spawnLink and child exits via exitWithError" $ applySchedulerFactory schedulerFactory $ do         me <- self-        u <- spawn $ do+        u <- spawn "unlinker" $ do           logCritical "unlinked child started"-          l <- spawnLink $ do+          l <- spawnLink "linker" $ do             logCritical "linked child started"             () <- receiveMessage             logCritical "linked child done"@@ -861,17 +863,17 @@                 x <- receiveMessage                 linkProcess x                 sendMessage mainProc True-        linker <- spawnLink linkingServer+        linker <- spawnLink "link-server" linkingServer         logNotice "mainProc"         do-          x <- spawnLink (logNotice "x 1" >> void (receiveMessage @Void))+          x <- spawnLink "x1" (logNotice "x 1" >> void (receiveMessage @Void))           withMonitor x $ \xRef -> do             sendMessage linker x             void $ receiveSelectedMessage (filterMessage id)             sendShutdown x ExitNormally             void (receiveSelectedMessage (selectProcessDown xRef))         do-          x <- spawnLink (logNotice "x 2" >> void (receiveMessage @Void))+          x <- spawnLink "x2" (logNotice "x 2" >> void (receiveMessage @Void))           withMonitor x $ \xRef -> do             sendMessage linker x             void $ receiveSelectedMessage (filterMessage id)@@ -912,10 +914,10 @@               sendMessage foo2Pid ("the end" :: String)               void receiveAnyMessage             )-      foo1Pid <- spawn foo1-      foo2Pid <- spawn (foo2 foo1Pid)+      foo1Pid <- spawn "foo1" foo1+      foo2Pid <- spawn "foo2" (foo2 foo1Pid)       me      <- self-      barPid  <- spawn (bar foo2Pid me)+      barPid  <- spawn "bar" (bar foo2Pid me)       handleInterrupts         (\er -> lift (LinkedProcessCrashed barPid @?= er))         (do@@ -945,7 +947,7 @@     , testCase "monitored process exit normally"     $ applySchedulerFactory schedulerFactory     $ do-        target <- spawn (receiveMessage >>= exitBecause)+        target <- spawn "reciever-exit-value" (receiveMessage >>= exitBecause)         ref    <- monitor target         sendMessage target ExitNormally         pd <- receiveSelectedMessage (selectProcessDown ref)@@ -954,7 +956,7 @@     , testCase "multiple monitors some demonitored"     $ applySchedulerFactory schedulerFactory     $ do-        target <- spawn (receiveMessage >>= exitBecause)+        target <- spawn "reciever-exit-value" (receiveMessage >>= exitBecause)         ref1   <- monitor target         ref2   <- monitor target         ref3   <- monitor target@@ -973,7 +975,7 @@     , testCase "monitored process killed"     $ applySchedulerFactory schedulerFactory     $ do-        target <- spawn (receiveMessage >>= exitBecause)+        target <- spawn "reciever-exit-value" (receiveMessage >>= exitBecause)         ref    <- monitor target         sendMessage target ExitProcessCancelled         pd <- receiveSelectedMessage (selectProcessDown ref)@@ -982,12 +984,12 @@     , testCase "demonitored process killed"     $ applySchedulerFactory schedulerFactory     $ do-        target <- spawn (receiveMessage >>= exitBecause)+        target <- spawn "reciever-exit-value" (receiveMessage >>= exitBecause)         ref    <- monitor target         demonitor ref         sendMessage target ExitProcessCancelled         me <- self-        spawn_ (lift (threadDelay 10000) >> sendMessage me ())+        spawn_ "wait-and-send" (lift (threadDelay 10000) >> sendMessage me ())         pd <- receiveSelectedMessage            (Right <$> selectProcessDown ref <|> Left <$> selectMessage @())@@ -1014,7 +1016,7 @@     $ applySchedulerFactory schedulerFactory     $ do         me    <- self-        other <- spawn+        other <- spawn "reciever"           (do             r <- receiveMessage @()             lift (r @?= ())@@ -1033,7 +1035,7 @@             testMsg :: Float             testMsg = 123         me    <- self-        other <- spawn+        other <- spawn "test"           (do             replicateM_ n $ sendMessage me ("bad message" :: String)             r <- receiveMessage @()@@ -1052,3 +1054,56 @@         lift (threadDelay 100)     ]   )++processDetailsTests ::+     forall r. (Lifted IO r, LogsTo IO r)+  => IO (Eff (InterruptableProcess r) () -> IO ())+  -> TestTree+processDetailsTests schedulerFactory =+  setTravisTestOptions+    (testGroup+       "process info tests"+       [ testCase+           "no infos for a non-existant process"+           (applySchedulerFactory+              schedulerFactory+              (do let nonExistentPid = ProcessId 123+                  me <- self+                  pInfo <- getProcessState nonExistentPid+                  lift (assertEqual "" (nonExistentPid /= me) (isNothing pInfo))))+       , testCase+           "no infos for a dead process"+           (applySchedulerFactory+              schedulerFactory+              (do deadProc <- spawn "dead" (return ())+                  mr <- monitor deadProc+                  void $ receiveSelectedMessage (selectProcessDown mr)+                  pInfo <- getProcessState deadProc+                  lift (assertBool "no process state of a dead process expected" (isNothing pInfo))))+       , testGroup+           "process title tests"+           [ testCase+               "getProcessState returns the title passed to spawn"+               (applySchedulerFactory+                  schedulerFactory+                  (do let expectedTitle = "expected title"+                      p <- spawn expectedTitle (void receiveAnyMessage)+                      (actualTitle, _, _) <- fromJust <$> getProcessState p+                      lift (assertEqual "unexpected process title" expectedTitle actualTitle)))+           ]+       , testGroup "process details tests" [+            testCase+                       "update"+                       (applySchedulerFactory+                          schedulerFactory+                          (do let expected1 = "test details 1"+                                  expected2 = "test details 2"+                              updateProcessDetails expected1+                              (_, actual1 , _) <- fromJust <$> (self >>= getProcessState)+                              lift (assertEqual "1" expected1 actual1)+                              updateProcessDetails expected2+                              (_, actual2 , _) <- fromJust <$> (self >>= getProcessState)+                              lift (assertEqual "2" expected2 actual2)+                              ))+       ]+       ])

@@ -16,12 +16,12 @@       $   Right (42 :: Int)       @=? Scheduler.schedulePure               (do-                  adderChild <- spawn $ do+                  adderChild <- spawn "test" $ do                       (from, arg1, arg2) <- receiveMessage                       sendMessage from ((arg1 + arg2) :: Int)                       foreverCheap $ void $ receiveAnyMessage -                  multiplierChild <- spawn $ do+                  multiplierChild <- spawn "test" $ do                       (from, arg1, arg2) <- receiveMessage                       sendMessage from ((arg1 * arg2) :: Int) @@ -40,7 +40,7 @@         "spawn a child and exit normally"         (Scheduler.defaultMainSingleThreaded             (do-                void (spawn (void receiveAnyMessage))+                void (spawn "test" (void receiveAnyMessage))                 void exitNormally                 fail "This should not happen!!"             )@@ -54,7 +54,7 @@         "spawn a child and let it exit and exit"         (Scheduler.defaultMainSingleThreaded             (do-                child <- spawn+                child <- spawn "test"                     (do                         void (receiveMessage @String)                         void exitNormally

@@ -7,7 +7,7 @@ import Control.DeepSeq import Control.Eff import Control.Eff.Concurrent-import Control.Eff.Concurrent.Protocol.Server as Server+import Control.Eff.Concurrent.Protocol.StatefulServer as Server import Control.Eff.Concurrent.Protocol.Supervisor as Sup import Control.Eff.Concurrent.Process.Timer import Data.Either (fromRight, isLeft, isRight)@@ -29,7 +29,7 @@       in [ runTestCase "The supervisor starts and is shut down" $ do              outerSelf <- self              testWorker <--               spawn $ do+               spawn "test-worker" $ do                  sup <- startTestSup                  sendMessage outerSelf sup                  () <- receiveMessage@@ -264,15 +264,15 @@   | ExitWhenRequested   deriving Eq -instance Server TestProtocol InterruptableProcEff where+instance Server TestProtocol SchedulerIO where   update (TestServerArgs testMode tId) evt =     case evt of-      OnRequest (Cast (TestInterruptWith i)) -> do+      OnCast (TestInterruptWith i) -> do         logInfo (pack (show tId) <> ": stopping with: " <> pack (show i))         interrupt i-      OnRequest (Call orig (TestGetStringLength str)) -> do+      OnCall ser orig (TestGetStringLength str) -> do         logInfo (pack (show tId) <> ": calculating length of: " <> pack str)-        sendReply orig (length str)+        sendReply ser orig (length str)       OnInterrupt x -> do         logNotice (pack (show tId) <> ": " <> pack (show x))         if testMode == IgnoreNormalExitRequest@@ -283,6 +283,6 @@             exitBecause (interruptToExit x)       _ ->         logDebug (pack (show tId) <> ": got some info: " <> pack (show evt))-  data instance StartArgument TestProtocol InterruptableProcEff = TestServerArgs TestProtocolServerMode Int+  data instance StartArgument TestProtocol SchedulerIO = TestServerArgs TestProtocolServerMode Int  type instance ChildId TestProtocol = Int