diff --git a/ChangeLog.md b/ChangeLog.md --- a/ChangeLog.md +++ b/ChangeLog.md @@ -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 diff --git a/extensible-effects-concurrent.cabal b/extensible-effects-concurrent.cabal --- a/extensible-effects-concurrent.cabal +++ b/extensible-effects-concurrent.cabal @@ -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 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 diff --git a/src/Control/Eff/Concurrent/Api.hs b/src/Control/Eff/Concurrent/Api.hs --- a/src/Control/Eff/Concurrent/Api.hs +++ b/src/Control/Eff/Concurrent/Api.hs @@ -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 diff --git a/src/Control/Eff/Concurrent/Api/Client.hs b/src/Control/Eff/Concurrent/Api/Client.hs --- a/src/Control/Eff/Concurrent/Api/Client.hs +++ b/src/Control/Eff/Concurrent/Api/Client.hs @@ -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 diff --git a/src/Control/Eff/Concurrent/Api/Internal.hs b/src/Control/Eff/Concurrent/Api/Internal.hs --- a/src/Control/Eff/Concurrent/Api/Internal.hs +++ b/src/Control/Eff/Concurrent/Api/Internal.hs @@ -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 diff --git a/src/Control/Eff/Concurrent/Api/Observer.hs b/src/Control/Eff/Concurrent/Api/Observer.hs new file mode 100644 --- /dev/null +++ b/src/Control/Eff/Concurrent/Api/Observer.hs @@ -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 diff --git a/src/Control/Eff/Concurrent/Api/Server.hs b/src/Control/Eff/Concurrent/Api/Server.hs --- a/src/Control/Eff/Concurrent/Api/Server.hs +++ b/src/Control/Eff/Concurrent/Api/Server.hs @@ -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 diff --git a/src/Control/Eff/Concurrent/Dispatcher.hs b/src/Control/Eff/Concurrent/Dispatcher.hs deleted file mode 100644 --- a/src/Control/Eff/Concurrent/Dispatcher.hs +++ /dev/null @@ -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)) diff --git a/src/Control/Eff/Concurrent/Examples.hs b/src/Control/Eff/Concurrent/Examples.hs --- a/src/Control/Eff/Concurrent/Examples.hs +++ b/src/Control/Eff/Concurrent/Examples.hs @@ -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 diff --git a/src/Control/Eff/Concurrent/Examples2.hs b/src/Control/Eff/Concurrent/Examples2.hs --- a/src/Control/Eff/Concurrent/Examples2.hs +++ b/src/Control/Eff/Concurrent/Examples2.hs @@ -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 diff --git a/src/Control/Eff/Concurrent/MessagePassing.hs b/src/Control/Eff/Concurrent/MessagePassing.hs deleted file mode 100644 --- a/src/Control/Eff/Concurrent/MessagePassing.hs +++ /dev/null @@ -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) diff --git a/src/Control/Eff/Concurrent/Observer.hs b/src/Control/Eff/Concurrent/Observer.hs deleted file mode 100644 --- a/src/Control/Eff/Concurrent/Observer.hs +++ /dev/null @@ -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 diff --git a/src/Control/Eff/Concurrent/Process.hs b/src/Control/Eff/Concurrent/Process.hs new file mode 100644 --- /dev/null +++ b/src/Control/Eff/Concurrent/Process.hs @@ -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 diff --git a/src/Control/Eff/Concurrent/Process/ForkIOScheduler.hs b/src/Control/Eff/Concurrent/Process/ForkIOScheduler.hs new file mode 100644 --- /dev/null +++ b/src/Control/Eff/Concurrent/Process/ForkIOScheduler.hs @@ -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 diff --git a/src/Control/Eff/Concurrent/Process/SingleThreadedScheduler.hs b/src/Control/Eff/Concurrent/Process/SingleThreadedScheduler.hs new file mode 100644 --- /dev/null +++ b/src/Control/Eff/Concurrent/Process/SingleThreadedScheduler.hs @@ -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 () diff --git a/src/Control/Eff/ExceptionExtra.hs b/src/Control/Eff/ExceptionExtra.hs --- a/src/Control/Eff/ExceptionExtra.hs +++ b/src/Control/Eff/ExceptionExtra.hs @@ -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) diff --git a/src/Control/Eff/InternalExtra.hs b/src/Control/Eff/InternalExtra.hs new file mode 100644 --- /dev/null +++ b/src/Control/Eff/InternalExtra.hs @@ -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 #-} diff --git a/src/Control/Eff/Log.hs b/src/Control/Eff/Log.hs --- a/src/Control/Eff/Log.hs +++ b/src/Control/Eff/Log.hs @@ -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) diff --git a/test/Driver.hs b/test/Driver.hs new file mode 100644 --- /dev/null +++ b/test/Driver.hs @@ -0,0 +1,1 @@ +{-# OPTIONS_GHC -F -pgmF tasty-discover -optF --tree-display #-} diff --git a/test/ForkIOScheduler.hs b/test/ForkIOScheduler.hs new file mode 100644 --- /dev/null +++ b/test/ForkIOScheduler.hs @@ -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!!" + ))) diff --git a/test/ProcessBehaviourTestCases.hs b/test/ProcessBehaviourTestCases.hs new file mode 100644 --- /dev/null +++ b/test/ProcessBehaviourTestCases.hs @@ -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 diff --git a/test/SingleThreadedScheduler.hs b/test/SingleThreadedScheduler.hs new file mode 100644 --- /dev/null +++ b/test/SingleThreadedScheduler.hs @@ -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!!" + ))) diff --git a/test/Spec.hs b/test/Spec.hs deleted file mode 100644 --- a/test/Spec.hs +++ /dev/null @@ -1,2 +0,0 @@ -main :: IO () -main = putStrLn "Test suite not yet implemented"