extensible-effects-concurrent 0.2.0.3 → 0.3.0.0

raw patch · 19 files changed

+1640/−1303 lines, 19 filesPVP ok

version bump matches the API change (PVP)

API changes (from Hackage documentation)

Files

• ChangeLog.md +51/−33
• extensible-effects-concurrent.cabal +8/−2
• src/Control/Eff/Concurrent/Api.hs +15/−13
• src/Control/Eff/Concurrent/Api/Client.hs +30/−22
• src/Control/Eff/Concurrent/Api/Observer.hs +21/−5
• src/Control/Eff/Concurrent/Api/Server.hs +134/−145
• src/Control/Eff/Concurrent/Examples.hs +1/−1
• src/Control/Eff/Concurrent/Examples2.hs +4/−4
• src/Control/Eff/Concurrent/Process.hs +165/−79
• src/Control/Eff/Concurrent/Process/ForkIOScheduler.hs +248/−256
• src/Control/Eff/Concurrent/Process/Interactive.hs +128/−0
• src/Control/Eff/Concurrent/Process/SingleThreadedScheduler.hs +130/−132
• src/Control/Eff/ExceptionExtra.hs +2/−59
• src/Control/Eff/Interactive.hs +0/−77
• test/Common.hs +2/−2
• test/Debug.hs +9/−0
• test/ForkIOScheduler.hs +136/−93
• test/ProcessBehaviourTestCases.hs +513/−348
• test/SingleThreadedScheduler.hs +43/−32

@@ -1,67 +1,85 @@ # Changelog for extensible-effects-concurrent +## 0.3.0.0++- Add support for running and interacting with a scheduler+  and it's processes from IO, for example from ghci+- Rename `yieldProcess` to `executeAndResume`+- Add an actual `yieldProcess`, that behaves like `yield`+- Change the return type of function to `()` where applicable+  to avoid all these `_ <- sendMessage...` or `void $ sendMessage`+  dances.+- Add a simple logging observer: `spawnLoggingObserver`+- Removed `Control.Eff.Interactive`+- Removed most functions in `Control.Eff.ExceptionExtra`+- Make `sendMessage` and the reply in `call` strict with respect to the message+  payload+ ## 0.2.0.3 -  * Improve 'Api' documentation-  * Improve `LogChannel` API-  * Reorganize unit tests-  * Hopefully tune travis ci test parameter enough to get a stable build result+- Improve 'Api' documentation+- Improve `LogChannel` API+- Reorganize unit tests+- Hopefully tune travis ci test parameter enough to get a stable build result  ## 0.2.0.2 -  * Fix minor `stack upload` complaints about the cabal file+- Fix minor `stack upload` complaints about the cabal file  ## 0.2.0.1 -  * Simplify IO Exception handling in `ForkIoScheduler`,-  * Add many unit tests for exception, exit and shutdown+- Simplify IO Exception handling in `ForkIoScheduler`,+- Add many unit tests for exception, exit and shutdown  ## 0.2.0.0 -  * Add `Spawn` to `Process`-  * Merge `MessagePassing` and `Process`-  * Add initial test suite-  * Fix shutdown error in `ForkIoScheduler`-  * Rename `Dispatcher` to `Scheduler`-  * Add `receiveLoop` function to `Process`-  * Change `Api.Server` `serve` to loop instead of handling just one request-  * Allow combining multiple `ApiHandler` such that one process can handle-    multiple APIs+- Add `Spawn` to `Process`+- Merge `MessagePassing` and `Process`+- Add initial test suite+- Fix shutdown error in `ForkIoScheduler`+- Rename `Dispatcher` to `Scheduler`+- Add `receiveLoop` function to `Process`+- Change `Api.Server` `serve` to loop instead of handling just one request+- Allow combining multiple `ApiHandler` such that one process can handle+  multiple APIs  ### TODO-  * Add `Kill` action to `Process` +- Add `Kill` action to `Process`+ ## 0.1.3.0 -  * Rename and split `GenServer` to `Api`, `Api.Client`, `Api.Server`-  * Add `registerServer`, `callRegistered` and `castRegistered`-  * Remove the variant of `cast` that returns a boolean+- Rename and split `GenServer` to `Api`, `Api.Client`, `Api.Server`+- Add `registerServer`, `callRegistered` and `castRegistered`+- Remove the variant of `cast` that returns a boolean  ## 0.1.2.2-  * Try to fix version bounds for hackage +- Try to fix version bounds for hackage+ ## 0.1.2.1-  * Add more documentation-  * Simplify Scheduler API-  * Make more exception safe +- Add more documentation+- Simplify Scheduler API+- Make more exception safe+ ## 0.1.2.0 -  * Add Observer module-  * Implement Exception handling-  * Improve Scheduler shutdown-  * Add logging support via the logging-effect library+- Add Observer module+- Implement Exception handling+- Improve Scheduler shutdown+- Add logging support via the logging-effect library  ## 0.1.1.0 -  * Substantial API reoganization-  * Rename/Move modules+- Substantial API reoganization+- Rename/Move modules  ## 0.1.0.1 -  * Stack/Cabal/Github Cosmetics-  * Travis build job+- Stack/Cabal/Github Cosmetics+- Travis build job  ## 0.1.0.0 -  * Initial Version+- Initial Version

@@ -1,5 +1,5 @@ name:           extensible-effects-concurrent-version:        0.2.0.3+version:        0.3.0.0 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@@ -43,13 +43,13 @@       tagged   exposed-modules:                   Control.Eff.ExceptionExtra,-                  Control.Eff.Interactive,                   Control.Eff.Log,                   Control.Eff.Concurrent.Api,                   Control.Eff.Concurrent.Api.Client,                   Control.Eff.Concurrent.Api.Server,                   Control.Eff.Concurrent.Process,                   Control.Eff.Concurrent.Process.ForkIOScheduler,+                  Control.Eff.Concurrent.Process.Interactive,                   Control.Eff.Concurrent.Process.SingleThreadedScheduler,                   Control.Eff.Concurrent.Api.Observer   other-modules:@@ -90,6 +90,7 @@               , ForkIOScheduler               , SingleThreadedScheduler               , ProcessBehaviourTestCases+              , Debug               , Common   ghc-options: -threaded -rtsopts -with-rtsopts=-N   build-depends:@@ -99,6 +100,9 @@               , tasty               , tasty-discover               , tasty-hunit+              , containers+              , QuickCheck <2.11+              , lens               , HUnit               , stm   default-language: Haskell2010@@ -106,6 +110,8 @@                       , RankNTypes                       , ScopedTypeVariables                       , TypeOperators+                      , GADTs+                      , TemplateHaskell                       , DataKinds                       , FlexibleContexts                       , BangPatterns

@@ -13,15 +13,16 @@ {-# LANGUAGE TypeApplications #-} {-# LANGUAGE GADTs #-} --- | This module contains a mechanisms to specify what kind of messages a--- process can receive and possible answer by sending a message back to the--- orginator.------ A message can be either a blocking or a non-blocking request.+-- | This module contains a mechanism to specify what kind of messages (aka+-- /requests/) a 'Server' ('Process') can handle, and if the caller blocks and+-- waits for an answer, which the server process provides. ----- The type magic in the 'Api' type familiy allows to define a related set of messages along+-- The type magic in the 'Api' type familiy allows to define a related set of /requests/ along -- with the corresponding responses. --+-- Request handling can be either blocking, if a response is requred, or+-- non-blocking.+-- -- A process can /serve/ a specific 'Api' instance by using the functions provided by -- the "Control.Eff.Concurrent.Api.Server" module. --@@ -44,8 +45,8 @@ import           Control.Eff.Concurrent.Process  -- | This data family defines an API, a communication interface description--- between to processes, where one process acts as a __server__ and the other(s)--- as __client(s)__.+-- between at least two processes. The processes act as __servers__ or+-- __client(s)__ regarding a specific instance of this type. -- -- The first parameter is usually a user defined phantom type that identifies -- the 'Api' instance.@@ -70,12 +71,13 @@ data family Api (api :: Type) (reply :: Synchronicity)  --- | This __data kind__ is used to indicate at the type level if a specific--- constructor of an @Api@ instance has a result for which some caller has to--- wait, or if it is asynchronous.+-- | The (promoted) constructors of this type specify (at the type level) the+-- reply behavior of a specific constructor of an @Api@ instance. data Synchronicity =-  Synchronous Type -- ^ Blocking operation with a specific return type, e.g. @('Synchronous (Either RentalError RentalId))@-  | Asynchronous -- ^ Non-blocking async operation+  Synchronous Type -- ^ Specify that handling a request is a blocking operation+                   -- with a specific return type, e.g. @('Synchronous (Either+                   -- RentalError RentalId))@+  | Asynchronous -- ^ Non-blocking, asynchronous, request handling     deriving (Typeable)  -- | This is a tag-type that wraps around a 'ProcessId' and holds an 'Api' index

@@ -36,7 +36,9 @@ import           Control.Eff.Concurrent.Api.Internal import           Control.Eff.Concurrent.Process import           Data.Dynamic-import           Data.Typeable (Typeable, typeRep)+import           Data.Typeable                  ( Typeable+                                                , typeRep+                                                ) import           GHC.Stack  -- | Send an 'Api' request that has no return value and return as fast as@@ -55,8 +57,8 @@   -> Server o   -> Api o 'Asynchronous   -> Eff r Bool-castChecked px (Server pid) callMsg =-  sendMessage px pid (toDyn  (Cast callMsg))+castChecked px (Server pid) castMsg =+  sendMessageChecked px pid (toDyn $! (Cast $! castMsg))  -- | Send an 'Api' request that has no return value and return as fast as -- possible. The type signature enforces that the corresponding 'Api' clause is@@ -72,9 +74,9 @@   -> Server o   -> Api o 'Asynchronous   -> Eff r ()-cast px toServer apiRequest =-  do _ <- castChecked px toServer apiRequest-     return ()+cast px toServer apiRequest = do+  _ <- castChecked px toServer apiRequest+  return ()  -- | Send an 'Api' request and wait for the server to return a result value. --@@ -90,18 +92,19 @@      )   => SchedulerProxy q   -> Server api-  -> Api api ('Synchronous result)+  -> Api api ( 'Synchronous result)   -> Eff r result call px (Server pidInt) req = do   fromPid <- self px-  let requestMessage = Call fromPid req-  wasSent <- sendMessage px pidInt (toDyn requestMessage)+  let requestMessage = Call fromPid $! req+  wasSent <- sendMessageChecked px pidInt (toDyn $! requestMessage)   if wasSent     then       let extractResult :: Response api result -> result           extractResult (Response _pxResult result) = result-      in extractResult <$> receiveMessageAs px-    else raiseError px+      in  extractResult <$> receiveMessageAs px+    else raiseError+      px       ("Could not send request message " ++ show (typeRep requestMessage))  -- * Registered Servers@@ -119,13 +122,16 @@  -- | Run a reader effect that contains __the one__ server handling a specific -- 'Api' instance.-registerServer :: Server o -> Eff ( Reader (Server o) ': r ) a -> Eff r a+registerServer :: Server o -> Eff (Reader (Server o) ': r) a -> Eff r a registerServer = flip runReader  -- | Like 'call' but take the 'Server' from the reader provided by -- 'registerServer'.-callRegistered :: (Typeable reply, ServesApi o r q)-      => SchedulerProxy q -> Api o ('Synchronous reply) -> Eff r reply+callRegistered+  :: (Typeable reply, ServesApi o r q)+  => SchedulerProxy q+  -> Api o ( 'Synchronous reply)+  -> Eff r reply callRegistered px method = do   serverPid <- ask   call px serverPid method@@ -135,19 +141,21 @@ -- application level errors can be combined with errors rising from inter -- process communication. callRegisteredA-  :: forall r q o f reply .-    (Alternative f, Typeable f, Typeable reply, ServesApi o r q)-  => SchedulerProxy q -> Api o ('Synchronous (f reply))+  :: forall r q o f reply+   . (Alternative f, Typeable f, Typeable reply, ServesApi o r q)+  => SchedulerProxy q+  -> Api o ( 'Synchronous (f reply))   -> Eff r (f reply) callRegisteredA px method = do-  catchRaisedError px-    (const (return (empty @f)))-    (callRegistered px method)+  catchRaisedError px (const (return (empty @f))) (callRegistered px method)  -- | Like 'cast' but take the 'Server' from the reader provided by -- 'registerServer'.-castRegistered :: (Typeable o, ServesApi o r q)-               => SchedulerProxy q -> Api o 'Asynchronous -> Eff r ()+castRegistered+  :: (Typeable o, ServesApi o r q)+  => SchedulerProxy q+  -> Api o 'Asynchronous+  -> Eff r () castRegistered px method = do   serverPid <- ask   cast px serverPid method

@@ -35,11 +35,11 @@   , notifyObservers   , CallbackObserver   , spawnCallbackObserver+  , spawnLoggingObserver   ) where  import GHC.Stack import Data.Dynamic-import Data.Maybe import Data.Set (Set) import qualified Data.Set as Set import Control.Eff@@ -206,16 +206,32 @@   asServer @(CallbackObserver o)   <$>   (spawn @r @q $ do-      me <- asServer @(CallbackObserver o) <$> self px       let loopUntil =             serve px             (ApiHandler @(CallbackObserver o)               (handleCast loopUntil)               (unhandledCallError px)-              (\e ->-                 do logMsg (show me ++ " observer terminating " ++ fromMaybe "normally" e)-                    defaultTermination px e))+              (defaultTermination px))       loopUntil)  where    handleCast k (CbObserved fromSvr v) =      onObserve fromSvr v >>= flip when k++-- | Use 'spawnCallbackObserver' to create a universal logging observer,+-- using the 'Show' instance of the 'Observation'.+-- | Start a new process for an 'Observer' that schedules+-- all observations to an effectful callback.+--+-- @since 0.3.0.0+spawnLoggingObserver+  :: forall o r q .+  ( SetMember Process (Process q) r+  , Typeable o+  , Show (Observation o)+  , Observable o+  , Member (Logs String) q)+  => SchedulerProxy q+  -> Eff r (Server (CallbackObserver o))+spawnLoggingObserver px =+  spawnCallbackObserver px+  (\s o -> logMsg (show s ++ " OBSERVED: " ++ show o) >> return True)

@@ -19,7 +19,7 @@   (   -- * Api Server     serve-  , ApiHandler (..)+  , ApiHandler(..)   -- * ApiHandler default callbacks   , unhandledCallError   , unhandledCastError@@ -31,19 +31,21 @@   , UnhandledRequest()   , catchUnhandled   , ensureAllHandled-  , castMessage+  , requestFromDynamic   , exitUnhandled   ) where  import           Control.Eff import           Control.Eff.InternalExtra-import qualified Control.Eff.Exception as Exc+import qualified Control.Eff.Exception         as Exc import           Control.Eff.Concurrent.Api import           Control.Eff.Concurrent.Api.Internal import           Control.Eff.Concurrent.Process import           Data.Proxy-import           Data.Typeable (Typeable, typeRep)+import           Data.Typeable                  ( Typeable+                                                , typeRep+                                                ) import           Data.Dynamic import           GHC.Stack @@ -54,20 +56,18 @@   => SchedulerProxy q   -> ApiHandler p r   -> Eff r ()-serve px handlers =-  receiveLoop px-   $ \ mReq ->-       case mReq of-        Left Nothing ->-          applyApiHandler px handlers (Terminate Nothing)-        Left (Just reason) ->-          applyApiHandler px handlers (Terminate (Just reason))-        Right dyn ->-          ensureAllHandled px-            (do msg <- castMessage dyn-                raise (applyApiHandler px handlers msg))+serve px handlers = receiveLoop px $ \mReq -> case mReq of+  Left  Nothing       -> applyApiHandler px handlers (Terminate Nothing)+  Left  (Just reason) -> applyApiHandler px handlers (Terminate (Just reason))+  Right dyn           -> ensureAllHandled+    px+    (do+      msg <- requestFromDynamic dyn+      raise (applyApiHandler px handlers msg)+    ) --- | A record of callbacks requests to a /server/, serving a specific 'Api' family instance.+-- | A record of callbacks, handling requests sent to a /server/ 'Process', all+-- belonging to a specific 'Api' family instance. data ApiHandler p r where   ApiHandler ::      { -- | A cast will not return a result directly. This is used for async@@ -79,7 +79,7 @@       -- handler calls the reply continuation.      , _handleCall          :: forall x . HasCallStack-         => Api p ('Synchronous x) -> (x -> Eff r Bool) -> Eff r ()+         => Api p ('Synchronous x) -> (x -> Eff r ()) -> Eff r ()      -- | This callback is called with @Nothing@ if the process exits      -- peacefully, or @Just "error message..."@ if the process exits with an      -- error. This function is responsible to exit the process if necessary.@@ -91,124 +91,117 @@  -- | Apply either the '_handleCall', '_handleCast' or the '_handleTerminate' -- callback to an incoming 'Request'. Note, it is unlikely that this function must be used.-applyApiHandler :: forall r q p-                . ( Typeable p-                  , SetMember Process (Process q) r-                  , HasCallStack)-              => SchedulerProxy q-              -> ApiHandler p r-              -> Request p-              -> Eff r ()-applyApiHandler _px handlers (Terminate e) =-  _handleTerminate handlers e-applyApiHandler _ handlers (Cast request        ) =-  _handleCast handlers request-applyApiHandler px handlers (Call fromPid request) =-  _handleCall handlers request sendReply-      where-       sendReply :: Typeable x => x -> Eff r Bool-       sendReply reply =-         sendMessage px fromPid (toDyn (Response (Proxy @p) reply))+applyApiHandler+  :: forall r q p+   . (Typeable p, SetMember Process (Process q) r, HasCallStack)+  => SchedulerProxy q+  -> ApiHandler p r+  -> Request p+  -> Eff r ()+applyApiHandler _px handlers (Terminate e) = _handleTerminate handlers e+applyApiHandler _ handlers (Cast request) = _handleCast handlers request+applyApiHandler px handlers (Call fromPid request) = _handleCall handlers+                                                                 request+                                                                 sendReply+ where+  sendReply :: Typeable x => x -> Eff r ()+  sendReply reply =+    sendMessage px fromPid (toDyn $! (Response (Proxy @p) $! reply))  -- | A default handler to use in '_handleCall' in 'ApiHandler'. It will call -- 'raiseError' with a nice error message. unhandledCallError-  :: forall p x r q .-    ( Show (Api p ( 'Synchronous x))-    , Typeable p-    , HasCallStack-    , SetMember Process (Process q) r-    )+  :: forall p x r q+   . ( Show (Api p ( 'Synchronous x))+     , Typeable p+     , HasCallStack+     , SetMember Process (Process q) r+     )   => SchedulerProxy q   -> Api p ( 'Synchronous x)-  -> (x -> Eff r Bool)+  -> (x -> Eff r ())   -> Eff r ()-unhandledCallError px api _ = raiseError px-  ("Unhandled call: ("-    ++ show api-    ++ " :: "-    ++ show (typeRep (Proxy @p)) ++ ")")+unhandledCallError px api _ = raiseError+  px+  ("Unhandled call: (" ++ show api ++ " :: " ++ show (typeRep (Proxy @p)) ++ ")"+  )  -- | A default handler to use in '_handleCast' in 'ApiHandler'. It will call -- 'raiseError' with a nice error message. unhandledCastError-  :: forall p r q .-    ( Show (Api p 'Asynchronous)-    , Typeable p-    , HasCallStack-    , SetMember Process (Process q) r-    )+  :: forall p r q+   . ( Show (Api p 'Asynchronous)+     , Typeable p+     , HasCallStack+     , SetMember Process (Process q) r+     )   => SchedulerProxy q   -> Api p 'Asynchronous   -> Eff r ()-unhandledCastError px api = raiseError px-  ("Unhandled cast: ("-    ++ show api-    ++ " :: "-    ++ show (typeRep (Proxy @p)) ++ ")")+unhandledCastError px api = raiseError+  px+  ("Unhandled cast: (" ++ show api ++ " :: " ++ show (typeRep (Proxy @p)) ++ ")"+  )  -- | Exit the process either normally of the error message is @Nothing@ -- or with 'exitWithError' otherwise.-defaultTermination :: forall q r .-                     ( HasCallStack, SetMember Process (Process q) r )-                   => SchedulerProxy q-                   -> Maybe String-                   -> Eff r ()-defaultTermination px e =-   maybe (exitNormally px) (exitWithError px) e+defaultTermination+  :: forall q r+   . (HasCallStack, SetMember Process (Process q) r)+  => SchedulerProxy q+  -> Maybe String+  -> Eff r ()+defaultTermination px e = maybe (exitNormally px) (exitWithError px) e   -- | 'serve' two 'ApiHandler's at once. The first handler is used for -- termination handling.-serveBoth :: forall r q p1 p2-          . ( Typeable p1, Typeable p2-            , SetMember Process (Process q) r-            , HasCallStack)-       => SchedulerProxy q-       -> ApiHandler p1 r -> ApiHandler p2 r-       -> Eff r ()-serveBoth px h1 h2 =-  receiveLoop px-   $ \ mReq ->-       case mReq of-        Left Nothing ->-          applyApiHandler px h1 (Terminate Nothing)-        Left (Just reason) ->-          applyApiHandler px h1 (Terminate (Just reason))-        Right dyn ->-          ensureAllHandled px-            (tryApiHandler px h1 dyn-              `catchUnhandled`-              tryApiHandler px h2)+serveBoth+  :: forall r q p1 p2+   . ( Typeable p1+     , Typeable p2+     , SetMember Process (Process q) r+     , HasCallStack+     )+  => SchedulerProxy q+  -> ApiHandler p1 r+  -> ApiHandler p2 r+  -> Eff r ()+serveBoth px h1 h2 = receiveLoop px $ \mReq -> case mReq of+  Left  Nothing       -> applyApiHandler px h1 (Terminate Nothing)+  Left  (Just reason) -> applyApiHandler px h1 (Terminate (Just reason))+  Right dyn           -> ensureAllHandled+    px+    (tryApiHandler px h1 dyn `catchUnhandled` tryApiHandler px h2)  -- | 'serve' three 'ApiHandler's at once. The first handler is used for -- termination handling.-serve3 :: forall r q p1 p2 p3-          . ( Typeable p1, Typeable p2, Typeable p3-            , SetMember Process (Process q) r-            , HasCallStack)-       => SchedulerProxy q-       -> ApiHandler p1 r -> ApiHandler p2 r -> ApiHandler p3 r-       -> Eff r ()-serve3 px h1 h2 h3 =-  receiveLoop px-   $ \ mReq ->-       case mReq of-        Left Nothing ->-          applyApiHandler px h1 (Terminate Nothing)-        Left (Just reason) ->-          applyApiHandler px h1 (Terminate (Just reason))-        Right dyn ->-          ensureAllHandled px-            (tryApiHandler px h1 dyn-              `catchUnhandled`-              tryApiHandler px h2-              `catchUnhandled`-              tryApiHandler px h3)+serve3+  :: forall r q p1 p2 p3+   . ( Typeable p1+     , Typeable p2+     , Typeable p3+     , SetMember Process (Process q) r+     , HasCallStack+     )+  => SchedulerProxy q+  -> ApiHandler p1 r+  -> ApiHandler p2 r+  -> ApiHandler p3 r+  -> Eff r ()+serve3 px h1 h2 h3 = receiveLoop px $ \mReq -> case mReq of+  Left  Nothing       -> applyApiHandler px h1 (Terminate Nothing)+  Left  (Just reason) -> applyApiHandler px h1 (Terminate (Just reason))+  Right dyn           -> ensureAllHandled+    px+    (                tryApiHandler px h1 dyn+    `catchUnhandled` tryApiHandler px h2+    `catchUnhandled` tryApiHandler px h3+    )  -- | The basic building block of the combination of 'ApiHandler's is this -- function, which can not only be passed to 'receiveLoop', but also throws an--- 'UnhandledRequest' exception if 'castMessage' failed, such that multiple+-- 'UnhandledRequest' exception if 'requestFromDynamic' failed, such that multiple -- invokation of this function for different 'ApiHandler's can be tried, by using 'catchUnhandled'. -- -- > tryApiHandler px handlers1 message@@ -217,34 +210,32 @@ -- >   `catchUnhandled` -- > tryApiHandler px handlers3 ---tryApiHandler  :: forall r q p-                . ( Typeable p-                  , SetMember Process (Process q) r-                  , HasCallStack)-              => SchedulerProxy q-              -> ApiHandler p r-              -> Dynamic-              -> Eff (Exc.Exc UnhandledRequest ': r) ()-tryApiHandler px handlers message =-  do request <- castMessage message-     raise (applyApiHandler px handlers request)+tryApiHandler+  :: forall r q p+   . (Typeable p, SetMember Process (Process q) r, HasCallStack)+  => SchedulerProxy q+  -> ApiHandler p r+  -> Dynamic+  -> Eff (Exc.Exc UnhandledRequest ': r) ()+tryApiHandler px handlers message = do+  request <- requestFromDynamic message+  raise (applyApiHandler px handlers request)  -- | An exception that is used by the mechanism that chains together multiple -- different 'ApiHandler' allowing a single process to implement multiple--- 'Api's. This exception is thrown by 'castMessage'. This+-- 'Api's. This exception is thrown by 'requestFromDynamic'. This -- exception can be caught with 'catchUnhandled', this way, several -- distinct 'ApiHandler' can be tried until one /fits/ or until the -- 'exitUnhandled' is invoked. newtype UnhandledRequest = UnhandledRequest { fromUnhandledRequest :: Dynamic } --- | If 'castMessage' failes to cast the message to a 'Request' for a+-- | If 'requestFromDynamic' failes to cast the message to a 'Request' for a -- certain 'ApiHandler' it throws an 'UnhandledRequest' exception. That -- exception is caught by this function and the raw message is given to the -- handler function. This is the basis for chaining 'ApiHandler's. catchUnhandled-  :: forall r a .-    ( Member (Exc.Exc UnhandledRequest) r-    , HasCallStack)+  :: forall r a+   . (Member (Exc.Exc UnhandledRequest) r, HasCallStack)   => Eff r a   -> (Dynamic -> Eff r a)   -> Eff r a@@ -253,36 +244,34 @@  -- | Catch 'UnhandledRequest's and terminate the process with an error, if necessary. ensureAllHandled-  :: forall r q .-    (HasCallStack, SetMember Process (Process q) r)+  :: forall r q+   . (HasCallStack, SetMember Process (Process q) r)   => SchedulerProxy q   -> Eff (Exc.Exc UnhandledRequest ': r) ()   -> Eff r ()-ensureAllHandled px effect =-  do result <- Exc.runError effect-     either (exitUnhandled px . fromUnhandledRequest) return result+ensureAllHandled px effect = do+  result <- Exc.runError effect+  either (exitUnhandled px . fromUnhandledRequest) return result  -- | Cast a 'Dynamic' value, and if that fails, throw an 'UnhandledRequest' -- error.-castMessage :: forall r a .-              ( HasCallStack-              , Typeable a-              , Member (Exc.Exc UnhandledRequest) r)-            => Dynamic -> Eff r a-castMessage message =-  maybe-  (Exc.throwError (UnhandledRequest message))-  return-  (fromDynamic message)+requestFromDynamic+  :: forall r a+   . (HasCallStack, Typeable a, Member (Exc.Exc UnhandledRequest) r)+  => Dynamic+  -> Eff r a+requestFromDynamic message =+  maybe (Exc.throwError (UnhandledRequest message)) return (fromDynamic message)  -- | If an incoming message could not be casted to a 'Request' corresponding to--- an 'ApiHandler' (e.g. with 'castMessage') one should use this function to+-- an 'ApiHandler' (e.g. with 'requestFromDynamic') one should use this function to -- exit the process with a corresponding error. exitUnhandled-  :: forall r q . (SetMember Process (Process q) r, HasCallStack)+  :: forall r q+   . (SetMember Process (Process q) r, HasCallStack)   => SchedulerProxy q   -> Dynamic   -> Eff r ()-exitUnhandled px message =-  do let reason = "unhandled message: " ++ show message-     exitWithError px reason+exitUnhandled px message = do+  let reason = "unhandled message: " ++ show message+  exitWithError px reason

@@ -108,7 +108,7 @@     handleCast (Shout x) = do       me <- self px       logMsg (show me ++ " Shouting: " ++ x)-    handleCall :: Api TestApi ('Synchronous x) -> (x -> Eff (Process r ': r) Bool) -> Eff (Process r ': r) ()+    handleCall :: Api TestApi ('Synchronous x) -> (x -> Eff (Process r ': r) ()) -> Eff (Process r ': r) ()     handleCall (SayHello "e1") _reply = do       me <- self px       logMsg (show me ++ " raising an error")

@@ -87,11 +87,11 @@      modify (+ (1 :: Integer))      currentCount <- get      notifyObservers px (CountChanged currentCount)-   handleCall :: Api Counter ('Synchronous x) -> (x -> Eff r Bool) -> Eff r ()+   handleCall :: Api Counter ('Synchronous x) -> (x -> Eff r ()) -> Eff r ()    handleCall Cnt reply = do      c <- get      logMsg ("Cnt is " ++ show c)-     _ <- reply c+     reply c      return ()  -- * Second API@@ -119,13 +119,13 @@      modify (+ x)      c <- get @Integer      logMsg ("Accumulator is " ++ show c)-   handleCall :: Api PocketCalc ('Synchronous x) -> (x -> Eff r Bool) -> Eff r ()+   handleCall :: Api PocketCalc ('Synchronous x) -> (x -> Eff r ()) -> Eff r ()    handleCall (Add x) reply = do      logMsg ("Add " ++ show x)      modify (+ x)      c <- get      logMsg ("Accumulator is " ++ show c)-     void (reply c)+     reply c  serverLoop :: forall r q .                  ( Member (Logs String) r

@@ -3,8 +3,13 @@ -- This module describes an abstract message passing effect, and a process -- effect, mimicking Erlang's process and message semantics. ----- An implementation of a handler for these effects can be found in--- 'Control.Eff.Concurrent.Process.ForkIOScheduler'.+-- Two __scheduler__ implementations for the 'Process' effect are provided:+--+--  * A scheduler using @forkIO@, i.e. relying on the multi threaded GHC runtime:+--    "Control.Eff.Concurrent.Process.ForkIOScheduler"+--+--  * And a /pure/(rer) coroutine based scheduler in:+--    "Control.Eff.Concurrent.Process.SingleThreadedScheduler" {-# LANGUAGE DeriveTraversable #-} {-# LANGUAGE DeriveFoldable #-} {-# LANGUAGE GeneralizedNewtypeDeriving #-}@@ -30,14 +35,19 @@   , ResumeProcess(..)   , SchedulerProxy(..)   , thisSchedulerProxy-  , yieldAndCatchProcess+  , executeAndCatch+  , yieldProcess   , sendMessage+  , sendMessageAs+  , sendMessageChecked   , spawn+  , spawn_   , receiveMessage   , receiveMessageAs   , receiveLoop   , self   , sendShutdown+  , sendShutdownChecked   , exitWithError   , exitNormally   , raiseError@@ -49,7 +59,9 @@ import           GHC.Stack import           Control.Eff import           Control.Lens-import           Control.Monad (forever)+import           Control.Monad                  ( forever+                                                , void+                                                ) import           Data.Dynamic import           Data.Kind import           Text.Printf@@ -81,6 +93,9 @@ -- -- * when the first process exists, all process should be killed immediately data Process (r :: [Type -> Type]) b where+  -- | In cooperative schedulers, this will give processing time to the+  -- scheduler. Every other operation implicitly serves the same purpose.+  YieldProcess :: Process r (ResumeProcess ())   -- | Return the current 'ProcessId'   SelfPid :: Process r (ResumeProcess ProcessId)   -- | Start a new process, the new process will execute an effect, the function@@ -131,43 +146,57 @@  -- | Execute a 'Process' action and resume the process, retry the action or exit -- the process when a shutdown was requested.-yieldProcess :: forall r q v .-               ( SetMember Process (Process q) r-               , HasCallStack)-             => Process q (ResumeProcess v)-             -> Eff r v-yieldProcess processAction =-  do result <- send processAction-     case result of-       ResumeWith value -> return value-       RetryLastAction -> yieldProcess processAction-       ShutdownRequested -> send (Shutdown @q)-       OnError e -> send (ExitWithError @q e)+executeAndResume+  :: forall r q v+   . (SetMember Process (Process q) r, HasCallStack)+  => Process q (ResumeProcess v)+  -> Eff r v+executeAndResume processAction = do+  result <- send processAction+  case result of+    ResumeWith value  -> return value+    RetryLastAction   -> executeAndResume processAction+    ShutdownRequested -> send (Shutdown @q)+    OnError e         -> send (ExitWithError @q e)  -- | Execute a and action and resume the process, retry the action, shutdown the process or return an error.-yieldAndCatchProcess :: forall q r v.-                       (SetMember Process (Process q) r, HasCallStack)-                     => SchedulerProxy q-                     -> Eff r (ResumeProcess v)-                     -> Eff r (Either String v)-yieldAndCatchProcess px processAction =-  do result <- processAction-     case result of-       ResumeWith value -> return (Right value)-       RetryLastAction -> yieldAndCatchProcess px processAction-       ShutdownRequested -> send (Shutdown @q)-       OnError e -> return (Left e)+executeAndCatch+  :: forall q r v+   . (SetMember Process (Process q) r, HasCallStack)+  => SchedulerProxy q+  -> Eff r (ResumeProcess v)+  -> Eff r (Either String v)+executeAndCatch px processAction = do+  result <- processAction+  case result of+    ResumeWith value  -> return (Right value)+    RetryLastAction   -> executeAndCatch px processAction+    ShutdownRequested -> send (Shutdown @q)+    OnError e         -> return (Left e) --- | Every function for 'Process' things needs proxy for the low-level--- effect list depending on the scheduler implementation.--- I don't know a smarter way yet to do this.+-- | Every function for 'Process' things needs such a proxy value+-- for the low-level effect list, i.e. the effects identified by+-- @__r__@ in @'Process' r : r@, this might be dependent on the+-- scheduler implementation. data SchedulerProxy :: [Type -> Type] -> Type where+  -- | Tell the typechecker what effects we have below 'Process'   SchedulerProxy :: SchedulerProxy q+  -- | Like 'SchedulerProxy' but shorter+  SP :: SchedulerProxy q  -- | Return a 'SchedulerProxy' for a 'Process' effect. thisSchedulerProxy :: Eff (Process r ': r) (SchedulerProxy r) thisSchedulerProxy = return SchedulerProxy +-- | Use 'executeAndResume' to execute 'YieldProcess'. Refer to 'YieldProcess'+-- for more information.+yieldProcess+  :: forall r q+   . (HasCallStack, SetMember Process (Process q) r)+  => SchedulerProxy q+  -> Eff r ()+yieldProcess _ = executeAndResume YieldProcess+ -- | Send a message to a process addressed by the 'ProcessId'. -- See 'SendMessage'. sendMessage@@ -176,96 +205,150 @@   => SchedulerProxy q   -> ProcessId   -> Dynamic+  -> Eff r ()+sendMessage px pid message = void (sendMessageChecked px pid message)++-- | Send a message to a process addressed by the 'ProcessId'.+-- See 'SendMessage'. Return @True@ if the process existed.+-- I you don't care, just 'sendMessage' instead.+sendMessageChecked+  :: forall r q+   . (HasCallStack, SetMember Process (Process q) r)+  => SchedulerProxy q+  -> ProcessId+  -> Dynamic   -> Eff r Bool-sendMessage _ pid message =-  yieldProcess (SendMessage pid message)+sendMessageChecked _ pid message =+  executeAndResume (SendMessage pid $! message)  -- | Send a message to a process addressed by the 'ProcessId'. -- See 'SendMessage'.+sendMessageAs+  :: forall o r q+   . (HasCallStack, SetMember Process (Process q) r, Typeable o)+  => SchedulerProxy q+  -> ProcessId+  -> o+  -> Eff r ()+sendMessageAs px pid = sendMessage px pid . toDyn++-- | Exit a process addressed by the 'ProcessId'.+-- See 'SendShutdown'. sendShutdown   :: forall r q    . (HasCallStack, SetMember Process (Process q) r)   => SchedulerProxy q   -> ProcessId+  -> Eff r ()+sendShutdown px pid = void (sendShutdownChecked px pid)++-- | Like 'sendShutdown', but also return @True@ iff the process to exit exists.+sendShutdownChecked+  :: forall r q+   . (HasCallStack, SetMember Process (Process q) r)+  => SchedulerProxy q+  -> ProcessId   -> Eff r Bool-sendShutdown _ pid =-  yieldProcess (SendShutdown pid)+sendShutdownChecked _ pid = executeAndResume (SendShutdown pid)  -- | Start a new process, the new process will execute an effect, the function -- will return immediately with a 'ProcessId'.-spawn :: forall r q .-        (HasCallStack, SetMember Process (Process q) r)-      => Eff (Process q ': q) ()-      -> Eff r ProcessId-spawn child =-  yieldProcess (Spawn @q child)+spawn+  :: forall r q+   . (HasCallStack, SetMember Process (Process q) r)+  => Eff (Process q ': q) ()+  -> Eff r ProcessId+spawn child = executeAndResume (Spawn @q child) +-- | Like 'spawn' but return @()@.+spawn_+  :: forall r q+   . (HasCallStack, SetMember Process (Process q) r)+  => Eff (Process q ': q) ()+  -> Eff r ()+spawn_ = void . spawn+ -- | Block until a message was received. receiveMessage   :: forall r q    . (HasCallStack, SetMember Process (Process q) r)-  => SchedulerProxy q -> Eff r Dynamic+  => SchedulerProxy q+  -> Eff r Dynamic receiveMessage _ = do-  yieldProcess ReceiveMessage+  executeAndResume ReceiveMessage  -- | Receive and cast the message to some 'Typeable' instance. receiveMessageAs-  :: forall a r q .-    (HasCallStack, Typeable a, SetMember Process (Process q) r)-  => SchedulerProxy q -> Eff r a-receiveMessageAs px =-  do messageDynamic <- receiveMessage px-     let castAndCheck dm =-          case fromDynamic dm of-           Nothing ->-             Left ("Invalid message type received: " ++ show dm)-           Just m ->-             Right m-         maybeMessage = castAndCheck messageDynamic-     either (raiseError px) return maybeMessage+  :: forall a r q+   . (HasCallStack, Typeable a, SetMember Process (Process q) r)+  => SchedulerProxy q+  -> Eff r a+receiveMessageAs px = do+  messageDynamic <- receiveMessage px+  let castAndCheck dm = case fromDynamic dm of+        Nothing -> Left ("Invalid message type received: " ++ show dm)+        Just m  -> Right m+      maybeMessage = castAndCheck messageDynamic+  either (raiseError px) return maybeMessage  -- | Enter a loop to receive messages and pass them to a callback, until the -- function returns 'False'. receiveLoop   :: forall r q-   . ( SetMember Process (Process q) r-     , HasCallStack)+   . (SetMember Process (Process q) r, HasCallStack)   => SchedulerProxy q   -> (Either (Maybe String) Dynamic -> Eff r ())   -> Eff r ()-receiveLoop _ handlers =-  forever $-  do mReq <- send (ReceiveMessage @q)-     case mReq of-       RetryLastAction ->-         return ()-       ShutdownRequested ->-         handlers (Left Nothing)-       OnError reason ->-         handlers (Left (Just reason))-       ResumeWith message ->-         handlers (Right message)+receiveLoop _ handlers = forever $ do+  mReq <- send (ReceiveMessage @q)+  case mReq of+    RetryLastAction    -> return ()+    ShutdownRequested  -> handlers (Left Nothing)+    OnError    reason  -> handlers (Left (Just reason))+    ResumeWith message -> handlers (Right message)  -- | Returns the 'ProcessId' of the current process.-self :: (HasCallStack, SetMember Process (Process q) r) => SchedulerProxy q -> Eff r ProcessId-self _px = yieldProcess SelfPid+self+  :: (HasCallStack, SetMember Process (Process q) r)+  => SchedulerProxy q+  -> Eff r ProcessId+self _px = executeAndResume SelfPid  -- | Exit the process.-exitNormally :: forall r q a. (HasCallStack, SetMember Process (Process q) r) => SchedulerProxy q -> Eff r a+exitNormally+  :: forall r q a+   . (HasCallStack, SetMember Process (Process q) r)+  => SchedulerProxy q+  -> Eff r a exitNormally _ = send (Shutdown @q)  -- | Exit the process with an error.-exitWithError :: forall r q a. (HasCallStack, SetMember Process (Process q) r) => SchedulerProxy q -> String -> Eff r a-exitWithError _ = send . ExitWithError @q+exitWithError+  :: forall r q a+   . (HasCallStack, SetMember Process (Process q) r)+  => SchedulerProxy q+  -> String+  -> Eff r a+exitWithError _ = send . (ExitWithError @q $!)  -- | Thrown an error, can be caught by 'catchRaisedError'.-raiseError :: forall r q b. (HasCallStack, SetMember Process (Process q) r) => SchedulerProxy q -> String -> Eff r b-raiseError _ = send . RaiseError @q+raiseError+  :: forall r q b+   . (HasCallStack, SetMember Process (Process q) r)+  => SchedulerProxy q+  -> String+  -> Eff r b+raiseError _ = send . (RaiseError @q $!)  -- | Catch and handle an error raised by 'raiseError'. Works independent of the -- handler implementation. catchRaisedError-  :: forall r q w . (HasCallStack, SetMember Process (Process q) r) => SchedulerProxy q -> (String -> Eff r w) -> Eff r w -> Eff r w+  :: forall r q w+   . (HasCallStack, SetMember Process (Process q) r)+  => SchedulerProxy q+  -> (String -> Eff r w)+  -> Eff r w+  -> Eff r w catchRaisedError _ onErr = interpose return go  where   go :: forall b . Process q b -> (b -> Eff r w) -> Eff r w@@ -275,7 +358,10 @@ -- | Like 'catchRaisedError' it catches 'raiseError', but instead of invoking a -- user provided handler, the result is wrapped into an 'Either'. ignoreProcessError-  :: (HasCallStack, SetMember Process (Process q) r) => SchedulerProxy q -> Eff r a -> Eff r (Either String a)+  :: (HasCallStack, SetMember Process (Process q) r)+  => SchedulerProxy q+  -> Eff r a+  -> Eff r (Either String a) ignoreProcessError px = catchRaisedError px (return . Left) . fmap Right  -- | Each process is identified by a single process id, that stays constant

@@ -15,7 +15,6 @@ -- -- 'spawn' uses 'forkFinally' and 'STM.TQueue's and tries to catch -- most exceptions.-{-# LANGUAGE GeneralizedNewtypeDeriving #-} {-# LANGUAGE ConstraintKinds #-} {-# LANGUAGE FlexibleContexts #-} {-# LANGUAGE FlexibleInstances #-}@@ -46,7 +45,7 @@ import           GHC.Stack import           Data.Bifunctor import           Data.Maybe-import           Data.Kind ()+import           Data.Kind                      ( ) import qualified Control.Exception             as Exc import           Control.Concurrent            as Concurrent import           Control.Concurrent.STM        as STM@@ -142,145 +141,144 @@ -- application. This function takes a 'Process' on top of the 'SchedulerIO' -- effect and a 'LogChannel' for concurrent logging. schedule :: Eff (ConsProcess SchedulerIO) () -> LogChannel String -> IO ()-schedule e logC =-  void-  $ withNewSchedulerState-  $ \schedulerStateVar ->-      do pidVar <- newEmptyTMVarIO-         scheduleProcessWithShutdownAction schedulerStateVar pidVar-           $ do mt <- lift myThreadId-                mp <- lift (atomically (readTMVar pidVar))-                logMsg (show mp ++ " main process started in thread " ++ show mt)-                e-                logMsg (show mp ++ " main process returned")-  where-    withNewSchedulerState :: (SchedulerVar -> IO a) -> IO a-    withNewSchedulerState mainProcessAction = do-      myTId <- myThreadId-      Exc.bracket-        (newTVarIO (Scheduler myPid Map.empty Map.empty False logC))-        (tearDownScheduler myTId)-        (mainProcessAction . SchedulerVar)--     where-      myPid = 1-      tearDownScheduler myTId v = do-        logChannelPutIO logC (show myTId ++" begin scheduler tear down")-        sch <--          (atomically-            (do-              sch <- readTVar v-              let sch' = sch & schedulerShuttingDown .~ True-              writeTVar v sch'-              return sch'-            )-          )-        logChannelPutIO logC (show myTId ++ " killing " ++-                                   let ts = (sch ^.. threadIdTable. traversed)-                                   in if length ts > 100-                                      then show (length ts) ++ " threads"-                                      else show ts )-        imapM_ (killProcThread myTId) (sch ^. threadIdTable)-        Concurrent.yield-        atomically+schedule e logC = void $ withNewSchedulerState $ \schedulerStateVar -> do+  pidVar <- newEmptyTMVarIO+  scheduleProcessWithShutdownAction schedulerStateVar pidVar $ do+    mt <- lift myThreadId+    mp <- lift (atomically (readTMVar pidVar))+    logMsg (show mp ++ " main process started in thread " ++ show mt)+    e+    logMsg (show mp ++ " main process returned")+ where+  withNewSchedulerState :: (SchedulerVar -> IO a) -> IO a+  withNewSchedulerState mainProcessAction = do+    myTId <- myThreadId+    Exc.bracket (newTVarIO (Scheduler myPid Map.empty Map.empty False logC))+                (tearDownScheduler myTId)+                (mainProcessAction . SchedulerVar)+   where+    myPid = 1+    tearDownScheduler myTId v = do+      logChannelPutIO logC (show myTId ++ " begin scheduler tear down")+      sch <-+        (atomically           (do-              scheduler <- readTVar v-              let allThreadsDead = scheduler^.threadIdTable.to Map.null-                                   && scheduler^.processTable.to Map.null-              STM.check allThreadsDead)-        logChannelPutIO logC "all threads dead"+            sch <- readTVar v+            let sch' = sch & schedulerShuttingDown .~ True+            writeTVar v sch'+            return sch'+          )+        )+      logChannelPutIO+        logC+        (  show myTId+        ++ " killing "+        ++ let ts = (sch ^.. threadIdTable . traversed)+           in  if length ts > 100+                 then show (length ts) ++ " threads"+                 else show ts+        )+      imapM_ (killProcThread myTId) (sch ^. threadIdTable)+      Concurrent.yield+      atomically+        (do+          scheduler <- readTVar v+          let allThreadsDead =+                scheduler+                  ^. threadIdTable+                  .  to Map.null+                  && scheduler+                  ^. processTable+                  .  to Map.null+          STM.check allThreadsDead+        )+      logChannelPutIO logC "all threads dead"  -      killProcThread myTId _pid tid =-        when (myTId /= tid) ( killThread tid)+    killProcThread myTId _pid tid = when (myTId /= tid) (killThread tid)  -- | Start the message passing concurrency system then execute a 'Process' on -- top of 'SchedulerIO' effect. All logging is sent to standard output. defaultMain :: Eff (ConsProcess SchedulerIO) () -> IO ()-defaultMain c =-  runLoggingT-    (logChannelBracket-      128-      (Just "~~~~~~ main process started")-      (Just "====== main process exited")-      (schedule c))-    (print :: String -> IO ())+defaultMain c = runLoggingT+  (logChannelBracket 128+                     (Just "~~~~~~ main process started")+                     (Just "====== main process exited")+                     (schedule c)+  )+  (print :: String -> IO ())  -- | Start the message passing concurrency system then execute a 'Process' on -- top of 'SchedulerIO' effect. All logging is sent to standard output.-defaultMainWithLogChannel :: LogChannel String -> Eff (ConsProcess SchedulerIO) () -> IO ()-defaultMainWithLogChannel logC c =-  closeLogChannelAfter-      (Just (fromString "====== main process exited"))-      logC-      (schedule c logC)+defaultMainWithLogChannel+  :: LogChannel String -> Eff (ConsProcess SchedulerIO) () -> IO ()+defaultMainWithLogChannel logC c = closeLogChannelAfter+  (Just (fromString "====== main process exited"))+  logC+  (schedule c logC)  scheduleProcessWithShutdownAction   :: SchedulerVar   -> STM.TMVar ProcessId   -> Eff (ConsProcess SchedulerIO) ()   -> IO (Either Exc.SomeException ())-scheduleProcessWithShutdownAction schedulerVar pidVar procAction =-  do cleanupVar <- newEmptyTMVarIO-     logC <-  getLogChannelIO schedulerVar-     mTid <- myThreadId-     eeres <- Exc.try (runProcEffects cleanupVar logC)-     let eres = join eeres-     getAndExecCleanup cleanupVar eres logC mTid-     logChannelPutIO logC (show mTid ++ " <~< process cleanup finished")-     return eres-  where+scheduleProcessWithShutdownAction schedulerVar pidVar procAction = do+  cleanupVar <- newEmptyTMVarIO+  logC       <- getLogChannelIO schedulerVar+  mTid       <- myThreadId+  eeres      <- Exc.try (runProcEffects cleanupVar logC)+  let eres = join eeres+  getAndExecCleanup cleanupVar eres logC mTid+  logChannelPutIO logC (show mTid ++ " <~< process cleanup finished")+  return eres+ where -    runProcEffects cleanupVar l =-      Data.Bifunctor.first Exc.SomeException-       <$> runLift-           (logToChannel l-            (runReader-              (scheduleProcessWithCleanup-                shutdownAction-                saveCleanupAndSchedule)-              schedulerVar))-      where-        shutdownAction =-          ShutdownAction-          (\eres ->-             do let ex = either id (const ProcessShuttingDown) eres-                Exc.throw ex)-        saveCleanupAndSchedule cleanUpAction pid =-          do lift (atomically (do STM.putTMVar cleanupVar cleanUpAction-                                  STM.putTMVar pidVar pid))-             mTid <- lift myThreadId-             logMsg (show mTid ++ " >~> begin process " ++ show pid)-             procAction-    getAndExecCleanup cleanupVar eres lc mt =-      do mcleanup <- atomically (STM.tryTakeTMVar cleanupVar)-         traverse_ execCleanup mcleanup-      where-        execCleanup ca =-          do runCleanUpAction ca-             logChannelPutIO lc-               $ show mt ++-               case eres of-                 Left se ->-                   case Exc.fromException se of-                     Nothing ->-                       " process caught exception: "-                       ++ Exc.displayException se-                     Just schedulerErr ->-                       (case schedulerErr of-                         ProcessShuttingDown ->-                           " process shutdown"-                         ProcessExitError m ->-                           " process exited with error: " ++ show m-                         ProcessRaisedError m ->-                           " process raised error: " ++ show m-                         _ ->-                           " scheduler error: " ++ show schedulerErr)-                       ++ " - full exception message: "-                       ++ Exc.displayException se+  runProcEffects cleanupVar l =+    Data.Bifunctor.first Exc.SomeException <$> runLift+      (logToChannel+        l+        (runReader+          (scheduleProcessWithCleanup shutdownAction saveCleanupAndSchedule)+          schedulerVar+        )+      )+   where+    shutdownAction = ShutdownAction+      (\eres -> do+        let ex = either id (const ProcessShuttingDown) eres+        Exc.throw ex+      )+    saveCleanupAndSchedule cleanUpAction pid = do+      lift+        (atomically+          (do+            STM.putTMVar cleanupVar cleanUpAction+            STM.putTMVar pidVar pid+          )+        )+      mTid <- lift myThreadId+      logMsg (show mTid ++ " >~> begin process " ++ show pid)+      procAction+  getAndExecCleanup cleanupVar eres lc mt = do+    mcleanup <- atomically (STM.tryTakeTMVar cleanupVar)+    traverse_ execCleanup mcleanup+   where+    execCleanup ca = do+      runCleanUpAction ca+      logChannelPutIO lc $ show mt ++ case eres of+        Left se -> case Exc.fromException se of+          Nothing -> " process caught exception: " ++ Exc.displayException se+          Just schedulerErr ->+            (case schedulerErr of+                ProcessShuttingDown  -> " process shutdown"+                ProcessExitError   m -> " process exited with error: " ++ show m+                ProcessRaisedError m -> " process raised error: " ++ show m+                _ -> " scheduler error: " ++ show schedulerErr+              )+              ++ " - full exception message: "+              ++ Exc.displayException se -                 Right _ ->-                   " process function returned"+        Right _ -> " process function returned"   getLogChannel :: HasSchedulerIO r => Eff r (LogChannel String)@@ -297,8 +295,7 @@   => ProcessId   -> Mtl.StateT ProcessInfo STM.STM a   -> Eff r (Either SchedulerError a)-overProcessInfo pid stAction =-  overScheduler+overProcessInfo pid stAction = overScheduler   (do     res <- use (processTable . at pid)     case res of@@ -306,24 +303,22 @@       Just pinfo -> do         (x, pinfoOut) <- Mtl.lift (Mtl.runStateT stAction pinfo)         processTable . at pid . _Just .= pinfoOut-        return (Right x))+        return (Right x)+  )  -- ** MessagePassing execution -spawnImpl :: HasCallStack-          => Eff (ConsProcess SchedulerIO) ()-          -> Eff SchedulerIO ProcessId+spawnImpl+  :: HasCallStack+  => Eff (ConsProcess SchedulerIO) ()+  -> Eff SchedulerIO ProcessId spawnImpl mfa = do   schedulerVar <- ask-  pidVar <- lift STM.newEmptyTMVarIO-  void-    $ lift-    $ Concurrent.forkIO-    $ void-    $ scheduleProcessWithShutdownAction-       schedulerVar-       pidVar-       mfa+  pidVar       <- lift STM.newEmptyTMVarIO+  void $ lift $ Concurrent.forkIO $ void $ scheduleProcessWithShutdownAction+    schedulerVar+    pidVar+    mfa   lift Concurrent.yield -- this is important, removing this causes test failures   lift (atomically (STM.readTMVar pidVar)) @@ -334,117 +329,113 @@   => ShutdownAction   -> (CleanUpAction -> ProcessId -> Eff (ConsProcess SchedulerIO) ())   -> Eff SchedulerIO (Either SchedulerError ())-scheduleProcessWithCleanup shutdownAction processAction =-  withMessageQueue-  (\cleanUpAction pinfo ->-     handle_relay-       (\x -> return x)-       (go (pinfo ^. processId))-       (processAction cleanUpAction (pinfo ^. processId)))+scheduleProcessWithCleanup shutdownAction processAction = withMessageQueue+  (\cleanUpAction pinfo -> handle_relay+    (\x -> return x)+    (go (pinfo ^. processId))+    (processAction cleanUpAction (pinfo ^. processId))+  )  where-  shutdownOrGo :: forall v a+  shutdownOrGo+    :: forall v a      . HasCallStack     => ProcessId     -> (ResumeProcess v -> Eff SchedulerIO a)     -> Eff SchedulerIO a     -> Eff SchedulerIO a-  shutdownOrGo pid k ok =-    do psVar <- getSchedulerTVar-       eHasShutdowReq <--         lift-           (do p <- atomically (readTVar psVar)-               let mPinfo = p ^. processTable . at pid-               case mPinfo of-                 Just pinfo ->-                   atomically-                     (do wasRequested <- readTVar (pinfo ^. shutdownRequested)-                         -- reset the shutdwown request flag, in case the shutdown-                         -- is prevented by the process-                         writeTVar (pinfo ^. shutdownRequested) False-                         return (Right wasRequested))-                 Nothing ->-                   return (Left (ProcessNotFound pid)))-       case eHasShutdowReq of-         Right True ->-           k ShutdownRequested-         Right False ->-           ok-         Left e ->-           k (OnError (show e))+  shutdownOrGo pid k ok = do+    psVar          <- getSchedulerTVar+    eHasShutdowReq <- lift+      (do+        p <- atomically (readTVar psVar)+        let mPinfo = p ^. processTable . at pid+        case mPinfo of+          Just pinfo -> atomically+            (do+              wasRequested <- readTVar (pinfo ^. shutdownRequested)+              -- reset the shutdwown request flag, in case the shutdown+              -- is prevented by the process+              writeTVar (pinfo ^. shutdownRequested) False+              return (Right wasRequested)+            )+          Nothing -> return (Left (ProcessNotFound pid))+      )+    case eHasShutdowReq of+      Right True  -> k ShutdownRequested+      Right False -> ok+      Left  e     -> k (OnError (show e)) -  go :: forall v a+  go+    :: forall v a      . HasCallStack     => ProcessId     -> Process SchedulerIO v     -> (v -> Eff SchedulerIO a)     -> Eff SchedulerIO a-  go pid (SendMessage toPid reqIn) k =-      shutdownOrGo pid k $-      do eres <--           do psVar <- getSchedulerTVar-              lift-                (Right <$>-                  (do p <- atomically (readTVar psVar)-                      let mto = p ^. processTable . at toPid-                      case mto of-                          Just toProc ->-                            do atomically-                                (writeTQueue-                                 (toProc ^. messageQ)-                                 (Just reqIn))-                               return True-                          Nothing ->-                            return False))-         lift Concurrent.yield-         let kArg = either (OnError . show @SchedulerError) ResumeWith eres-         k kArg+  go pid (SendMessage toPid reqIn) k = shutdownOrGo pid k $ do+    eres <- do+      psVar <- getSchedulerTVar+      lift+        (   Right+        <$> (do+              p <- atomically (readTVar psVar)+              let mto = p ^. processTable . at toPid+              case mto of+                Just toProc -> do+                  atomically (writeTQueue (toProc ^. messageQ) (Just reqIn))+                  return True+                Nothing -> return False+            )+        )+    lift Concurrent.yield+    let kArg = either (OnError . show @SchedulerError) ResumeWith eres+    k kArg -  go pid (SendShutdown toPid) k =-    shutdownOrGo pid k-    $ do eres <--           do psVar <- getSchedulerTVar-              lift-                (Right <$>-                  (do p <- atomically (readTVar psVar)-                      let mto = p ^. processTable . at toPid-                      case mto of-                        Just toProc ->-                          atomically $-                           do writeTVar (toProc ^. shutdownRequested) True-                              writeTQueue (toProc ^. messageQ) Nothing-                              return True-                        Nothing ->-                          return False))-         let kArg = either (OnError . show @SchedulerError) resume eres-             resume = if toPid == pid-                      then const ShutdownRequested-                      else ResumeWith-         lift Concurrent.yield-         k kArg+  go pid (SendShutdown toPid) k = shutdownOrGo pid k $ do+    eres <- do+      psVar <- getSchedulerTVar+      lift+        (   Right+        <$> (do+              p <- atomically (readTVar psVar)+              let mto = p ^. processTable . at toPid+              case mto of+                Just toProc -> atomically $ do+                  writeTVar (toProc ^. shutdownRequested) True+                  writeTQueue (toProc ^. messageQ) Nothing+                  return True+                Nothing -> return False+            )+        )+    let kArg   = either (OnError . show @SchedulerError) resume eres+        resume = if toPid == pid then const ShutdownRequested else ResumeWith+    lift Concurrent.yield+    k kArg -  go pid (Spawn child) k =-    shutdownOrGo pid k $-    do res <- spawnImpl child-       k (ResumeWith res)+  go pid (Spawn child) k = shutdownOrGo pid k $ do+    res <- spawnImpl child+    k (ResumeWith res) -  go pid ReceiveMessage k =-    shutdownOrGo pid k $-      do emq <- overProcessInfo pid (use messageQ)-         case emq of-           Left e ->-             k (OnError (show @SchedulerError e))-           Right mq -> do-             emdynMsg <- lift (Right <$> (atomically (readTQueue mq)))-             k (either-                (OnError . show @SchedulerError)-                (maybe RetryLastAction ResumeWith)-                emdynMsg)+  go pid ReceiveMessage k = shutdownOrGo pid k $ do+    emq <- overProcessInfo pid (use messageQ)+    case emq of+      Left  e  -> k (OnError (show @SchedulerError e))+      Right mq -> do+        emdynMsg <- lift (Right <$> (atomically (readTQueue mq)))+        k+          (either (OnError . show @SchedulerError)+                  (maybe RetryLastAction ResumeWith)+                  emdynMsg+          ) -  go pid SelfPid k =-    shutdownOrGo pid k $-    do lift Concurrent.yield-       k (ResumeWith pid)+  go pid SelfPid k = shutdownOrGo pid k $ do+    lift Concurrent.yield+    k (ResumeWith pid) +  go pid YieldProcess k = shutdownOrGo pid k $ do+    lift Concurrent.yield+    k (ResumeWith ())+   go _pid Shutdown _k = do     invokeShutdownAction shutdownAction (Right ()) @@ -457,12 +448,12 @@ data ShutdownAction =   ShutdownAction (forall a . Either SchedulerError () -> IO a) -invokeShutdownAction :: ( HasCallStack-                       , SetMember Lift (Lift IO) r-                       , Member (Logs String) r)-                     => ShutdownAction -> Either SchedulerError () -> Eff r a-invokeShutdownAction (ShutdownAction a) res =-  lift (a res)+invokeShutdownAction+  :: (HasCallStack, SetMember Lift (Lift IO) r, Member (Logs String) r)+  => ShutdownAction+  -> Either SchedulerError ()+  -> Eff r a+invokeShutdownAction (ShutdownAction a) res = lift (a res)  withMessageQueue   :: HasSchedulerIO r@@ -478,8 +469,7 @@           lc           (pinfo ^. processId)       Right <$> m cleanUpAction pinfo-    Left e ->-      return $ Left e+    Left e -> return $ Left e  where   createQueue = do     myTId <- lift myThreadId@@ -489,8 +479,8 @@         if abortNow           then return (Left SchedulerShuttingDown)           else do-            pid     <- nextPid <<+= 1-            channel <- Mtl.lift newTQueue+            pid               <- nextPid <<+= 1+            channel           <- Mtl.lift newTQueue             shutdownIndicator <- Mtl.lift (newTVar False)             let pinfo = ProcessInfo pid channel shutdownIndicator             threadIdTable . at pid .= Just myTId@@ -501,9 +491,12 @@     didWork <- Exc.try       (overSchedulerIO         psVar-        (do os <- processTable . at pid <<.= Nothing-            ot <- threadIdTable . at pid <<.= Nothing-            return (os, isJust os || isJust ot)))+        (do+          os <- processTable . at pid <<.= Nothing+          ot <- threadIdTable . at pid <<.= Nothing+          return (os, isJust os || isJust ot)+        )+      )     let getCause =           Exc.try @Exc.SomeException               (overSchedulerIO psVar (preuse (processTable . at pid)))@@ -512,8 +505,7 @@                   (return . (maybe (show pid) show))      case didWork of-      Right (_pinfo, True) ->-        return ()+      Right (_pinfo, True) -> return ()       Right (pinfo, False) ->         logChannelPutIO lc ("queue already destroyed: " ++ show pinfo)       Left (e :: Exc.SomeException) ->@@ -528,17 +520,17 @@   -> Eff r (Either SchedulerError a) overScheduler stAction = do   psVar <- getSchedulerTVar-  -- liftCatch (Left . LowLevelIOException) (overSchedulerIO psVar stAction)   lift (overSchedulerIO psVar stAction) -overSchedulerIO-  :: STM.TVar Scheduler -> Mtl.StateT Scheduler STM.STM a -> IO a-overSchedulerIO psVar stAction =-  do STM.atomically-       (do ps <- STM.readTVar psVar-           (result, psModified) <- Mtl.runStateT stAction ps-           STM.writeTVar psVar psModified-           return result)+overSchedulerIO :: STM.TVar Scheduler -> Mtl.StateT Scheduler STM.STM a -> IO a+overSchedulerIO psVar stAction = do+  STM.atomically+    (do+      ps                   <- STM.readTVar psVar+      (result, psModified) <- Mtl.runStateT stAction ps+      STM.writeTVar psVar psModified+      return result+    )  getSchedulerTVar :: HasSchedulerIO r => Eff r (TVar Scheduler) getSchedulerTVar = fromSchedulerVar <$> ask

@@ -0,0 +1,128 @@+module Control.Eff.Concurrent.Process.Interactive+where++import           Control.Arrow+import           Control.Concurrent+import           Control.Eff+import           Control.Eff.Lift+import           Control.Eff.Concurrent.Api+import           Control.Eff.Concurrent.Api.Client+import           Control.Eff.Concurrent.Process+import           Control.Monad+import           Data.Typeable                  ( Typeable )+++-- | This module provides support for executing 'Process' actions from 'IO'.+--+-- One use case is interacting with processes from the REPL, e.g.:+--+-- >>> import Control.Eff.Concurrent.Process.SingleThreadedScheduler (defaultMain)+--+-- >>> import Data.Dynamic+--+-- >>> import Data.Maybe+--+-- >>> s <- forkInteractiveScheduler Control.Eff.Concurrent.Process.SingleThreadedScheduler.defaultMain+--+-- >>> fooPid <- submit s (spawn (forever (receiveMessage SP >>= (logMsg . fromMaybe "Huh!??" . fromDynamic))))+--+-- >>> fooPid+-- <0.1.0>+--+-- >>> submit s (sendMessageAs SP fooPid "test")+-- test+--+-- >>> submit s (sendShutdown SP fooPid)+--+--+--+-- @since 0.3.0.0++data SchedulerVar r =+  SchedulerVar { _schedulerThreadId :: ThreadId+               , _schedulerInQueue :: MVar (Eff (Process r ': r) (Maybe String))+               }+++-- | Fork a scheduler with a process that communicates with it via 'MVar',+-- which is also the reason for the @Lift IO@ constraint.+forkInteractiveScheduler+  :: forall r+   . (SetMember Lift (Lift IO) r)+  => (Eff (Process r ': r) () -> IO ())+  -> IO (SchedulerVar r)+forkInteractiveScheduler ioScheduler = do+  inQueue <- newEmptyMVar+  tid     <- forkIO (ioScheduler (readEvalPrintLoop inQueue))+  return (SchedulerVar tid inQueue)+ where+  readEvalPrintLoop = forever . (readAction >>> evalAction >=> printResult)+   where+    readAction v = do+      mr <- lift (tryTakeMVar v)+      case mr of+        Nothing -> do+          yieldProcess SP+          readAction v+        Just r -> return r+    evalAction  = join+    printResult = mapM_ (lift . putStrLn)++-- | Exit the schedulder immediately using an asynchronous exception.+killInteractiveScheduler :: SchedulerVar r -> IO ()+killInteractiveScheduler = killThread . _schedulerThreadId++-- | Send a 'Process' effect to the main process of a scheduler, this blocks+-- until the effect is executed.+submit+  :: forall r a+   . (SetMember Lift (Lift IO) r)+  => SchedulerVar r+  -> Eff (Process r ': r) a+  -> IO a+submit r theAction = do+  resVar <- newEmptyMVar+  worked <- tryPutMVar (_schedulerInQueue r) (runAndPutResult resVar)+  if worked then takeMVar resVar else fail "ERROR: Scheduler still busy"+ where+  runAndPutResult resVar = do+    res <- theAction+    lift (putMVar resVar $! res)+    return Nothing++-- | Send a 'Process' effect to the main process of a scheduler, this blocks+-- until the effect is executed, then the result is printed by the thread,+-- that runs the process 0 in the scheduler.+submitPrint+  :: forall r a+   . (Show a, SetMember Lift (Lift IO) r)+  => SchedulerVar r+  -> Eff (Process r ': r) a+  -> IO ()+submitPrint r theAction = do+  worked <- tryPutMVar (_schedulerInQueue r) runAndShowResult+  if worked then return () else fail "ERROR: Scheduler still busy"+ where+  runAndShowResult = do+    res <- theAction+    return (Just $! (show res))++-- | Combination of 'submit' and 'cast'.+submitCast+  :: forall o r+   . (SetMember Lift (Lift IO) r, Typeable o)+  => SchedulerVar r+  -> Server o+  -> Api o 'Asynchronous+  -> IO ()+submitCast sc svr request = submit sc (cast SchedulerProxy svr request)++-- | Combination of 'submit' and 'cast'.+submitCall+  :: forall o q r+   . (SetMember Lift (Lift IO) r, Typeable o, Typeable q)+  => SchedulerVar r+  -> Server o+  -> Api o ( 'Synchronous q)+  -> IO q+submitCall sc svr request = submit sc (call SchedulerProxy svr request)

@@ -1,129 +1,143 @@ -- | A coroutine based, single threaded scheduler for 'Process'es. module Control.Eff.Concurrent.Process.SingleThreadedScheduler   ( schedule+  , scheduleWithYield   , defaultMain   , singleThreadedIoScheduler-  , LoggingAndIo) where+  , LoggingAndIo+  )+where -import Control.Eff-import Control.Eff.Lift-import Control.Eff.Log-import Control.Eff.Concurrent.Process-import Control.Lens hiding ((|>), Empty)-import qualified Data.Sequence as Seq-import Data.Sequence (Seq (..))-import qualified Data.Map.Strict as Map-import GHC.Stack-import Data.Kind ()-import Data.Dynamic-import Data.Maybe-import Control.Monad+import           Control.Concurrent             ( yield )+import           Control.Eff+import           Control.Eff.Lift+import           Control.Eff.Log+import           Control.Eff.Concurrent.Process+import           Control.Lens            hiding ( (|>)+                                                , Empty+                                                )+import qualified Data.Sequence                 as Seq+import           Data.Sequence                  ( Seq(..) )+import qualified Data.Map.Strict               as Map+import           GHC.Stack+import           Data.Kind                      ( )+import           Data.Dynamic+import           Data.Maybe +-- | Execute a 'Process' in the current thread, all child processes spawned by+-- 'spawn' will be executed concurrently using a co-routine based, round-robin+-- scheduler.+schedule :: forall r . Eff (Process r ': r) () -> Eff r ()+schedule = scheduleWithYield (return ())  -- | Execute a 'Process' in the current thread, all child processes spawned by -- 'spawn' will be executed concurrently using a co-routine based, round-robin -- scheduler.-schedule-  :: forall r . Eff (Process r ': r) ()-  -> Eff r ()-schedule mainProcessAction =-  do-    y <- runAsCoroutine mainProcessAction-    go 1 (Map.singleton 0 Seq.empty) (Seq.singleton (y, 0))-  where+scheduleWithYield :: forall r . Eff r () -> Eff (Process r ': r) () -> Eff r ()+scheduleWithYield yieldEff mainProcessAction = do+  y <- runAsCoroutine mainProcessAction+  go 1 (Map.singleton 0 Seq.empty) (Seq.singleton (y, 0))+ where -    go :: ProcessId -> Map.Map ProcessId (Seq Dynamic) -> Seq (OnYield r, ProcessId) -> Eff r ()-    go _newPid _msgQs Empty = return ()+  go+    :: ProcessId+    -> Map.Map ProcessId (Seq Dynamic)+    -> Seq (OnYield r, ProcessId)+    -> Eff r ()+  go _newPid _msgQs Empty = return () -    go newPid msgQs allProcs@((processState, pid) :<| rest) =-      let handleExit =-            if pid == 0 then return ()-            else go newPid-                    (msgQs & at pid .~ Nothing)-                    rest+  go newPid msgQs allProcs@((processState, pid) :<| rest)+    = let+        handleExit = if pid == 0+          then return ()+          else go newPid (msgQs & at pid .~ Nothing) rest+        maybeYield = if pid == 0 then yieldEff else return ()+      in+        case processState of+          OnDone                     -> handleExit -      in case processState of-             OnDone -> handleExit+          OnRaiseError _             -> handleExit -             OnRaiseError _ -> handleExit+          OnExitError  _             -> handleExit -             OnExitError _ -> handleExit+          OnSendShutdown targetPid k -> do+            let allButTarget =+                  Seq.filter (\(_, e) -> e /= pid && e /= targetPid) allProcs+                targets     = Seq.filter (\(_, e) -> e == targetPid) allProcs+                suicide     = targetPid == pid+                targetFound = suicide || not (Seq.null targets)+            if suicide+              then do+                nextK <- k ShutdownRequested+                go newPid msgQs (rest :|> (nextK, pid))+              else do+                let deliverTheGoodNews (targetState, tPid) = do+                      nextTargetState <- case targetState of+                        OnSendShutdown _ tk -> tk ShutdownRequested+                        OnYield tk          -> tk ShutdownRequested+                        OnSelf  tk          -> tk ShutdownRequested+                        OnSend _ _ tk       -> tk ShutdownRequested+                        OnRecv tk           -> tk ShutdownRequested+                        OnSpawn _ tk        -> tk ShutdownRequested+                        OnDone              -> return OnDone+                        OnExitError  er     -> return (OnExitError er)+                        OnRaiseError er     -> return (OnExitError er) -- return (error ("TODO write test "++er))+                      return (nextTargetState, tPid)+                nextTargets <- traverse deliverTheGoodNews targets+                nextK       <- k (ResumeWith targetFound)+                maybeYield+                go newPid+                   msgQs+                   (allButTarget Seq.>< (nextTargets :|> (nextK, pid))) -             OnSendShutdown targetPid k ->-               do let allButTarget =-                        Seq.filter (\(_,e) -> e /= pid && e /= targetPid) allProcs-                      targets =-                        Seq.filter (\(_,e) -> e == targetPid) allProcs-                      suicide = targetPid == pid-                      targetFound = suicide || not (Seq.null targets)-                  if suicide-                   then-                    do nextK <- k ShutdownRequested-                       go newPid-                          msgQs-                          (rest :|> (nextK, pid))-                   else-                    do let deliverTheGoodNews (targetState, tPid) =-                             do nextTargetState <--                                   case targetState of-                                     OnSendShutdown _ tk ->-                                       tk ShutdownRequested-                                     OnSelf tk ->-                                       tk ShutdownRequested-                                     OnSend _ _ tk ->-                                       tk ShutdownRequested-                                     OnRecv tk ->-                                       tk ShutdownRequested-                                     OnSpawn _ tk ->-                                       tk ShutdownRequested-                                     OnDone -> return OnDone-                                     OnExitError er -> return (OnExitError er)-                                     OnRaiseError er -> return (OnExitError er) -- return (error ("TODO write test "++er))-                                return (nextTargetState, tPid)-                       nextTargets <- traverse deliverTheGoodNews targets-                       nextK <- k (ResumeWith targetFound)-                       go newPid-                          msgQs-                          (allButTarget Seq.>< (nextTargets :|> (nextK, pid))) -             OnSelf k ->-               do nextK <- k (ResumeWith pid)-                  go newPid-                     msgQs-                     (rest :|> (nextK, pid))+          OnSelf k -> do+            nextK <- k (ResumeWith pid)+            maybeYield+            go newPid msgQs (rest :|> (nextK, pid)) -             OnSend toPid msg k ->-               do nextK <- k (ResumeWith (msgQs ^. at toPid . to isJust))-                  go newPid-                     (msgQs & at toPid . _Just %~ (:|> msg))-                     (rest :|> (nextK, pid))+          OnYield k -> do+            yieldEff+            nextK <- k (ResumeWith ())+            go newPid msgQs (rest :|> (nextK, pid)) -             recv@(OnRecv k) ->-               case msgQs ^. at pid of-                 Nothing ->-                   do nextK <- k (OnError (show pid ++ " has no message queue!"))-                      go newPid msgQs (rest :|> (nextK, pid))-                 Just Empty ->-                   if Seq.length rest == 0 then-                     do nextK <- k (OnError ("Process " ++ show pid ++ " deadlocked!"))-                        go newPid msgQs (rest :|> (nextK, pid))-                   else-                     go newPid msgQs (rest :|> (recv, pid))+          OnSend toPid msg k -> do+            nextK <- k (ResumeWith (msgQs ^. at toPid . to isJust))+            maybeYield+            go newPid+               (msgQs & at toPid . _Just %~ (:|> msg))+               (rest :|> (nextK, pid)) -                 Just (nextMessage :<| restMessages) ->-                   do nextK <- k (ResumeWith nextMessage)-                      go newPid-                         (msgQs & at pid . _Just .~ restMessages)-                         (rest :|> (nextK, pid))+          recv@(OnRecv k) -> case msgQs ^. at pid of+            Nothing -> do+              nextK <- k (OnError (show pid ++ " has no message queue!"))+              maybeYield+              go newPid msgQs (rest :|> (nextK, pid))+            Just Empty -> if Seq.length rest == 0+              then do+                nextK <- k (OnError ("Process " ++ show pid ++ " deadlocked!"))+                maybeYield+                go newPid msgQs (rest :|> (nextK, pid))+              else go newPid msgQs (rest :|> (recv, pid)) -             OnSpawn f k ->-               do nextK <- k (ResumeWith newPid)-                  fk <- runAsCoroutine f-                  go (newPid + 1)-                     (msgQs & at newPid .~ Just Seq.empty)-                     ( rest :|> (nextK, pid) :|> (fk, newPid))+            Just (nextMessage :<| restMessages) -> do+              nextK <- k (ResumeWith nextMessage)+              maybeYield+              go newPid+                 (msgQs & at pid . _Just .~ restMessages)+                 (rest :|> (nextK, pid)) +          OnSpawn f k -> do+            nextK <- k (ResumeWith newPid)+            fk    <- runAsCoroutine f+            maybeYield+            go (newPid + 1)+               (msgQs & at newPid .~ Just Seq.empty)+               (rest :|> (nextK, pid) :|> (fk, newPid))+ data OnYield r where+  OnYield :: (ResumeProcess () -> Eff r (OnYield r))+         -> OnYield r   OnSelf :: (ResumeProcess ProcessId -> Eff r (OnYield r))          -> OnYield r   OnSpawn :: Eff (Process r ': r) ()@@ -139,19 +153,19 @@          -> OnYield r   OnSendShutdown :: ProcessId -> (ResumeProcess Bool -> Eff r (OnYield r)) -> OnYield r -runAsCoroutine :: forall r v . Eff (Process r ': r) v-               -> Eff r (OnYield r)+runAsCoroutine :: forall r v . Eff (Process r ': r) v -> Eff r (OnYield r) runAsCoroutine m = handle_relay (const $ return OnDone) cont m-  where-    cont :: Process r x -> (x -> Eff r (OnYield r)) -> Eff r (OnYield r)-    cont SelfPid k = return (OnSelf k)-    cont (Spawn e) k = return (OnSpawn e k)-    cont Shutdown _k = return OnDone-    cont (ExitWithError e) _k = return (OnExitError e)-    cont (RaiseError e) _k = return (OnRaiseError e)-    cont (SendMessage tp msg) k = return (OnSend tp msg k)-    cont ReceiveMessage k = return (OnRecv k)-    cont (SendShutdown pid) k = return (OnSendShutdown pid k)+ where+  cont :: Process r x -> (x -> Eff r (OnYield r)) -> Eff r (OnYield r)+  cont YieldProcess         k  = return (OnYield k)+  cont SelfPid              k  = return (OnSelf k)+  cont (Spawn e)            k  = return (OnSpawn e k)+  cont Shutdown             _k = return OnDone+  cont (ExitWithError e   ) _k = return (OnExitError e)+  cont (RaiseError    e   ) _k = return (OnRaiseError e)+  cont (SendMessage tp msg) k  = return (OnSend tp msg k)+  cont ReceiveMessage       k  = return (OnRecv k)+  cont (SendShutdown pid)   k  = return (OnSendShutdown pid k)  -- | The concrete list of 'Eff'ects for running this pure scheduler on @IO@ and -- with string logging.@@ -170,21 +184,5 @@   :: HasCallStack   => Eff '[Process '[Logs String, Lift IO], Logs String, Lift IO] ()   -> IO ()-defaultMain go = runLift $ handleLogsWith (schedule go) ($ putStrLn)--_example :: IO ()-_example = defaultMain go-  where-    px :: SchedulerProxy '[Logs String, Lift IO]-    px = SchedulerProxy-    go :: Eff (ConsProcess '[Logs String, Lift IO]) ()-    go = do-           p1 <- spawn $ do-                    logMsg "Hello World!"-                    me <- self px-                    logMsg ("I am: " ++ show me)-                    d <- receiveMessage px-                    let m = fromDyn @String d ""-                    logMsg ("Got: " ++ m)-           void $ sendMessage px p1 (toDyn "huhu")-           return ()+defaultMain go =+  runLift $ handleLogsWith (scheduleWithYield (lift yield) go) ($ putStrLn)

@@ -5,78 +5,21 @@ {-# LANGUAGE TypeOperators #-} -- | Add-ons to 'Control.Eff.Exception' module Control.Eff.ExceptionExtra-  ( try-  , liftRethrow-  , runErrorRethrowIO-  , liftCatch-  , liftCatchEff-  , liftTry+  ( liftTry   , module X   ) where -import           Control.Monad                  ( (>=>) ) import qualified Control.Exception             as Exc import           Control.Eff import           Control.Eff.Lift import           Control.Eff.Exception         as X import           GHC.Stack --- * Effects combining "Control.Eff.Exception" and 'Exc.try'---- | Catch an exception and return it in an 'Either'.-try :: forall e r a . Member (Exc e) r => Eff r a -> Eff r (Either e a)-try e = catchError (Right <$> e) (return . Left)---- | Lift an IO action and catch all error using 'Exc.try' then wrap the--- 'Exc.Exception' using a given wrapper function and rethrow it using--- 'X.throwError'.-liftRethrow-  :: forall proxy e r a-   . (Exc.Exception e, SetMember Lift (Lift IO) r, SetMember Exc (Exc e) r)-  => proxy e-  -> IO a-  -> Eff r a-liftRethrow _ m = lift (Exc.try m) >>= either (throwError @e) return---- | Run an effect with exceptions like 'X.runError' and rethrow it as--- 'Exc.SomeException' using 'Exc.throw'-runErrorRethrowIO-  :: forall e r a-   . (Exc.Exception e, SetMember Lift (Lift IO) r)-  => Eff (Exc e ': r) a-  -> Eff r a-runErrorRethrowIO = runError >=> either (Exc.throw . Exc.SomeException) return---- * Effect Utilities around 'Exc.try'---- | Lift an IO action and catch all errors with 'Exc.try' with a pure--- handler.-liftCatch-  :: forall e r a-   . (HasCallStack, Exc.Exception e, SetMember Lift (Lift IO) r)-  => (e -> a)-  -> IO a-  -> Eff r a-liftCatch handleE m =-  lift (Exc.try m) >>= either (return . handleE) return---- | Lift an IO action and catch all errors with 'Exc.try' with an effect--- handler.-liftCatchEff-  :: forall e r a-   . (SetMember Lift (Lift IO) r, HasCallStack, Exc.Exception e)-  => (e -> Eff r a)-  -> IO a-  -> Eff r a-liftCatchEff handleE m =-  lift (Exc.try m) >>= either handleE return- -- | Catch 'Exc.Exception' thrown by an effect. liftTry   :: forall e r a    . (HasCallStack, Exc.Exception e, SetMember Lift (Lift IO) r)   => Eff r a   -> Eff r (Either e a)-liftTry m =-   (Right <$> m) `catchDynE` (return . Left)+liftTry m = (Right <$> m) `catchDynE` (return . Left)

@@ -1,77 +0,0 @@-{-# LANGUAGE FlexibleContexts #-}-{-# LANGUAGE DataKinds #-}-{-# LANGUAGE TypeOperators #-}-{-# LANGUAGE GADTs #-}-module Control.Eff.Interactive-  ( Interaction(..)-  , printLine-  , printStep-  , promptStep-  , step-  , Interactive(..)-  , interactiveProgram-  , runInteractionIOE-  , runInteractionIO-  )-where--import           Control.Eff-import           Control.Eff.Lift-import           Control.Eff.Operational-import           Control.Monad                  ( void )-import           GHC.Stack--data Interaction a where-  PrintLine :: String -> Interaction ()-  ReadLine :: (String -> a) -> Interaction a-  Step :: Interaction ()--printLine-  :: (Member (Program Interaction) r, HasCallStack) => String -> Eff r ()-printLine = singleton . PrintLine--printStep-  :: (Member (Program Interaction) r, HasCallStack) => String -> Eff r ()-printStep m = do-  printLine m-  step--step :: (Member (Program Interaction) r, HasCallStack) => Eff r ()-step = do-  printLine "\nPress ENTER to continue."-  singleton Step--promptStep-  :: (Member (Program Interaction) r, HasCallStack)-  => String-  -> (String -> a)-  -> Eff r a-promptStep m p = do-  singleton (PrintLine m)-  singleton (ReadLine p)---runInteractionIOE-  :: (SetMember Lift (Lift IO) r, HasCallStack)-  => Eff (Program Interaction ': r) a-  -> Eff r a-runInteractionIOE = runProgram go- where-  go :: (SetMember Lift (Lift IO) r, HasCallStack) => Interaction a -> Eff r a-  go (PrintLine m)    = lift (putStrLn m)-  go Step             = lift (void getLine)-  go (ReadLine parse) = lift (parse <$> getLine)--runInteractionIO :: Eff '[Program Interaction, Lift IO] a -> IO a-runInteractionIO = runLift . runInteractionIOE---- ** Interactively work with 'Control.Eff.Operational.Program's.--class Interactive f where-  singleSteps :: (Member (Program Interaction) r, HasCallStack) => f a -> Eff r a--interactiveProgram-  :: (HasCallStack, Member (Program Interaction) r, Interactive f)-  => Eff (Program f ': r) a-  -> Eff r a-interactiveProgram = runProgram singleSteps

@@ -32,10 +32,10 @@ testLogC =   filterLogChannel (\m -> take (length logPrefix) m == logPrefix)   <$>-  forkLogger 1 (putStrLn)  (Just (logPrefix ++ "^^^^^^^^^^^^^^ LogChannel Start ^^^^^^^^^^^^^^"))+  forkLogger 1 (putStrLn)  Nothing  testLogJoin :: LogChannel String -> IO ()-testLogJoin = joinLogChannel (Just (logPrefix ++ "^^^^^^^^^^^^^^ LogChannel Done ^^^^^^^^^^^^^^"))+testLogJoin = joinLogChannel Nothing  tlog :: Member (Logs String) r => String -> Eff r () tlog = logMsg . (logPrefix ++)

@@ -0,0 +1,9 @@+module Debug+where++import           ProcessBehaviourTestCases+import           Test.Tasty+import           Test.Tasty.HUnit++debugMain :: IO ()+debugMain = defaultMain test_forkIo

@@ -1,111 +1,154 @@-module ForkIOScheduler where--import Control.Exception-import Control.Concurrent-import Control.Concurrent.STM-import Control.Eff-import Control.Eff.Lift-import Control.Eff.Concurrent.Process-import Control.Eff.Concurrent.Process.ForkIOScheduler as Scheduler-import Control.Monad (void, forever)-import Test.Tasty-import Test.Tasty.HUnit-import Data.Dynamic-import Common+module ForkIOScheduler+where -timeoutSeconds :: Integer -> Timeout-timeoutSeconds seconds = Timeout (seconds * 1000000) (show seconds ++ "s")+import           Control.Exception+import           Control.Concurrent+import           Control.Concurrent.STM+import           Control.Eff+import           Control.Eff.Lift+import           Control.Eff.Concurrent.Process+import           Control.Eff.Concurrent.Process.ForkIOScheduler+                                               as Scheduler+import           Control.Monad                  ( void+                                                , forever+                                                )+import           Test.Tasty+import           Test.Tasty.HUnit+import           Data.Dynamic+import           Common  test_IOExceptionsIsolated :: TestTree-test_IOExceptionsIsolated =-  setTravisTestOptions-    $ testGroup "one process throws an IO exception, the other continues unimpaired"-       [ testCase ("process 2 exits with: "++ howToExit-                    ++ " - while process 1 is busy with: " ++ busyWith)-        $ do aVar <- newEmptyTMVarIO-             lc <- testLogC-             Scheduler.defaultMainWithLogChannel lc-               $ do p1 <- spawn $ forever busyEffect-                    lift (threadDelay 1000)-                    void $ spawn $ do lift (threadDelay 1000)-                                      doExit-                    lift (threadDelay 100000)-                    wasStillRunningP1 <- sendShutdown forkIoScheduler p1-                    lift (atomically (putTMVar aVar wasStillRunningP1))--             wasStillRunningP1 <- atomically (takeTMVar aVar)-             assertBool "the other process was still running" wasStillRunningP1+test_IOExceptionsIsolated = setTravisTestOptions $ testGroup+  "one process throws an IO exception, the other continues unimpaired"+  [ testCase+        (  "process 2 exits with: "+        ++ howToExit+        ++ " - while process 1 is busy with: "+        ++ busyWith+        )+      $ do+          aVar <- newEmptyTMVarIO+          lc   <- testLogC+          Scheduler.defaultMainWithLogChannel lc $ do+            p1 <- spawn $ forever busyEffect+            lift (threadDelay 1000)+            void $ spawn $ do+              lift (threadDelay 1000)+              doExit+            lift (threadDelay 100000)+            wasStillRunningP1 <- sendShutdownChecked forkIoScheduler p1+            lift (atomically (putTMVar aVar wasStillRunningP1)) -       | (busyWith, busyEffect) <--         [ ("receiving", void (send (ReceiveMessage @SchedulerIO)))-         , ("sending", void (send (SendMessage @SchedulerIO 44444 (toDyn "test message"))))-         , ("sending shutdown", void (send (SendShutdown @SchedulerIO 44444)))-         , ("selfpid-ing", void (send (SelfPid @SchedulerIO)))-         , ("spawn-ing", void (send (Spawn @SchedulerIO (void (send (ReceiveMessage @SchedulerIO))))))-         ]-       , (howToExit, doExit) <--         [ ("throw async exception", void (lift (throw UserInterrupt)))-         , ("division by zero", void (return ((123::Int) `div` 0) >>= lift . putStrLn . show ))-         , ("call 'fail'", void (fail "test fail"))-         , ("call 'error'", void (error "test error"))-         ]-       ]+          wasStillRunningP1 <- atomically (takeTMVar aVar)+          assertBool "the other process was still running" wasStillRunningP1+  | (busyWith , busyEffect) <-+    [ ("receiving", void (send (ReceiveMessage @SchedulerIO)))+    , ( "sending"+      , void (send (SendMessage @SchedulerIO 44444 (toDyn "test message")))+      )+    , ("sending shutdown", void (send (SendShutdown @SchedulerIO 44444)))+    , ("selfpid-ing"     , void (send (SelfPid @SchedulerIO)))+    , ( "spawn-ing"+      , void+        (send (Spawn @SchedulerIO (void (send (ReceiveMessage @SchedulerIO)))))+      )+    ]+  , (howToExit, doExit    ) <-+    [ ("throw async exception", void (lift (throw UserInterrupt)))+    , ( "division by zero"+      , void (return ((123 :: Int) `div` 0) >>= lift . putStrLn . show)+      )+    , ("call 'fail'" , void (fail "test fail"))+    , ("call 'error'", void (error "test error"))+    ]+  ]  test_mainProcessSpawnsAChildAndReturns :: TestTree-test_mainProcessSpawnsAChildAndReturns =-  setTravisTestOptions-  (testCase "spawn a child and return"-   (do lc <- testLogC-       Scheduler.defaultMainWithLogChannel lc-         (void (spawn (void (receiveMessage forkIoScheduler))))))+test_mainProcessSpawnsAChildAndReturns = setTravisTestOptions+  (testCase+    "spawn a child and return"+    (do+      lc <- testLogC+      Scheduler.defaultMainWithLogChannel+        lc+        (void (spawn (void (receiveMessage forkIoScheduler))))+    )+  )  test_mainProcessSpawnsAChildAndExitsNormally :: TestTree-test_mainProcessSpawnsAChildAndExitsNormally =-  setTravisTestOptions-  (testCase "spawn a child and exit normally"-   (do lc <- testLogC-       Scheduler.defaultMainWithLogChannel lc-         (do void (spawn (void (receiveMessage forkIoScheduler)))-             void (exitNormally forkIoScheduler)-             fail "This should not happen!!"-         )))+test_mainProcessSpawnsAChildAndExitsNormally = setTravisTestOptions+  (testCase+    "spawn a child and exit normally"+    (do+      lc <- testLogC+      Scheduler.defaultMainWithLogChannel+        lc+        (do+          void (spawn (void (receiveMessage forkIoScheduler)))+          void (exitNormally forkIoScheduler)+          fail "This should not happen!!"+        )+    )+  )   test_mainProcessSpawnsAChildInABusySendLoopAndExitsNormally :: TestTree test_mainProcessSpawnsAChildInABusySendLoopAndExitsNormally =   setTravisTestOptions-  (testCase "spawn a child with a busy send loop and exit normally"-   (do lc <- testLogC-       Scheduler.defaultMainWithLogChannel lc-         (do void (spawn (forever (void (sendMessage forkIoScheduler 1000 (toDyn "test")))))-             void (exitNormally forkIoScheduler)-             fail "This should not happen!!"-         )))+    (testCase+      "spawn a child with a busy send loop and exit normally"+      (do+        lc <- testLogC+        Scheduler.defaultMainWithLogChannel+          lc+          (do+            void+              (spawn+                (forever+                  (void (sendMessage forkIoScheduler 1000 (toDyn "test")))+                )+              )+            void (exitNormally forkIoScheduler)+            fail "This should not happen!!"+          )+      )+    )   test_mainProcessSpawnsAChildBothReturn :: TestTree-test_mainProcessSpawnsAChildBothReturn =-  setTravisTestOptions-  (testCase "spawn a child and let it return and return"-   (do lc <- testLogC-       Scheduler.defaultMainWithLogChannel lc-         (do child <- spawn (void (receiveMessageAs @String forkIoScheduler))-             True <- sendMessage forkIoScheduler child (toDyn "test")-             return ()-         )))+test_mainProcessSpawnsAChildBothReturn = setTravisTestOptions+  (testCase+    "spawn a child and let it return and return"+    (do+      lc <- testLogC+      Scheduler.defaultMainWithLogChannel+        lc+        (do+          child <- spawn (void (receiveMessageAs @String forkIoScheduler))+          True  <- sendMessageChecked forkIoScheduler child (toDyn "test")+          return ()+        )+    )+  )  test_mainProcessSpawnsAChildBothExitNormally :: TestTree-test_mainProcessSpawnsAChildBothExitNormally =-  setTravisTestOptions-  (testCase "spawn a child and let it exit and exit"-   (do lc <- testLogC-       Scheduler.defaultMainWithLogChannel lc-         (do child <- spawn-                     (do void (receiveMessageAs @String forkIoScheduler)-                         void (exitNormally forkIoScheduler)-                         error "This should not happen (child)!!"-                     )-             True <- sendMessage forkIoScheduler child (toDyn "test")-             void (exitNormally forkIoScheduler)-             error "This should not happen!!"-         )))+test_mainProcessSpawnsAChildBothExitNormally = setTravisTestOptions+  (testCase+    "spawn a child and let it exit and exit"+    (do+      lc <- testLogC+      Scheduler.defaultMainWithLogChannel+        lc+        (do+          child <- spawn+            (do+              void (receiveMessageAs @String forkIoScheduler)+              void (exitNormally forkIoScheduler)+              error "This should not happen (child)!!"+            )+          True <- sendMessageChecked forkIoScheduler child (toDyn "test")+          void (exitNormally forkIoScheduler)+          error "This should not happen!!"+        )+    )+  )

@@ -1,393 +1,558 @@-module ProcessBehaviourTestCases where+module ProcessBehaviourTestCases+where -import Data.List (sort)-import Data.Dynamic-import Data.Foldable (traverse_)-import Control.Exception-import Control.Concurrent-import Control.Concurrent.STM-import Control.Eff.Concurrent.Process-import qualified Control.Eff.Concurrent.Process.ForkIOScheduler as ForkIO-import qualified Control.Eff.Concurrent.Process.SingleThreadedScheduler as SingleThreaded-import Control.Eff-import Control.Eff.Log-import Control.Eff.Lift-import Control.Monad (void, replicateM, forever, when)-import Test.Tasty-import Test.Tasty.HUnit-import Common+import           Data.List                      ( sort )+import           Data.Dynamic+import           Data.Foldable                  ( traverse_ )+import           Control.Exception+import           Control.Concurrent+import           Control.Concurrent.STM+import           Control.Eff.Concurrent.Process+import qualified Control.Eff.Concurrent.Process.ForkIOScheduler+                                               as ForkIO+import qualified Control.Eff.Concurrent.Process.SingleThreadedScheduler+                                               as SingleThreaded+import           Control.Eff+import           Control.Eff.Log+import           Control.Eff.Lift+import           Control.Monad                  ( void+                                                , replicateM+                                                , forever+                                                , when+                                                )+import           Test.Tasty+import           Test.Tasty.HUnit+import           Common  test_forkIo :: TestTree-test_forkIo =-  setTravisTestOptions+test_forkIo = setTravisTestOptions   (withTestLogC ForkIO.schedule-    (\factory ->-       testGroup "ForkIOScheduler"-       [allTests factory]))+                (\factory -> testGroup "ForkIOScheduler" [allTests factory])+  )  test_singleThreaded :: TestTree-test_singleThreaded =-  setTravisTestOptions-  (withTestLogC (const . SingleThreaded.defaultMain)-    (\factory ->-       testGroup "SingleThreadedScheduler"-       [allTests factory]))+test_singleThreaded = setTravisTestOptions+  (withTestLogC+    (const . SingleThreaded.defaultMain)+    (\factory -> testGroup "SingleThreadedScheduler" [allTests factory])+  ) -allTests :: forall r . (Member (Logs String) r, SetMember Lift (Lift IO) r)-           => IO (Eff (Process r ': r) () -> IO ())-           -> TestTree-allTests schedulerFactory =-    localOption (timeoutSeconds 300)-    (testGroup "Process"-      [ errorTests schedulerFactory-      , sendShutdownTests schedulerFactory-      , concurrencyTests schedulerFactory-      , exitTests schedulerFactory-      ])+allTests+  :: forall r+   . (Member (Logs String) r, SetMember Lift (Lift IO) r)+  => IO (Eff (Process r ': r) () -> IO ())+  -> TestTree+allTests schedulerFactory = localOption+  (timeoutSeconds 300)+  (testGroup+    "Process"+    [ errorTests schedulerFactory+    , sendShutdownTests schedulerFactory+    , concurrencyTests schedulerFactory+    , exitTests schedulerFactory+    , pingPongTests schedulerFactory+    ]+  ) -errorTests :: forall r . (Member (Logs String) r, SetMember Lift (Lift IO) r)-             => IO (Eff (Process r ': r) () -> IO ()) -> TestTree-errorTests schedulerFactory =-  let px :: SchedulerProxy r+data Ping = Ping ProcessId+data Pong = Pong+  deriving (Eq, Show)++pingPongTests+  :: forall r+   . (Member (Logs String) r, SetMember Lift (Lift IO) r)+  => IO (Eff (Process r ': r) () -> IO ())+  -> TestTree+pingPongTests schedulerFactory = testGroup+  "Yield Tests"+  [ testCase "ping pong a message between two processes, both don't yield"+  $ applySchedulerFactory schedulerFactory+  $ do+      let pongProc = forever $ do+            Ping pinger <- receiveMessageAs SP+            sendMessageAs SP pinger Pong+          pingProc ponger parent = do+            me <- self SP+            sendMessageAs SP ponger (Ping me)+            Pong <- receiveMessageAs SP+            sendMessageAs SP parent True+      pongPid <- spawn pongProc+      me      <- self SP+      spawn_ (pingProc pongPid me)+      ok <- receiveMessageAs @Bool SP+      lift (ok @? "ping pong failed")+  , testCase "ping pong a message between two processes, with massive yielding"+  $ applySchedulerFactory schedulerFactory+  $ do+      yieldProcess SP+      let pongProc = forever $ do+            yieldProcess SP+            Ping pinger <- receiveMessageAs SP+            yieldProcess SP+            sendMessageAs SP pinger Pong+            yieldProcess SP+          pingProc ponger parent = do+            yieldProcess SP+            me <- self SP+            yieldProcess SP+            sendMessageAs SP ponger (Ping me)+            yieldProcess SP+            Pong <- receiveMessageAs SP+            yieldProcess SP+            sendMessageAs SP parent True+            yieldProcess SP+      yieldProcess SP+      pongPid <- spawn pongProc+      yieldProcess SP+      me <- self SP+      yieldProcess SP+      spawn_ (pingProc pongPid me)+      yieldProcess SP+      ok <- receiveMessageAs @Bool SP+      yieldProcess SP+      lift (ok @? "ping pong failed")+      yieldProcess SP+  , testCase+    "the first message is not delayed, not even in cooperative scheduling (because of yield)"+  $ applySchedulerFactory schedulerFactory+  $ do+      pongVar <- lift newEmptyMVar+      let pongProc = forever $ do+            Pong <- receiveMessageAs SP+            lift (putMVar pongVar Pong)+      ponger <- spawn pongProc+      sendMessageAs SP ponger Pong+      let waitLoop = do+            p <- lift (tryTakeMVar pongVar)+            case p of+              Nothing -> do+                yieldProcess SP+                waitLoop+              Just r -> return r+      p <- waitLoop+      lift (p == Pong @? "ping pong failed")+  ]++errorTests+  :: forall r+   . (Member (Logs String) r, SetMember Lift (Lift IO) r)+  => IO (Eff (Process r ': r) () -> IO ())+  -> TestTree+errorTests schedulerFactory+  = let+      px :: SchedulerProxy r       px = SchedulerProxy-  in-    testGroup "causing and handling errors"-    [ testGroup "raiseError"-       [-         testCase "unhandled raiseError"-         $ applySchedulerFactory schedulerFactory-         $ do void $ raiseError px "test error"+    in+      testGroup+        "causing and handling errors"+        [ testGroup+          "raiseError"+          [ testCase "unhandled raiseError"+          $ applySchedulerFactory schedulerFactory+          $ do+              void $ raiseError px "test error"               error "This should not happen"-       , testCase "catch raiseError"-         $ scheduleAndAssert schedulerFactory-         $ \assertEff ->-             catchRaisedError-                px-                (assertEff "error must be caught" . (== "test error 2"))-                (void (raiseError px "test error 2"))-       , testCase "catch raiseError from a long sub block"-         $ scheduleAndAssert schedulerFactory-         $ \assertEff ->-             catchRaisedError-                px-                (assertEff "error must be caught" . (== "test error 3"))-                (do void (replicateM 100000 (void (self px)))-                    void (raiseError px "test error 3"))-       ]-    , testGroup "exitWithError"-      [ testCase "unhandled exitWithError"-         $ applySchedulerFactory schedulerFactory-         $ do void $ exitWithError px "test error"+          , testCase "catch raiseError"+          $ scheduleAndAssert schedulerFactory+          $ \assertEff -> catchRaisedError+              px+              (assertEff "error must be caught" . (== "test error 2"))+              (void (raiseError px "test error 2"))+          , testCase "catch raiseError from a long sub block"+          $ scheduleAndAssert schedulerFactory+          $ \assertEff -> catchRaisedError+              px+              (assertEff "error must be caught" . (== "test error 3"))+              (do+                void (replicateM 100000 (void (self px)))+                void (raiseError px "test error 3")+              )+          ]+        , testGroup+          "exitWithError"+          [ testCase "unhandled exitWithError"+          $ applySchedulerFactory schedulerFactory+          $ do+              void $ exitWithError px "test error"               error "This should not happen"-      , testCase "cannot catch exitWithError"-         $ applySchedulerFactory schedulerFactory-         $ do void $ ignoreProcessError px $ exitWithError px "test error 4"+          , testCase "cannot catch exitWithError"+          $ applySchedulerFactory schedulerFactory+          $ do+              void $ ignoreProcessError px $ exitWithError px "test error 4"               error "This should not happen"-      , testCase "multi process exitWithError"-        $ scheduleAndAssert schedulerFactory-        $ \assertEff ->-           do me <- self px+          , testCase "multi process exitWithError"+          $ scheduleAndAssert schedulerFactory+          $ \assertEff -> do+              me <- self px               let n = 15               traverse_-                (\(i :: Int) ->-                    spawn $-                      if i `rem` 5 == 0 then do-                        void $ sendMessage px me (toDyn i)-                        void (exitWithError px (show i ++ " died"))-                        assertEff "this should not be reached" False-                      else-                        forever-                          (void (sendMessage px 888 (toDyn "test message to 888"))))+                (\(i :: Int) -> spawn $ if i `rem` 5 == 0+                  then do+                    void $ sendMessage px me (toDyn i)+                    void (exitWithError px (show i ++ " died"))+                    assertEff "this should not be reached" False+                  else+                    forever+                      (void (sendMessage px 888 (toDyn "test message to 888"))+                      )+                )                 [0, 5 .. n]               oks <- replicateM-                     (length [0,5 .. n])-                     (do j <- receiveMessageAs px-                         return j)-              assertEff "" (sort oks == [0,5 .. n])--      ]-    ]+                (length [0, 5 .. n])+                (do+                  j <- receiveMessageAs px+                  return j+                )+              assertEff "" (sort oks == [0, 5 .. n])+          ]+        ] -concurrencyTests :: forall r . (Member (Logs String) r, SetMember Lift (Lift IO) r)-                  => IO (Eff (Process r ': r) () -> IO ()) -> TestTree-concurrencyTests schedulerFactory =-  let px :: SchedulerProxy r+concurrencyTests+  :: forall r+   . (Member (Logs String) r, SetMember Lift (Lift IO) r)+  => IO (Eff (Process r ': r) () -> IO ())+  -> TestTree+concurrencyTests schedulerFactory+  = let+      px :: SchedulerProxy r       px = SchedulerProxy-      n = 100-  in-    testGroup "concurrency tests"-    [ testCase "when main process exits the scheduler kills/cleans and returns"-      $ applySchedulerFactory schedulerFactory-      $ do me <- self px-           traverse_-             (const-               (spawn-                 (do m <- receiveMessage px-                     void (sendMessage px me m))))-             [1..n]-           lift (threadDelay 1000)-    ,-      testCase "new processes are executed before the parent process"-      $ scheduleAndAssert schedulerFactory-      $ \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 px))) [1..n]-             assertEff "" True-    ,-      testCase "two concurrent processes"-      $ scheduleAndAssert schedulerFactory-      $ \assertEff ->-          do me <- self px-             child1 <--               spawn (do m <- receiveMessage px-                         void (sendMessage px me m))-             child2 <--               spawn (forever (void (sendMessage px 888 (toDyn ""))))-             True <- sendMessage px child1 (toDyn "test")-             i <- receiveMessageAs px-             True <- sendShutdown px child2-             assertEff "" (i == "test")-    ,-      testCase "most processes send forever"-      $ scheduleAndAssert schedulerFactory-      $ \assertEff ->-          do me <- self px-             traverse_-               (\(i :: Int) ->-                   spawn-                   $ do when (i `rem` 5 == 0) $-                          void $ sendMessage px me (toDyn i)-                        forever $-                          void (sendMessage px 888 (toDyn "test message to 888"))-               ) [0 .. n]-             oks <- replicateM-                     (length [0,5 .. n])-                     (do j <- receiveMessageAs px-                         return j)-             assertEff "" (sort oks == [0,5 .. n])-    , testCase "most processes self forever"-      $ scheduleAndAssert schedulerFactory-      $ \assertEff ->-          do me <- self px-             traverse_-               (\(i :: Int) ->-                   spawn-                   $ do when (i `rem` 5 == 0) $-                          void $ sendMessage px me (toDyn i)-                        forever $ void (self px)-               ) [0 .. n]-             oks <- replicateM-                     (length [0,5 .. n])-                     (do j <- receiveMessageAs px-                         return j)-             assertEff "" (sort oks == [0,5 .. n])-    , testCase "most processes sendShutdown forever"-      $ scheduleAndAssert schedulerFactory-      $ \assertEff ->-          do me <- self px-             traverse_-               (\(i :: Int) ->-                   spawn-                   $ do when (i `rem` 5 == 0) $-                          void $ sendMessage px me (toDyn i)-                        forever $ void (sendShutdown px 999)-               ) [0 .. n]-             oks <- replicateM-                     (length [0,5 .. n])-                     (do j <- receiveMessageAs px-                         return j)-             assertEff "" (sort oks == [0,5 .. n])-    , testCase "most processes spawn forever"-      $ scheduleAndAssert schedulerFactory-      $ \assertEff ->-          do me <- self px-             traverse_-               (\(i :: Int) ->-                   spawn-                   $ do when (i `rem` 5 == 0) $-                          void $ sendMessage px me (toDyn i)-                        parent <- self px-                        forever $-                          void (spawn (void (sendMessage px parent (toDyn "test msg from child"))))-               ) [0 .. n]-             oks <- replicateM-                     (length [0,5 .. n])-                     (do j <- receiveMessageAs px-                         return j)-             assertEff "" (sort oks == [0,5 .. n])-    , testCase "most processes receive forever"-      $ scheduleAndAssert schedulerFactory-      $ \assertEff ->-          do me <- self px-             traverse_-               (\(i :: Int) ->-                   spawn-                   $ do when (i `rem` 5 == 0) $-                          void $ sendMessage px me (toDyn i)-                        forever $-                          void (receiveMessage px)-               ) [0 .. n]-             oks <- replicateM-                     (length [0,5 .. n])-                     (do j <- receiveMessageAs px-                         return j)-             assertEff "" (sort oks == [0,5 .. n])-       ]+      n  = 100+    in+      testGroup+        "concurrency tests"+        [ testCase+          "when main process exits the scheduler kills/cleans and returns"+        $ applySchedulerFactory schedulerFactory+        $ do+            me <- self px+            traverse_+              (const+                (spawn+                  (do+                    m <- receiveMessage px+                    void (sendMessage px me m)+                  )+                )+              )+              [1 .. n]+            lift (threadDelay 1000)+        , testCase "new processes are executed before the parent process"+        $ scheduleAndAssert schedulerFactory+        $ \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 px))) [1 .. n]+            assertEff "" True+        , testCase "two concurrent processes"+        $ scheduleAndAssert schedulerFactory+        $ \assertEff -> do+            me     <- self px+            child1 <- spawn+              (do+                m <- receiveMessage px+                void (sendMessage px me m)+              )+            child2 <- spawn (forever (void (sendMessage px 888 (toDyn ""))))+            True   <- sendMessageChecked px child1 (toDyn "test")+            i      <- receiveMessageAs px+            True   <- sendShutdownChecked px child2+            assertEff "" (i == "test")+        , testCase "most processes send forever"+        $ scheduleAndAssert schedulerFactory+        $ \assertEff -> do+            me <- self px+            traverse_+              (\(i :: Int) -> spawn $ do+                when (i `rem` 5 == 0) $ void $ sendMessage px me (toDyn i)+                forever+                  $ void (sendMessage px 888 (toDyn "test message to 888"))+              )+              [0 .. n]+            oks <- replicateM+              (length [0, 5 .. n])+              (do+                j <- receiveMessageAs px+                return j+              )+            assertEff "" (sort oks == [0, 5 .. n])+        , testCase "most processes self forever"+        $ scheduleAndAssert schedulerFactory+        $ \assertEff -> do+            me <- self px+            traverse_+              (\(i :: Int) -> spawn $ do+                when (i `rem` 5 == 0) $ void $ sendMessage px me (toDyn i)+                forever $ void (self px)+              )+              [0 .. n]+            oks <- replicateM+              (length [0, 5 .. n])+              (do+                j <- receiveMessageAs px+                return j+              )+            assertEff "" (sort oks == [0, 5 .. n])+        , testCase "most processes sendShutdown forever"+        $ scheduleAndAssert schedulerFactory+        $ \assertEff -> do+            me <- self px+            traverse_+              (\(i :: Int) -> spawn $ do+                when (i `rem` 5 == 0) $ void $ sendMessage px me (toDyn i)+                forever $ void (sendShutdown px 999)+              )+              [0 .. n]+            oks <- replicateM+              (length [0, 5 .. n])+              (do+                j <- receiveMessageAs px+                return j+              )+            assertEff "" (sort oks == [0, 5 .. n])+        , testCase "most processes spawn forever"+        $ scheduleAndAssert schedulerFactory+        $ \assertEff -> do+            me <- self px+            traverse_+              (\(i :: Int) -> spawn $ do+                when (i `rem` 5 == 0) $ void $ sendMessage px me (toDyn i)+                parent <- self px+                forever+                  $ void+                      (spawn+                        (void+                          (sendMessage px parent (toDyn "test msg from child")+                          )+                        )+                      )+              )+              [0 .. n]+            oks <- replicateM+              (length [0, 5 .. n])+              (do+                j <- receiveMessageAs px+                return j+              )+            assertEff "" (sort oks == [0, 5 .. n])+        , testCase "most processes receive forever"+        $ scheduleAndAssert schedulerFactory+        $ \assertEff -> do+            me <- self px+            traverse_+              (\(i :: Int) -> spawn $ do+                when (i `rem` 5 == 0) $ void $ sendMessage px me (toDyn i)+                forever $ void (receiveMessage px)+              )+              [0 .. n]+            oks <- replicateM+              (length [0, 5 .. n])+              (do+                j <- receiveMessageAs px+                return j+              )+            assertEff "" (sort oks == [0, 5 .. n])+        ] -exitTests :: forall r . (Member (Logs String) r, SetMember Lift (Lift IO) r)-          => IO (Eff (Process r ': r) () -> IO ()) -> TestTree+exitTests+  :: forall r+   . (Member (Logs String) r, SetMember Lift (Lift IO) r)+  => IO (Eff (Process r ': r) () -> IO ())+  -> TestTree exitTests schedulerFactory =-  let px :: SchedulerProxy r-      px = SchedulerProxy+  let+    px :: SchedulerProxy r+    px = SchedulerProxy   in-    testGroup "process exit tests" $-    [ testGroup "async exceptions"-       [ testCase ("a process dies immediately if a "-                   ++ show e ++ " is thrown, while " ++ busyWith )-         $ do tidVar <- newEmptyTMVarIO-              schedulerDoneVar <- newEmptyTMVarIO-              void $ forkIO $-                do void $ try @SomeException $ void $ applySchedulerFactory schedulerFactory-                     $ do tid <- lift $ myThreadId+    testGroup "process exit tests"+      $ [ testGroup+          "async exceptions"+          [ testCase+                (  "a process dies immediately if a "+                ++ show e+                ++ " is thrown, while "+                ++ busyWith+                )+              $ do+                  tidVar           <- newEmptyTMVarIO+                  schedulerDoneVar <- newEmptyTMVarIO+                  void $ forkIO $ do+                    void+                      $ try @SomeException+                      $ void+                      $ applySchedulerFactory schedulerFactory+                      $ do+                          tid <- lift $ myThreadId                           lift $ atomically $ putTMVar tidVar tid                           forever busyEffect-                   atomically $ putTMVar schedulerDoneVar ()-              tid <- atomically $ takeTMVar tidVar-              threadDelay 1000-              throwTo tid e-              void $ atomically $ takeTMVar schedulerDoneVar-       | e <- [ ThreadKilled, UserInterrupt, HeapOverflow, StackOverflow ]-       , (busyWith, busyEffect) <--         [ ("receiving", void (send (ReceiveMessage @r)))-         , ("sending", void (send (SendMessage @r 44444 (toDyn "test message"))))-         , ("sending shutdown", void (send (SendShutdown @r 44444)))-         , ("selfpid-ing", void (send (SelfPid @r)))-         , ("spawn-ing", void (send (Spawn @r (void (send (ReceiveMessage @r))))))-         , ("sleeping", lift (threadDelay 100000))-         ]-       ]-    , testGroup "main thread exit not blocked by"-       [ testCase ("a child process, busy with "++ busyWith)-        $ applySchedulerFactory schedulerFactory-        $ do void $ spawn $ forever busyEffect-             lift (threadDelay 10000)-       |  (busyWith, busyEffect) <--        [ ("receiving", void (send (ReceiveMessage @r)))-        , ("sending", void (send (SendMessage @r 44444 (toDyn "test message"))))-        , ("sending shutdown", void (send (SendShutdown @r 44444)))-        , ("selfpid-ing", void (send (SelfPid @r)))-        , ("spawn-ing", void (send (Spawn @r (void (send (ReceiveMessage @r))))))+                    atomically $ putTMVar schedulerDoneVar ()+                  tid <- atomically $ takeTMVar tidVar+                  threadDelay 1000+                  throwTo tid e+                  void $ atomically $ takeTMVar schedulerDoneVar+          | e <- [ThreadKilled, UserInterrupt, HeapOverflow, StackOverflow]+          , (busyWith, busyEffect) <-+            [ ("receiving", void (send (ReceiveMessage @r)))+            , ( "sending"+              , void (send (SendMessage @r 44444 (toDyn "test message")))+              )+            , ("sending shutdown", void (send (SendShutdown @r 44444)))+            , ("selfpid-ing"     , void (send (SelfPid @r)))+            , ( "spawn-ing"+              , void (send (Spawn @r (void (send (ReceiveMessage @r)))))+              )+            , ("sleeping", lift (threadDelay 100000))+            ]+          ]+        , testGroup+          "main thread exit not blocked by"+          [ testCase ("a child process, busy with " ++ busyWith)+            $ applySchedulerFactory schedulerFactory+            $ do+                void $ spawn $ forever busyEffect+                lift (threadDelay 10000)+          | (busyWith, busyEffect) <-+            [ ("receiving", void (send (ReceiveMessage @r)))+            , ( "sending"+              , void (send (SendMessage @r 44444 (toDyn "test message")))+              )+            , ("sending shutdown", void (send (SendShutdown @r 44444)))+            , ("selfpid-ing"     , void (send (SelfPid @r)))+            , ( "spawn-ing"+              , void (send (Spawn @r (void (send (ReceiveMessage @r)))))+              )+            ]+          ]+        , testGroup+          "one process exits, the other continues unimpaired"+          [ testCase+              (  "process 2 exits with: "+              ++ howToExit+              ++ " - while process 1 is busy with: "+              ++ busyWith+              )+            $ scheduleAndAssert schedulerFactory+            $ \assertEff -> do+                p1 <- spawn $ forever busyEffect+                lift (threadDelay 1000)+                void $ spawn $ do+                  lift (threadDelay 1000)+                  doExit+                lift (threadDelay 100000)+                wasStillRunningP1 <- sendShutdownChecked px p1+                assertEff "the other process was still running"+                          wasStillRunningP1+          | (busyWith , busyEffect) <-+            [ ("receiving", void (send (ReceiveMessage @r)))+            , ( "sending"+              , void (send (SendMessage @r 44444 (toDyn "test message")))+              )+            , ("sending shutdown", void (send (SendShutdown @r 44444)))+            , ("selfpid-ing"     , void (send (SelfPid @r)))+            , ( "spawn-ing"+              , void (send (Spawn @r (void (send (ReceiveMessage @r)))))+              )+            ]+          , (howToExit, doExit    ) <-+            [ ("normally"        , void (exitNormally px))+            , ("simply returning", return ())+            , ("raiseError"      , void (raiseError px "test error raised"))+            , ("exitWithError"   , void (exitWithError px "test error exit"))+            , ( "sendShutdown to self"+              , do+                me <- self px+                void (sendShutdown px me)+              )+            ]+          ]         ] -       ]-    , testGroup "one process exits, the other continues unimpaired"-       [ testCase ("process 2 exits with: "++ howToExit-                    ++ " - while process 1 is busy with: " ++ busyWith)-        $ scheduleAndAssert schedulerFactory-        $ \assertEff ->-          do p1 <- spawn $ forever busyEffect-             lift (threadDelay 1000)-             tlog "Yeah"-             void $ spawn $ do lift (threadDelay 1000)-                               doExit-             lift (threadDelay 100000)-             wasStillRunningP1 <- sendShutdown px p1-             assertEff "the other process was still running" wasStillRunningP1 -       | (busyWith, busyEffect) <--         [ ("receiving", void (send (ReceiveMessage @r)))-         , ("sending", void (send (SendMessage @r 44444 (toDyn "test message"))))-         , ("sending shutdown", void (send (SendShutdown @r 44444)))-         , ("selfpid-ing", void (send (SelfPid @r)))-         , ("spawn-ing", void (send (Spawn @r (void (send (ReceiveMessage @r))))))-         ]-       , (howToExit, doExit) <--         [ ("normally", void (exitNormally px))-         , ("simply returning", return ())-         , ("raiseError", void (raiseError px "test error raised"))-         , ("exitWithError", void (exitWithError px "test error exit"))-         , ("sendShutdown to self", do me <- self px-                                       void (sendShutdown px me))-         ]-       ]-     ]---sendShutdownTests :: forall r . (Member (Logs String) r, SetMember Lift (Lift IO) r)-                  => IO (Eff (Process r ': r) () -> IO ()) -> TestTree+sendShutdownTests+  :: forall r+   . (Member (Logs String) r, SetMember Lift (Lift IO) r)+  => IO (Eff (Process r ': r) () -> IO ())+  -> TestTree sendShutdownTests schedulerFactory =   let px :: SchedulerProxy r       px = SchedulerProxy-  in-    testGroup "sendShutdown"-    [ testCase "... self"-      $ applySchedulerFactory schedulerFactory-      $ do me <- self px-           void $ sendShutdown px me-           raiseError px "sendShutdown must not return"-    , testCase "... self low-level"-      $ scheduleAndAssert schedulerFactory-      $ \assertEff ->-          do  me <- self px-              r <- send (SendShutdown @r me)-              assertEff-                "ShutdownRequested must be returned"-                (case r of-                    ShutdownRequested -> True-                    _ -> False)--    , testGroup "... other process"-      [ testCase "while it is sending"+  in  testGroup+        "sendShutdown"+        [ testCase "... self" $ applySchedulerFactory schedulerFactory $ do+          me <- self px+          void $ sendShutdown px me+          raiseError px "sendShutdown must not return"+        , testCase "... self low-level"         $ scheduleAndAssert schedulerFactory-        $ \assertEff ->-          do  me <- self px-              other <--                spawn (do untilShutdown (SendMessage @r 666 (toDyn "test"))-                          void (sendMessage px me (toDyn "OK")))+        $ \assertEff -> do+            me <- self px+            r  <- send (SendShutdown @r me)+            assertEff+              "ShutdownRequested must be returned"+              (case r of+                ShutdownRequested -> True+                _                 -> False+              )+        , testGroup+          "... other process"+          [ testCase "while it is sending"+          $ scheduleAndAssert schedulerFactory+          $ \assertEff -> do+              me    <- self px+              other <- spawn+                (do+                  untilShutdown (SendMessage @r 666 (toDyn "test"))+                  void (sendMessage px me (toDyn "OK"))+                )               void (sendShutdown px other)               a <- receiveMessageAs px               assertEff "" (a == "OK")-      , testCase "while it is receiving"-        $ scheduleAndAssert schedulerFactory-        $ \assertEff ->-          do  me <- self px-              other <--                spawn (do untilShutdown (ReceiveMessage @r)-                          void (sendMessage px me (toDyn "OK")))+          , testCase "while it is receiving"+          $ scheduleAndAssert schedulerFactory+          $ \assertEff -> do+              me    <- self px+              other <- spawn+                (do+                  untilShutdown (ReceiveMessage @r)+                  void (sendMessage px me (toDyn "OK"))+                )               void (sendShutdown px other)               a <- receiveMessageAs px               assertEff "" (a == "OK")-      , testCase "while it is self'ing"-        $ scheduleAndAssert schedulerFactory-        $ \assertEff ->-          do  me <- self px-              other <--                spawn (do untilShutdown (SelfPid @r)-                          void (sendMessage px me (toDyn "OK")))+          , testCase "while it is self'ing"+          $ scheduleAndAssert schedulerFactory+          $ \assertEff -> do+              me    <- self px+              other <- spawn+                (do+                  untilShutdown (SelfPid @r)+                  void (sendMessage px me (toDyn "OK"))+                )               void (sendShutdown px other)               a <- receiveMessageAs px               assertEff "" (a == "OK")-      , testCase "while it is spawning"-        $ scheduleAndAssert schedulerFactory-        $ \assertEff ->-          do  me <- self px-              other <--                spawn (do untilShutdown (Spawn @r (return ()))-                          void (sendMessage px me (toDyn "OK")))+          , testCase "while it is spawning"+          $ scheduleAndAssert schedulerFactory+          $ \assertEff -> do+              me    <- self px+              other <- spawn+                (do+                  untilShutdown (Spawn @r (return ()))+                  void (sendMessage px me (toDyn "OK"))+                )               void (sendShutdown px other)               a <- receiveMessageAs px               assertEff "" (a == "OK")-      , testCase "while it is sending shutdown messages"-        $ scheduleAndAssert schedulerFactory-        $ \assertEff ->-          do  me <- self px-              other <--                spawn (do untilShutdown (SendShutdown @r 777)-                          void (sendMessage px me (toDyn "OK")))+          , testCase "while it is sending shutdown messages"+          $ scheduleAndAssert schedulerFactory+          $ \assertEff -> do+              me    <- self px+              other <- spawn+                (do+                  untilShutdown (SendShutdown @r 777)+                  void (sendMessage px me (toDyn "OK"))+                )               void (sendShutdown px other)               a <- receiveMessageAs px               assertEff "" (a == "OK")-      ]-    ]+          ]+        ]

@@ -1,37 +1,48 @@-module SingleThreadedScheduler where--import Control.Eff.Concurrent.Process-import Control.Eff.Concurrent.Process.SingleThreadedScheduler as Scheduler-import Control.Monad (void)-import Test.Tasty-import Test.Tasty.HUnit-import Data.Dynamic-import Common+module SingleThreadedScheduler+where -timeoutSeconds :: Integer -> Timeout-timeoutSeconds seconds = Timeout (seconds * 1000000) (show seconds ++ "s")+import           Control.Eff.Concurrent.Process+import           Control.Eff.Concurrent.Process.SingleThreadedScheduler+                                               as Scheduler+import           Control.Monad                  ( void+                                                )+import           Test.Tasty+import           Test.Tasty.HUnit+import           Data.Dynamic+import           Common  test_mainProcessSpawnsAChildAndExitsNormally :: TestTree-test_mainProcessSpawnsAChildAndExitsNormally =-  setTravisTestOptions-  (testCase "spawn a child and exit normally"-   (Scheduler.defaultMain-     (do void (spawn (void (receiveMessage singleThreadedIoScheduler)))-         void (exitNormally singleThreadedIoScheduler)-         fail "This should not happen!!"-     )))+test_mainProcessSpawnsAChildAndExitsNormally = setTravisTestOptions+    (testCase+        "spawn a child and exit normally"+        (Scheduler.defaultMain+            (do+                void (spawn (void (receiveMessage singleThreadedIoScheduler)))+                void (exitNormally singleThreadedIoScheduler)+                fail "This should not happen!!"+            )+        )+    ) + test_mainProcessSpawnsAChildBothExitNormally :: TestTree-test_mainProcessSpawnsAChildBothExitNormally =-  setTravisTestOptions-  (testCase "spawn a child and let it exit and exit"-   (Scheduler.defaultMain-     (do child <- spawn-                 (do void (receiveMessageAs @String singleThreadedIoScheduler)-                     void (exitNormally singleThreadedIoScheduler)-                     error "This should not happen (child)!!"-                 )-         True <- sendMessage singleThreadedIoScheduler child (toDyn "test")-         void (exitNormally singleThreadedIoScheduler)-         error "This should not happen!!"-     )))+test_mainProcessSpawnsAChildBothExitNormally = setTravisTestOptions+    (testCase+        "spawn a child and let it exit and exit"+        (Scheduler.defaultMain+            (do+                child <- spawn+                    (do+                        void+                            (receiveMessageAs @String singleThreadedIoScheduler)+                        void (exitNormally singleThreadedIoScheduler)+                        error "This should not happen (child)!!"+                    )+                True <- sendMessageChecked singleThreadedIoScheduler+                                           child+                                           (toDyn "test")+                void (exitNormally singleThreadedIoScheduler)+                error "This should not happen!!"+            )+        )+    )