extensible-effects-concurrent 0.1.3.0 → 0.2.0.2

raw patch · 23 files changed

+2501/−1190 lines, 23 filesdep +HUnitdep +tastydep +tasty-discoverdep ~extensible-effectsdep ~stmPVP ok

version bump matches the API change (PVP)

Dependencies added: HUnit, tasty, tasty-discover, tasty-hunit

Dependency ranges changed: extensible-effects, stm

API changes (from Hackage documentation)

Files

• ChangeLog.md +42/−16
• extensible-effects-concurrent.cabal +58/−39
• src/Control/Eff/Concurrent/Api.hs +1/−6
• src/Control/Eff/Concurrent/Api/Client.hs +43/−41
• src/Control/Eff/Concurrent/Api/Internal.hs +4/−10
• src/Control/Eff/Concurrent/Api/Observer.hs +222/−0
• src/Control/Eff/Concurrent/Api/Server.hs +251/−62
• src/Control/Eff/Concurrent/Dispatcher.hs +0/−490
• src/Control/Eff/Concurrent/Examples.hs +67/−60
• src/Control/Eff/Concurrent/Examples2.hs +119/−52
• src/Control/Eff/Concurrent/MessagePassing.hs +0/−173
• src/Control/Eff/Concurrent/Observer.hs +0/−217
• src/Control/Eff/Concurrent/Process.hs +298/−0
• src/Control/Eff/Concurrent/Process/ForkIOScheduler.hs +535/−0
• src/Control/Eff/Concurrent/Process/SingleThreadedScheduler.hs +190/−0
• src/Control/Eff/ExceptionExtra.hs +43/−4
• src/Control/Eff/InternalExtra.hs +17/−0
• src/Control/Eff/Log.hs +37/−18
• test/Driver.hs +1/−0
• test/ForkIOScheduler.hs +110/−0
• test/ProcessBehaviourTestCases.hs +425/−0
• test/SingleThreadedScheduler.hs +38/−0
• test/Spec.hs +0/−2

@@ -1,35 +1,61 @@ # Changelog for extensible-effects-concurrent ++## 0.2.0.2++  * 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++## 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++### TODO+  * 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 Dispatcher 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 Dispatcher 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.1.3.0+version:        0.2.0.2 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@@ -42,50 +42,69 @@       stm >= 2.4.5,       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.Dispatcher,-      Control.Eff.Concurrent.MessagePassing,-      Control.Eff.Concurrent.Observer+                  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.SingleThreadedScheduler,+                  Control.Eff.Concurrent.Api.Observer   other-modules:-      Control.Eff.Concurrent.Api.Internal,-      Paths_extensible_effects_concurrent,-      Control.Eff.Concurrent.Examples,-      Control.Eff.Concurrent.Examples2-  other-extensions:-      ConstraintKinds,-      DeriveFoldable,-      DeriveFunctor,-      DeriveFunctor,-      DeriveGeneric,-      DeriveTraversable,-      GADTs,-      FlexibleContexts,-      GeneralizedNewtypeDeriving,-      RankNTypes,-      ScopedTypeVariables,-      StandaloneDeriving,-      TemplateHaskell,-      TupleSections,-      TypeApplications,-      TypeInType,-      TypeOperators+                Control.Eff.InternalExtra,+                Control.Eff.Concurrent.Api.Internal,+                Paths_extensible_effects_concurrent,+                Control.Eff.Concurrent.Examples,+                Control.Eff.Concurrent.Examples2+  default-extensions:+                     ConstraintKinds,+                     DeriveFoldable,+                     DeriveFunctor,+                     DeriveFunctor,+                     DeriveGeneric,+                     DeriveTraversable,+                     GADTs,+                     FlexibleContexts,+                     GeneralizedNewtypeDeriving,+                     RankNTypes,+                     ScopedTypeVariables,+                     StandaloneDeriving,+                     TemplateHaskell,+                     TupleSections,+                     TypeApplications,+                     TypeInType,+                     TypeOperators,+                     ViewPatterns   default-language: Haskell2010+  ghc-options: -Wall   test-suite extensible-effects-concurrent-test   type: exitcode-stdio-1.0-  main-is: Spec.hs-  hs-source-dirs:-      test+  main-is: Driver.hs+  hs-source-dirs: test+  other-modules:+                Paths_extensible_effects_concurrent+              , ForkIOScheduler+              , SingleThreadedScheduler+              , ProcessBehaviourTestCases   ghc-options: -threaded -rtsopts -with-rtsopts=-N   build-depends:-      base >=4.9 && <5-    , extensible-effects-concurrent-  other-modules:-      Paths_extensible_effects_concurrent+                base >=4.9 && <5+              , extensible-effects-concurrent+              , extensible-effects+              , tasty+              , tasty-discover+              , tasty-hunit+              , HUnit+              , stm   default-language: Haskell2010+  default-extensions:   TypeApplications+                      , RankNTypes+                      , ScopedTypeVariables+                      , TypeOperators+                      , DataKinds+                      , FlexibleContexts+                      , BangPatterns

@@ -25,15 +25,10 @@   ) where -import           GHC.Stack import           Data.Kind-import           Control.Eff import           Control.Lens-import           Control.Monad import           Data.Typeable (Typeable, typeRep)-import           Data.Proxy-import           Control.Eff.Reader.Lazy-import           Control.Eff.Concurrent.MessagePassing+import           Control.Eff.Concurrent.Process  -- | This data family defines an API implemented by a server. -- The first parameter is the API /index/ and the second parameter

@@ -32,90 +32,92 @@ import           Control.Eff.Reader.Lazy import           Control.Eff.Concurrent.Api import           Control.Eff.Concurrent.Api.Internal-import           Control.Eff.Concurrent.MessagePassing-import           Control.Monad-import           Data.Proxy+import           Control.Eff.Concurrent.Process+import           Data.Dynamic import           Data.Typeable (Typeable, typeRep) import           GHC.Stack  castChecked-  :: forall r o+  :: forall r q o    . ( HasCallStack-     , Member MessagePassing r+     , SetMember Process (Process q) r      , Typeable o      , Typeable (Api o 'Asynchronous)      )-  => Server o+  => SchedulerProxy q+  -> Server o   -> Api o 'Asynchronous   -> Eff r Bool-castChecked (Server pid) callMsg = sendMessage pid (Cast callMsg)+castChecked px (Server pid) callMsg =+  sendMessage px pid (toDyn  (Cast callMsg))  cast-  :: forall r o+  :: forall r q o    . ( HasCallStack-     , Member MessagePassing r+     , SetMember Process (Process q) r      , Typeable o      , Typeable (Api o 'Asynchronous)      )-  => Server o+  => SchedulerProxy q+  -> Server o   -> Api o 'Asynchronous   -> Eff r ()-cast = ((.) . (.)) void castChecked+cast px toServer apiRequest =+  do _ <- castChecked px toServer apiRequest+     return ()  call-  :: forall result api r-   . ( Member MessagePassing r-     , Member Process r+  :: forall result api r q+   . ( SetMember Process (Process q) r      , Typeable api      , Typeable (Api api ( 'Synchronous result))      , Typeable result      , HasCallStack      )-  => Server api-  -> Api api ( 'Synchronous result)-  -> Eff r (Message result)-call (Server pidInt) req = do-  fromPid <- self+  => SchedulerProxy q+  -> Server api+  -> Api api ('Synchronous result)+  -> Eff r result+call px (Server pidInt) req = do+  fromPid <- self px   let requestMessage = Call fromPid req-  wasSent <- sendMessage pidInt requestMessage+  wasSent <- sendMessage px pidInt (toDyn requestMessage)   if wasSent     then       let extractResult :: Response api result -> result           extractResult (Response _pxResult result) = result-      in  do-            mResp <- receiveMessage (Proxy @(Response api result))-            return (extractResult <$> mResp)-    else raiseError+      in extractResult <$> receiveMessageAs px+    else raiseError px       ("Could not send request message " ++ show (typeRep requestMessage))  -type ServesApi o r =+type ServesApi o r q =   ( Typeable o-  , Member MessagePassing r-  , Member Process r+  , SetMember Process (Process q) r   , Member (Reader (Server o)) r   )  registerServer :: Server o -> Eff ( Reader (Server o) ': r ) a -> Eff r a registerServer = flip runReader -callRegistered :: (Typeable reply, ServesApi o r)-      => Api o ('Synchronous reply) -> Eff r (Message reply)-callRegistered method = do+callRegistered :: (Typeable reply, ServesApi o r q)+      => SchedulerProxy q -> Api o ('Synchronous reply) -> Eff r reply+callRegistered px method = do   serverPid <- ask-  call serverPid method+  call px serverPid method  callRegisteredA-  :: (Alternative f, Typeable f, Typeable reply, ServesApi o r)-  => 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 method = do-  m <- callRegistered method-  case m of-    ProcessControlMessage _pm -> return empty-    Message r                 -> return r+callRegisteredA px method = do+  catchRaisedError px+    (const (return (empty @f)))+    (callRegistered px method) -castRegistered :: (Typeable o, ServesApi o r) => Api o 'Asynchronous -> Eff r ()-castRegistered method = do+castRegistered :: (Typeable o, ServesApi o r q)+               => SchedulerProxy q -> Api o 'Asynchronous -> Eff r ()+castRegistered px method = do   serverPid <- ask-  cast serverPid method+  cast px serverPid method

@@ -13,7 +13,7 @@ {-# LANGUAGE TypeApplications #-} {-# LANGUAGE GADTs #-} --- | Type safe /server/ API processes+-- | Internal request and response type for casts and calls  module Control.Eff.Concurrent.Api.Internal   ( Request (..)@@ -21,23 +21,17 @@   ) where -import           GHC.Stack-import           Data.Kind-import           Control.Eff-import           Control.Lens-import           Control.Monad-import           Data.Typeable (Typeable, typeRep)+import           Data.Typeable (Typeable) import           Data.Proxy- import           Control.Eff.Concurrent.Api-import           Control.Eff.Concurrent.MessagePassing-+import           Control.Eff.Concurrent.Process  data Request api where   Call :: forall api apiCallReplyType . (Typeable api, Typeable apiCallReplyType, Typeable (Api api ('Synchronous apiCallReplyType)))          => ProcessId -> Api api ('Synchronous apiCallReplyType) -> Request api   Cast :: forall api . (Typeable api, Typeable (Api api 'Asynchronous))          => Api api 'Asynchronous -> Request api+  Terminate :: Maybe String -> Request api   deriving Typeable  data Response api apiCallReplyType where

@@ -0,0 +1,222 @@+-- | Observer Effects+--+-- This module supports the implementation of observers and observables. One+-- more concrete perspective might be to understand observers as event listeners+-- and observables as event sources. The tools in this module are tailored+-- towards 'Control.Eff.Concurrent.Api.Api' endpoints+{-# LANGUAGE IncoherentInstances #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE ConstraintKinds #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE TypeFamilies #-}+{-# LANGUAGE KindSignatures #-}+{-# LANGUAGE DataKinds #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE TemplateHaskell #-}+{-# LANGUAGE DeriveFunctor #-}+{-# LANGUAGE StandaloneDeriving #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE TypeOperators #-}+{-# LANGUAGE TypeApplications #-}+{-# LANGUAGE GADTs #-}+module Control.Eff.Concurrent.Api.Observer+  ( Observer(..)+  , Observable(..)+  , notifyObserver+  , registerObserver+  , forgetObserver+  , SomeObserver(..)+  , notifySomeObserver+  , Observers()+  , manageObservers+  , addObserver+  , removeObserver+  , notifyObservers+  , CallbackObserver+  , spawnCallbackObserver+  ) where++import GHC.Stack+import Data.Dynamic+import Data.Maybe+import Data.Set (Set)+import qualified Data.Set as Set+import Control.Eff+import Control.Eff.Concurrent.Process+import Control.Eff.Concurrent.Api+import Control.Eff.Concurrent.Api.Client+import Control.Eff.Concurrent.Api.Server+import Control.Eff.Log+import Control.Eff.State.Lazy+import Control.Lens+import Control.Monad++-- * Observation API++-- | An 'Api' index that support observation of the+-- another 'Api' that is 'Observable'.+class (Typeable p, Observable o) => Observer p o where+  -- | Wrap the 'Observation' and the 'ProcessId' (i.e. the 'Server')+  -- that caused the observation into a 'Api' value that the+  -- 'Observable' understands.+  observationMessage :: Server o -> Observation o -> Api p 'Asynchronous++-- | An 'Api' index that supports registration and de-registration of+-- 'Observer's.+class (Typeable o, Typeable (Observation o)) => Observable o where+  -- | Type of observations visible on this observable+  data Observation o+  -- | Return the 'Api' value for the 'cast_' that registeres an observer+  registerObserverMessage :: SomeObserver o -> Api o 'Asynchronous+  -- | Return the 'Api' value for the 'cast_' that de-registeres an observer+  forgetObserverMessage :: SomeObserver o -> Api o 'Asynchronous++-- | Send an 'Observation' to an 'Observer'+notifyObserver :: ( SetMember Process (Process q) r+                 , Observable o+                 , Observer p o+                 , HasCallStack+                 )+               => SchedulerProxy q -> Server p -> Server o -> Observation o -> Eff r ()+notifyObserver px observer observed observation =+  cast px observer (observationMessage observed observation)++-- | Send the 'registerObserverMessage'+registerObserver :: ( SetMember Process (Process q) r+                 , Observable o+                 , Observer p o+                 , HasCallStack+                 )+               => SchedulerProxy q -> Server p -> Server o -> Eff r ()+registerObserver px observer observed =+  cast px observed (registerObserverMessage (SomeObserver observer))++-- | Send the 'forgetObserverMessage'+forgetObserver :: ( SetMember Process (Process q) r+                , Observable o+                , Observer p o)+              => SchedulerProxy q -> Server p -> Server o -> Eff r ()+forgetObserver px observer observed =+  cast px observed (forgetObserverMessage (SomeObserver observer))++-- ** Generalized observation++-- | An existential wrapper around a 'Server' of an 'Observer'.+-- Needed to support different types of observers to observe the+-- same 'Observable' in a general fashion.+data SomeObserver o where+  SomeObserver :: (Show (Server p), Typeable p, Observer p o) => Server p -> SomeObserver o++deriving instance Show (SomeObserver o)++instance Ord (SomeObserver o) where+  compare (SomeObserver (Server o1)) (SomeObserver (Server o2)) =+    compare o1 o2++instance Eq (SomeObserver o) where+  (==) (SomeObserver (Server o1)) (SomeObserver (Server o2)) =+    o1 == o2++-- | Send an 'Observation' to 'SomeObserver'.+notifySomeObserver :: ( SetMember Process (Process q) r+                     , Observable o+                     , HasCallStack+                     )+                   => SchedulerProxy q+                   -> Server o+                   -> Observation o+                   -> SomeObserver o+                   -> Eff r ()+notifySomeObserver px observed observation (SomeObserver observer) =+  notifyObserver px observer observed observation++-- ** Manage 'Observers's++-- | Internal state for 'manageobservers'+data Observers o =+  Observers { _observers :: Set (SomeObserver o) }++observers :: Iso' (Observers o) (Set (SomeObserver o))+observers = iso _observers Observers++-- | Keep track of registered 'Observer's Observers can be added and removed,+-- and an 'Observation' can be sent to all registerd observers at once.+manageObservers :: Eff (State (Observers o) ': r) a -> Eff r a+manageObservers = flip evalState (Observers Set.empty)++-- | Add an 'Observer' to the 'Observers' managed by 'manageObservers'.+addObserver+  :: ( SetMember Process (Process q) r+    , Member (State (Observers o)) r+    , Observable o)+  => SomeObserver o -> Eff r ()+addObserver = modify . over observers . Set.insert++-- | Delete an 'Observer' from the 'Observers' managed by 'manageObservers'.+removeObserver+  ::  ( SetMember Process (Process q) r+    , Member (State (Observers o)) r+    , Observable o)+  => SomeObserver o -> Eff r ()+removeObserver = modify . over observers . Set.delete+++-- | Send an 'Observation' to all 'SomeObserver's in the 'Observers' state.+notifyObservers+  :: forall o r q+    . ( Observable o+      , SetMember Process (Process q) r+      , Member (State (Observers o)) r)+  => SchedulerProxy q -> Observation o -> Eff r ()+notifyObservers px observation = do+  me <- asServer @o <$> self px+  os <- view observers <$> get+  mapM_ (notifySomeObserver px me observation) os++-- * Callback 'Observer'++-- | An 'Observer' that schedules the observations to an effectful callback.+data CallbackObserver o+  deriving Typeable++data instance Api (CallbackObserver o) r where+  CbObserved :: (Typeable o, Typeable (Observation o)) =>+             Server o -> Observation o -> Api (CallbackObserver o) 'Asynchronous+  deriving Typeable++deriving instance Show (Observation o) => Show (Api (CallbackObserver o) r)++instance (Observable o) => Observer (CallbackObserver o) o where+  observationMessage = CbObserved++-- | Start a new process for an 'Observer' that schedules+-- all observations to an effectful callback.+spawnCallbackObserver+  :: forall o r q .+  ( SetMember Process (Process q) r+  , Typeable o+  , Show (Observation o)+  , Observable o+  , Member (Logs String) q)+  => SchedulerProxy q+  -> (Server o -> Observation o -> Eff (Process q ': q) Bool)+  -> Eff r (Server (CallbackObserver o))+spawnCallbackObserver px onObserve =+  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))+      loopUntil)+ where+   handleCast k (CbObserved fromSvr v) =+     onObserve fromSvr v >>= flip when k

@@ -16,84 +16,273 @@ -- | Type safe /server/ API processes  module Control.Eff.Concurrent.Api.Server-  ( ApiHandler (..), serve, serve_, unhandledCallError, unhandledCastError )+  ( ApiHandler (..)+  , serve+  , unhandledCallError+  , unhandledCastError+  , defaultTermination+  , serveBoth+  , serve3+  , tryApiHandler+  , UnhandledRequest()+  , catchUnhandled+  , ensureAllHandled+  , castMessage+  , exitUnhandled+  ) where  import           Control.Eff+import           Control.Eff.InternalExtra+import qualified Control.Eff.Exception as Exc import           Control.Eff.Concurrent.Api import           Control.Eff.Concurrent.Api.Internal-import           Control.Eff.Concurrent.MessagePassing-import           Control.Lens-import           Control.Monad-import           Data.Kind+import           Control.Eff.Concurrent.Process import           Data.Proxy import           Data.Typeable (Typeable, typeRep)+import           Data.Dynamic import           GHC.Stack -data ApiHandler p r e where+-- | Receive messages until the process exits and invoke the callback on each+-- message.+serve+  :: forall r q p+   . (Typeable p, SetMember Process (Process q) r, HasCallStack)+  => 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))++-- | A record of callbacks requests to a /server/, serving a specific 'Api' family instance.+data ApiHandler p r where   ApiHandler ::-     { _handleCast-         :: (Typeable p, Typeable (Api p 'Asynchronous), HasCallStack)-         => Api p 'Asynchronous -> Eff r e+     { -- | A cast will not return a result directly. This is used for async+       -- methods.+       _handleCast+         :: HasCallStack+         => Api p 'Asynchronous -> Eff r ()+      -- | A call is a blocking operation, the caller is blocked until this+      -- handler calls the reply continuation.      , _handleCall-         :: forall x . (Typeable p, Typeable (Api p ('Synchronous x)), Typeable x, HasCallStack)-         => Api p ('Synchronous x) -> (x -> Eff r Bool) -> Eff r e+         :: forall x . HasCallStack+         => Api p ('Synchronous x) -> (x -> Eff r Bool) -> 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.+     -- The default behavior is defined in 'defaultTermination'.      , _handleTerminate-         :: (Typeable p, HasCallStack)-         => String -> Eff r ()-     } -> ApiHandler p r e--serve_-  :: forall r p-   . (Typeable p, Member MessagePassing r, Member Process r, HasCallStack)-  => ApiHandler p r ()-  -> Eff r ()-serve_ = void . serve+         :: HasCallStack+         => Maybe String -> Eff r ()+     } -> ApiHandler p r -serve-  :: forall r p e-   . (Typeable p, Member MessagePassing r, Member Process r, HasCallStack)-  => ApiHandler p r e-  -> Eff r (Message e)-serve (ApiHandler handleCast handleCall handleTerminate) = do-  mReq <- receiveMessage (Proxy @(Request p))-  mapM receiveCallReq mReq >>= catchProcessControlMessage- where-  catchProcessControlMessage :: Message e -> Eff r (Message e)-  catchProcessControlMessage s@(ProcessControlMessage msg) =-    handleTerminate msg >> return s-  catchProcessControlMessage s = return s+-- | Apply either the '_handleCall', '_handleCast' or the '_handleTerminate'+-- callback to an incoming 'Request'.+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)) -  receiveCallReq :: Request p -> Eff r e-  receiveCallReq (Cast request        ) = handleCast request-  receiveCallReq (Call fromPid request) = handleCall request-                                                     (sendReply request)-   where-    sendReply :: Typeable x => Api p ( 'Synchronous x) -> x -> Eff r Bool-    sendReply _ reply = sendMessage fromPid (Response (Proxy :: Proxy p) reply)+-- * ApiHandler default callbacks +-- | A default handler to use in '_handleCall' in 'ApiHandler'. It will call+-- 'raiseError' with a nice error message. unhandledCallError-  :: ( Show (Api p ( 'Synchronous x))-     , Typeable p-     , Typeable (Api p ( 'Synchronous x))-     , Typeable x-     , HasCallStack-     , Member Process r-     )-  => Api p ( 'Synchronous x)+  :: 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)-  -> Eff r e-unhandledCallError api _ = raiseError-  ("Unhandled call: (" ++ show api ++ " :: " ++ show (typeRep api) ++ ")")+  -> Eff r ()+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-  :: ( Show (Api p 'Asynchronous)-     , Typeable p-     , Typeable (Api p 'Asynchronous)-     , HasCallStack-     , Member Process r-     )-  => Api p 'Asynchronous-  -> Eff r e-unhandledCastError api = raiseError-  ("Unhandled cast: (" ++ show api ++ " :: " ++ show (typeRep api) ++ ")")+  :: 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)) ++ ")")++-- | 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++-- * Multiple ApiHandler combined++-- | '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)++-- | '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)++-- | 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+-- invokation of this function for different 'ApiHandler's can be tried, by using 'catchUnhandled'.+--+-- > tryApiHandler px handlers1 message+-- >   `catchUnhandled`+-- > tryApiHandler px handlers2+-- >   `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)++-- | 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+-- 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+-- 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)+  => Eff r a+  -> (Dynamic -> Eff r a)+  -> Eff r a+catchUnhandled effect handler =+  effect `Exc.catchError` (handler . fromUnhandledRequest)++-- | Catch 'UnhandledRequest's and terminate the process with an error, if necessary.+ensureAllHandled+  :: 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++-- | 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)++-- | 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+-- exit the process with a corresponding error.+exitUnhandled+  :: 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

@@ -1,490 +0,0 @@--- | Implement Erlang style message passing concurrency.------ This handles the 'MessagePassing' and 'Process' effects, using--- 'STM.TQueue's and 'forkIO'.------ This aims to be a pragmatic implementation, so even logging is--- supported.------ At the core is a /main process/ that enters 'runMainProcess'--- and creates all of the internal state stored in 'STM.TVar's--- to manage processes with message queues.------ The 'Eff' handler for 'Process' and 'MessagePassing' use--- are implemented and available through 'spawn'.------ 'spawn' uses 'forkFinally' and 'STM.TQueue's and tries to catch--- most exceptions.-{-# LANGUAGE GeneralizedNewtypeDeriving #-}-{-# LANGUAGE ConstraintKinds #-}-{-# LANGUAGE FlexibleContexts #-}-{-# LANGUAGE FlexibleInstances #-}-{-# LANGUAGE MultiParamTypeClasses #-}-{-# LANGUAGE TypeFamilies #-}-{-# LANGUAGE KindSignatures #-}-{-# LANGUAGE DataKinds #-}-{-# LANGUAGE RankNTypes #-}-{-# LANGUAGE TemplateHaskell #-}-{-# LANGUAGE DeriveFunctor #-}-{-# LANGUAGE StandaloneDeriving #-}-{-# LANGUAGE ScopedTypeVariables #-}-{-# LANGUAGE TypeOperators #-}-{-# LANGUAGE TypeApplications #-}-{-# LANGUAGE GADTs #-}-module Control.Eff.Concurrent.Dispatcher-  ( runMainProcess-  , defaultMain-  , spawn-  , DispatcherError(..)-  , DispatcherIO-  , HasDispatcherIO-  , ProcIO-  )-where--import           GHC.Stack-import           Data.Maybe-import           Data.Kind-import qualified Control.Exception             as Exc-import           Control.Concurrent            as Concurrent-import           Control.Concurrent.STM        as STM-import           Control.Eff-import           Control.Eff.Concurrent.MessagePassing-import           Control.Eff.ExceptionExtra-import           Control.Eff.Lift-import           Control.Eff.Log-import           Control.Eff.Reader.Strict     as Reader-import           Control.Lens-import           Control.Monad                  ( when-                                                , void-                                                )-import qualified Control.Monad.State           as Mtl-import           Data.Dynamic-import           Data.Typeable                  ( typeRep )-import           Data.Map                       ( Map )-import qualified Data.Map                      as Map---- | Information about a process, needed to implement 'MessagePassing' and--- 'Process' handlers. The message queue is backed by a 'STM.TQueue'.-data ProcessInfo =-                 ProcessInfo { _processId       :: ProcessId-                             , _messageQ        :: STM.TQueue Dynamic-                             , _exitOnShutdown  :: Bool-                             }--makeLenses ''ProcessInfo--instance Show ProcessInfo where-  show p =-    "ProcessInfo: " ++ show (p ^. processId) ++ " trapExit: "-                      ++ show (not (p^.exitOnShutdown))---- | Contains all 'ProcessInfo' elements, as well as the state needed to--- implement inter process communication. It contains also a 'LogChannel' to--- which the logs of all processes are forwarded to.-data Dispatcher =-               Dispatcher { _nextPid :: ProcessId-                          , _processTable :: Map ProcessId ProcessInfo-                          , _threadIdTable :: Map ProcessId ThreadId-                          , _schedulerShuttingDown :: Bool-                          , _logChannel :: LogChannel String-                          }--makeLenses ''Dispatcher---- | A newtype wrapper around an 'STM.TVar' holding a 'Dispatcher' state.--- This is needed by 'spawn' and provided by 'runDispatcher'.-newtype DispatcherVar = DispatcherVar { fromDispatcherVar :: STM.TVar Dispatcher }-  deriving Typeable---- | A sum-type with errors that can occur when dispatching messages.-data DispatcherError =-   UnhandledMessageReceived Dynamic ProcessId-   -- ^ A process message queue contained a bad message and the 'Dynamic' value-   -- could not be converted to the expected value using 'fromDynamic'.-  | ProcessNotFound ProcessId-    -- ^ No 'ProcessInfo' was found for a 'ProcessId' during internal-    -- processing. NOTE: This is **ONLY** caused by internal errors, probably by-    -- an incorrect 'MessagePassing' handler in this module. **Sending a message-    -- to a process ALWAYS succeeds!** Even if the process does not exist.-  | ProcessException String ProcessId-    -- ^ A process called 'raiseError'.-  | DispatcherShuttingDown-    -- ^ An action was not performed while the dispatcher was exiting.-  | LowLevelIOException Exc.SomeException-    -- ^ 'Control.Exception.SomeException' was caught while dispatching-    -- messages.-  deriving (Typeable, Show)--instance Exc.Exception DispatcherError---- | An alias for the constraints for the effects essential to this dispatcher--- implementation, i.e. these effects allow 'spawn'ing new 'Process'es.--- @see DispatcherIO-type HasDispatcherIO r = ( HasCallStack-                        , SetMember Lift (Lift IO) r-                        , Member (Exc DispatcherError) r-                        , Member (Logs String) r-                        , Member (Reader DispatcherVar) r)---- | The concrete list of 'Eff'ects for this scheduler implementation.--- @see HasDispatcherIO-type DispatcherIO =-              '[ Exc DispatcherError-               , Reader DispatcherVar-               , Logs String-               , Lift IO-               ]---- | The concrete list of 'Eff'ects that provide 'MessagePassing' and--- 'Process'es ontop of 'DispatcherIO'-type ProcIO = ConsProcIO DispatcherIO---- | /Cons/ 'ProcIO' onto a list of effects.-type ConsProcIO r = MessagePassing ': Process ': r--instance MonadLog String (Eff ProcIO) where-  logMessageFree = logMessageFreeEff---- | This is the main entry point to running a message passing concurrency--- application. This function takes a 'ProcIO' effect and a 'LogChannel' for--- concurrent logging.-runMainProcess :: Eff ProcIO a -> LogChannel String -> IO a-runMainProcess e logC = withDispatcher-  (dispatchMessages-    (\cleanup -> do-      mt <- lift myThreadId-      mp <- self-      logMessage (show mp ++ " main process started in thread " ++ show mt)-      res <- try e-      case res of-        Left ex ->-          do-              logMessage-                (  show mp-                ++ " main process exception: "-                ++ ((show :: DispatcherError -> String) ex)-                )-              lift (runCleanUpAction cleanup)-            >> throwError ex-        Right rres -> do-          logMessage (show mp ++ " main process exited")-          lift (runCleanUpAction cleanup)-          return rres-    )-  )-  where-    withDispatcher :: Eff DispatcherIO a -> IO a-    withDispatcher mainProcessAction = do-      myTId <- myThreadId-      Exc.bracket-        (newTVarIO (Dispatcher myPid Map.empty Map.empty False logC))-        (tearDownDispatcher myTId)-        (\sch -> runLift-          (forwardLogsToChannel-            logC-            (runReader (runErrorRethrowIO mainProcessAction) (DispatcherVar sch))-          )-        )-     where-      myPid = 1-      tearDownDispatcher myTId v = do-        logChannelPutIO logC (show myTId ++" begin dispatcher tear down")-        sch <--          (atomically-            (do-              sch <- readTVar v-              let sch' = sch & schedulerShuttingDown .~ True-              writeTVar v sch'-              return sch-            )-          )-        logChannelPutIO logC (show myTId ++ " killing threads: " ++-                                   show (sch ^.. threadIdTable. traversed))-        imapM_ (killProcThread myTId) (sch ^. threadIdTable)-        atomically-          (do-              dispatcher <- readTVar v-              let allThreadsDead = dispatcher^.threadIdTable.to Map.null-                                   && dispatcher^.processTable.to Map.null-              STM.check allThreadsDead)-        logChannelPutIO logC "all threads dead"---      killProcThread myTId pid tid = when-        (myTId /= tid)-        (  logChannelPutIO logC ("killing thread " ++ show pid)-        >> killThread tid-        )----- | Start the message passing concurrency system then execute a 'ProcIO' effect.--- All logging is sent to standard output.-defaultMain :: Eff ProcIO a -> IO a-defaultMain c =-  runLoggingT-    (logChannelBracket-      (Just "~~~~~~ main process started")-      (Just "====== main process exited")-      (runMainProcess c))-    (print :: String -> IO ())---runChildProcess-  :: DispatcherVar-  -> (CleanUpAction -> Eff ProcIO a)-  -> IO (Either DispatcherError a)-runChildProcess s procAction = do-  l <- (view logChannel) <$> atomically (readTVar (fromDispatcherVar s))-  runLift-    (forwardLogsToChannel-      l-      (runReader (runError (dispatchMessages procAction)) s)-    )--getLogChannel :: HasDispatcherIO r => Eff r (LogChannel String)-getLogChannel = do-  s <- getDispatcherTVar-  lift ((view logChannel) <$> atomically (readTVar s))---overProcessInfo-  :: HasDispatcherIO r-  => ProcessId-  -> Mtl.StateT ProcessInfo STM.STM a-  -> Eff r a-overProcessInfo pid stAction = liftEither =<< overDispatcher-  (do-    res <- use (processTable . at pid)-    case res of-      Nothing    -> return (Left (ProcessNotFound pid))-      Just pinfo -> do-        (x, pinfoOut) <- Mtl.lift (Mtl.runStateT stAction pinfo)-        processTable . at pid . _Just .= pinfoOut-        return (Right x)-  )---- ** MessagePassing execution--spawn :: HasDispatcherIO r => Eff ProcIO () -> Eff r ProcessId-spawn mfa = do-  schedulerVar <- ask-  pidVar       <- lift newEmptyTMVarIO-  cleanupVar   <- lift newEmptyTMVarIO-  lc           <- getLogChannel-  void-    (lift-      (Concurrent.forkFinally-        (runChildProcess-          schedulerVar-          (\cleanUpAction -> do-            lift (atomically (STM.putTMVar cleanupVar cleanUpAction))-            pid <- self-            lift (atomically (STM.putTMVar pidVar (Just pid)))-            catchError-              mfa-              ( logMessage-              . ("process exception: " ++)-              . (show :: DispatcherError -> String)-              )-          )-        )-        (\eres -> do-          mcleanup <- atomically (STM.tryTakeTMVar cleanupVar)-          void (atomically (tryPutTMVar pidVar Nothing))-          case mcleanup of-            Nothing -> return ()-            Just ca -> do-              runCleanUpAction ca-              mt <- myThreadId-              case eres of-                Left se -> logChannelPutIO-                  lc-                  ("thread " ++ show mt ++ " killed by exception: " ++ show se)-                Right _ ->-                  logChannelPutIO lc ("thread " ++ show mt ++ " exited")-        )-      )-    )-  mPid <- lift (atomically (STM.takeTMVar pidVar))-  maybe (throwError DispatcherShuttingDown) return mPid--newtype CleanUpAction = CleanUpAction { runCleanUpAction :: IO () }--dispatchMessages-  :: forall r a-   . (HasDispatcherIO r, HasCallStack)-  => (CleanUpAction -> Eff (ConsProcIO r) a)-  -> Eff r a-dispatchMessages processAction = withMessageQueue-  (\cleanUpAction pinfo ->-     try-     (handle_relay-      return-      (goProc (pinfo ^. processId))-       (handle_relay return (go (pinfo ^. processId))-        (processAction cleanUpAction)-       ))-     >>=-     either-     (\(e :: DispatcherError) ->-        do logMsg (show (pinfo^.processId)-                    ++ " cleanup on exception: "-                    ++ show e)-           lift (runCleanUpAction cleanUpAction)-           throwError e)-     return-  )- where-  go-    :: forall v-     . HasCallStack-    => ProcessId-    -> MessagePassing v-    -> (v -> Eff (Process ': r) a)-    -> Eff (Process ': r) a-  go _pid (SendMessage toPid reqIn) k = do-    psVar <- getDispatcherTVar-    liftRethrow-        LowLevelIOException-        (atomically-          (do-            p <- readTVar psVar-            let mto = p ^. processTable . at toPid-            case mto of-              Just toProc ->-                let dReq = toDyn reqIn-                in  do-                      writeTQueue (toProc ^. messageQ) dReq-                      return True-              Nothing -> return False-          )-        )-      >>= k-  go pid (ReceiveMessage onMsg) k = do-    catchError-      (do-        mDynMsg <- overProcessInfo pid-          (do-            mq <- use messageQ-            Mtl.lift (readTQueue mq))-        case fromDynamic mDynMsg of-          Just req -> let result = onMsg req in k (Message result)-          nix@Nothing ->-            let-              msg =-                "unexpected message: " ++ show mDynMsg ++ " expected: " ++ show-                  (typeRep nix)-            in  do-                  isExitOnShutdown <- overProcessInfo pid (use exitOnShutdown)-                  if isExitOnShutdown-                    then throwError (UnhandledMessageReceived mDynMsg pid)-                    else k (ProcessControlMessage msg)-      )-      (\(se :: DispatcherError) -> do-        isExitOnShutdown <- overProcessInfo pid (use exitOnShutdown)-        if isExitOnShutdown-          then throwError se-          else k (ProcessControlMessage (show se))-      )--  goProc-    :: forall v x-     . HasCallStack-    => ProcessId-    -> Process v-    -> (v -> Eff r x)-    -> Eff r x-  goProc pid SelfPid k = k pid-  goProc pid (TrapExit s) k =-    overProcessInfo pid (exitOnShutdown .= (not s)) >>= k-  goProc pid GetTrapExit k =-    overProcessInfo pid (use exitOnShutdown) >>= k . not-  goProc pid (RaiseError msg) _k = do-    logMsg (show pid ++ " error raised: " ++ msg)-    throwError (ProcessException msg pid)--withMessageQueue-  :: HasDispatcherIO r => (CleanUpAction -> ProcessInfo -> Eff r a) -> Eff r a-withMessageQueue m = do-  mpinfo <- createQueue-  lc     <- getLogChannel-  case mpinfo of-    Just pinfo -> do-      cleanUpAction <--        getDispatcherTVar >>= return . CleanUpAction . destroyQueue-          lc-          (pinfo ^. processId)-      m cleanUpAction pinfo-    Nothing -> throwError DispatcherShuttingDown- where-  createQueue = do-    myTId <- lift myThreadId-    overDispatcher-      (do-        abortNow <- use schedulerShuttingDown-        if abortNow-          then return Nothing-          else do-            pid     <- nextPid <<+= 1-            channel <- Mtl.lift newTQueue-            let pinfo = ProcessInfo pid channel True-            threadIdTable . at pid .= Just myTId-            processTable . at pid .= Just pinfo-            return (Just pinfo)-      )-  destroyQueue lc pid psVar = do-    didWork <- Exc.try-      (overDispatcherIO-        psVar-        (do-          abortNow <- use schedulerShuttingDown-          if abortNow-            then return (Nothing, False)-            else do-              os <- processTable . at pid <<.= Nothing-              ot <- threadIdTable . at pid <<.= Nothing-              return (os, isJust os || isJust ot)-        )-      )-    let getCause =-          Exc.try @Exc.SomeException-              (overDispatcherIO psVar (preuse (processTable . at pid)))-            >>= either-                  (return . (show pid ++) . show)-                  (return . (maybe (show pid) show))--    case didWork of-      Right (pinfo, True) ->-        logChannelPutIO lc ("destroying queue: " ++ show pinfo)-      Right (pinfo, False) ->-        logChannelPutIO lc ("queue already destroyed: " ++ show pinfo)-      Left (e :: Exc.SomeException) ->-        getCause-          >>= logChannelPutIO lc-          .   (("failed to destroy queue: " ++ show e ++ " ") ++)---overDispatcher-  :: HasDispatcherIO r => Mtl.StateT Dispatcher STM.STM a -> Eff r a-overDispatcher stAction = do-  psVar <- getDispatcherTVar-  liftRethrow LowLevelIOException (overDispatcherIO psVar stAction)--overDispatcherIO-  :: STM.TVar Dispatcher -> Mtl.StateT Dispatcher STM.STM a -> IO a-overDispatcherIO psVar stAction = STM.atomically-  (do-    ps                   <- STM.readTVar psVar-    (result, psModified) <- Mtl.runStateT stAction ps-    STM.writeTVar psVar psModified-    return result-  )--getDispatcherTVar :: HasDispatcherIO r => Eff r (TVar Dispatcher)-getDispatcherTVar = fromDispatcherVar <$> ask--getDispatcher :: HasDispatcherIO r => Eff r Dispatcher-getDispatcher = do-  processesVar <- getDispatcherTVar-  lift (atomically (readTVar processesVar))

@@ -24,8 +24,8 @@ import Control.Eff.Concurrent.Api import Control.Eff.Concurrent.Api.Client import Control.Eff.Concurrent.Api.Server-import Control.Eff.Concurrent.MessagePassing-import Control.Eff.Concurrent.Dispatcher+import Control.Eff.Concurrent.Process+import Control.Eff.Concurrent.Process.ForkIOScheduler as Scheduler import Control.Eff.Log import qualified Control.Exception as Exc @@ -35,8 +35,8 @@ data instance Api TestApi x where   SayHello :: String -> Api TestApi ('Synchronous Bool)   Shout :: String -> Api TestApi 'Asynchronous-  SetTrapExit :: Bool -> Api TestApi ('Synchronous ())   Terminate :: Api TestApi ('Synchronous ())+  TerminateError :: String -> Api TestApi ('Synchronous ())   deriving (Typeable)  data MyException = MyException@@ -47,96 +47,103 @@ deriving instance Show (Api TestApi x)  main :: IO ()-main = defaultMain example+main = defaultMain (example forkIoScheduler) +mainProcessSpawnsAChildAndReturns+  :: ( HasCallStack, SetMember Process (Process q) r)+  => SchedulerProxy q -> Eff r ()+mainProcessSpawnsAChildAndReturns px =+  void (spawn (void (receiveMessage px)))+ example   :: ( HasCallStack+    , SetMember Process (Process q) r     , Member (Logs String) r-    , HasDispatcherIO r-    , Member MessagePassing r-    , Member Process r-    , MonadLog String (Eff r)+    , Member (Logs String) q+    , SetMember Lift (Lift IO) q     , SetMember Lift (Lift IO) r)-  => Eff r ()-example = do-  me <- self-  trapExit True-  logMessage ("I am " ++ show me)+  => SchedulerProxy q -> Eff r ()+example px = do+  me <- self px+  logMsg ("I am " ++ show me)   server <- asServer @TestApi <$> spawn testServerLoop-  logMessage ("Started server " ++ show server)+  logMsg ("Started server " ++ show server)   let go = do         x <- lift getLine         case x of-          ('k':_) -> do-            callRegistered Terminate+          ('k':rest) -> do+            callRegistered px (TerminateError rest)             go-          ('c':_) -> do-            castRegistered (Shout x)+          ('s':_) -> do+            callRegistered px Terminate             go-          ('t':'0':_) -> do-            callRegistered (SetTrapExit False)+          ('c':_) -> do+            castRegistered px (Shout x)             go-          ('t':'1':_) -> do-            callRegistered (SetTrapExit True)+          ('r':rest) -> do+            void (replicateM+                  (read rest)+                  (castRegistered px (Shout x)))             go           ('q':_) ->             logMsg "Done."           _ ->-            do res <- ignoreProcessError (callRegistered (SayHello x))+            do res <- ignoreProcessError px (callRegistered px (SayHello x))                logMsg ("Result: " ++ show res)                go   registerServer server go  testServerLoop-  :: forall r. (HasCallStack, Member MessagePassing r, Member Process r-         , MonadLog String (Eff r)-         , SetMember Lift (Lift IO) r)-  => Eff r ()+  :: forall r .+    (HasCallStack+    , Member (Logs String) r+    , SetMember Lift (Lift IO) r)+  => Eff (Process r ': r) () testServerLoop =-  -- trapExit True-    -- >>-    (forever $ serve_ $ ApiHandler handleCast handleCall handleTerminate)+    serve px $ ApiHandler handleCast handleCall handleTerminate   where-    handleCast :: Api TestApi 'Asynchronous -> Eff r ()+    px :: SchedulerProxy r+    px = SchedulerProxy+    handleCast :: Api TestApi 'Asynchronous -> Eff (Process r ': r) ()     handleCast (Shout x) = do-      me <- self-      logMessage (show me ++ " Shouting: " ++ x)-    handleCall :: Api TestApi ('Synchronous x) -> (x -> Eff r Bool) -> Eff r ()+      me <- self px+      logMsg (show me ++ " Shouting: " ++ x)+    handleCall :: Api TestApi ('Synchronous x) -> (x -> Eff (Process r ': r) Bool) -> Eff (Process r ': r) ()     handleCall (SayHello "e1") _reply = do-      me <- self-      logMessage (show me ++ " raising an error")-      raiseError "No body loves me... :,("+      me <- self px+      logMsg (show me ++ " raising an error")+      raiseError px "No body loves me... :,("     handleCall (SayHello "e2") _reply = do-      me <- self-      logMessage (show me ++ " throwing a MyException ")+      me <- self px+      logMsg (show me ++ " throwing a MyException ")       lift (Exc.throw MyException)     handleCall (SayHello "self") reply = do-      me <- self-      logMessage (show me ++ " casting to self")-      cast (asServer @TestApi me) (Shout "from me")+      me <- self px+      logMsg (show me ++ " casting to self")+      cast px (asServer @TestApi me) (Shout "from me")       void (reply False)     handleCall (SayHello "die") reply = do-      me <- self-      logMessage (show me ++ " throwing and catching ")-      catchProcessError-        (\ er -> logMessage ("WOW: " ++ show er ++ " - No. This is wrong!"))-        (raiseError "No body loves me... :,(")+      me <- self px+      logMsg (show me ++ " throwing and catching ")+      catchRaisedError px+        (\ er -> logMsg ("WOW: " ++ show er ++ " - No. This is wrong!"))+        (raiseError px "No body loves me... :,(")       void (reply True)     handleCall (SayHello x) reply = do-      me <- self-      logMessage (show me ++ " Got Hello: " ++ x)+      me <- self px+      logMsg (show me ++ " Got Hello: " ++ x)       void (reply (length x > 3))-    handleCall (SetTrapExit x) reply = do-      me <- self-      logMessage (show me ++ " setting trap exit to " ++ show x)-      trapExit x-      void (reply ())     handleCall Terminate reply = do-      me <- self-      logMessage (show me ++ " exitting")+      me <- self px+      logMsg (show me ++ " exiting")       void (reply ())-      raiseError "DONE"+      exitNormally px+    handleCall (TerminateError msg) reply = do+      me <- self px+      logMsg (show me ++ " exiting with error: " ++ msg)+      void (reply ())+      exitWithError px msg     handleTerminate msg = do-      me <- self-      logMessage (show me ++ " is exiting: " ++ msg)-      raiseError msg+      me <- self px+      logMsg (show me ++ " is exiting: " ++ show msg)+      maybe (exitNormally px) (exitWithError px) msg

@@ -1,4 +1,3 @@-{-# LANGUAGE IncoherentInstances #-} {-# LANGUAGE GeneralizedNewtypeDeriving #-} {-# LANGUAGE ConstraintKinds #-} {-# LANGUAGE FlexibleContexts #-}@@ -19,16 +18,24 @@  import Data.Dynamic import Control.Eff-import Control.Eff.Concurrent.Dispatcher+import qualified Control.Eff.Concurrent.Process.ForkIOScheduler as ForkIO+import qualified Control.Eff.Concurrent.Process.SingleThreadedScheduler as Pure import Control.Eff.Concurrent.Api import Control.Eff.Concurrent.Api.Server import Control.Eff.Concurrent.Api.Client-import Control.Eff.Concurrent.MessagePassing-import Control.Eff.Concurrent.Observer+import Control.Eff.Concurrent.Process+import Control.Eff.Concurrent.Api.Observer import Control.Eff.Log import Control.Eff.State.Lazy import Control.Monad +main :: IO ()+main =+  do ForkIO.defaultMain (counterExample SchedulerProxy)+     Pure.defaultMain (counterExample SchedulerProxy)++-- * First API+ data Counter deriving Typeable  data instance Api Counter x where@@ -46,28 +53,31 @@   registerObserverMessage os = ObserveCounter os   forgetObserverMessage os = UnobserveCounter os -logCounterObservations :: Eff ProcIO (Server (CallbackObserver Counter))-logCounterObservations =-  spawnCallbackObserver+logCounterObservations+  :: (SetMember Process (Process q) r, Member (Logs String) q)+    => SchedulerProxy q -> Eff r (Server (CallbackObserver Counter))+logCounterObservations px =+  spawnCallbackObserver px   (\fromSvr msg ->-     do me <- self-        logMsg (show me ++ " observed on: " ++ show fromSvr ++ ": " ++ show msg)+     do me <- self px+        logMsg (show me+                 ++ " observed on: "+                 ++ show fromSvr+                 ++ ": "+                 ++ show msg)         return True) -type CounterEff = State (Observers Counter) ': State Integer ': ProcIO--data ServerState st a where-  ServerState :: State st a -> ServerState st a--counterServerLoop :: Eff ProcIO ()-counterServerLoop = do-  trapExit True-  evalState (manageObservers-             $ forever-             $ serve_-             $ ApiHandler @Counter handleCast handleCall error) 0+counterHandler :: forall r q .+                 ( Member (State (Observers Counter)) r+                 , Member (State Integer) r+                 , Member (Logs String) r+                 , SetMember Process (Process q) r)+               => SchedulerProxy q+               -> ApiHandler Counter r+counterHandler px =+  ApiHandler @Counter handleCast handleCall (defaultTermination px)  where-   handleCast :: Api Counter 'Asynchronous -> Eff CounterEff ()+   handleCast :: Api Counter 'Asynchronous -> Eff r ()    handleCast (ObserveCounter o) = do      addObserver o    handleCast (UnobserveCounter o) = do@@ -76,41 +86,98 @@      logMsg "Inc"      modify (+ (1 :: Integer))      currentCount <- get-     notifyObservers (CountChanged currentCount)-   handleCall :: Api Counter ('Synchronous x) -> (x -> Eff CounterEff Bool) -> Eff CounterEff ()+     notifyObservers px (CountChanged currentCount)+   handleCall :: Api Counter ('Synchronous x) -> (x -> Eff r Bool) -> Eff r ()    handleCall Cnt reply = do      c <- get      logMsg ("Cnt is " ++ show c)      _ <- reply c      return () --- ** Counter client+-- * Second API -counterExample :: Eff ProcIO ()-counterExample = do-  let cnt sv = do r <- call sv Cnt+data PocketCalc deriving Typeable++data instance Api PocketCalc x where+  Add :: Integer -> Api PocketCalc ('Synchronous Integer)+  AAdd :: Integer -> Api PocketCalc 'Asynchronous++deriving instance Show (Api PocketCalc x)++pocketCalcHandler :: forall r q .+                 ( Member (State Integer) r+                 , Member (Logs String) r+                 , SetMember Process (Process q) r)+               => SchedulerProxy q+               -> ApiHandler PocketCalc r+pocketCalcHandler px =+  ApiHandler @PocketCalc handleCast handleCall (defaultTermination px)+ where+   handleCast :: Api PocketCalc 'Asynchronous -> Eff r ()+   handleCast (AAdd x) = do+     logMsg ("AsyncAdd " ++ show x)+     modify (+ x)+     c <- get @Integer+     logMsg ("Accumulator is " ++ show c)+   handleCall :: Api PocketCalc ('Synchronous x) -> (x -> Eff r Bool) -> Eff r ()+   handleCall (Add x) reply = do+     logMsg ("Add " ++ show x)+     modify (+ x)+     c <- get+     logMsg ("Accumulator is " ++ show c)+     void (reply c)++serverLoop :: forall r q .+                 ( Member (Logs String) r+                 , SetMember Process (Process q) r)+               => SchedulerProxy q+               -> Eff r ()+serverLoop px = do+  evalState @Integer+    (manageObservers @Counter+     (serveBoth px+       (counterHandler px)+       (pocketCalcHandler px)))+    0+++-- ** Counter client+counterExample :: ( Member (Logs String) r+                 , Member (Logs String) q+                 , SetMember Process (Process q) r)+               => SchedulerProxy q+               -> Eff r ()+counterExample px = do+  let cnt sv = do r <- call px sv Cnt                   logMsg (show sv ++ " " ++ show r)-  server1 <- asServer @Counter <$> spawn counterServerLoop-  server2 <- asServer @Counter <$> spawn counterServerLoop-  cast server1 Inc-  cnt server1-  cnt server2-  co1 <- logCounterObservations-  co2 <- logCounterObservations-  registerObserver co1 server1-  registerObserver co2 server2-  cast server1 Inc-  cnt server1-  cast server2 Inc-  cnt server2-  registerObserver co2 server1-  registerObserver co1 server2-  cast server1 Inc-  cnt server1-  cast server2 Inc-  cnt server2-  forgetObserver co2 server1-  cast server1 Inc-  cnt server1-  cast server2 Inc-  cnt server2+  pid1 <- spawn (serverLoop px)+  pid2 <- spawn (serverLoop px)+  let cntServer1 = asServer @Counter pid1+      cntServer2 = asServer @Counter pid2+      calcServer1 = asServer @PocketCalc pid1+      calcServer2 = asServer @PocketCalc pid2+  cast px cntServer1 Inc+  cnt cntServer1+  cnt cntServer2+  co1 <- logCounterObservations px+  co2 <- logCounterObservations px+  registerObserver px co1 cntServer1+  registerObserver px co2 cntServer2+  cast px calcServer1 (AAdd 12313)+  cast px cntServer1 Inc+  cnt cntServer1+  cast px cntServer2 Inc+  cnt cntServer2+  registerObserver px co2 cntServer1+  registerObserver px co1 cntServer2+  cast px cntServer1 Inc+  void (call px calcServer2 (Add 878))+  cnt cntServer1+  cast px cntServer2 Inc+  cnt cntServer2+  forgetObserver px co2 cntServer1+  cast px cntServer1 Inc+  cnt cntServer1+  cast px cntServer2 Inc+  cnt cntServer2+  void $ sendShutdown px pid2

@@ -1,173 +0,0 @@--- | The message passing effect.------ 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.Dispatcher'.-{-# LANGUAGE DeriveTraversable #-}-{-# LANGUAGE DeriveFoldable #-}-{-# LANGUAGE GeneralizedNewtypeDeriving #-}-{-# LANGUAGE ConstraintKinds #-}-{-# LANGUAGE FlexibleContexts #-}-{-# LANGUAGE TypeFamilies #-}-{-# LANGUAGE KindSignatures #-}-{-# LANGUAGE DataKinds #-}-{-# LANGUAGE RankNTypes #-}-{-# LANGUAGE TemplateHaskell #-}-{-# LANGUAGE DeriveFunctor #-}-{-# LANGUAGE StandaloneDeriving #-}-{-# LANGUAGE ScopedTypeVariables #-}-{-# LANGUAGE TypeOperators #-}-{-# LANGUAGE TypeApplications #-}-{-# LANGUAGE GADTs #-}-module Control.Eff.Concurrent.MessagePassing-  ( ProcessId(..)-  , fromProcessId-  , Process(..)-  , self-  , trapExit-  , getTrapExit-  , raiseError-  , catchProcessError-  , ignoreProcessError-  , MessagePassing(..)-  , Message(..)-  , sendMessage-  , receiveMessage-  )-where--import           GHC.Stack-import           Control.Eff-import           Control.Lens-import           Data.Dynamic-import           Data.Kind-import           Data.Proxy-import           Text.Printf----- * Process Effects---- | The process effect is the basis for message passing concurrency. This binds--- the semantics of a process with a process-id, and some process flags, and the--- ability to leave a process early with an error.-data Process b where-  SelfPid :: Process ProcessId-  TrapExit :: Bool -> Process ()-  GetTrapExit :: Process Bool-  RaiseError :: String -> Process b-  --  LinkProcesses :: ProcessId -> ProcessId -> Process ()---- | Returns the 'ProcessId' of the current process.-self :: Member Process r => Eff r ProcessId-self = send SelfPid---- | Set the flag that controls a process reaction to--- exit messages from linked/monitored processes.-trapExit :: Member Process r => Bool -> Eff r ()-trapExit = send . TrapExit---- | Return the 'trapExit' flag.-getTrapExit :: Member Process r => Eff r Bool-getTrapExit = send GetTrapExit---- | Thrown an error, can be caught by 'catchProcessError'.-raiseError :: Member Process r => String -> Eff r b-raiseError = send . RaiseError---- | Catch and handle an error raised by 'raiseError'. Works independent of the--- handler implementation.-catchProcessError-  :: forall r w . Member Process r => (String -> Eff r w) -> Eff r w -> Eff r w-catchProcessError onErr = interpose return go- where-  go :: forall b . Process b -> (b -> Eff r w) -> Eff r w-  go (RaiseError emsg) _k = onErr emsg-  go s                 k  = send s >>= k---- | Like 'catchProcessError' it catches 'raiseError', but instead of invoking a--- user provided handler, the result is wrapped into an 'Either'.-ignoreProcessError-  :: (HasCallStack, Member Process r) => Eff r a -> Eff r (Either String a)-ignoreProcessError = catchProcessError (return . Left) . fmap Right---- | Each process is identified by a single process id, that stays constant--- throughout the life cycle of a process. Also, message sending relies on these--- values to address messages to processes.-newtype ProcessId = ProcessId { _fromProcessId :: Int }-  deriving (Eq,Ord,Typeable,Bounded,Num, Enum, Integral, Real)--instance Read ProcessId where-  readsPrec _ ('<':'0':'.':rest1) =-    case reads rest1 of-      [(c, '.':'0':'>':rest2)] -> [(ProcessId c, rest2)]-      _ -> []-  readsPrec _ _ = []--instance Show ProcessId where-  show (ProcessId c) =-    printf "<0.%d.0>" c--makeLenses ''ProcessId----- * 'MessagePassing' Effect---- | An effect for sending and receiving messages.-data MessagePassing b where-  -- | Send a message to a process addressed by the 'ProcessId'. Sending a-  -- message should **always succeed** and return **immediately**, even if the-  -- destination process does not exist, or does not accept messages of the-  -- given type.-  SendMessage :: Typeable m-              => ProcessId-              -> m-              -> MessagePassing Bool-  -- | Receive a message. This should block until an a message was received. The-  -- pure function may convert the incoming message into something, and the-  -- result is returned as 'Message' value. Another reason why this function-  -- returns, is if a process control message was sent to the process. This can-  -- only occur from inside the runtime system, aka the effect handler-  -- implementation. (Currently there is one in 'Control.Eff.Concurrent.Dispatcher'.)-  ReceiveMessage :: forall e m . (Typeable m, Typeable (Message m))-                 => (m -> e)-                 -> MessagePassing (Message e)---- | When a process invokes 'receiveMessage' a value of this type is returned.--- There are more reasons that 'receiveMessage' might return, one is that a--- message was sent to the process, another might be that in internal, handler--- specific, event occurred for which the process should /wake-up/.-data Message m where-  ProcessControlMessage :: String -> Message m-  Message :: m -> Message m-  deriving (Typeable, Functor, Show, Eq, Ord, Foldable, Traversable)---- | Send a message to a process addressed by the 'ProcessId'.--- @see 'SendMessage'.-sendMessage-  :: forall o r-   . (HasCallStack, Member MessagePassing r, Typeable o)-  => ProcessId-  -> o-  -> Eff r Bool-sendMessage pid message = send (SendMessage pid message)---- | Block until a message was received. Expect a message of the type annotated--- by the 'Proxy'.--- Depending on 'trapExit', this will 'raiseError'.--- @see 'ReceiveMessage'.-receiveMessage-  :: forall o r-   . (HasCallStack, Member MessagePassing r, Member Process r, Typeable o)-  => Proxy o-  -> Eff r (Message o)-receiveMessage _ = do-  res <- send (ReceiveMessage id)-  case res of-    Message               _   -> return res-    ProcessControlMessage msg -> do-      isTrapExit <- getTrapExit-      if isTrapExit-        then return res-        else raiseError ("received exit message: " ++ msg)

@@ -1,217 +0,0 @@--- | Observer Effects------ This module supports the implementation of observers and observables. One--- more concrete perspective might be to understand observers as event listeners--- and observables as event sources. The tools in this module are tailored--- towards 'Control.Eff.Concurrent.Api.Api' endpoints-{-# LANGUAGE IncoherentInstances #-}-{-# LANGUAGE GeneralizedNewtypeDeriving #-}-{-# LANGUAGE ConstraintKinds #-}-{-# LANGUAGE FlexibleContexts #-}-{-# LANGUAGE FlexibleInstances #-}-{-# LANGUAGE MultiParamTypeClasses #-}-{-# LANGUAGE TypeFamilies #-}-{-# LANGUAGE KindSignatures #-}-{-# LANGUAGE DataKinds #-}-{-# LANGUAGE RankNTypes #-}-{-# LANGUAGE TemplateHaskell #-}-{-# LANGUAGE DeriveFunctor #-}-{-# LANGUAGE StandaloneDeriving #-}-{-# LANGUAGE ScopedTypeVariables #-}-{-# LANGUAGE TypeOperators #-}-{-# LANGUAGE TypeApplications #-}-{-# LANGUAGE GADTs #-}-module Control.Eff.Concurrent.Observer-  ( Observer(..)-  , Observable(..)-  , notifyObserver-  , registerObserver-  , forgetObserver-  , SomeObserver(..)-  , notifySomeObserver-  , Observers()-  , manageObservers-  , addObserver-  , removeObserver-  , notifyObservers-  , CallbackObserver-  , spawnCallbackObserver-  ) where--import GHC.Stack-import Data.Dynamic-import Data.Set (Set)-import qualified Data.Set as Set-import Control.Eff-import Control.Eff.Concurrent.MessagePassing-import Control.Eff.Concurrent.Api-import Control.Eff.Concurrent.Api.Client-import Control.Eff.Concurrent.Api.Server-import Control.Eff.Concurrent.Dispatcher-import Control.Eff.Log-import Control.Eff.State.Lazy-import Control.Lens-import Control.Monad---- * Observation API---- | An 'Api' index that support observation of the--- another 'Api' that is 'Observable'.-class (Typeable p, Observable o) => Observer p o where-  -- | Wrap the 'Observation' and the 'ProcessId' (i.e. the 'Server')-  -- that caused the observation into a 'Api' value that the-  -- 'Observable' understands.-  observationMessage :: Server o -> Observation o -> Api p 'Asynchronous---- | An 'Api' index that supports registration and de-registration of--- 'Observer's.-class (Typeable o, Typeable (Observation o)) => Observable o where-  -- | Type of observations visible on this observable-  data Observation o-  -- | Return the 'Api' value for the 'cast_' that registeres an observer-  registerObserverMessage :: SomeObserver o -> Api o 'Asynchronous-  -- | Return the 'Api' value for the 'cast_' that de-registeres an observer-  forgetObserverMessage :: SomeObserver o -> Api o 'Asynchronous---- | Send an 'Observation' to an 'Observer'-notifyObserver :: ( Member Process r-                 , Member MessagePassing r-                 , Observable o-                 , Observer p o-                 , HasCallStack-                 )-               => Server p -> Server o -> Observation o -> Eff r ()-notifyObserver observer observed observation =-  cast observer (observationMessage observed observation)---- | Send the 'registerObserverMessage'-registerObserver :: ( Member Process r-                 , Member MessagePassing r-                 , Observable o-                 , Observer p o-                 , HasCallStack-                 )-               => Server p -> Server o -> Eff r ()-registerObserver observer observed =-  cast observed (registerObserverMessage (SomeObserver observer))---- | Send the 'forgetObserverMessage'-forgetObserver :: ( Member Process r-                , Member MessagePassing r-                , Observable o-                , Observer p o)-              => Server p -> Server o -> Eff r ()-forgetObserver observer observed =-  cast observed (forgetObserverMessage (SomeObserver observer))---- ** Generalized observation---- | An existential wrapper around a 'Server' of an 'Observer'.--- Needed to support different types of observers to observe the--- same 'Observable' in a general fashion.-data SomeObserver o where-  SomeObserver :: (Show (Server p), Typeable p, Observer p o) => Server p -> SomeObserver o--deriving instance Show (SomeObserver o)--instance Ord (SomeObserver o) where-  compare (SomeObserver (Server o1)) (SomeObserver (Server o2)) =-    compare o1 o2--instance Eq (SomeObserver o) where-  (==) (SomeObserver (Server o1)) (SomeObserver (Server o2)) =-    o1 == o2---- | Send an 'Observation' to 'SomeObserver'.-notifySomeObserver :: ( Member Process r-                     , Member MessagePassing r-                     , Observable o-                     , HasCallStack-                     )-                   => Server o-                   -> Observation o-                   -> SomeObserver o-                   -> Eff r ()-notifySomeObserver observed observation (SomeObserver observer) =-  notifyObserver observer observed observation---- ** Manage 'Observers's---- | Internal state for 'manageobservers'-data Observers o =-  Observers { _observers :: Set (SomeObserver o) }--observers :: Iso' (Observers o) (Set (SomeObserver o))-observers = iso _observers Observers---- | Keep track of registered 'Observer's Observers can be added and removed,--- and an 'Observation' can be sent to all registerd observers at once.-manageObservers :: Eff (State (Observers o) ': r) a -> Eff r a-manageObservers = flip evalState (Observers Set.empty)---- | Add an 'Observer' to the 'Observers' managed by 'manageObservers'.-addObserver-  :: ( Member MessagePassing r-    , Member (State (Observers o)) r-    , Observable o)-  => SomeObserver o -> Eff r ()-addObserver = modify . over observers . Set.insert---- | Delete an 'Observer' from the 'Observers' managed by 'manageObservers'.-removeObserver-  ::  ( Member MessagePassing r-    , Member (State (Observers o)) r-    , Observable o)-  => SomeObserver o -> Eff r ()-removeObserver = modify . over observers . Set.delete---- | Send an 'Observation' to all 'SomeObserver's in the 'Observers' state.-notifyObservers-  :: forall o r . ( Observable o-            , Member MessagePassing r-            , Member Process r-            , Member (State (Observers o)) r)-  => Observation o -> Eff r ()-notifyObservers observation = do-  me <- asServer @o <$> self-  os <- view observers <$> get-  mapM_ (notifySomeObserver me observation) os---- * Callback 'Observer'---- | An 'Observer' that dispatches the observations to an effectful callback.-data CallbackObserver o-  deriving Typeable--data instance Api (CallbackObserver o) r where-  CbObserved :: (Typeable o, Typeable (Observation o)) =>-             Server o -> Observation o -> Api (CallbackObserver o) 'Asynchronous-  deriving Typeable--deriving instance Show (Observation o) => Show (Api (CallbackObserver o) r)--instance (Observable o) => Observer (CallbackObserver o) o where-  observationMessage = CbObserved---- | Start a new process for an 'Observer' that dispatches--- all observations to an effectful callback.-spawnCallbackObserver-  :: forall o r . (HasDispatcherIO r, Typeable o, Show (Observation o), Observable o)-  => (Server o -> Observation o -> Eff ProcIO Bool)-  -> Eff r (Server (CallbackObserver o))-spawnCallbackObserver onObserve =-  asServer @(CallbackObserver o)-  <$>-  (spawn $ do-      trapExit True-      me <- asServer @(CallbackObserver o) <$> self-      let loopUntil =-            serve_ (ApiHandler @(CallbackObserver o)-                     (handleCast loopUntil)-                     unhandledCallError-                     (logMsg . ((show me ++ " observer terminating ") ++)))-      loopUntil-  )- where-   handleCast :: Eff ProcIO () -> Api (CallbackObserver o) 'Asynchronous -> Eff ProcIO ()-   handleCast k (CbObserved fromSvr v) = onObserve fromSvr v >>= flip when k

@@ -0,0 +1,298 @@+-- | The message passing effect.+--+-- 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'.+{-# LANGUAGE DeriveTraversable #-}+{-# LANGUAGE DeriveFoldable #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE ConstraintKinds #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE TypeFamilies #-}+{-# LANGUAGE KindSignatures #-}+{-# LANGUAGE DataKinds #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE TemplateHaskell #-}+{-# LANGUAGE DeriveFunctor #-}+{-# LANGUAGE StandaloneDeriving #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE TypeOperators #-}+{-# LANGUAGE TypeApplications #-}+{-# LANGUAGE GADTs #-}+module Control.Eff.Concurrent.Process+  ( ProcessId(..)+  , fromProcessId+  , Process(..)+  , ConsProcess+  , ResumeProcess(..)+  , SchedulerProxy(..)+  , thisSchedulerProxy+  , yieldAndCatchProcess+  , sendMessage+  , spawn+  , receiveMessage+  , receiveMessageAs+  , receiveLoop+  , self+  , sendShutdown+  , exitWithError+  , exitNormally+  , raiseError+  , catchRaisedError+  , ignoreProcessError+  )+where++import           GHC.Stack+import           Control.Eff+import           Control.Lens+import           Control.Monad (forever)+import           Data.Dynamic+import           Data.Kind+import           Text.Printf+++-- * Process Effects++-- | The process effect is the basis for message passing concurrency. This+-- effect describes an interface for concurrent, communicating isolated+-- processes identified uniquely by a process-id.+--+-- Processes can raise exceptions that can be caught, exit gracefully or with an+-- error, or be killed by other processes, with the option of ignoring the+-- shutdown request.+--+-- Process Scheduling is implemented in different modules. All scheduler+-- implementations should follow some basic rules:+--+-- * fair scheduling+--+-- * sending a message does not block+--+-- * receiving a message does block+--+-- * spawning a child blocks only a very moment+--+-- * a newly spawned process shall be scheduled before the parent process after+-- * the spawn+--+-- * when the first process exists, all process should be killed immediately+data Process (r :: [Type -> Type]) b where+  -- | Return the current 'ProcessId'+  SelfPid :: Process r (ResumeProcess ProcessId)+  -- | Start a new process, the new process will execute an effect, the function+  -- will return immediately with a 'ProcessId'.+  Spawn :: Eff (Process r ': r) () -> Process r (ResumeProcess ProcessId)+  -- | Process exit, this is the same as if the function that was applied to a+  -- spawn function returned.+  Shutdown :: Process r a+  -- | Exit the process due to an error, this cannot be caught.+  ExitWithError :: String -> Process  r b+  -- | Raise an error, that can be handled.+  RaiseError :: String -> Process r b+  -- | Request that another a process exits. The targeted process is interrupted+  -- and gets a 'ShutdownRequested', the target process may decide to ignore the shutdown+  -- requests.+  SendShutdown :: ProcessId -> Process r (ResumeProcess Bool)+  --  LinkProcesses :: ProcessId -> ProcessId -> Process ()+  -- | Send a message to a process addressed by the 'ProcessId'. Sending a+  -- message should **always succeed** and return **immediately**, even if the+  -- destination process does not exist, or does not accept messages of the+  -- given type.+  SendMessage :: ProcessId -> Dynamic -> Process r (ResumeProcess Bool)+  -- | Receive a message. This should block until an a message was received. The+  -- pure function may convert the incoming message into something, and the+  -- result is returned as 'ProcessMessage' value. Another reason why this function+  -- returns, is if a process control message was sent to the process. This can+  -- only occur from inside the runtime system, aka the effect handler+  -- implementation. (Currently there is one in 'Control.Eff.Concurrent.Process.ForkIOScheduler'.)+  ReceiveMessage :: Process r (ResumeProcess Dynamic)++-- | Every 'Process' action returns it's actual result wrapped in this type. It+-- will allow to signal errors as well as pass on normal results such as+-- incoming messages.+data ResumeProcess v where+  -- | The process is required to exit.+  ShutdownRequested :: ResumeProcess v+  -- | The process is required to exit from an error condition, that cannot be+  -- recovered from.+  OnError :: String -> ResumeProcess v+  -- | The process may resume to do work, using the given result.+  ResumeWith :: a -> ResumeProcess a+  -- | This indicates that the action did not complete, and maybe retried+  RetryLastAction :: ResumeProcess v+  deriving (Typeable, Foldable, Functor, Show, Eq, Ord, Traversable)++-- | /Cons/ 'Process' onto a list of effects.+type ConsProcess r = Process r ': r++-- | 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)++-- | 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)++-- | 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.+data SchedulerProxy :: [Type -> Type] -> Type where+  SchedulerProxy :: SchedulerProxy q++-- | Return a 'SchedulerProxy' for a 'Process' effect.+thisSchedulerProxy :: Eff (Process r ': r) (SchedulerProxy r)+thisSchedulerProxy = return SchedulerProxy++-- | Send a message to a process addressed by the 'ProcessId'.+-- See 'SendMessage'.+sendMessage+  :: forall r q+   . (HasCallStack, SetMember Process (Process q) r)+  => SchedulerProxy q+  -> ProcessId+  -> Dynamic+  -> Eff r Bool+sendMessage _ pid message =+  yieldProcess (SendMessage pid message)++-- | Send a message to a process addressed by the 'ProcessId'.+-- See 'SendMessage'.+sendShutdown+  :: forall r q+   . (HasCallStack, SetMember Process (Process q) r)+  => SchedulerProxy q+  -> ProcessId+  -> Eff r Bool+sendShutdown _ pid =+  yieldProcess (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)++-- | Block until a message was received.+receiveMessage+  :: forall r q+   . (HasCallStack, SetMember Process (Process q) r)+  => SchedulerProxy q -> Eff r Dynamic+receiveMessage _ = do+  yieldProcess 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++-- | 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)+  => 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)++-- | Returns the 'ProcessId' of the current process.+self :: (HasCallStack, SetMember Process (Process q) r) => SchedulerProxy q -> Eff r ProcessId+self _px = yieldProcess SelfPid++-- | Exit the process.+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++-- | 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++-- | 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+catchRaisedError _ onErr = interpose return go+ where+  go :: forall b . Process q b -> (b -> Eff r w) -> Eff r w+  go (RaiseError emsg) _k = onErr emsg+  go s                 k  = send s >>= k++-- | 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)+ignoreProcessError px = catchRaisedError px (return . Left) . fmap Right++-- | Each process is identified by a single process id, that stays constant+-- throughout the life cycle of a process. Also, message sending relies on these+-- values to address messages to processes.+newtype ProcessId = ProcessId { _fromProcessId :: Int }+  deriving (Eq,Ord,Typeable,Bounded,Num, Enum, Integral, Real)++instance Read ProcessId where+  readsPrec _ ('<':'0':'.':rest1) =+    case reads rest1 of+      [(c, '.':'0':'>':rest2)] -> [(ProcessId c, rest2)]+      _ -> []+  readsPrec _ _ = []++instance Show ProcessId where+  show (ProcessId c) =+    printf "<0.%d.0>" c++makeLenses ''ProcessId

@@ -0,0 +1,535 @@+-- | Implement Erlang style message passing concurrency.+--+-- This handles the 'MessagePassing' and 'Process' effects, using+-- 'STM.TQueue's and 'forkIO'.+--+-- This aims to be a pragmatic implementation, so even logging is+-- supported.+--+-- At the core is a /main process/ that enters 'schedule'+-- and creates all of the internal state stored in 'STM.TVar's+-- to manage processes with message queues.+--+-- The 'Eff' handler for 'Process' and 'MessagePassing' use+-- are implemented and available through 'spawn'.+--+-- 'spawn' uses 'forkFinally' and 'STM.TQueue's and tries to catch+-- most exceptions.+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE ConstraintKinds #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE TypeFamilies #-}+{-# LANGUAGE KindSignatures #-}+{-# LANGUAGE DataKinds #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE TemplateHaskell #-}+{-# LANGUAGE DeriveFunctor #-}+{-# LANGUAGE StandaloneDeriving #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE TypeOperators #-}+{-# LANGUAGE TypeApplications #-}+{-# LANGUAGE GADTs #-}+module Control.Eff.Concurrent.Process.ForkIOScheduler+  ( schedule+  , defaultMain+  , SchedulerError(..)+  , SchedulerIO+  , forkIoScheduler+  , HasSchedulerIO+  )+where++import           Data.Foldable+import           GHC.Stack+import           Data.Bifunctor+import           Data.Maybe+import           Data.Kind ()+import qualified Control.Exception             as Exc+import           Control.Concurrent            as Concurrent+import           Control.Concurrent.STM        as STM+import           Control.Eff+import           Control.Eff.Concurrent.Process+-- import           Control.Eff.ExceptionExtra+import           Control.Eff.Lift+import           Control.Eff.Log+import           Control.Eff.Reader.Strict     as Reader+import           Control.Lens+import           Control.Monad                  ( when+                                                , void+                                                , join+                                                )+import qualified Control.Monad.State           as Mtl+import           Data.Dynamic+import           Data.Map                       ( Map )+import qualified Data.Map                      as Map++-- | Information about a process, needed to implement 'MessagePassing' and+-- 'Process' handlers. The message queue is backed by a 'STM.TQueue' and contains+-- 'MessageQEntry' values.+data ProcessInfo =+                 ProcessInfo { _processId         :: ProcessId+                             , _messageQ          :: STM.TQueue (Maybe Dynamic)+                             , _shutdownRequested :: STM.TVar Bool+                             }++makeLenses ''ProcessInfo++instance Show ProcessInfo where+  show p =  "ProcessInfo: " ++ show (p ^. processId)++-- | Contains all 'ProcessInfo' elements, as well as the state needed to+-- implement inter process communication. It contains also a 'LogChannel' to+-- which the logs of all processes are forwarded to.+data Scheduler =+               Scheduler { _nextPid :: ProcessId+                          , _processTable :: Map ProcessId ProcessInfo+                          , _threadIdTable :: Map ProcessId ThreadId+                          , _schedulerShuttingDown :: Bool+                          , _logChannel :: LogChannel String+                          }++makeLenses ''Scheduler++-- | A newtype wrapper around an 'STM.TVar' holding a 'Scheduler' state.+-- This is needed by 'spawn' and provided by 'runScheduler'.+newtype SchedulerVar = SchedulerVar { fromSchedulerVar :: STM.TVar Scheduler }+  deriving Typeable++-- | A sum-type with errors that can occur when scheduleing messages.+data SchedulerError =+    ProcessNotFound ProcessId+    -- ^ No 'ProcessInfo' was found for a 'ProcessId' during internal+    -- processing. NOTE: This is **ONLY** caused by internal errors, probably by+    -- an incorrect 'MessagePassing' handler in this module. **Sending a message+    -- to a process ALWAYS succeeds!** Even if the process does not exist.+  | ProcessRaisedError String+    -- ^ A process called 'raiseError'.+  | ProcessExitError String+    -- ^ A process called 'exitWithError'.+  | ProcessShuttingDown+    -- ^ A process exits.+  | SchedulerShuttingDown+    -- ^ An action was not performed while the scheduler was exiting.+  deriving (Typeable, Show)++instance Exc.Exception SchedulerError++-- | An alias for the constraints for the effects essential to this scheduler+-- implementation, i.e. these effects allow 'spawn'ing new 'Process'es.+-- See SchedulerIO+type HasSchedulerIO r = ( HasCallStack+                        , SetMember Lift (Lift IO) r+                        , Member (Logs String) r+                        , Member (Reader SchedulerVar) r)++-- | The concrete list of 'Eff'ects for this scheduler implementation.+-- See HasSchedulerIO+type SchedulerIO =+              '[ Reader SchedulerVar+               , Logs String+               , Lift IO+               ]++-- | A 'SchedulerProxy' for 'SchedulerIO'+forkIoScheduler :: SchedulerProxy SchedulerIO+forkIoScheduler = SchedulerProxy++-- | This is the main entry point to running a message passing concurrency+-- 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+          (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)++-- | 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+      (Just "~~~~~~ main process started")+      (Just "====== main process exited")+      (schedule c))+    (print :: String -> IO ())++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++    runProcEffects cleanupVar l =+      Data.Bifunctor.first Exc.SomeException+       <$> runLift+           (forwardLogsToChannel 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"+++getLogChannel :: HasSchedulerIO r => Eff r (LogChannel String)+getLogChannel = do+  s <- getSchedulerVar+  lift (getLogChannelIO s)++getLogChannelIO :: SchedulerVar -> IO (LogChannel String)+getLogChannelIO s =+  (view logChannel) <$> atomically (readTVar (fromSchedulerVar s))++overProcessInfo+  :: HasSchedulerIO r+  => ProcessId+  -> Mtl.StateT ProcessInfo STM.STM a+  -> Eff r (Either SchedulerError a)+overProcessInfo pid stAction =+  overScheduler+  (do+    res <- use (processTable . at pid)+    case res of+      Nothing    -> return (Left (ProcessNotFound pid))+      Just pinfo -> do+        (x, pinfoOut) <- Mtl.lift (Mtl.runStateT stAction pinfo)+        processTable . at pid . _Just .= pinfoOut+        return (Right x))++-- ** MessagePassing execution++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+  lift Concurrent.yield -- this is important, removing this causes test failures+  lift (atomically (STM.readTMVar pidVar))++newtype CleanUpAction = CleanUpAction { runCleanUpAction :: IO () }++scheduleProcessWithCleanup+  :: HasCallStack+  => 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)))+ where+  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))++  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 (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 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 Shutdown _k = do+    invokeShutdownAction shutdownAction (Right ())++  go _pid (ExitWithError msg) _k = do+    invokeShutdownAction shutdownAction (Left (ProcessExitError msg))++  go _pid (RaiseError msg) _k = do+    invokeShutdownAction shutdownAction (Left (ProcessExitError msg))++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)++withMessageQueue+  :: HasSchedulerIO r+  => (CleanUpAction -> ProcessInfo -> Eff r a)+  -> Eff r (Either SchedulerError a)+withMessageQueue m = do+  mpinfo <- createQueue+  lc     <- getLogChannel+  case mpinfo of+    Right pinfo -> do+      cleanUpAction <-+        getSchedulerTVar >>= return . CleanUpAction . destroyQueue+          lc+          (pinfo ^. processId)+      Right <$> m cleanUpAction pinfo+    Left e ->+      return $ Left e+ where+  createQueue = do+    myTId <- lift myThreadId+    overScheduler+      (do+        abortNow <- use schedulerShuttingDown+        if abortNow+          then return (Left SchedulerShuttingDown)+          else do+            pid     <- nextPid <<+= 1+            channel <- Mtl.lift newTQueue+            shutdownIndicator <- Mtl.lift (newTVar False)+            let pinfo = ProcessInfo pid channel shutdownIndicator+            threadIdTable . at pid .= Just myTId+            processTable . at pid .= Just pinfo+            return (Right pinfo)+      )+  destroyQueue lc pid psVar = do+    didWork <- Exc.try+      (overSchedulerIO+        psVar+        (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)))+            >>= either+                  (return . (show pid ++) . show)+                  (return . (maybe (show pid) show))++    case didWork of+      Right (_pinfo, True) ->+        return ()+      Right (pinfo, False) ->+        logChannelPutIO lc ("queue already destroyed: " ++ show pinfo)+      Left (e :: Exc.SomeException) ->+        getCause+          >>= logChannelPutIO lc+          .   (("failed to destroy queue: " ++ show e ++ " ") ++)+++overScheduler+  :: HasSchedulerIO r+  => Mtl.StateT Scheduler STM.STM (Either SchedulerError a)+  -> 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)++getSchedulerTVar :: HasSchedulerIO r => Eff r (TVar Scheduler)+getSchedulerTVar = fromSchedulerVar <$> ask++getSchedulerVar :: HasSchedulerIO r => Eff r SchedulerVar+getSchedulerVar = ask

@@ -0,0 +1,190 @@+-- | A coroutine based, single threaded scheduler for 'Process'es.+module Control.Eff.Concurrent.Process.SingleThreadedScheduler+  ( schedule+  , defaultMain+  , singleThreadedIoScheduler+  , 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+++-- | 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++    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++      in case processState of+             OnDone -> handleExit++             OnRaiseError _ -> 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+                                     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))++             OnSend toPid msg k ->+               do nextK <- k (ResumeWith (msgQs ^. at toPid . to isJust))+                  go newPid+                     (msgQs & at toPid . _Just %~ (:|> msg))+                     (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))++                 Just (nextMessage :<| restMessages) ->+                   do nextK <- k (ResumeWith nextMessage)+                      go newPid+                         (msgQs & at pid . _Just .~ restMessages)+                         (rest :|> (nextK, 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))++data OnYield r where+  OnSelf :: (ResumeProcess ProcessId -> Eff r (OnYield r))+         -> OnYield r+  OnSpawn :: Eff (Process r ': r) ()+          -> (ResumeProcess ProcessId -> Eff r (OnYield r))+          -> OnYield r+  OnDone :: OnYield r+  OnExitError :: String -> OnYield r+  OnRaiseError :: String -> OnYield r+  OnSend :: ProcessId -> Dynamic+         -> (ResumeProcess Bool -> Eff r (OnYield r))+         -> OnYield r+  OnRecv :: (ResumeProcess Dynamic -> Eff r (OnYield r))+         -> 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 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)++-- | The concrete list of 'Eff'ects for running this pure scheduler on @IO@ and+-- with string logging.+type LoggingAndIo =+              '[ Logs String+               , Lift IO+               ]++-- | A 'SchedulerProxy' for 'LoggingAndIo'.+singleThreadedIoScheduler :: SchedulerProxy LoggingAndIo+singleThreadedIoScheduler = SchedulerProxy++-- | Execute a 'Process' using 'schedule' on top of 'Lift' @IO@ and 'Logs'+-- @String@ effects.+defaultMain+  :: 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 ()

@@ -8,6 +8,9 @@   ( try   , liftRethrow   , runErrorRethrowIO+  , liftCatch+  , liftCatchEff+  , liftTry   , module X   ) where@@ -17,7 +20,10 @@ 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)@@ -26,12 +32,12 @@ -- 'Exc.Exception' using a given wrapper function and rethrow it using -- 'X.throwError'. liftRethrow-  :: forall e r a-   . (Exc.Exception e, SetMember Lift (Lift IO) r, Member (Exc e) r)-  => (Exc.SomeException -> e)+  :: 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 liftE m = lift (Exc.try m) >>= either (throwError . liftE) return+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'@@ -41,3 +47,36 @@   => 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)

@@ -0,0 +1,17 @@+-- | "Control.Eff" utilities, some stolen from future versions of+-- @extensible-effects@. In future versions, there might be something like a+-- @Control.Eff.Extend@ module that might contain a lot of it, according to+-- the discussion here:+-- https://github.com/suhailshergill/extensible-effects/issues/98+module Control.Eff.InternalExtra where++import Control.Eff++-- | Embeds a less-constrained 'Eff' into a more-constrained one. Analogous to+-- MTL's 'lift'.+raise :: Eff r a -> Eff (e ': r) a+raise = loop+  where+    loop (Val x) = pure x+    loop (E u q) = E (weaken u) $ qComps q loop+{-# INLINE raise #-}

@@ -14,11 +14,12 @@ module Control.Eff.Log   ( handleLogsWith   , Logs(..)-  , logMessageFreeEff   , logMsg   , module ExtLog   , LogChannel()   , logChannelPutIO+  , captureLogs+  , ignoreLogs   , forwardLogsToChannel   , forkLogChannel   , joinLogChannel@@ -36,26 +37,47 @@ import           Control.Monad.Log             as ExtLog                                          hiding ( ) import           Control.Monad.Trans.Control-import           Data.Kind+import           Data.Kind() import qualified Control.Eff.Lift              as Eff import qualified Control.Monad.Log             as Log+import           Data.Sequence                 (Seq())+import qualified Data.Sequence                 as Seq  -- | The 'Eff'ect type to wrap 'ExtLog.MonadLog'. -- This is a data Logs message a where-  LogMessageFree :: (forall n . Monoid n => (message -> n) -> n) -> Logs message ()---- | Effectful version of the /strange/ 'ExtLog.logMessageFree' function.-logMessageFreeEff-  :: Member (Logs message) r-  => (forall n . Monoid n => (message -> n) -> n)-  -> Eff r ()-logMessageFreeEff foldMapish = send (LogMessageFree foldMapish)+  LogMsg :: message -> Logs message ()  -- | Effectful version of the 'ExtLog.logMessage' function. logMsg :: Member (Logs m) r => m -> Eff r ()-logMsg msg = logMessageFreeEff ($ msg)+logMsg msg = send (LogMsg msg) +-- | Capture the logs in a 'Seq'.+captureLogs :: Eff (Logs message ':  r) a+            -> Eff r (a, Seq message)+captureLogs actionThatLogs =+  Eff.handle_relay_s+     Seq.empty+     (\logs result -> return (result, logs))+     handleLogs+     actionThatLogs+  where+    handleLogs :: Seq message+               -> Logs message x+               -> (Seq message -> Arr r x y)+               -> Eff r y+    handleLogs logs (LogMsg m) k =+      k (logs Seq.:|> m) ()++-- | Throw away all log messages.+ignoreLogs :: Eff (Logs message ':  r) a+           -> Eff r a+ignoreLogs actionThatLogs =+  Eff.handle_relay return handleLogs actionThatLogs+  where+    handleLogs :: Logs m x -> Arr r x y -> Eff r y+    handleLogs (LogMsg _) k = k ()+ -- | Handle 'Logs' effects using 'Log.LoggingT' 'Log.Handler's. handleLogsWith   :: forall m r message a@@ -68,9 +90,8 @@                                                              actionThatLogs  where   go :: Logs message b -> (b -> Eff r c) -> Eff r c-  go (LogMessageFree foldMapish) k =-    Eff.lift (foldHandler (Log.runLoggingT (Log.logMessageFree foldMapish)))-      >>= k+  go (LogMsg m) k =+    Eff.lift (foldHandler (\doLog -> doLog m)) >>= k  -- * Concurrent Logging @@ -78,7 +99,7 @@ data LogChannel message =   LogChannel { fromLogChannel :: TQueue message              , logChannelOpen :: TVar Bool-             , logChannelThread :: ThreadId+             , _logChannelThread :: ThreadId              }  -- | Send the log messages to a 'LogChannel'.@@ -167,7 +188,5 @@   -> LoggingT message IO a logChannelBracket mWelcome mGoodbye f = control   (\runInIO -> do-    myTId <- myThreadId     let logHandler = void . runInIO . logMessage-    bracket (forkLogChannel logHandler mWelcome) (joinLogChannel mGoodbye) f-  )+    bracket (forkLogChannel logHandler mWelcome) (joinLogChannel mGoodbye) f)

@@ -0,0 +1,1 @@+{-# OPTIONS_GHC -F -pgmF tasty-discover -optF --tree-display #-}

@@ -0,0 +1,110 @@+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++timeoutSeconds :: Integer -> Timeout+timeoutSeconds seconds = Timeout (seconds * 1000000) (show seconds ++ "s")++test_IOExceptionsIsolated :: TestTree+test_IOExceptionsIsolated =+  localOption+  (timeoutSeconds 30)+    $ 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+             Scheduler.defaultMain+               $ 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++       | (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 =+  localOption+  (timeoutSeconds 30)+  (testCase "spawn a child and return"+   (Scheduler.defaultMain+     (void (spawn (void (receiveMessage forkIoScheduler))))))++test_mainProcessSpawnsAChildAndExitsNormally :: TestTree+test_mainProcessSpawnsAChildAndExitsNormally =+  localOption+  (timeoutSeconds 30)+  (testCase "spawn a child and exit normally"+   (Scheduler.defaultMain+     (do void (spawn (void (receiveMessage forkIoScheduler)))+         void (exitNormally forkIoScheduler)+         fail "This should not happen!!"+     )))+++test_mainProcessSpawnsAChildInABusySendLoopAndExitsNormally :: TestTree+test_mainProcessSpawnsAChildInABusySendLoopAndExitsNormally =+  localOption+  (timeoutSeconds 300)+  (testCase "spawn a child with a busy send loop and exit normally"+   (Scheduler.defaultMain+     (do void (spawn (forever (void (sendMessage forkIoScheduler 1000 (toDyn "test")))))+         void (exitNormally forkIoScheduler)+         fail "This should not happen!!"+     )))+++test_mainProcessSpawnsAChildBothReturn :: TestTree+test_mainProcessSpawnsAChildBothReturn =+  localOption+  (timeoutSeconds 30)+  (testCase "spawn a child and let it return and return"+   (Scheduler.defaultMain+     (do child <- spawn (void (receiveMessageAs @String forkIoScheduler))+         True <- sendMessage forkIoScheduler child (toDyn "test")+         return ()+     )))++test_mainProcessSpawnsAChildBothExitNormally :: TestTree+test_mainProcessSpawnsAChildBothExitNormally =+  localOption+  (timeoutSeconds 30)+  (testCase "spawn a child and let it exit and exit"+   (Scheduler.defaultMain+     (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!!"+     )))

@@ -0,0 +1,425 @@+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.Runners+import Test.Tasty.HUnit++timeoutSeconds :: Integer -> Timeout+timeoutSeconds seconds = Timeout (seconds * 1000000) (show seconds ++ "s")++test_forkIo :: TestTree+test_forkIo =+  localOption (NumThreads 1)+  (withResource+   (forkLogChannel (const (return ())) Nothing)+   (joinLogChannel Nothing)+   (\ logCFactory ->+       testGroup "ForkIOScheduler"+       [allTests+        (return+          (\e ->+             do logC <- logCFactory+                ForkIO.schedule e logC))]))++test_singleThreaded :: TestTree+test_singleThreaded =+  testGroup "SingleThreadedScheduler"+  [allTests (return SingleThreaded.defaultMain)]++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+      ])++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"+              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"+              error "This should not happen"+      , 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+              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"))))+                [0, 5 .. n]+              oks <- replicateM+                     (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+      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])+       ]++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+  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+                          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))))))+        ]++       ]+    , 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 <- 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 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"+        $ 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")))+              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")))+              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")))+              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")))+              void (sendShutdown px other)+              a <- receiveMessageAs px+              assertEff "" (a == "OK")+      ]+    ]++untilShutdown+  :: Member t r => t (ResumeProcess v) -> Eff r ()+untilShutdown pa = do+  r <- send pa+  case r of+    ShutdownRequested -> return ()+    _ -> untilShutdown pa++scheduleAndAssert :: forall r .+                    (SetMember Lift (Lift IO) r, Member (Logs String) r)+                  => IO (Eff (Process r ': r) () -> IO ())+                  -> ((String -> Bool -> Eff (Process r ': r) ()) -> Eff (Process r ': r) ())+                  -> IO ()+scheduleAndAssert schedulerFactory testCaseAction =+  do resultVar <- newEmptyTMVarIO+     void (applySchedulerFactory schedulerFactory+            (testCaseAction (\ title cond ->+                               do lift (atomically (putTMVar resultVar (title, cond))))))+     (title, result) <- atomically (takeTMVar resultVar)+     assertBool title result++applySchedulerFactory :: forall r . (Member (Logs String) r, SetMember Lift (Lift IO) r)+                      => IO (Eff (Process r ': r) () -> IO ()) -> Eff (Process r ': r) () -> IO ()+applySchedulerFactory factory procAction =+  do scheduler <- factory+     scheduler procAction

@@ -0,0 +1,38 @@+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++timeoutSeconds :: Integer -> Timeout+timeoutSeconds seconds = Timeout (seconds * 1000000) (show seconds ++ "s")++test_mainProcessSpawnsAChildAndExitsNormally :: TestTree+test_mainProcessSpawnsAChildAndExitsNormally =+  localOption+  (timeoutSeconds 2)+  (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 =+  localOption+  (timeoutSeconds 30)+  (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!!"+     )))

@@ -1,2 +0,0 @@-main :: IO ()-main = putStrLn "Test suite not yet implemented"