diff --git a/ChangeLog.md b/ChangeLog.md --- a/ChangeLog.md +++ b/ChangeLog.md @@ -1,67 +1,85 @@ # Changelog for extensible-effects-concurrent +## 0.3.0.0 + +- Add support for running and interacting with a scheduler + and it's processes from IO, for example from ghci +- Rename `yieldProcess` to `executeAndResume` +- Add an actual `yieldProcess`, that behaves like `yield` +- Change the return type of function to `()` where applicable + to avoid all these `_ <- sendMessage...` or `void $ sendMessage` + dances. +- Add a simple logging observer: `spawnLoggingObserver` +- Removed `Control.Eff.Interactive` +- Removed most functions in `Control.Eff.ExceptionExtra` +- Make `sendMessage` and the reply in `call` strict with respect to the message + payload + ## 0.2.0.3 - * Improve 'Api' documentation - * Improve `LogChannel` API - * Reorganize unit tests - * Hopefully tune travis ci test parameter enough to get a stable build result +- Improve 'Api' documentation +- Improve `LogChannel` API +- Reorganize unit tests +- Hopefully tune travis ci test parameter enough to get a stable build result ## 0.2.0.2 - * Fix minor `stack upload` complaints about the cabal file +- Fix minor `stack upload` complaints about the cabal file ## 0.2.0.1 - * Simplify IO Exception handling in `ForkIoScheduler`, - * Add many unit tests for exception, exit and shutdown +- Simplify IO Exception handling in `ForkIoScheduler`, +- Add many unit tests for exception, exit and shutdown ## 0.2.0.0 - * Add `Spawn` to `Process` - * Merge `MessagePassing` and `Process` - * Add initial test suite - * Fix shutdown error in `ForkIoScheduler` - * Rename `Dispatcher` to `Scheduler` - * Add `receiveLoop` function to `Process` - * Change `Api.Server` `serve` to loop instead of handling just one request - * Allow combining multiple `ApiHandler` such that one process can handle - multiple APIs +- Add `Spawn` to `Process` +- Merge `MessagePassing` and `Process` +- Add initial test suite +- Fix shutdown error in `ForkIoScheduler` +- Rename `Dispatcher` to `Scheduler` +- Add `receiveLoop` function to `Process` +- Change `Api.Server` `serve` to loop instead of handling just one request +- Allow combining multiple `ApiHandler` such that one process can handle + multiple APIs ### TODO - * Add `Kill` action to `Process` +- Add `Kill` action to `Process` + ## 0.1.3.0 - * Rename and split `GenServer` to `Api`, `Api.Client`, `Api.Server` - * Add `registerServer`, `callRegistered` and `castRegistered` - * Remove the variant of `cast` that returns a boolean +- Rename and split `GenServer` to `Api`, `Api.Client`, `Api.Server` +- Add `registerServer`, `callRegistered` and `castRegistered` +- Remove the variant of `cast` that returns a boolean ## 0.1.2.2 - * Try to fix version bounds for hackage +- Try to fix version bounds for hackage + ## 0.1.2.1 - * Add more documentation - * Simplify Scheduler API - * Make more exception safe +- Add more documentation +- Simplify Scheduler API +- Make more exception safe + ## 0.1.2.0 - * Add Observer module - * Implement Exception handling - * Improve Scheduler shutdown - * Add logging support via the logging-effect library +- Add Observer module +- Implement Exception handling +- Improve Scheduler shutdown +- Add logging support via the logging-effect library ## 0.1.1.0 - * Substantial API reoganization - * Rename/Move modules +- Substantial API reoganization +- Rename/Move modules ## 0.1.0.1 - * Stack/Cabal/Github Cosmetics - * Travis build job +- Stack/Cabal/Github Cosmetics +- Travis build job ## 0.1.0.0 - * Initial Version +- Initial Version 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.2.0.3 +version: 0.3.0.0 description: Please see the README on GitHub at synopsis: Message passing concurrency as extensible-effect homepage: https://github.com/sheyll/extensible-effects-concurrent#readme @@ -43,13 +43,13 @@ tagged exposed-modules: Control.Eff.ExceptionExtra, - Control.Eff.Interactive, Control.Eff.Log, Control.Eff.Concurrent.Api, Control.Eff.Concurrent.Api.Client, Control.Eff.Concurrent.Api.Server, Control.Eff.Concurrent.Process, Control.Eff.Concurrent.Process.ForkIOScheduler, + Control.Eff.Concurrent.Process.Interactive, Control.Eff.Concurrent.Process.SingleThreadedScheduler, Control.Eff.Concurrent.Api.Observer other-modules: @@ -90,6 +90,7 @@ , ForkIOScheduler , SingleThreadedScheduler , ProcessBehaviourTestCases + , Debug , Common ghc-options: -threaded -rtsopts -with-rtsopts=-N build-depends: @@ -99,6 +100,9 @@ , tasty , tasty-discover , tasty-hunit + , containers + , QuickCheck <2.11 + , lens , HUnit , stm default-language: Haskell2010 @@ -106,6 +110,8 @@ , RankNTypes , ScopedTypeVariables , TypeOperators + , GADTs + , TemplateHaskell , DataKinds , FlexibleContexts , BangPatterns 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 @@ -13,15 +13,16 @@ {-# LANGUAGE TypeApplications #-} {-# LANGUAGE GADTs #-} --- | This module contains a mechanisms to specify what kind of messages a --- process can receive and possible answer by sending a message back to the --- orginator. --- --- A message can be either a blocking or a non-blocking request. +-- | This module contains a mechanism to specify what kind of messages (aka +-- /requests/) a 'Server' ('Process') can handle, and if the caller blocks and +-- waits for an answer, which the server process provides. -- --- The type magic in the 'Api' type familiy allows to define a related set of messages along +-- The type magic in the 'Api' type familiy allows to define a related set of /requests/ along -- with the corresponding responses. -- +-- Request handling can be either blocking, if a response is requred, or +-- non-blocking. +-- -- A process can /serve/ a specific 'Api' instance by using the functions provided by -- the "Control.Eff.Concurrent.Api.Server" module. -- @@ -44,8 +45,8 @@ import Control.Eff.Concurrent.Process -- | This data family defines an API, a communication interface description --- between to processes, where one process acts as a __server__ and the other(s) --- as __client(s)__. +-- between at least two processes. The processes act as __servers__ or +-- __client(s)__ regarding a specific instance of this type. -- -- The first parameter is usually a user defined phantom type that identifies -- the 'Api' instance. @@ -70,12 +71,13 @@ data family Api (api :: Type) (reply :: Synchronicity) --- | This __data kind__ is used to indicate at the type level if a specific --- constructor of an @Api@ instance has a result for which some caller has to --- wait, or if it is asynchronous. +-- | The (promoted) constructors of this type specify (at the type level) the +-- reply behavior of a specific constructor of an @Api@ instance. data Synchronicity = - Synchronous Type -- ^ Blocking operation with a specific return type, e.g. @('Synchronous (Either RentalError RentalId))@ - | Asynchronous -- ^ Non-blocking async operation + Synchronous Type -- ^ Specify that handling a request is a blocking operation + -- with a specific return type, e.g. @('Synchronous (Either + -- RentalError RentalId))@ + | Asynchronous -- ^ Non-blocking, asynchronous, request handling deriving (Typeable) -- | This is a tag-type that wraps around a 'ProcessId' and holds an 'Api' index 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 @@ -36,7 +36,9 @@ import Control.Eff.Concurrent.Api.Internal import Control.Eff.Concurrent.Process import Data.Dynamic -import Data.Typeable (Typeable, typeRep) +import Data.Typeable ( Typeable + , typeRep + ) import GHC.Stack -- | Send an 'Api' request that has no return value and return as fast as @@ -55,8 +57,8 @@ -> Server o -> Api o 'Asynchronous -> Eff r Bool -castChecked px (Server pid) callMsg = - sendMessage px pid (toDyn (Cast callMsg)) +castChecked px (Server pid) castMsg = + sendMessageChecked px pid (toDyn $! (Cast $! castMsg)) -- | Send an 'Api' request that has no return value and return as fast as -- possible. The type signature enforces that the corresponding 'Api' clause is @@ -72,9 +74,9 @@ -> Server o -> Api o 'Asynchronous -> Eff r () -cast px toServer apiRequest = - do _ <- castChecked px toServer apiRequest - return () +cast px toServer apiRequest = do + _ <- castChecked px toServer apiRequest + return () -- | Send an 'Api' request and wait for the server to return a result value. -- @@ -90,18 +92,19 @@ ) => SchedulerProxy q -> Server api - -> Api api ('Synchronous result) + -> Api api ( 'Synchronous result) -> Eff r result call px (Server pidInt) req = do fromPid <- self px - let requestMessage = Call fromPid req - wasSent <- sendMessage px pidInt (toDyn requestMessage) + let requestMessage = Call fromPid $! req + wasSent <- sendMessageChecked px pidInt (toDyn $! requestMessage) if wasSent then let extractResult :: Response api result -> result extractResult (Response _pxResult result) = result - in extractResult <$> receiveMessageAs px - else raiseError px + in extractResult <$> receiveMessageAs px + else raiseError + px ("Could not send request message " ++ show (typeRep requestMessage)) -- * Registered Servers @@ -119,13 +122,16 @@ -- | Run a reader effect that contains __the one__ server handling a specific -- 'Api' instance. -registerServer :: Server o -> Eff ( Reader (Server o) ': r ) a -> Eff r a +registerServer :: Server o -> Eff (Reader (Server o) ': r) a -> Eff r a registerServer = flip runReader -- | Like 'call' but take the 'Server' from the reader provided by -- 'registerServer'. -callRegistered :: (Typeable reply, ServesApi o r q) - => SchedulerProxy q -> Api o ('Synchronous reply) -> Eff r reply +callRegistered + :: (Typeable reply, ServesApi o r q) + => SchedulerProxy q + -> Api o ( 'Synchronous reply) + -> Eff r reply callRegistered px method = do serverPid <- ask call px serverPid method @@ -135,19 +141,21 @@ -- application level errors can be combined with errors rising from inter -- process communication. callRegisteredA - :: forall r q o f reply . - (Alternative f, Typeable f, Typeable reply, ServesApi o r q) - => SchedulerProxy q -> Api o ('Synchronous (f reply)) + :: forall r q o f reply + . (Alternative f, Typeable f, Typeable reply, ServesApi o r q) + => SchedulerProxy q + -> Api o ( 'Synchronous (f reply)) -> Eff r (f reply) callRegisteredA px method = do - catchRaisedError px - (const (return (empty @f))) - (callRegistered px method) + catchRaisedError px (const (return (empty @f))) (callRegistered px method) -- | Like 'cast' but take the 'Server' from the reader provided by -- 'registerServer'. -castRegistered :: (Typeable o, ServesApi o r q) - => SchedulerProxy q -> Api o 'Asynchronous -> Eff r () +castRegistered + :: (Typeable o, ServesApi o r q) + => SchedulerProxy q + -> Api o 'Asynchronous + -> Eff r () castRegistered px method = do serverPid <- ask cast px serverPid method diff --git a/src/Control/Eff/Concurrent/Api/Observer.hs b/src/Control/Eff/Concurrent/Api/Observer.hs --- a/src/Control/Eff/Concurrent/Api/Observer.hs +++ b/src/Control/Eff/Concurrent/Api/Observer.hs @@ -35,11 +35,11 @@ , notifyObservers , CallbackObserver , spawnCallbackObserver + , spawnLoggingObserver ) where import GHC.Stack import Data.Dynamic -import Data.Maybe import Data.Set (Set) import qualified Data.Set as Set import Control.Eff @@ -206,16 +206,32 @@ asServer @(CallbackObserver o) <$> (spawn @r @q $ do - me <- asServer @(CallbackObserver o) <$> self px let loopUntil = serve px (ApiHandler @(CallbackObserver o) (handleCast loopUntil) (unhandledCallError px) - (\e -> - do logMsg (show me ++ " observer terminating " ++ fromMaybe "normally" e) - defaultTermination px e)) + (defaultTermination px)) loopUntil) where handleCast k (CbObserved fromSvr v) = onObserve fromSvr v >>= flip when k + +-- | Use 'spawnCallbackObserver' to create a universal logging observer, +-- using the 'Show' instance of the 'Observation'. +-- | Start a new process for an 'Observer' that schedules +-- all observations to an effectful callback. +-- +-- @since 0.3.0.0 +spawnLoggingObserver + :: forall o r q . + ( SetMember Process (Process q) r + , Typeable o + , Show (Observation o) + , Observable o + , Member (Logs String) q) + => SchedulerProxy q + -> Eff r (Server (CallbackObserver o)) +spawnLoggingObserver px = + spawnCallbackObserver px + (\s o -> logMsg (show s ++ " OBSERVED: " ++ show o) >> return True) 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 @@ -19,7 +19,7 @@ ( -- * Api Server serve - , ApiHandler (..) + , ApiHandler(..) -- * ApiHandler default callbacks , unhandledCallError , unhandledCastError @@ -31,19 +31,21 @@ , UnhandledRequest() , catchUnhandled , ensureAllHandled - , castMessage + , requestFromDynamic , exitUnhandled ) where import Control.Eff import Control.Eff.InternalExtra -import qualified Control.Eff.Exception as Exc +import qualified Control.Eff.Exception as Exc import Control.Eff.Concurrent.Api import Control.Eff.Concurrent.Api.Internal import Control.Eff.Concurrent.Process import Data.Proxy -import Data.Typeable (Typeable, typeRep) +import Data.Typeable ( Typeable + , typeRep + ) import Data.Dynamic import GHC.Stack @@ -54,20 +56,18 @@ => SchedulerProxy q -> ApiHandler p r -> Eff r () -serve px handlers = - receiveLoop px - $ \ mReq -> - case mReq of - Left Nothing -> - applyApiHandler px handlers (Terminate Nothing) - Left (Just reason) -> - applyApiHandler px handlers (Terminate (Just reason)) - Right dyn -> - ensureAllHandled px - (do msg <- castMessage dyn - raise (applyApiHandler px handlers msg)) +serve px handlers = receiveLoop px $ \mReq -> case mReq of + Left Nothing -> applyApiHandler px handlers (Terminate Nothing) + Left (Just reason) -> applyApiHandler px handlers (Terminate (Just reason)) + Right dyn -> ensureAllHandled + px + (do + msg <- requestFromDynamic dyn + raise (applyApiHandler px handlers msg) + ) --- | A record of callbacks requests to a /server/, serving a specific 'Api' family instance. +-- | A record of callbacks, handling requests sent to a /server/ 'Process', all +-- belonging to a specific 'Api' family instance. data ApiHandler p r where ApiHandler :: { -- | A cast will not return a result directly. This is used for async @@ -79,7 +79,7 @@ -- handler calls the reply continuation. , _handleCall :: forall x . HasCallStack - => Api p ('Synchronous x) -> (x -> Eff r Bool) -> Eff r () + => Api p ('Synchronous x) -> (x -> Eff r ()) -> Eff r () -- | This callback is called with @Nothing@ if the process exits -- peacefully, or @Just "error message..."@ if the process exits with an -- error. This function is responsible to exit the process if necessary. @@ -91,124 +91,117 @@ -- | Apply either the '_handleCall', '_handleCast' or the '_handleTerminate' -- callback to an incoming 'Request'. Note, it is unlikely that this function must be used. -applyApiHandler :: forall r q p - . ( Typeable p - , SetMember Process (Process q) r - , HasCallStack) - => SchedulerProxy q - -> ApiHandler p r - -> Request p - -> Eff r () -applyApiHandler _px handlers (Terminate e) = - _handleTerminate handlers e -applyApiHandler _ handlers (Cast request ) = - _handleCast handlers request -applyApiHandler px handlers (Call fromPid request) = - _handleCall handlers request sendReply - where - sendReply :: Typeable x => x -> Eff r Bool - sendReply reply = - sendMessage px fromPid (toDyn (Response (Proxy @p) reply)) +applyApiHandler + :: forall r q p + . (Typeable p, SetMember Process (Process q) r, HasCallStack) + => SchedulerProxy q + -> ApiHandler p r + -> Request p + -> Eff r () +applyApiHandler _px handlers (Terminate e) = _handleTerminate handlers e +applyApiHandler _ handlers (Cast request) = _handleCast handlers request +applyApiHandler px handlers (Call fromPid request) = _handleCall handlers + request + sendReply + where + sendReply :: Typeable x => x -> Eff r () + sendReply reply = + sendMessage px fromPid (toDyn $! (Response (Proxy @p) $! reply)) -- | A default handler to use in '_handleCall' in 'ApiHandler'. It will call -- 'raiseError' with a nice error message. unhandledCallError - :: forall p x r q . - ( Show (Api p ( 'Synchronous x)) - , Typeable p - , HasCallStack - , SetMember Process (Process q) r - ) + :: forall p x r q + . ( Show (Api p ( 'Synchronous x)) + , Typeable p + , HasCallStack + , SetMember Process (Process q) r + ) => SchedulerProxy q -> Api p ( 'Synchronous x) - -> (x -> Eff r Bool) + -> (x -> Eff r ()) -> Eff r () -unhandledCallError px api _ = raiseError px - ("Unhandled call: (" - ++ show api - ++ " :: " - ++ show (typeRep (Proxy @p)) ++ ")") +unhandledCallError px api _ = raiseError + px + ("Unhandled call: (" ++ show api ++ " :: " ++ show (typeRep (Proxy @p)) ++ ")" + ) -- | A default handler to use in '_handleCast' in 'ApiHandler'. It will call -- 'raiseError' with a nice error message. unhandledCastError - :: forall p r q . - ( Show (Api p 'Asynchronous) - , Typeable p - , HasCallStack - , SetMember Process (Process q) r - ) + :: forall p r q + . ( Show (Api p 'Asynchronous) + , Typeable p + , HasCallStack + , SetMember Process (Process q) r + ) => SchedulerProxy q -> Api p 'Asynchronous -> Eff r () -unhandledCastError px api = raiseError px - ("Unhandled cast: (" - ++ show api - ++ " :: " - ++ show (typeRep (Proxy @p)) ++ ")") +unhandledCastError px api = raiseError + px + ("Unhandled cast: (" ++ show api ++ " :: " ++ show (typeRep (Proxy @p)) ++ ")" + ) -- | Exit the process either normally of the error message is @Nothing@ -- or with 'exitWithError' otherwise. -defaultTermination :: forall q r . - ( HasCallStack, SetMember Process (Process q) r ) - => SchedulerProxy q - -> Maybe String - -> Eff r () -defaultTermination px e = - maybe (exitNormally px) (exitWithError px) e +defaultTermination + :: forall q r + . (HasCallStack, SetMember Process (Process q) r) + => SchedulerProxy q + -> Maybe String + -> Eff r () +defaultTermination px e = maybe (exitNormally px) (exitWithError px) e -- | 'serve' two 'ApiHandler's at once. The first handler is used for -- termination handling. -serveBoth :: forall r q p1 p2 - . ( Typeable p1, Typeable p2 - , SetMember Process (Process q) r - , HasCallStack) - => SchedulerProxy q - -> ApiHandler p1 r -> ApiHandler p2 r - -> Eff r () -serveBoth px h1 h2 = - receiveLoop px - $ \ mReq -> - case mReq of - Left Nothing -> - applyApiHandler px h1 (Terminate Nothing) - Left (Just reason) -> - applyApiHandler px h1 (Terminate (Just reason)) - Right dyn -> - ensureAllHandled px - (tryApiHandler px h1 dyn - `catchUnhandled` - tryApiHandler px h2) +serveBoth + :: forall r q p1 p2 + . ( Typeable p1 + , Typeable p2 + , SetMember Process (Process q) r + , HasCallStack + ) + => SchedulerProxy q + -> ApiHandler p1 r + -> ApiHandler p2 r + -> Eff r () +serveBoth px h1 h2 = receiveLoop px $ \mReq -> case mReq of + Left Nothing -> applyApiHandler px h1 (Terminate Nothing) + Left (Just reason) -> applyApiHandler px h1 (Terminate (Just reason)) + Right dyn -> ensureAllHandled + px + (tryApiHandler px h1 dyn `catchUnhandled` tryApiHandler px h2) -- | 'serve' three 'ApiHandler's at once. The first handler is used for -- termination handling. -serve3 :: forall r q p1 p2 p3 - . ( Typeable p1, Typeable p2, Typeable p3 - , SetMember Process (Process q) r - , HasCallStack) - => SchedulerProxy q - -> ApiHandler p1 r -> ApiHandler p2 r -> ApiHandler p3 r - -> Eff r () -serve3 px h1 h2 h3 = - receiveLoop px - $ \ mReq -> - case mReq of - Left Nothing -> - applyApiHandler px h1 (Terminate Nothing) - Left (Just reason) -> - applyApiHandler px h1 (Terminate (Just reason)) - Right dyn -> - ensureAllHandled px - (tryApiHandler px h1 dyn - `catchUnhandled` - tryApiHandler px h2 - `catchUnhandled` - tryApiHandler px h3) +serve3 + :: forall r q p1 p2 p3 + . ( Typeable p1 + , Typeable p2 + , Typeable p3 + , SetMember Process (Process q) r + , HasCallStack + ) + => SchedulerProxy q + -> ApiHandler p1 r + -> ApiHandler p2 r + -> ApiHandler p3 r + -> Eff r () +serve3 px h1 h2 h3 = receiveLoop px $ \mReq -> case mReq of + Left Nothing -> applyApiHandler px h1 (Terminate Nothing) + Left (Just reason) -> applyApiHandler px h1 (Terminate (Just reason)) + Right dyn -> ensureAllHandled + px + ( tryApiHandler px h1 dyn + `catchUnhandled` tryApiHandler px h2 + `catchUnhandled` tryApiHandler px h3 + ) -- | The basic building block of the combination of 'ApiHandler's is this -- function, which can not only be passed to 'receiveLoop', but also throws an --- 'UnhandledRequest' exception if 'castMessage' failed, such that multiple +-- 'UnhandledRequest' exception if 'requestFromDynamic' failed, such that multiple -- invokation of this function for different 'ApiHandler's can be tried, by using 'catchUnhandled'. -- -- > tryApiHandler px handlers1 message @@ -217,34 +210,32 @@ -- > `catchUnhandled` -- > tryApiHandler px handlers3 -- -tryApiHandler :: forall r q p - . ( Typeable p - , SetMember Process (Process q) r - , HasCallStack) - => SchedulerProxy q - -> ApiHandler p r - -> Dynamic - -> Eff (Exc.Exc UnhandledRequest ': r) () -tryApiHandler px handlers message = - do request <- castMessage message - raise (applyApiHandler px handlers request) +tryApiHandler + :: forall r q p + . (Typeable p, SetMember Process (Process q) r, HasCallStack) + => SchedulerProxy q + -> ApiHandler p r + -> Dynamic + -> Eff (Exc.Exc UnhandledRequest ': r) () +tryApiHandler px handlers message = do + request <- requestFromDynamic message + raise (applyApiHandler px handlers request) -- | An exception that is used by the mechanism that chains together multiple -- different 'ApiHandler' allowing a single process to implement multiple --- 'Api's. This exception is thrown by 'castMessage'. This +-- 'Api's. This exception is thrown by 'requestFromDynamic'. This -- exception can be caught with 'catchUnhandled', this way, several -- distinct 'ApiHandler' can be tried until one /fits/ or until the -- 'exitUnhandled' is invoked. newtype UnhandledRequest = UnhandledRequest { fromUnhandledRequest :: Dynamic } --- | If 'castMessage' failes to cast the message to a 'Request' for a +-- | If 'requestFromDynamic' failes to cast the message to a 'Request' for a -- certain 'ApiHandler' it throws an 'UnhandledRequest' exception. That -- exception is caught by this function and the raw message is given to the -- handler function. This is the basis for chaining 'ApiHandler's. catchUnhandled - :: forall r a . - ( Member (Exc.Exc UnhandledRequest) r - , HasCallStack) + :: forall r a + . (Member (Exc.Exc UnhandledRequest) r, HasCallStack) => Eff r a -> (Dynamic -> Eff r a) -> Eff r a @@ -253,36 +244,34 @@ -- | Catch 'UnhandledRequest's and terminate the process with an error, if necessary. ensureAllHandled - :: forall r q . - (HasCallStack, SetMember Process (Process q) r) + :: forall r q + . (HasCallStack, SetMember Process (Process q) r) => SchedulerProxy q -> Eff (Exc.Exc UnhandledRequest ': r) () -> Eff r () -ensureAllHandled px effect = - do result <- Exc.runError effect - either (exitUnhandled px . fromUnhandledRequest) return result +ensureAllHandled px effect = do + result <- Exc.runError effect + either (exitUnhandled px . fromUnhandledRequest) return result -- | Cast a 'Dynamic' value, and if that fails, throw an 'UnhandledRequest' -- error. -castMessage :: forall r a . - ( HasCallStack - , Typeable a - , Member (Exc.Exc UnhandledRequest) r) - => Dynamic -> Eff r a -castMessage message = - maybe - (Exc.throwError (UnhandledRequest message)) - return - (fromDynamic message) +requestFromDynamic + :: forall r a + . (HasCallStack, Typeable a, Member (Exc.Exc UnhandledRequest) r) + => Dynamic + -> Eff r a +requestFromDynamic message = + maybe (Exc.throwError (UnhandledRequest message)) return (fromDynamic message) -- | If an incoming message could not be casted to a 'Request' corresponding to --- an 'ApiHandler' (e.g. with 'castMessage') one should use this function to +-- an 'ApiHandler' (e.g. with 'requestFromDynamic') one should use this function to -- exit the process with a corresponding error. exitUnhandled - :: forall r q . (SetMember Process (Process q) r, HasCallStack) + :: forall r q + . (SetMember Process (Process q) r, HasCallStack) => SchedulerProxy q -> Dynamic -> Eff r () -exitUnhandled px message = - do let reason = "unhandled message: " ++ show message - exitWithError px reason +exitUnhandled px message = do + let reason = "unhandled message: " ++ show message + exitWithError px reason 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 @@ -108,7 +108,7 @@ handleCast (Shout x) = do me <- self px logMsg (show me ++ " Shouting: " ++ x) - handleCall :: Api TestApi ('Synchronous x) -> (x -> Eff (Process r ': r) Bool) -> Eff (Process r ': r) () + handleCall :: Api TestApi ('Synchronous x) -> (x -> Eff (Process r ': r) ()) -> Eff (Process r ': r) () handleCall (SayHello "e1") _reply = do me <- self px logMsg (show me ++ " raising an error") 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 @@ -87,11 +87,11 @@ modify (+ (1 :: Integer)) currentCount <- get notifyObservers px (CountChanged currentCount) - handleCall :: Api Counter ('Synchronous x) -> (x -> Eff r Bool) -> Eff r () + handleCall :: Api Counter ('Synchronous x) -> (x -> Eff r ()) -> Eff r () handleCall Cnt reply = do c <- get logMsg ("Cnt is " ++ show c) - _ <- reply c + reply c return () -- * Second API @@ -119,13 +119,13 @@ modify (+ x) c <- get @Integer logMsg ("Accumulator is " ++ show c) - handleCall :: Api PocketCalc ('Synchronous x) -> (x -> Eff r Bool) -> Eff r () + handleCall :: Api PocketCalc ('Synchronous x) -> (x -> Eff r ()) -> Eff r () handleCall (Add x) reply = do logMsg ("Add " ++ show x) modify (+ x) c <- get logMsg ("Accumulator is " ++ show c) - void (reply c) + reply c serverLoop :: forall r q . ( Member (Logs String) r diff --git a/src/Control/Eff/Concurrent/Process.hs b/src/Control/Eff/Concurrent/Process.hs --- a/src/Control/Eff/Concurrent/Process.hs +++ b/src/Control/Eff/Concurrent/Process.hs @@ -3,8 +3,13 @@ -- This module describes an abstract message passing effect, and a process -- effect, mimicking Erlang's process and message semantics. -- --- An implementation of a handler for these effects can be found in --- 'Control.Eff.Concurrent.Process.ForkIOScheduler'. +-- Two __scheduler__ implementations for the 'Process' effect are provided: +-- +-- * A scheduler using @forkIO@, i.e. relying on the multi threaded GHC runtime: +-- "Control.Eff.Concurrent.Process.ForkIOScheduler" +-- +-- * And a /pure/(rer) coroutine based scheduler in: +-- "Control.Eff.Concurrent.Process.SingleThreadedScheduler" {-# LANGUAGE DeriveTraversable #-} {-# LANGUAGE DeriveFoldable #-} {-# LANGUAGE GeneralizedNewtypeDeriving #-} @@ -30,14 +35,19 @@ , ResumeProcess(..) , SchedulerProxy(..) , thisSchedulerProxy - , yieldAndCatchProcess + , executeAndCatch + , yieldProcess , sendMessage + , sendMessageAs + , sendMessageChecked , spawn + , spawn_ , receiveMessage , receiveMessageAs , receiveLoop , self , sendShutdown + , sendShutdownChecked , exitWithError , exitNormally , raiseError @@ -49,7 +59,9 @@ import GHC.Stack import Control.Eff import Control.Lens -import Control.Monad (forever) +import Control.Monad ( forever + , void + ) import Data.Dynamic import Data.Kind import Text.Printf @@ -81,6 +93,9 @@ -- -- * when the first process exists, all process should be killed immediately data Process (r :: [Type -> Type]) b where + -- | In cooperative schedulers, this will give processing time to the + -- scheduler. Every other operation implicitly serves the same purpose. + YieldProcess :: Process r (ResumeProcess ()) -- | Return the current 'ProcessId' SelfPid :: Process r (ResumeProcess ProcessId) -- | Start a new process, the new process will execute an effect, the function @@ -131,43 +146,57 @@ -- | Execute a 'Process' action and resume the process, retry the action or exit -- the process when a shutdown was requested. -yieldProcess :: forall r q v . - ( SetMember Process (Process q) r - , HasCallStack) - => Process q (ResumeProcess v) - -> Eff r v -yieldProcess processAction = - do result <- send processAction - case result of - ResumeWith value -> return value - RetryLastAction -> yieldProcess processAction - ShutdownRequested -> send (Shutdown @q) - OnError e -> send (ExitWithError @q e) +executeAndResume + :: forall r q v + . (SetMember Process (Process q) r, HasCallStack) + => Process q (ResumeProcess v) + -> Eff r v +executeAndResume processAction = do + result <- send processAction + case result of + ResumeWith value -> return value + RetryLastAction -> executeAndResume processAction + ShutdownRequested -> send (Shutdown @q) + OnError e -> send (ExitWithError @q e) -- | Execute a and action and resume the process, retry the action, shutdown the process or return an error. -yieldAndCatchProcess :: forall q r v. - (SetMember Process (Process q) r, HasCallStack) - => SchedulerProxy q - -> Eff r (ResumeProcess v) - -> Eff r (Either String v) -yieldAndCatchProcess px processAction = - do result <- processAction - case result of - ResumeWith value -> return (Right value) - RetryLastAction -> yieldAndCatchProcess px processAction - ShutdownRequested -> send (Shutdown @q) - OnError e -> return (Left e) +executeAndCatch + :: forall q r v + . (SetMember Process (Process q) r, HasCallStack) + => SchedulerProxy q + -> Eff r (ResumeProcess v) + -> Eff r (Either String v) +executeAndCatch px processAction = do + result <- processAction + case result of + ResumeWith value -> return (Right value) + RetryLastAction -> executeAndCatch px processAction + ShutdownRequested -> send (Shutdown @q) + OnError e -> return (Left e) --- | Every function for 'Process' things needs proxy for the low-level --- effect list depending on the scheduler implementation. --- I don't know a smarter way yet to do this. +-- | Every function for 'Process' things needs such a proxy value +-- for the low-level effect list, i.e. the effects identified by +-- @__r__@ in @'Process' r : r@, this might be dependent on the +-- scheduler implementation. data SchedulerProxy :: [Type -> Type] -> Type where + -- | Tell the typechecker what effects we have below 'Process' SchedulerProxy :: SchedulerProxy q + -- | Like 'SchedulerProxy' but shorter + SP :: SchedulerProxy q -- | Return a 'SchedulerProxy' for a 'Process' effect. thisSchedulerProxy :: Eff (Process r ': r) (SchedulerProxy r) thisSchedulerProxy = return SchedulerProxy +-- | Use 'executeAndResume' to execute 'YieldProcess'. Refer to 'YieldProcess' +-- for more information. +yieldProcess + :: forall r q + . (HasCallStack, SetMember Process (Process q) r) + => SchedulerProxy q + -> Eff r () +yieldProcess _ = executeAndResume YieldProcess + -- | Send a message to a process addressed by the 'ProcessId'. -- See 'SendMessage'. sendMessage @@ -176,96 +205,150 @@ => SchedulerProxy q -> ProcessId -> Dynamic + -> Eff r () +sendMessage px pid message = void (sendMessageChecked px pid message) + +-- | Send a message to a process addressed by the 'ProcessId'. +-- See 'SendMessage'. Return @True@ if the process existed. +-- I you don't care, just 'sendMessage' instead. +sendMessageChecked + :: forall r q + . (HasCallStack, SetMember Process (Process q) r) + => SchedulerProxy q + -> ProcessId + -> Dynamic -> Eff r Bool -sendMessage _ pid message = - yieldProcess (SendMessage pid message) +sendMessageChecked _ pid message = + executeAndResume (SendMessage pid $! message) -- | Send a message to a process addressed by the 'ProcessId'. -- See 'SendMessage'. +sendMessageAs + :: forall o r q + . (HasCallStack, SetMember Process (Process q) r, Typeable o) + => SchedulerProxy q + -> ProcessId + -> o + -> Eff r () +sendMessageAs px pid = sendMessage px pid . toDyn + +-- | Exit a process addressed by the 'ProcessId'. +-- See 'SendShutdown'. sendShutdown :: forall r q . (HasCallStack, SetMember Process (Process q) r) => SchedulerProxy q -> ProcessId + -> Eff r () +sendShutdown px pid = void (sendShutdownChecked px pid) + +-- | Like 'sendShutdown', but also return @True@ iff the process to exit exists. +sendShutdownChecked + :: forall r q + . (HasCallStack, SetMember Process (Process q) r) + => SchedulerProxy q + -> ProcessId -> Eff r Bool -sendShutdown _ pid = - yieldProcess (SendShutdown pid) +sendShutdownChecked _ pid = executeAndResume (SendShutdown pid) -- | Start a new process, the new process will execute an effect, the function -- will return immediately with a 'ProcessId'. -spawn :: forall r q . - (HasCallStack, SetMember Process (Process q) r) - => Eff (Process q ': q) () - -> Eff r ProcessId -spawn child = - yieldProcess (Spawn @q child) +spawn + :: forall r q + . (HasCallStack, SetMember Process (Process q) r) + => Eff (Process q ': q) () + -> Eff r ProcessId +spawn child = executeAndResume (Spawn @q child) +-- | Like 'spawn' but return @()@. +spawn_ + :: forall r q + . (HasCallStack, SetMember Process (Process q) r) + => Eff (Process q ': q) () + -> Eff r () +spawn_ = void . spawn + -- | Block until a message was received. receiveMessage :: forall r q . (HasCallStack, SetMember Process (Process q) r) - => SchedulerProxy q -> Eff r Dynamic + => SchedulerProxy q + -> Eff r Dynamic receiveMessage _ = do - yieldProcess ReceiveMessage + executeAndResume ReceiveMessage -- | Receive and cast the message to some 'Typeable' instance. receiveMessageAs - :: forall a r q . - (HasCallStack, Typeable a, SetMember Process (Process q) r) - => SchedulerProxy q -> Eff r a -receiveMessageAs px = - do messageDynamic <- receiveMessage px - let castAndCheck dm = - case fromDynamic dm of - Nothing -> - Left ("Invalid message type received: " ++ show dm) - Just m -> - Right m - maybeMessage = castAndCheck messageDynamic - either (raiseError px) return maybeMessage + :: forall a r q + . (HasCallStack, Typeable a, SetMember Process (Process q) r) + => SchedulerProxy q + -> Eff r a +receiveMessageAs px = do + messageDynamic <- receiveMessage px + let castAndCheck dm = case fromDynamic dm of + Nothing -> Left ("Invalid message type received: " ++ show dm) + Just m -> Right m + maybeMessage = castAndCheck messageDynamic + either (raiseError px) return maybeMessage -- | Enter a loop to receive messages and pass them to a callback, until the -- function returns 'False'. receiveLoop :: forall r q - . ( SetMember Process (Process q) r - , HasCallStack) + . (SetMember Process (Process q) r, HasCallStack) => SchedulerProxy q -> (Either (Maybe String) Dynamic -> Eff r ()) -> Eff r () -receiveLoop _ handlers = - forever $ - do mReq <- send (ReceiveMessage @q) - case mReq of - RetryLastAction -> - return () - ShutdownRequested -> - handlers (Left Nothing) - OnError reason -> - handlers (Left (Just reason)) - ResumeWith message -> - handlers (Right message) +receiveLoop _ handlers = forever $ do + mReq <- send (ReceiveMessage @q) + case mReq of + RetryLastAction -> return () + ShutdownRequested -> handlers (Left Nothing) + OnError reason -> handlers (Left (Just reason)) + ResumeWith message -> handlers (Right message) -- | Returns the 'ProcessId' of the current process. -self :: (HasCallStack, SetMember Process (Process q) r) => SchedulerProxy q -> Eff r ProcessId -self _px = yieldProcess SelfPid +self + :: (HasCallStack, SetMember Process (Process q) r) + => SchedulerProxy q + -> Eff r ProcessId +self _px = executeAndResume SelfPid -- | Exit the process. -exitNormally :: forall r q a. (HasCallStack, SetMember Process (Process q) r) => SchedulerProxy q -> Eff r a +exitNormally + :: forall r q a + . (HasCallStack, SetMember Process (Process q) r) + => SchedulerProxy q + -> Eff r a exitNormally _ = send (Shutdown @q) -- | Exit the process with an error. -exitWithError :: forall r q a. (HasCallStack, SetMember Process (Process q) r) => SchedulerProxy q -> String -> Eff r a -exitWithError _ = send . ExitWithError @q +exitWithError + :: forall r q a + . (HasCallStack, SetMember Process (Process q) r) + => SchedulerProxy q + -> String + -> Eff r a +exitWithError _ = send . (ExitWithError @q $!) -- | Thrown an error, can be caught by 'catchRaisedError'. -raiseError :: forall r q b. (HasCallStack, SetMember Process (Process q) r) => SchedulerProxy q -> String -> Eff r b -raiseError _ = send . RaiseError @q +raiseError + :: forall r q b + . (HasCallStack, SetMember Process (Process q) r) + => SchedulerProxy q + -> String + -> Eff r b +raiseError _ = send . (RaiseError @q $!) -- | Catch and handle an error raised by 'raiseError'. Works independent of the -- handler implementation. catchRaisedError - :: forall r q w . (HasCallStack, SetMember Process (Process q) r) => SchedulerProxy q -> (String -> Eff r w) -> Eff r w -> Eff r w + :: forall r q w + . (HasCallStack, SetMember Process (Process q) r) + => SchedulerProxy q + -> (String -> Eff r w) + -> Eff r w + -> Eff r w catchRaisedError _ onErr = interpose return go where go :: forall b . Process q b -> (b -> Eff r w) -> Eff r w @@ -275,7 +358,10 @@ -- | Like 'catchRaisedError' it catches 'raiseError', but instead of invoking a -- user provided handler, the result is wrapped into an 'Either'. ignoreProcessError - :: (HasCallStack, SetMember Process (Process q) r) => SchedulerProxy q -> Eff r a -> Eff r (Either String a) + :: (HasCallStack, SetMember Process (Process q) r) + => SchedulerProxy q + -> Eff r a + -> Eff r (Either String a) ignoreProcessError px = catchRaisedError px (return . Left) . fmap Right -- | Each process is identified by a single process id, that stays constant diff --git a/src/Control/Eff/Concurrent/Process/ForkIOScheduler.hs b/src/Control/Eff/Concurrent/Process/ForkIOScheduler.hs --- a/src/Control/Eff/Concurrent/Process/ForkIOScheduler.hs +++ b/src/Control/Eff/Concurrent/Process/ForkIOScheduler.hs @@ -15,7 +15,6 @@ -- -- 'spawn' uses 'forkFinally' and 'STM.TQueue's and tries to catch -- most exceptions. -{-# LANGUAGE GeneralizedNewtypeDeriving #-} {-# LANGUAGE ConstraintKinds #-} {-# LANGUAGE FlexibleContexts #-} {-# LANGUAGE FlexibleInstances #-} @@ -46,7 +45,7 @@ import GHC.Stack import Data.Bifunctor import Data.Maybe -import Data.Kind () +import Data.Kind ( ) import qualified Control.Exception as Exc import Control.Concurrent as Concurrent import Control.Concurrent.STM as STM @@ -142,145 +141,144 @@ -- application. This function takes a 'Process' on top of the 'SchedulerIO' -- effect and a 'LogChannel' for concurrent logging. schedule :: Eff (ConsProcess SchedulerIO) () -> LogChannel String -> IO () -schedule e logC = - void - $ withNewSchedulerState - $ \schedulerStateVar -> - do pidVar <- newEmptyTMVarIO - scheduleProcessWithShutdownAction schedulerStateVar pidVar - $ do mt <- lift myThreadId - mp <- lift (atomically (readTMVar pidVar)) - logMsg (show mp ++ " main process started in thread " ++ show mt) - e - logMsg (show mp ++ " main process returned") - where - withNewSchedulerState :: (SchedulerVar -> IO a) -> IO a - withNewSchedulerState mainProcessAction = do - myTId <- myThreadId - Exc.bracket - (newTVarIO (Scheduler myPid Map.empty Map.empty False logC)) - (tearDownScheduler myTId) - (mainProcessAction . SchedulerVar) - - where - myPid = 1 - tearDownScheduler myTId v = do - logChannelPutIO logC (show myTId ++" begin scheduler tear down") - sch <- - (atomically - (do - sch <- readTVar v - let sch' = sch & schedulerShuttingDown .~ True - writeTVar v sch' - return sch' - ) - ) - logChannelPutIO logC (show myTId ++ " killing " ++ - let ts = (sch ^.. threadIdTable. traversed) - in if length ts > 100 - then show (length ts) ++ " threads" - else show ts ) - imapM_ (killProcThread myTId) (sch ^. threadIdTable) - Concurrent.yield - atomically +schedule e logC = void $ withNewSchedulerState $ \schedulerStateVar -> do + pidVar <- newEmptyTMVarIO + scheduleProcessWithShutdownAction schedulerStateVar pidVar $ do + mt <- lift myThreadId + mp <- lift (atomically (readTMVar pidVar)) + logMsg (show mp ++ " main process started in thread " ++ show mt) + e + logMsg (show mp ++ " main process returned") + where + withNewSchedulerState :: (SchedulerVar -> IO a) -> IO a + withNewSchedulerState mainProcessAction = do + myTId <- myThreadId + Exc.bracket (newTVarIO (Scheduler myPid Map.empty Map.empty False logC)) + (tearDownScheduler myTId) + (mainProcessAction . SchedulerVar) + where + myPid = 1 + tearDownScheduler myTId v = do + logChannelPutIO logC (show myTId ++ " begin scheduler tear down") + sch <- + (atomically (do - scheduler <- readTVar v - let allThreadsDead = scheduler^.threadIdTable.to Map.null - && scheduler^.processTable.to Map.null - STM.check allThreadsDead) - logChannelPutIO logC "all threads dead" + sch <- readTVar v + let sch' = sch & schedulerShuttingDown .~ True + writeTVar v sch' + return sch' + ) + ) + logChannelPutIO + logC + ( show myTId + ++ " killing " + ++ let ts = (sch ^.. threadIdTable . traversed) + in if length ts > 100 + then show (length ts) ++ " threads" + else show ts + ) + imapM_ (killProcThread myTId) (sch ^. threadIdTable) + Concurrent.yield + atomically + (do + scheduler <- readTVar v + let allThreadsDead = + scheduler + ^. threadIdTable + . to Map.null + && scheduler + ^. processTable + . to Map.null + STM.check allThreadsDead + ) + logChannelPutIO logC "all threads dead" - killProcThread myTId _pid tid = - when (myTId /= tid) ( killThread tid) + killProcThread myTId _pid tid = when (myTId /= tid) (killThread tid) -- | Start the message passing concurrency system then execute a 'Process' on -- top of 'SchedulerIO' effect. All logging is sent to standard output. defaultMain :: Eff (ConsProcess SchedulerIO) () -> IO () -defaultMain c = - runLoggingT - (logChannelBracket - 128 - (Just "~~~~~~ main process started") - (Just "====== main process exited") - (schedule c)) - (print :: String -> IO ()) +defaultMain c = runLoggingT + (logChannelBracket 128 + (Just "~~~~~~ main process started") + (Just "====== main process exited") + (schedule c) + ) + (print :: String -> IO ()) -- | Start the message passing concurrency system then execute a 'Process' on -- top of 'SchedulerIO' effect. All logging is sent to standard output. -defaultMainWithLogChannel :: LogChannel String -> Eff (ConsProcess SchedulerIO) () -> IO () -defaultMainWithLogChannel logC c = - closeLogChannelAfter - (Just (fromString "====== main process exited")) - logC - (schedule c logC) +defaultMainWithLogChannel + :: LogChannel String -> Eff (ConsProcess SchedulerIO) () -> IO () +defaultMainWithLogChannel logC c = closeLogChannelAfter + (Just (fromString "====== main process exited")) + logC + (schedule c logC) scheduleProcessWithShutdownAction :: SchedulerVar -> STM.TMVar ProcessId -> Eff (ConsProcess SchedulerIO) () -> IO (Either Exc.SomeException ()) -scheduleProcessWithShutdownAction schedulerVar pidVar procAction = - do cleanupVar <- newEmptyTMVarIO - logC <- getLogChannelIO schedulerVar - mTid <- myThreadId - eeres <- Exc.try (runProcEffects cleanupVar logC) - let eres = join eeres - getAndExecCleanup cleanupVar eres logC mTid - logChannelPutIO logC (show mTid ++ " <~< process cleanup finished") - return eres - where +scheduleProcessWithShutdownAction schedulerVar pidVar procAction = do + cleanupVar <- newEmptyTMVarIO + logC <- getLogChannelIO schedulerVar + mTid <- myThreadId + eeres <- Exc.try (runProcEffects cleanupVar logC) + let eres = join eeres + getAndExecCleanup cleanupVar eres logC mTid + logChannelPutIO logC (show mTid ++ " <~< process cleanup finished") + return eres + where - runProcEffects cleanupVar l = - Data.Bifunctor.first Exc.SomeException - <$> runLift - (logToChannel l - (runReader - (scheduleProcessWithCleanup - shutdownAction - saveCleanupAndSchedule) - schedulerVar)) - where - shutdownAction = - ShutdownAction - (\eres -> - do let ex = either id (const ProcessShuttingDown) eres - Exc.throw ex) - saveCleanupAndSchedule cleanUpAction pid = - do lift (atomically (do STM.putTMVar cleanupVar cleanUpAction - STM.putTMVar pidVar pid)) - mTid <- lift myThreadId - logMsg (show mTid ++ " >~> begin process " ++ show pid) - procAction - getAndExecCleanup cleanupVar eres lc mt = - do mcleanup <- atomically (STM.tryTakeTMVar cleanupVar) - traverse_ execCleanup mcleanup - where - execCleanup ca = - do runCleanUpAction ca - logChannelPutIO lc - $ show mt ++ - case eres of - Left se -> - case Exc.fromException se of - Nothing -> - " process caught exception: " - ++ Exc.displayException se - Just schedulerErr -> - (case schedulerErr of - ProcessShuttingDown -> - " process shutdown" - ProcessExitError m -> - " process exited with error: " ++ show m - ProcessRaisedError m -> - " process raised error: " ++ show m - _ -> - " scheduler error: " ++ show schedulerErr) - ++ " - full exception message: " - ++ Exc.displayException se + runProcEffects cleanupVar l = + Data.Bifunctor.first Exc.SomeException <$> runLift + (logToChannel + l + (runReader + (scheduleProcessWithCleanup shutdownAction saveCleanupAndSchedule) + schedulerVar + ) + ) + where + shutdownAction = ShutdownAction + (\eres -> do + let ex = either id (const ProcessShuttingDown) eres + Exc.throw ex + ) + saveCleanupAndSchedule cleanUpAction pid = do + lift + (atomically + (do + STM.putTMVar cleanupVar cleanUpAction + STM.putTMVar pidVar pid + ) + ) + mTid <- lift myThreadId + logMsg (show mTid ++ " >~> begin process " ++ show pid) + procAction + getAndExecCleanup cleanupVar eres lc mt = do + mcleanup <- atomically (STM.tryTakeTMVar cleanupVar) + traverse_ execCleanup mcleanup + where + execCleanup ca = do + runCleanUpAction ca + logChannelPutIO lc $ show mt ++ case eres of + Left se -> case Exc.fromException se of + Nothing -> " process caught exception: " ++ Exc.displayException se + Just schedulerErr -> + (case schedulerErr of + ProcessShuttingDown -> " process shutdown" + ProcessExitError m -> " process exited with error: " ++ show m + ProcessRaisedError m -> " process raised error: " ++ show m + _ -> " scheduler error: " ++ show schedulerErr + ) + ++ " - full exception message: " + ++ Exc.displayException se - Right _ -> - " process function returned" + Right _ -> " process function returned" getLogChannel :: HasSchedulerIO r => Eff r (LogChannel String) @@ -297,8 +295,7 @@ => ProcessId -> Mtl.StateT ProcessInfo STM.STM a -> Eff r (Either SchedulerError a) -overProcessInfo pid stAction = - overScheduler +overProcessInfo pid stAction = overScheduler (do res <- use (processTable . at pid) case res of @@ -306,24 +303,22 @@ Just pinfo -> do (x, pinfoOut) <- Mtl.lift (Mtl.runStateT stAction pinfo) processTable . at pid . _Just .= pinfoOut - return (Right x)) + return (Right x) + ) -- ** MessagePassing execution -spawnImpl :: HasCallStack - => Eff (ConsProcess SchedulerIO) () - -> Eff SchedulerIO ProcessId +spawnImpl + :: HasCallStack + => Eff (ConsProcess SchedulerIO) () + -> Eff SchedulerIO ProcessId spawnImpl mfa = do schedulerVar <- ask - pidVar <- lift STM.newEmptyTMVarIO - void - $ lift - $ Concurrent.forkIO - $ void - $ scheduleProcessWithShutdownAction - schedulerVar - pidVar - mfa + pidVar <- lift STM.newEmptyTMVarIO + void $ lift $ Concurrent.forkIO $ void $ scheduleProcessWithShutdownAction + schedulerVar + pidVar + mfa lift Concurrent.yield -- this is important, removing this causes test failures lift (atomically (STM.readTMVar pidVar)) @@ -334,117 +329,113 @@ => ShutdownAction -> (CleanUpAction -> ProcessId -> Eff (ConsProcess SchedulerIO) ()) -> Eff SchedulerIO (Either SchedulerError ()) -scheduleProcessWithCleanup shutdownAction processAction = - withMessageQueue - (\cleanUpAction pinfo -> - handle_relay - (\x -> return x) - (go (pinfo ^. processId)) - (processAction cleanUpAction (pinfo ^. processId))) +scheduleProcessWithCleanup shutdownAction processAction = withMessageQueue + (\cleanUpAction pinfo -> handle_relay + (\x -> return x) + (go (pinfo ^. processId)) + (processAction cleanUpAction (pinfo ^. processId)) + ) where - shutdownOrGo :: forall v a + shutdownOrGo + :: forall v a . HasCallStack => ProcessId -> (ResumeProcess v -> Eff SchedulerIO a) -> Eff SchedulerIO a -> Eff SchedulerIO a - shutdownOrGo pid k ok = - do psVar <- getSchedulerTVar - eHasShutdowReq <- - lift - (do p <- atomically (readTVar psVar) - let mPinfo = p ^. processTable . at pid - case mPinfo of - Just pinfo -> - atomically - (do wasRequested <- readTVar (pinfo ^. shutdownRequested) - -- reset the shutdwown request flag, in case the shutdown - -- is prevented by the process - writeTVar (pinfo ^. shutdownRequested) False - return (Right wasRequested)) - Nothing -> - return (Left (ProcessNotFound pid))) - case eHasShutdowReq of - Right True -> - k ShutdownRequested - Right False -> - ok - Left e -> - k (OnError (show e)) + shutdownOrGo pid k ok = do + psVar <- getSchedulerTVar + eHasShutdowReq <- lift + (do + p <- atomically (readTVar psVar) + let mPinfo = p ^. processTable . at pid + case mPinfo of + Just pinfo -> atomically + (do + wasRequested <- readTVar (pinfo ^. shutdownRequested) + -- reset the shutdwown request flag, in case the shutdown + -- is prevented by the process + writeTVar (pinfo ^. shutdownRequested) False + return (Right wasRequested) + ) + Nothing -> return (Left (ProcessNotFound pid)) + ) + case eHasShutdowReq of + Right True -> k ShutdownRequested + Right False -> ok + Left e -> k (OnError (show e)) - go :: forall v a + go + :: forall v a . HasCallStack => ProcessId -> Process SchedulerIO v -> (v -> Eff SchedulerIO a) -> Eff SchedulerIO a - go pid (SendMessage toPid reqIn) k = - shutdownOrGo pid k $ - do eres <- - do psVar <- getSchedulerTVar - lift - (Right <$> - (do p <- atomically (readTVar psVar) - let mto = p ^. processTable . at toPid - case mto of - Just toProc -> - do atomically - (writeTQueue - (toProc ^. messageQ) - (Just reqIn)) - return True - Nothing -> - return False)) - lift Concurrent.yield - let kArg = either (OnError . show @SchedulerError) ResumeWith eres - k kArg + go pid (SendMessage toPid reqIn) k = shutdownOrGo pid k $ do + eres <- do + psVar <- getSchedulerTVar + lift + ( Right + <$> (do + p <- atomically (readTVar psVar) + let mto = p ^. processTable . at toPid + case mto of + Just toProc -> do + atomically (writeTQueue (toProc ^. messageQ) (Just reqIn)) + return True + Nothing -> return False + ) + ) + lift Concurrent.yield + let kArg = either (OnError . show @SchedulerError) ResumeWith eres + k kArg - go pid (SendShutdown toPid) k = - shutdownOrGo pid k - $ do eres <- - do psVar <- getSchedulerTVar - lift - (Right <$> - (do p <- atomically (readTVar psVar) - let mto = p ^. processTable . at toPid - case mto of - Just toProc -> - atomically $ - do writeTVar (toProc ^. shutdownRequested) True - writeTQueue (toProc ^. messageQ) Nothing - return True - Nothing -> - return False)) - let kArg = either (OnError . show @SchedulerError) resume eres - resume = if toPid == pid - then const ShutdownRequested - else ResumeWith - lift Concurrent.yield - k kArg + go pid (SendShutdown toPid) k = shutdownOrGo pid k $ do + eres <- do + psVar <- getSchedulerTVar + lift + ( Right + <$> (do + p <- atomically (readTVar psVar) + let mto = p ^. processTable . at toPid + case mto of + Just toProc -> atomically $ do + writeTVar (toProc ^. shutdownRequested) True + writeTQueue (toProc ^. messageQ) Nothing + return True + Nothing -> return False + ) + ) + let kArg = either (OnError . show @SchedulerError) resume eres + resume = if toPid == pid then const ShutdownRequested else ResumeWith + lift Concurrent.yield + k kArg - go pid (Spawn child) k = - shutdownOrGo pid k $ - do res <- spawnImpl child - k (ResumeWith res) + go pid (Spawn child) k = shutdownOrGo pid k $ do + res <- spawnImpl child + k (ResumeWith res) - go pid ReceiveMessage k = - shutdownOrGo pid k $ - do emq <- overProcessInfo pid (use messageQ) - case emq of - Left e -> - k (OnError (show @SchedulerError e)) - Right mq -> do - emdynMsg <- lift (Right <$> (atomically (readTQueue mq))) - k (either - (OnError . show @SchedulerError) - (maybe RetryLastAction ResumeWith) - emdynMsg) + go pid ReceiveMessage k = shutdownOrGo pid k $ do + emq <- overProcessInfo pid (use messageQ) + case emq of + Left e -> k (OnError (show @SchedulerError e)) + Right mq -> do + emdynMsg <- lift (Right <$> (atomically (readTQueue mq))) + k + (either (OnError . show @SchedulerError) + (maybe RetryLastAction ResumeWith) + emdynMsg + ) - go pid SelfPid k = - shutdownOrGo pid k $ - do lift Concurrent.yield - k (ResumeWith pid) + go pid SelfPid k = shutdownOrGo pid k $ do + lift Concurrent.yield + k (ResumeWith pid) + go pid YieldProcess k = shutdownOrGo pid k $ do + lift Concurrent.yield + k (ResumeWith ()) + go _pid Shutdown _k = do invokeShutdownAction shutdownAction (Right ()) @@ -457,12 +448,12 @@ data ShutdownAction = ShutdownAction (forall a . Either SchedulerError () -> IO a) -invokeShutdownAction :: ( HasCallStack - , SetMember Lift (Lift IO) r - , Member (Logs String) r) - => ShutdownAction -> Either SchedulerError () -> Eff r a -invokeShutdownAction (ShutdownAction a) res = - lift (a res) +invokeShutdownAction + :: (HasCallStack, SetMember Lift (Lift IO) r, Member (Logs String) r) + => ShutdownAction + -> Either SchedulerError () + -> Eff r a +invokeShutdownAction (ShutdownAction a) res = lift (a res) withMessageQueue :: HasSchedulerIO r @@ -478,8 +469,7 @@ lc (pinfo ^. processId) Right <$> m cleanUpAction pinfo - Left e -> - return $ Left e + Left e -> return $ Left e where createQueue = do myTId <- lift myThreadId @@ -489,8 +479,8 @@ if abortNow then return (Left SchedulerShuttingDown) else do - pid <- nextPid <<+= 1 - channel <- Mtl.lift newTQueue + pid <- nextPid <<+= 1 + channel <- Mtl.lift newTQueue shutdownIndicator <- Mtl.lift (newTVar False) let pinfo = ProcessInfo pid channel shutdownIndicator threadIdTable . at pid .= Just myTId @@ -501,9 +491,12 @@ didWork <- Exc.try (overSchedulerIO psVar - (do os <- processTable . at pid <<.= Nothing - ot <- threadIdTable . at pid <<.= Nothing - return (os, isJust os || isJust ot))) + (do + os <- processTable . at pid <<.= Nothing + ot <- threadIdTable . at pid <<.= Nothing + return (os, isJust os || isJust ot) + ) + ) let getCause = Exc.try @Exc.SomeException (overSchedulerIO psVar (preuse (processTable . at pid))) @@ -512,8 +505,7 @@ (return . (maybe (show pid) show)) case didWork of - Right (_pinfo, True) -> - return () + Right (_pinfo, True) -> return () Right (pinfo, False) -> logChannelPutIO lc ("queue already destroyed: " ++ show pinfo) Left (e :: Exc.SomeException) -> @@ -528,17 +520,17 @@ -> Eff r (Either SchedulerError a) overScheduler stAction = do psVar <- getSchedulerTVar - -- liftCatch (Left . LowLevelIOException) (overSchedulerIO psVar stAction) lift (overSchedulerIO psVar stAction) -overSchedulerIO - :: STM.TVar Scheduler -> Mtl.StateT Scheduler STM.STM a -> IO a -overSchedulerIO psVar stAction = - do STM.atomically - (do ps <- STM.readTVar psVar - (result, psModified) <- Mtl.runStateT stAction ps - STM.writeTVar psVar psModified - return result) +overSchedulerIO :: STM.TVar Scheduler -> Mtl.StateT Scheduler STM.STM a -> IO a +overSchedulerIO psVar stAction = do + STM.atomically + (do + ps <- STM.readTVar psVar + (result, psModified) <- Mtl.runStateT stAction ps + STM.writeTVar psVar psModified + return result + ) getSchedulerTVar :: HasSchedulerIO r => Eff r (TVar Scheduler) getSchedulerTVar = fromSchedulerVar <$> ask diff --git a/src/Control/Eff/Concurrent/Process/Interactive.hs b/src/Control/Eff/Concurrent/Process/Interactive.hs new file mode 100644 --- /dev/null +++ b/src/Control/Eff/Concurrent/Process/Interactive.hs @@ -0,0 +1,128 @@ +module Control.Eff.Concurrent.Process.Interactive +where + +import Control.Arrow +import Control.Concurrent +import Control.Eff +import Control.Eff.Lift +import Control.Eff.Concurrent.Api +import Control.Eff.Concurrent.Api.Client +import Control.Eff.Concurrent.Process +import Control.Monad +import Data.Typeable ( Typeable ) + + +-- | This module provides support for executing 'Process' actions from 'IO'. +-- +-- One use case is interacting with processes from the REPL, e.g.: +-- +-- >>> import Control.Eff.Concurrent.Process.SingleThreadedScheduler (defaultMain) +-- +-- >>> import Data.Dynamic +-- +-- >>> import Data.Maybe +-- +-- >>> s <- forkInteractiveScheduler Control.Eff.Concurrent.Process.SingleThreadedScheduler.defaultMain +-- +-- >>> fooPid <- submit s (spawn (forever (receiveMessage SP >>= (logMsg . fromMaybe "Huh!??" . fromDynamic)))) +-- +-- >>> fooPid +-- <0.1.0> +-- +-- >>> submit s (sendMessageAs SP fooPid "test") +-- test +-- +-- >>> submit s (sendShutdown SP fooPid) +-- +-- +-- +-- @since 0.3.0.0 + +data SchedulerVar r = + SchedulerVar { _schedulerThreadId :: ThreadId + , _schedulerInQueue :: MVar (Eff (Process r ': r) (Maybe String)) + } + + +-- | Fork a scheduler with a process that communicates with it via 'MVar', +-- which is also the reason for the @Lift IO@ constraint. +forkInteractiveScheduler + :: forall r + . (SetMember Lift (Lift IO) r) + => (Eff (Process r ': r) () -> IO ()) + -> IO (SchedulerVar r) +forkInteractiveScheduler ioScheduler = do + inQueue <- newEmptyMVar + tid <- forkIO (ioScheduler (readEvalPrintLoop inQueue)) + return (SchedulerVar tid inQueue) + where + readEvalPrintLoop = forever . (readAction >>> evalAction >=> printResult) + where + readAction v = do + mr <- lift (tryTakeMVar v) + case mr of + Nothing -> do + yieldProcess SP + readAction v + Just r -> return r + evalAction = join + printResult = mapM_ (lift . putStrLn) + +-- | Exit the schedulder immediately using an asynchronous exception. +killInteractiveScheduler :: SchedulerVar r -> IO () +killInteractiveScheduler = killThread . _schedulerThreadId + +-- | Send a 'Process' effect to the main process of a scheduler, this blocks +-- until the effect is executed. +submit + :: forall r a + . (SetMember Lift (Lift IO) r) + => SchedulerVar r + -> Eff (Process r ': r) a + -> IO a +submit r theAction = do + resVar <- newEmptyMVar + worked <- tryPutMVar (_schedulerInQueue r) (runAndPutResult resVar) + if worked then takeMVar resVar else fail "ERROR: Scheduler still busy" + where + runAndPutResult resVar = do + res <- theAction + lift (putMVar resVar $! res) + return Nothing + +-- | Send a 'Process' effect to the main process of a scheduler, this blocks +-- until the effect is executed, then the result is printed by the thread, +-- that runs the process 0 in the scheduler. +submitPrint + :: forall r a + . (Show a, SetMember Lift (Lift IO) r) + => SchedulerVar r + -> Eff (Process r ': r) a + -> IO () +submitPrint r theAction = do + worked <- tryPutMVar (_schedulerInQueue r) runAndShowResult + if worked then return () else fail "ERROR: Scheduler still busy" + where + runAndShowResult = do + res <- theAction + return (Just $! (show res)) + +-- | Combination of 'submit' and 'cast'. +submitCast + :: forall o r + . (SetMember Lift (Lift IO) r, Typeable o) + => SchedulerVar r + -> Server o + -> Api o 'Asynchronous + -> IO () +submitCast sc svr request = submit sc (cast SchedulerProxy svr request) + +-- | Combination of 'submit' and 'cast'. +submitCall + :: forall o q r + . (SetMember Lift (Lift IO) r, Typeable o, Typeable q) + => SchedulerVar r + -> Server o + -> Api o ( 'Synchronous q) + -> IO q +submitCall sc svr request = submit sc (call SchedulerProxy svr request) diff --git a/src/Control/Eff/Concurrent/Process/SingleThreadedScheduler.hs b/src/Control/Eff/Concurrent/Process/SingleThreadedScheduler.hs --- a/src/Control/Eff/Concurrent/Process/SingleThreadedScheduler.hs +++ b/src/Control/Eff/Concurrent/Process/SingleThreadedScheduler.hs @@ -1,129 +1,143 @@ -- | A coroutine based, single threaded scheduler for 'Process'es. module Control.Eff.Concurrent.Process.SingleThreadedScheduler ( schedule + , scheduleWithYield , defaultMain , singleThreadedIoScheduler - , LoggingAndIo) where + , LoggingAndIo + ) +where -import Control.Eff -import Control.Eff.Lift -import Control.Eff.Log -import Control.Eff.Concurrent.Process -import Control.Lens hiding ((|>), Empty) -import qualified Data.Sequence as Seq -import Data.Sequence (Seq (..)) -import qualified Data.Map.Strict as Map -import GHC.Stack -import Data.Kind () -import Data.Dynamic -import Data.Maybe -import Control.Monad +import Control.Concurrent ( yield ) +import Control.Eff +import Control.Eff.Lift +import Control.Eff.Log +import Control.Eff.Concurrent.Process +import Control.Lens hiding ( (|>) + , Empty + ) +import qualified Data.Sequence as Seq +import Data.Sequence ( Seq(..) ) +import qualified Data.Map.Strict as Map +import GHC.Stack +import Data.Kind ( ) +import Data.Dynamic +import Data.Maybe +-- | Execute a 'Process' in the current thread, all child processes spawned by +-- 'spawn' will be executed concurrently using a co-routine based, round-robin +-- scheduler. +schedule :: forall r . Eff (Process r ': r) () -> Eff r () +schedule = scheduleWithYield (return ()) -- | Execute a 'Process' in the current thread, all child processes spawned by -- 'spawn' will be executed concurrently using a co-routine based, round-robin -- scheduler. -schedule - :: forall r . Eff (Process r ': r) () - -> Eff r () -schedule mainProcessAction = - do - y <- runAsCoroutine mainProcessAction - go 1 (Map.singleton 0 Seq.empty) (Seq.singleton (y, 0)) - where +scheduleWithYield :: forall r . Eff r () -> Eff (Process r ': r) () -> Eff r () +scheduleWithYield yieldEff mainProcessAction = do + y <- runAsCoroutine mainProcessAction + go 1 (Map.singleton 0 Seq.empty) (Seq.singleton (y, 0)) + where - go :: ProcessId -> Map.Map ProcessId (Seq Dynamic) -> Seq (OnYield r, ProcessId) -> Eff r () - go _newPid _msgQs Empty = return () + go + :: ProcessId + -> Map.Map ProcessId (Seq Dynamic) + -> Seq (OnYield r, ProcessId) + -> Eff r () + go _newPid _msgQs Empty = return () - go newPid msgQs allProcs@((processState, pid) :<| rest) = - let handleExit = - if pid == 0 then return () - else go newPid - (msgQs & at pid .~ Nothing) - rest + go newPid msgQs allProcs@((processState, pid) :<| rest) + = let + handleExit = if pid == 0 + then return () + else go newPid (msgQs & at pid .~ Nothing) rest + maybeYield = if pid == 0 then yieldEff else return () + in + case processState of + OnDone -> handleExit - in case processState of - OnDone -> handleExit + OnRaiseError _ -> handleExit - OnRaiseError _ -> handleExit + OnExitError _ -> handleExit - OnExitError _ -> handleExit + OnSendShutdown targetPid k -> do + let allButTarget = + Seq.filter (\(_, e) -> e /= pid && e /= targetPid) allProcs + targets = Seq.filter (\(_, e) -> e == targetPid) allProcs + suicide = targetPid == pid + targetFound = suicide || not (Seq.null targets) + if suicide + then do + nextK <- k ShutdownRequested + go newPid msgQs (rest :|> (nextK, pid)) + else do + let deliverTheGoodNews (targetState, tPid) = do + nextTargetState <- case targetState of + OnSendShutdown _ tk -> tk ShutdownRequested + OnYield tk -> tk ShutdownRequested + OnSelf tk -> tk ShutdownRequested + OnSend _ _ tk -> tk ShutdownRequested + OnRecv tk -> tk ShutdownRequested + OnSpawn _ tk -> tk ShutdownRequested + OnDone -> return OnDone + OnExitError er -> return (OnExitError er) + OnRaiseError er -> return (OnExitError er) -- return (error ("TODO write test "++er)) + return (nextTargetState, tPid) + nextTargets <- traverse deliverTheGoodNews targets + nextK <- k (ResumeWith targetFound) + maybeYield + go newPid + msgQs + (allButTarget Seq.>< (nextTargets :|> (nextK, pid))) - OnSendShutdown targetPid k -> - do let allButTarget = - Seq.filter (\(_,e) -> e /= pid && e /= targetPid) allProcs - targets = - Seq.filter (\(_,e) -> e == targetPid) allProcs - suicide = targetPid == pid - targetFound = suicide || not (Seq.null targets) - if suicide - then - do nextK <- k ShutdownRequested - go newPid - msgQs - (rest :|> (nextK, pid)) - else - do let deliverTheGoodNews (targetState, tPid) = - do nextTargetState <- - case targetState of - OnSendShutdown _ tk -> - tk ShutdownRequested - OnSelf tk -> - tk ShutdownRequested - OnSend _ _ tk -> - tk ShutdownRequested - OnRecv tk -> - tk ShutdownRequested - OnSpawn _ tk -> - tk ShutdownRequested - OnDone -> return OnDone - OnExitError er -> return (OnExitError er) - OnRaiseError er -> return (OnExitError er) -- return (error ("TODO write test "++er)) - return (nextTargetState, tPid) - nextTargets <- traverse deliverTheGoodNews targets - nextK <- k (ResumeWith targetFound) - go newPid - msgQs - (allButTarget Seq.>< (nextTargets :|> (nextK, pid))) - OnSelf k -> - do nextK <- k (ResumeWith pid) - go newPid - msgQs - (rest :|> (nextK, pid)) + OnSelf k -> do + nextK <- k (ResumeWith pid) + maybeYield + go newPid msgQs (rest :|> (nextK, pid)) - OnSend toPid msg k -> - do nextK <- k (ResumeWith (msgQs ^. at toPid . to isJust)) - go newPid - (msgQs & at toPid . _Just %~ (:|> msg)) - (rest :|> (nextK, pid)) + OnYield k -> do + yieldEff + nextK <- k (ResumeWith ()) + go newPid msgQs (rest :|> (nextK, pid)) - recv@(OnRecv k) -> - case msgQs ^. at pid of - Nothing -> - do nextK <- k (OnError (show pid ++ " has no message queue!")) - go newPid msgQs (rest :|> (nextK, pid)) - Just Empty -> - if Seq.length rest == 0 then - do nextK <- k (OnError ("Process " ++ show pid ++ " deadlocked!")) - go newPid msgQs (rest :|> (nextK, pid)) - else - go newPid msgQs (rest :|> (recv, pid)) + OnSend toPid msg k -> do + nextK <- k (ResumeWith (msgQs ^. at toPid . to isJust)) + maybeYield + go newPid + (msgQs & at toPid . _Just %~ (:|> msg)) + (rest :|> (nextK, pid)) - Just (nextMessage :<| restMessages) -> - do nextK <- k (ResumeWith nextMessage) - go newPid - (msgQs & at pid . _Just .~ restMessages) - (rest :|> (nextK, pid)) + recv@(OnRecv k) -> case msgQs ^. at pid of + Nothing -> do + nextK <- k (OnError (show pid ++ " has no message queue!")) + maybeYield + go newPid msgQs (rest :|> (nextK, pid)) + Just Empty -> if Seq.length rest == 0 + then do + nextK <- k (OnError ("Process " ++ show pid ++ " deadlocked!")) + maybeYield + go newPid msgQs (rest :|> (nextK, pid)) + else go newPid msgQs (rest :|> (recv, pid)) - OnSpawn f k -> - do nextK <- k (ResumeWith newPid) - fk <- runAsCoroutine f - go (newPid + 1) - (msgQs & at newPid .~ Just Seq.empty) - ( rest :|> (nextK, pid) :|> (fk, newPid)) + Just (nextMessage :<| restMessages) -> do + nextK <- k (ResumeWith nextMessage) + maybeYield + go newPid + (msgQs & at pid . _Just .~ restMessages) + (rest :|> (nextK, pid)) + OnSpawn f k -> do + nextK <- k (ResumeWith newPid) + fk <- runAsCoroutine f + maybeYield + go (newPid + 1) + (msgQs & at newPid .~ Just Seq.empty) + (rest :|> (nextK, pid) :|> (fk, newPid)) + data OnYield r where + OnYield :: (ResumeProcess () -> Eff r (OnYield r)) + -> OnYield r OnSelf :: (ResumeProcess ProcessId -> Eff r (OnYield r)) -> OnYield r OnSpawn :: Eff (Process r ': r) () @@ -139,19 +153,19 @@ -> OnYield r OnSendShutdown :: ProcessId -> (ResumeProcess Bool -> Eff r (OnYield r)) -> OnYield r -runAsCoroutine :: forall r v . Eff (Process r ': r) v - -> Eff r (OnYield r) +runAsCoroutine :: forall r v . Eff (Process r ': r) v -> Eff r (OnYield r) runAsCoroutine m = handle_relay (const $ return OnDone) cont m - where - cont :: Process r x -> (x -> Eff r (OnYield r)) -> Eff r (OnYield r) - cont SelfPid k = return (OnSelf k) - cont (Spawn e) k = return (OnSpawn e k) - cont Shutdown _k = return OnDone - cont (ExitWithError e) _k = return (OnExitError e) - cont (RaiseError e) _k = return (OnRaiseError e) - cont (SendMessage tp msg) k = return (OnSend tp msg k) - cont ReceiveMessage k = return (OnRecv k) - cont (SendShutdown pid) k = return (OnSendShutdown pid k) + where + cont :: Process r x -> (x -> Eff r (OnYield r)) -> Eff r (OnYield r) + cont YieldProcess k = return (OnYield k) + cont SelfPid k = return (OnSelf k) + cont (Spawn e) k = return (OnSpawn e k) + cont Shutdown _k = return OnDone + cont (ExitWithError e ) _k = return (OnExitError e) + cont (RaiseError e ) _k = return (OnRaiseError e) + cont (SendMessage tp msg) k = return (OnSend tp msg k) + cont ReceiveMessage k = return (OnRecv k) + cont (SendShutdown pid) k = return (OnSendShutdown pid k) -- | The concrete list of 'Eff'ects for running this pure scheduler on @IO@ and -- with string logging. @@ -170,21 +184,5 @@ :: HasCallStack => Eff '[Process '[Logs String, Lift IO], Logs String, Lift IO] () -> IO () -defaultMain go = runLift $ handleLogsWith (schedule go) ($ putStrLn) - -_example :: IO () -_example = defaultMain go - where - px :: SchedulerProxy '[Logs String, Lift IO] - px = SchedulerProxy - go :: Eff (ConsProcess '[Logs String, Lift IO]) () - go = do - p1 <- spawn $ do - logMsg "Hello World!" - me <- self px - logMsg ("I am: " ++ show me) - d <- receiveMessage px - let m = fromDyn @String d "" - logMsg ("Got: " ++ m) - void $ sendMessage px p1 (toDyn "huhu") - return () +defaultMain go = + runLift $ handleLogsWith (scheduleWithYield (lift yield) go) ($ putStrLn) 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 @@ -5,78 +5,21 @@ {-# LANGUAGE TypeOperators #-} -- | Add-ons to 'Control.Eff.Exception' module Control.Eff.ExceptionExtra - ( try - , liftRethrow - , runErrorRethrowIO - , liftCatch - , liftCatchEff - , liftTry + ( liftTry , module X ) where -import Control.Monad ( (>=>) ) import qualified Control.Exception as Exc import Control.Eff import Control.Eff.Lift import Control.Eff.Exception as X import GHC.Stack --- * Effects combining "Control.Eff.Exception" and 'Exc.try' - --- | Catch an exception and return it in an 'Either'. -try :: forall e r a . Member (Exc e) r => Eff r a -> Eff r (Either e a) -try e = catchError (Right <$> e) (return . Left) - --- | Lift an IO action and catch all error using 'Exc.try' then wrap the --- 'Exc.Exception' using a given wrapper function and rethrow it using --- 'X.throwError'. -liftRethrow - :: forall proxy e r a - . (Exc.Exception e, SetMember Lift (Lift IO) r, SetMember Exc (Exc e) r) - => proxy e - -> IO a - -> Eff r a -liftRethrow _ m = lift (Exc.try m) >>= either (throwError @e) return - --- | Run an effect with exceptions like 'X.runError' and rethrow it as --- 'Exc.SomeException' using 'Exc.throw' -runErrorRethrowIO - :: forall e r a - . (Exc.Exception e, SetMember Lift (Lift IO) r) - => Eff (Exc e ': r) a - -> Eff r a -runErrorRethrowIO = runError >=> either (Exc.throw . Exc.SomeException) return - --- * Effect Utilities around 'Exc.try' - --- | Lift an IO action and catch all errors with 'Exc.try' with a pure --- handler. -liftCatch - :: forall e r a - . (HasCallStack, Exc.Exception e, SetMember Lift (Lift IO) r) - => (e -> a) - -> IO a - -> Eff r a -liftCatch handleE m = - lift (Exc.try m) >>= either (return . handleE) return - --- | Lift an IO action and catch all errors with 'Exc.try' with an effect --- handler. -liftCatchEff - :: forall e r a - . (SetMember Lift (Lift IO) r, HasCallStack, Exc.Exception e) - => (e -> Eff r a) - -> IO a - -> Eff r a -liftCatchEff handleE m = - lift (Exc.try m) >>= either handleE return - -- | Catch 'Exc.Exception' thrown by an effect. liftTry :: forall e r a . (HasCallStack, Exc.Exception e, SetMember Lift (Lift IO) r) => Eff r a -> Eff r (Either e a) -liftTry m = - (Right <$> m) `catchDynE` (return . Left) +liftTry m = (Right <$> m) `catchDynE` (return . Left) diff --git a/src/Control/Eff/Interactive.hs b/src/Control/Eff/Interactive.hs deleted file mode 100644 --- a/src/Control/Eff/Interactive.hs +++ /dev/null @@ -1,77 +0,0 @@ -{-# LANGUAGE FlexibleContexts #-} -{-# LANGUAGE DataKinds #-} -{-# LANGUAGE TypeOperators #-} -{-# LANGUAGE GADTs #-} -module Control.Eff.Interactive - ( Interaction(..) - , printLine - , printStep - , promptStep - , step - , Interactive(..) - , interactiveProgram - , runInteractionIOE - , runInteractionIO - ) -where - -import Control.Eff -import Control.Eff.Lift -import Control.Eff.Operational -import Control.Monad ( void ) -import GHC.Stack - -data Interaction a where - PrintLine :: String -> Interaction () - ReadLine :: (String -> a) -> Interaction a - Step :: Interaction () - -printLine - :: (Member (Program Interaction) r, HasCallStack) => String -> Eff r () -printLine = singleton . PrintLine - -printStep - :: (Member (Program Interaction) r, HasCallStack) => String -> Eff r () -printStep m = do - printLine m - step - -step :: (Member (Program Interaction) r, HasCallStack) => Eff r () -step = do - printLine "\nPress ENTER to continue." - singleton Step - -promptStep - :: (Member (Program Interaction) r, HasCallStack) - => String - -> (String -> a) - -> Eff r a -promptStep m p = do - singleton (PrintLine m) - singleton (ReadLine p) - - -runInteractionIOE - :: (SetMember Lift (Lift IO) r, HasCallStack) - => Eff (Program Interaction ': r) a - -> Eff r a -runInteractionIOE = runProgram go - where - go :: (SetMember Lift (Lift IO) r, HasCallStack) => Interaction a -> Eff r a - go (PrintLine m) = lift (putStrLn m) - go Step = lift (void getLine) - go (ReadLine parse) = lift (parse <$> getLine) - -runInteractionIO :: Eff '[Program Interaction, Lift IO] a -> IO a -runInteractionIO = runLift . runInteractionIOE - --- ** Interactively work with 'Control.Eff.Operational.Program's. - -class Interactive f where - singleSteps :: (Member (Program Interaction) r, HasCallStack) => f a -> Eff r a - -interactiveProgram - :: (HasCallStack, Member (Program Interaction) r, Interactive f) - => Eff (Program f ': r) a - -> Eff r a -interactiveProgram = runProgram singleSteps diff --git a/test/Common.hs b/test/Common.hs --- a/test/Common.hs +++ b/test/Common.hs @@ -32,10 +32,10 @@ testLogC = filterLogChannel (\m -> take (length logPrefix) m == logPrefix) <$> - forkLogger 1 (putStrLn) (Just (logPrefix ++ "^^^^^^^^^^^^^^ LogChannel Start ^^^^^^^^^^^^^^")) + forkLogger 1 (putStrLn) Nothing testLogJoin :: LogChannel String -> IO () -testLogJoin = joinLogChannel (Just (logPrefix ++ "^^^^^^^^^^^^^^ LogChannel Done ^^^^^^^^^^^^^^")) +testLogJoin = joinLogChannel Nothing tlog :: Member (Logs String) r => String -> Eff r () tlog = logMsg . (logPrefix ++) diff --git a/test/Debug.hs b/test/Debug.hs new file mode 100644 --- /dev/null +++ b/test/Debug.hs @@ -0,0 +1,9 @@ +module Debug +where + +import ProcessBehaviourTestCases +import Test.Tasty +import Test.Tasty.HUnit + +debugMain :: IO () +debugMain = defaultMain test_forkIo diff --git a/test/ForkIOScheduler.hs b/test/ForkIOScheduler.hs --- a/test/ForkIOScheduler.hs +++ b/test/ForkIOScheduler.hs @@ -1,111 +1,154 @@ -module ForkIOScheduler where - -import Control.Exception -import Control.Concurrent -import Control.Concurrent.STM -import Control.Eff -import Control.Eff.Lift -import Control.Eff.Concurrent.Process -import Control.Eff.Concurrent.Process.ForkIOScheduler as Scheduler -import Control.Monad (void, forever) -import Test.Tasty -import Test.Tasty.HUnit -import Data.Dynamic -import Common +module ForkIOScheduler +where -timeoutSeconds :: Integer -> Timeout -timeoutSeconds seconds = Timeout (seconds * 1000000) (show seconds ++ "s") +import Control.Exception +import Control.Concurrent +import Control.Concurrent.STM +import Control.Eff +import Control.Eff.Lift +import Control.Eff.Concurrent.Process +import Control.Eff.Concurrent.Process.ForkIOScheduler + as Scheduler +import Control.Monad ( void + , forever + ) +import Test.Tasty +import Test.Tasty.HUnit +import Data.Dynamic +import Common test_IOExceptionsIsolated :: TestTree -test_IOExceptionsIsolated = - setTravisTestOptions - $ testGroup "one process throws an IO exception, the other continues unimpaired" - [ testCase ("process 2 exits with: "++ howToExit - ++ " - while process 1 is busy with: " ++ busyWith) - $ do aVar <- newEmptyTMVarIO - lc <- testLogC - Scheduler.defaultMainWithLogChannel lc - $ do p1 <- spawn $ forever busyEffect - lift (threadDelay 1000) - void $ spawn $ do lift (threadDelay 1000) - doExit - lift (threadDelay 100000) - wasStillRunningP1 <- sendShutdown forkIoScheduler p1 - lift (atomically (putTMVar aVar wasStillRunningP1)) - - wasStillRunningP1 <- atomically (takeTMVar aVar) - assertBool "the other process was still running" wasStillRunningP1 +test_IOExceptionsIsolated = setTravisTestOptions $ testGroup + "one process throws an IO exception, the other continues unimpaired" + [ testCase + ( "process 2 exits with: " + ++ howToExit + ++ " - while process 1 is busy with: " + ++ busyWith + ) + $ do + aVar <- newEmptyTMVarIO + lc <- testLogC + Scheduler.defaultMainWithLogChannel lc $ do + p1 <- spawn $ forever busyEffect + lift (threadDelay 1000) + void $ spawn $ do + lift (threadDelay 1000) + doExit + lift (threadDelay 100000) + wasStillRunningP1 <- sendShutdownChecked forkIoScheduler p1 + lift (atomically (putTMVar aVar wasStillRunningP1)) - | (busyWith, busyEffect) <- - [ ("receiving", void (send (ReceiveMessage @SchedulerIO))) - , ("sending", void (send (SendMessage @SchedulerIO 44444 (toDyn "test message")))) - , ("sending shutdown", void (send (SendShutdown @SchedulerIO 44444))) - , ("selfpid-ing", void (send (SelfPid @SchedulerIO))) - , ("spawn-ing", void (send (Spawn @SchedulerIO (void (send (ReceiveMessage @SchedulerIO)))))) - ] - , (howToExit, doExit) <- - [ ("throw async exception", void (lift (throw UserInterrupt))) - , ("division by zero", void (return ((123::Int) `div` 0) >>= lift . putStrLn . show )) - , ("call 'fail'", void (fail "test fail")) - , ("call 'error'", void (error "test error")) - ] - ] + wasStillRunningP1 <- atomically (takeTMVar aVar) + assertBool "the other process was still running" wasStillRunningP1 + | (busyWith , busyEffect) <- + [ ("receiving", void (send (ReceiveMessage @SchedulerIO))) + , ( "sending" + , void (send (SendMessage @SchedulerIO 44444 (toDyn "test message"))) + ) + , ("sending shutdown", void (send (SendShutdown @SchedulerIO 44444))) + , ("selfpid-ing" , void (send (SelfPid @SchedulerIO))) + , ( "spawn-ing" + , void + (send (Spawn @SchedulerIO (void (send (ReceiveMessage @SchedulerIO))))) + ) + ] + , (howToExit, doExit ) <- + [ ("throw async exception", void (lift (throw UserInterrupt))) + , ( "division by zero" + , void (return ((123 :: Int) `div` 0) >>= lift . putStrLn . show) + ) + , ("call 'fail'" , void (fail "test fail")) + , ("call 'error'", void (error "test error")) + ] + ] test_mainProcessSpawnsAChildAndReturns :: TestTree -test_mainProcessSpawnsAChildAndReturns = - setTravisTestOptions - (testCase "spawn a child and return" - (do lc <- testLogC - Scheduler.defaultMainWithLogChannel lc - (void (spawn (void (receiveMessage forkIoScheduler)))))) +test_mainProcessSpawnsAChildAndReturns = setTravisTestOptions + (testCase + "spawn a child and return" + (do + lc <- testLogC + Scheduler.defaultMainWithLogChannel + lc + (void (spawn (void (receiveMessage forkIoScheduler)))) + ) + ) test_mainProcessSpawnsAChildAndExitsNormally :: TestTree -test_mainProcessSpawnsAChildAndExitsNormally = - setTravisTestOptions - (testCase "spawn a child and exit normally" - (do lc <- testLogC - Scheduler.defaultMainWithLogChannel lc - (do void (spawn (void (receiveMessage forkIoScheduler))) - void (exitNormally forkIoScheduler) - fail "This should not happen!!" - ))) +test_mainProcessSpawnsAChildAndExitsNormally = setTravisTestOptions + (testCase + "spawn a child and exit normally" + (do + lc <- testLogC + Scheduler.defaultMainWithLogChannel + lc + (do + void (spawn (void (receiveMessage forkIoScheduler))) + void (exitNormally forkIoScheduler) + fail "This should not happen!!" + ) + ) + ) test_mainProcessSpawnsAChildInABusySendLoopAndExitsNormally :: TestTree test_mainProcessSpawnsAChildInABusySendLoopAndExitsNormally = setTravisTestOptions - (testCase "spawn a child with a busy send loop and exit normally" - (do lc <- testLogC - Scheduler.defaultMainWithLogChannel lc - (do void (spawn (forever (void (sendMessage forkIoScheduler 1000 (toDyn "test"))))) - void (exitNormally forkIoScheduler) - fail "This should not happen!!" - ))) + (testCase + "spawn a child with a busy send loop and exit normally" + (do + lc <- testLogC + Scheduler.defaultMainWithLogChannel + lc + (do + void + (spawn + (forever + (void (sendMessage forkIoScheduler 1000 (toDyn "test"))) + ) + ) + void (exitNormally forkIoScheduler) + fail "This should not happen!!" + ) + ) + ) test_mainProcessSpawnsAChildBothReturn :: TestTree -test_mainProcessSpawnsAChildBothReturn = - setTravisTestOptions - (testCase "spawn a child and let it return and return" - (do lc <- testLogC - Scheduler.defaultMainWithLogChannel lc - (do child <- spawn (void (receiveMessageAs @String forkIoScheduler)) - True <- sendMessage forkIoScheduler child (toDyn "test") - return () - ))) +test_mainProcessSpawnsAChildBothReturn = setTravisTestOptions + (testCase + "spawn a child and let it return and return" + (do + lc <- testLogC + Scheduler.defaultMainWithLogChannel + lc + (do + child <- spawn (void (receiveMessageAs @String forkIoScheduler)) + True <- sendMessageChecked forkIoScheduler child (toDyn "test") + return () + ) + ) + ) test_mainProcessSpawnsAChildBothExitNormally :: TestTree -test_mainProcessSpawnsAChildBothExitNormally = - setTravisTestOptions - (testCase "spawn a child and let it exit and exit" - (do lc <- testLogC - Scheduler.defaultMainWithLogChannel lc - (do child <- spawn - (do void (receiveMessageAs @String forkIoScheduler) - void (exitNormally forkIoScheduler) - error "This should not happen (child)!!" - ) - True <- sendMessage forkIoScheduler child (toDyn "test") - void (exitNormally forkIoScheduler) - error "This should not happen!!" - ))) +test_mainProcessSpawnsAChildBothExitNormally = setTravisTestOptions + (testCase + "spawn a child and let it exit and exit" + (do + lc <- testLogC + Scheduler.defaultMainWithLogChannel + lc + (do + child <- spawn + (do + void (receiveMessageAs @String forkIoScheduler) + void (exitNormally forkIoScheduler) + error "This should not happen (child)!!" + ) + True <- sendMessageChecked forkIoScheduler child (toDyn "test") + void (exitNormally forkIoScheduler) + error "This should not happen!!" + ) + ) + ) diff --git a/test/ProcessBehaviourTestCases.hs b/test/ProcessBehaviourTestCases.hs --- a/test/ProcessBehaviourTestCases.hs +++ b/test/ProcessBehaviourTestCases.hs @@ -1,393 +1,558 @@ -module ProcessBehaviourTestCases where +module ProcessBehaviourTestCases +where -import Data.List (sort) -import Data.Dynamic -import Data.Foldable (traverse_) -import Control.Exception -import Control.Concurrent -import Control.Concurrent.STM -import Control.Eff.Concurrent.Process -import qualified Control.Eff.Concurrent.Process.ForkIOScheduler as ForkIO -import qualified Control.Eff.Concurrent.Process.SingleThreadedScheduler as SingleThreaded -import Control.Eff -import Control.Eff.Log -import Control.Eff.Lift -import Control.Monad (void, replicateM, forever, when) -import Test.Tasty -import Test.Tasty.HUnit -import Common +import Data.List ( sort ) +import Data.Dynamic +import Data.Foldable ( traverse_ ) +import Control.Exception +import Control.Concurrent +import Control.Concurrent.STM +import Control.Eff.Concurrent.Process +import qualified Control.Eff.Concurrent.Process.ForkIOScheduler + as ForkIO +import qualified Control.Eff.Concurrent.Process.SingleThreadedScheduler + as SingleThreaded +import Control.Eff +import Control.Eff.Log +import Control.Eff.Lift +import Control.Monad ( void + , replicateM + , forever + , when + ) +import Test.Tasty +import Test.Tasty.HUnit +import Common test_forkIo :: TestTree -test_forkIo = - setTravisTestOptions +test_forkIo = setTravisTestOptions (withTestLogC ForkIO.schedule - (\factory -> - testGroup "ForkIOScheduler" - [allTests factory])) + (\factory -> testGroup "ForkIOScheduler" [allTests factory]) + ) test_singleThreaded :: TestTree -test_singleThreaded = - setTravisTestOptions - (withTestLogC (const . SingleThreaded.defaultMain) - (\factory -> - testGroup "SingleThreadedScheduler" - [allTests factory])) +test_singleThreaded = setTravisTestOptions + (withTestLogC + (const . SingleThreaded.defaultMain) + (\factory -> testGroup "SingleThreadedScheduler" [allTests factory]) + ) -allTests :: forall r . (Member (Logs String) r, SetMember Lift (Lift IO) r) - => IO (Eff (Process r ': r) () -> IO ()) - -> TestTree -allTests schedulerFactory = - localOption (timeoutSeconds 300) - (testGroup "Process" - [ errorTests schedulerFactory - , sendShutdownTests schedulerFactory - , concurrencyTests schedulerFactory - , exitTests schedulerFactory - ]) +allTests + :: forall r + . (Member (Logs String) r, SetMember Lift (Lift IO) r) + => IO (Eff (Process r ': r) () -> IO ()) + -> TestTree +allTests schedulerFactory = localOption + (timeoutSeconds 300) + (testGroup + "Process" + [ errorTests schedulerFactory + , sendShutdownTests schedulerFactory + , concurrencyTests schedulerFactory + , exitTests schedulerFactory + , pingPongTests schedulerFactory + ] + ) -errorTests :: forall r . (Member (Logs String) r, SetMember Lift (Lift IO) r) - => IO (Eff (Process r ': r) () -> IO ()) -> TestTree -errorTests schedulerFactory = - let px :: SchedulerProxy r +data Ping = Ping ProcessId +data Pong = Pong + deriving (Eq, Show) + +pingPongTests + :: forall r + . (Member (Logs String) r, SetMember Lift (Lift IO) r) + => IO (Eff (Process r ': r) () -> IO ()) + -> TestTree +pingPongTests schedulerFactory = testGroup + "Yield Tests" + [ testCase "ping pong a message between two processes, both don't yield" + $ applySchedulerFactory schedulerFactory + $ do + let pongProc = forever $ do + Ping pinger <- receiveMessageAs SP + sendMessageAs SP pinger Pong + pingProc ponger parent = do + me <- self SP + sendMessageAs SP ponger (Ping me) + Pong <- receiveMessageAs SP + sendMessageAs SP parent True + pongPid <- spawn pongProc + me <- self SP + spawn_ (pingProc pongPid me) + ok <- receiveMessageAs @Bool SP + lift (ok @? "ping pong failed") + , testCase "ping pong a message between two processes, with massive yielding" + $ applySchedulerFactory schedulerFactory + $ do + yieldProcess SP + let pongProc = forever $ do + yieldProcess SP + Ping pinger <- receiveMessageAs SP + yieldProcess SP + sendMessageAs SP pinger Pong + yieldProcess SP + pingProc ponger parent = do + yieldProcess SP + me <- self SP + yieldProcess SP + sendMessageAs SP ponger (Ping me) + yieldProcess SP + Pong <- receiveMessageAs SP + yieldProcess SP + sendMessageAs SP parent True + yieldProcess SP + yieldProcess SP + pongPid <- spawn pongProc + yieldProcess SP + me <- self SP + yieldProcess SP + spawn_ (pingProc pongPid me) + yieldProcess SP + ok <- receiveMessageAs @Bool SP + yieldProcess SP + lift (ok @? "ping pong failed") + yieldProcess SP + , testCase + "the first message is not delayed, not even in cooperative scheduling (because of yield)" + $ applySchedulerFactory schedulerFactory + $ do + pongVar <- lift newEmptyMVar + let pongProc = forever $ do + Pong <- receiveMessageAs SP + lift (putMVar pongVar Pong) + ponger <- spawn pongProc + sendMessageAs SP ponger Pong + let waitLoop = do + p <- lift (tryTakeMVar pongVar) + case p of + Nothing -> do + yieldProcess SP + waitLoop + Just r -> return r + p <- waitLoop + lift (p == Pong @? "ping pong failed") + ] + +errorTests + :: forall r + . (Member (Logs String) r, SetMember Lift (Lift IO) r) + => IO (Eff (Process r ': r) () -> IO ()) + -> TestTree +errorTests schedulerFactory + = let + px :: SchedulerProxy r px = SchedulerProxy - in - testGroup "causing and handling errors" - [ testGroup "raiseError" - [ - testCase "unhandled raiseError" - $ applySchedulerFactory schedulerFactory - $ do void $ raiseError px "test error" + in + testGroup + "causing and handling errors" + [ testGroup + "raiseError" + [ testCase "unhandled raiseError" + $ applySchedulerFactory schedulerFactory + $ do + void $ raiseError px "test error" error "This should not happen" - , testCase "catch raiseError" - $ scheduleAndAssert schedulerFactory - $ \assertEff -> - catchRaisedError - px - (assertEff "error must be caught" . (== "test error 2")) - (void (raiseError px "test error 2")) - , testCase "catch raiseError from a long sub block" - $ scheduleAndAssert schedulerFactory - $ \assertEff -> - catchRaisedError - px - (assertEff "error must be caught" . (== "test error 3")) - (do void (replicateM 100000 (void (self px))) - void (raiseError px "test error 3")) - ] - , testGroup "exitWithError" - [ testCase "unhandled exitWithError" - $ applySchedulerFactory schedulerFactory - $ do void $ exitWithError px "test error" + , testCase "catch raiseError" + $ scheduleAndAssert schedulerFactory + $ \assertEff -> catchRaisedError + px + (assertEff "error must be caught" . (== "test error 2")) + (void (raiseError px "test error 2")) + , testCase "catch raiseError from a long sub block" + $ scheduleAndAssert schedulerFactory + $ \assertEff -> catchRaisedError + px + (assertEff "error must be caught" . (== "test error 3")) + (do + void (replicateM 100000 (void (self px))) + void (raiseError px "test error 3") + ) + ] + , testGroup + "exitWithError" + [ testCase "unhandled exitWithError" + $ applySchedulerFactory schedulerFactory + $ do + void $ exitWithError px "test error" error "This should not happen" - , testCase "cannot catch exitWithError" - $ applySchedulerFactory schedulerFactory - $ do void $ ignoreProcessError px $ exitWithError px "test error 4" + , testCase "cannot catch exitWithError" + $ applySchedulerFactory schedulerFactory + $ do + void $ ignoreProcessError px $ exitWithError px "test error 4" error "This should not happen" - , testCase "multi process exitWithError" - $ scheduleAndAssert schedulerFactory - $ \assertEff -> - do me <- self px + , testCase "multi process exitWithError" + $ scheduleAndAssert schedulerFactory + $ \assertEff -> do + me <- self px let n = 15 traverse_ - (\(i :: Int) -> - spawn $ - if i `rem` 5 == 0 then do - void $ sendMessage px me (toDyn i) - void (exitWithError px (show i ++ " died")) - assertEff "this should not be reached" False - else - forever - (void (sendMessage px 888 (toDyn "test message to 888")))) + (\(i :: Int) -> spawn $ if i `rem` 5 == 0 + then do + void $ sendMessage px me (toDyn i) + void (exitWithError px (show i ++ " died")) + assertEff "this should not be reached" False + else + forever + (void (sendMessage px 888 (toDyn "test message to 888")) + ) + ) [0, 5 .. n] oks <- replicateM - (length [0,5 .. n]) - (do j <- receiveMessageAs px - return j) - assertEff "" (sort oks == [0,5 .. n]) - - ] - ] + (length [0, 5 .. n]) + (do + j <- receiveMessageAs px + return j + ) + assertEff "" (sort oks == [0, 5 .. n]) + ] + ] -concurrencyTests :: forall r . (Member (Logs String) r, SetMember Lift (Lift IO) r) - => IO (Eff (Process r ': r) () -> IO ()) -> TestTree -concurrencyTests schedulerFactory = - let px :: SchedulerProxy r +concurrencyTests + :: forall r + . (Member (Logs String) r, SetMember Lift (Lift IO) r) + => IO (Eff (Process r ': r) () -> IO ()) + -> TestTree +concurrencyTests schedulerFactory + = let + px :: SchedulerProxy r px = SchedulerProxy - n = 100 - in - testGroup "concurrency tests" - [ testCase "when main process exits the scheduler kills/cleans and returns" - $ applySchedulerFactory schedulerFactory - $ do me <- self px - traverse_ - (const - (spawn - (do m <- receiveMessage px - void (sendMessage px me m)))) - [1..n] - lift (threadDelay 1000) - , - testCase "new processes are executed before the parent process" - $ scheduleAndAssert schedulerFactory - $ \assertEff -> - do -- start massive amount of children that exit as soon as they are - -- executed, this will only work smoothly when scheduler schedules - -- the new child before the parent - traverse_ (const (spawn (exitNormally px))) [1..n] - assertEff "" True - , - testCase "two concurrent processes" - $ scheduleAndAssert schedulerFactory - $ \assertEff -> - do me <- self px - child1 <- - spawn (do m <- receiveMessage px - void (sendMessage px me m)) - child2 <- - spawn (forever (void (sendMessage px 888 (toDyn "")))) - True <- sendMessage px child1 (toDyn "test") - i <- receiveMessageAs px - True <- sendShutdown px child2 - assertEff "" (i == "test") - , - testCase "most processes send forever" - $ scheduleAndAssert schedulerFactory - $ \assertEff -> - do me <- self px - traverse_ - (\(i :: Int) -> - spawn - $ do when (i `rem` 5 == 0) $ - void $ sendMessage px me (toDyn i) - forever $ - void (sendMessage px 888 (toDyn "test message to 888")) - ) [0 .. n] - oks <- replicateM - (length [0,5 .. n]) - (do j <- receiveMessageAs px - return j) - assertEff "" (sort oks == [0,5 .. n]) - , testCase "most processes self forever" - $ scheduleAndAssert schedulerFactory - $ \assertEff -> - do me <- self px - traverse_ - (\(i :: Int) -> - spawn - $ do when (i `rem` 5 == 0) $ - void $ sendMessage px me (toDyn i) - forever $ void (self px) - ) [0 .. n] - oks <- replicateM - (length [0,5 .. n]) - (do j <- receiveMessageAs px - return j) - assertEff "" (sort oks == [0,5 .. n]) - , testCase "most processes sendShutdown forever" - $ scheduleAndAssert schedulerFactory - $ \assertEff -> - do me <- self px - traverse_ - (\(i :: Int) -> - spawn - $ do when (i `rem` 5 == 0) $ - void $ sendMessage px me (toDyn i) - forever $ void (sendShutdown px 999) - ) [0 .. n] - oks <- replicateM - (length [0,5 .. n]) - (do j <- receiveMessageAs px - return j) - assertEff "" (sort oks == [0,5 .. n]) - , testCase "most processes spawn forever" - $ scheduleAndAssert schedulerFactory - $ \assertEff -> - do me <- self px - traverse_ - (\(i :: Int) -> - spawn - $ do when (i `rem` 5 == 0) $ - void $ sendMessage px me (toDyn i) - parent <- self px - forever $ - void (spawn (void (sendMessage px parent (toDyn "test msg from child")))) - ) [0 .. n] - oks <- replicateM - (length [0,5 .. n]) - (do j <- receiveMessageAs px - return j) - assertEff "" (sort oks == [0,5 .. n]) - , testCase "most processes receive forever" - $ scheduleAndAssert schedulerFactory - $ \assertEff -> - do me <- self px - traverse_ - (\(i :: Int) -> - spawn - $ do when (i `rem` 5 == 0) $ - void $ sendMessage px me (toDyn i) - forever $ - void (receiveMessage px) - ) [0 .. n] - oks <- replicateM - (length [0,5 .. n]) - (do j <- receiveMessageAs px - return j) - assertEff "" (sort oks == [0,5 .. n]) - ] + n = 100 + in + testGroup + "concurrency tests" + [ testCase + "when main process exits the scheduler kills/cleans and returns" + $ applySchedulerFactory schedulerFactory + $ do + me <- self px + traverse_ + (const + (spawn + (do + m <- receiveMessage px + void (sendMessage px me m) + ) + ) + ) + [1 .. n] + lift (threadDelay 1000) + , testCase "new processes are executed before the parent process" + $ scheduleAndAssert schedulerFactory + $ \assertEff -> do -- start massive amount of children that exit as soon as they are + -- executed, this will only work smoothly when scheduler schedules + -- the new child before the parent + traverse_ (const (spawn (exitNormally px))) [1 .. n] + assertEff "" True + , testCase "two concurrent processes" + $ scheduleAndAssert schedulerFactory + $ \assertEff -> do + me <- self px + child1 <- spawn + (do + m <- receiveMessage px + void (sendMessage px me m) + ) + child2 <- spawn (forever (void (sendMessage px 888 (toDyn "")))) + True <- sendMessageChecked px child1 (toDyn "test") + i <- receiveMessageAs px + True <- sendShutdownChecked px child2 + assertEff "" (i == "test") + , testCase "most processes send forever" + $ scheduleAndAssert schedulerFactory + $ \assertEff -> do + me <- self px + traverse_ + (\(i :: Int) -> spawn $ do + when (i `rem` 5 == 0) $ void $ sendMessage px me (toDyn i) + forever + $ void (sendMessage px 888 (toDyn "test message to 888")) + ) + [0 .. n] + oks <- replicateM + (length [0, 5 .. n]) + (do + j <- receiveMessageAs px + return j + ) + assertEff "" (sort oks == [0, 5 .. n]) + , testCase "most processes self forever" + $ scheduleAndAssert schedulerFactory + $ \assertEff -> do + me <- self px + traverse_ + (\(i :: Int) -> spawn $ do + when (i `rem` 5 == 0) $ void $ sendMessage px me (toDyn i) + forever $ void (self px) + ) + [0 .. n] + oks <- replicateM + (length [0, 5 .. n]) + (do + j <- receiveMessageAs px + return j + ) + assertEff "" (sort oks == [0, 5 .. n]) + , testCase "most processes sendShutdown forever" + $ scheduleAndAssert schedulerFactory + $ \assertEff -> do + me <- self px + traverse_ + (\(i :: Int) -> spawn $ do + when (i `rem` 5 == 0) $ void $ sendMessage px me (toDyn i) + forever $ void (sendShutdown px 999) + ) + [0 .. n] + oks <- replicateM + (length [0, 5 .. n]) + (do + j <- receiveMessageAs px + return j + ) + assertEff "" (sort oks == [0, 5 .. n]) + , testCase "most processes spawn forever" + $ scheduleAndAssert schedulerFactory + $ \assertEff -> do + me <- self px + traverse_ + (\(i :: Int) -> spawn $ do + when (i `rem` 5 == 0) $ void $ sendMessage px me (toDyn i) + parent <- self px + forever + $ void + (spawn + (void + (sendMessage px parent (toDyn "test msg from child") + ) + ) + ) + ) + [0 .. n] + oks <- replicateM + (length [0, 5 .. n]) + (do + j <- receiveMessageAs px + return j + ) + assertEff "" (sort oks == [0, 5 .. n]) + , testCase "most processes receive forever" + $ scheduleAndAssert schedulerFactory + $ \assertEff -> do + me <- self px + traverse_ + (\(i :: Int) -> spawn $ do + when (i `rem` 5 == 0) $ void $ sendMessage px me (toDyn i) + forever $ void (receiveMessage px) + ) + [0 .. n] + oks <- replicateM + (length [0, 5 .. n]) + (do + j <- receiveMessageAs px + return j + ) + assertEff "" (sort oks == [0, 5 .. n]) + ] -exitTests :: forall r . (Member (Logs String) r, SetMember Lift (Lift IO) r) - => IO (Eff (Process r ': r) () -> IO ()) -> TestTree +exitTests + :: forall r + . (Member (Logs String) r, SetMember Lift (Lift IO) r) + => IO (Eff (Process r ': r) () -> IO ()) + -> TestTree exitTests schedulerFactory = - let px :: SchedulerProxy r - px = SchedulerProxy + let + px :: SchedulerProxy r + px = SchedulerProxy in - testGroup "process exit tests" $ - [ testGroup "async exceptions" - [ testCase ("a process dies immediately if a " - ++ show e ++ " is thrown, while " ++ busyWith ) - $ do tidVar <- newEmptyTMVarIO - schedulerDoneVar <- newEmptyTMVarIO - void $ forkIO $ - do void $ try @SomeException $ void $ applySchedulerFactory schedulerFactory - $ do tid <- lift $ myThreadId + testGroup "process exit tests" + $ [ testGroup + "async exceptions" + [ testCase + ( "a process dies immediately if a " + ++ show e + ++ " is thrown, while " + ++ busyWith + ) + $ do + tidVar <- newEmptyTMVarIO + schedulerDoneVar <- newEmptyTMVarIO + void $ forkIO $ do + void + $ try @SomeException + $ void + $ applySchedulerFactory schedulerFactory + $ do + tid <- lift $ myThreadId lift $ atomically $ putTMVar tidVar tid forever busyEffect - atomically $ putTMVar schedulerDoneVar () - tid <- atomically $ takeTMVar tidVar - threadDelay 1000 - throwTo tid e - void $ atomically $ takeTMVar schedulerDoneVar - | e <- [ ThreadKilled, UserInterrupt, HeapOverflow, StackOverflow ] - , (busyWith, busyEffect) <- - [ ("receiving", void (send (ReceiveMessage @r))) - , ("sending", void (send (SendMessage @r 44444 (toDyn "test message")))) - , ("sending shutdown", void (send (SendShutdown @r 44444))) - , ("selfpid-ing", void (send (SelfPid @r))) - , ("spawn-ing", void (send (Spawn @r (void (send (ReceiveMessage @r)))))) - , ("sleeping", lift (threadDelay 100000)) - ] - ] - , testGroup "main thread exit not blocked by" - [ testCase ("a child process, busy with "++ busyWith) - $ applySchedulerFactory schedulerFactory - $ do void $ spawn $ forever busyEffect - lift (threadDelay 10000) - | (busyWith, busyEffect) <- - [ ("receiving", void (send (ReceiveMessage @r))) - , ("sending", void (send (SendMessage @r 44444 (toDyn "test message")))) - , ("sending shutdown", void (send (SendShutdown @r 44444))) - , ("selfpid-ing", void (send (SelfPid @r))) - , ("spawn-ing", void (send (Spawn @r (void (send (ReceiveMessage @r)))))) + atomically $ putTMVar schedulerDoneVar () + tid <- atomically $ takeTMVar tidVar + threadDelay 1000 + throwTo tid e + void $ atomically $ takeTMVar schedulerDoneVar + | e <- [ThreadKilled, UserInterrupt, HeapOverflow, StackOverflow] + , (busyWith, busyEffect) <- + [ ("receiving", void (send (ReceiveMessage @r))) + , ( "sending" + , void (send (SendMessage @r 44444 (toDyn "test message"))) + ) + , ("sending shutdown", void (send (SendShutdown @r 44444))) + , ("selfpid-ing" , void (send (SelfPid @r))) + , ( "spawn-ing" + , void (send (Spawn @r (void (send (ReceiveMessage @r))))) + ) + , ("sleeping", lift (threadDelay 100000)) + ] + ] + , testGroup + "main thread exit not blocked by" + [ testCase ("a child process, busy with " ++ busyWith) + $ applySchedulerFactory schedulerFactory + $ do + void $ spawn $ forever busyEffect + lift (threadDelay 10000) + | (busyWith, busyEffect) <- + [ ("receiving", void (send (ReceiveMessage @r))) + , ( "sending" + , void (send (SendMessage @r 44444 (toDyn "test message"))) + ) + , ("sending shutdown", void (send (SendShutdown @r 44444))) + , ("selfpid-ing" , void (send (SelfPid @r))) + , ( "spawn-ing" + , void (send (Spawn @r (void (send (ReceiveMessage @r))))) + ) + ] + ] + , testGroup + "one process exits, the other continues unimpaired" + [ testCase + ( "process 2 exits with: " + ++ howToExit + ++ " - while process 1 is busy with: " + ++ busyWith + ) + $ scheduleAndAssert schedulerFactory + $ \assertEff -> do + p1 <- spawn $ forever busyEffect + lift (threadDelay 1000) + void $ spawn $ do + lift (threadDelay 1000) + doExit + lift (threadDelay 100000) + wasStillRunningP1 <- sendShutdownChecked px p1 + assertEff "the other process was still running" + wasStillRunningP1 + | (busyWith , busyEffect) <- + [ ("receiving", void (send (ReceiveMessage @r))) + , ( "sending" + , void (send (SendMessage @r 44444 (toDyn "test message"))) + ) + , ("sending shutdown", void (send (SendShutdown @r 44444))) + , ("selfpid-ing" , void (send (SelfPid @r))) + , ( "spawn-ing" + , void (send (Spawn @r (void (send (ReceiveMessage @r))))) + ) + ] + , (howToExit, doExit ) <- + [ ("normally" , void (exitNormally px)) + , ("simply returning", return ()) + , ("raiseError" , void (raiseError px "test error raised")) + , ("exitWithError" , void (exitWithError px "test error exit")) + , ( "sendShutdown to self" + , do + me <- self px + void (sendShutdown px me) + ) + ] + ] ] - ] - , testGroup "one process exits, the other continues unimpaired" - [ testCase ("process 2 exits with: "++ howToExit - ++ " - while process 1 is busy with: " ++ busyWith) - $ scheduleAndAssert schedulerFactory - $ \assertEff -> - do p1 <- spawn $ forever busyEffect - lift (threadDelay 1000) - tlog "Yeah" - void $ spawn $ do lift (threadDelay 1000) - doExit - lift (threadDelay 100000) - wasStillRunningP1 <- sendShutdown px p1 - assertEff "the other process was still running" wasStillRunningP1 - | (busyWith, busyEffect) <- - [ ("receiving", void (send (ReceiveMessage @r))) - , ("sending", void (send (SendMessage @r 44444 (toDyn "test message")))) - , ("sending shutdown", void (send (SendShutdown @r 44444))) - , ("selfpid-ing", void (send (SelfPid @r))) - , ("spawn-ing", void (send (Spawn @r (void (send (ReceiveMessage @r)))))) - ] - , (howToExit, doExit) <- - [ ("normally", void (exitNormally px)) - , ("simply returning", return ()) - , ("raiseError", void (raiseError px "test error raised")) - , ("exitWithError", void (exitWithError px "test error exit")) - , ("sendShutdown to self", do me <- self px - void (sendShutdown px me)) - ] - ] - ] - - -sendShutdownTests :: forall r . (Member (Logs String) r, SetMember Lift (Lift IO) r) - => IO (Eff (Process r ': r) () -> IO ()) -> TestTree +sendShutdownTests + :: forall r + . (Member (Logs String) r, SetMember Lift (Lift IO) r) + => IO (Eff (Process r ': r) () -> IO ()) + -> TestTree sendShutdownTests schedulerFactory = let px :: SchedulerProxy r px = SchedulerProxy - in - testGroup "sendShutdown" - [ testCase "... self" - $ applySchedulerFactory schedulerFactory - $ do me <- self px - void $ sendShutdown px me - raiseError px "sendShutdown must not return" - , testCase "... self low-level" - $ scheduleAndAssert schedulerFactory - $ \assertEff -> - do me <- self px - r <- send (SendShutdown @r me) - assertEff - "ShutdownRequested must be returned" - (case r of - ShutdownRequested -> True - _ -> False) - - , testGroup "... other process" - [ testCase "while it is sending" + in testGroup + "sendShutdown" + [ testCase "... self" $ applySchedulerFactory schedulerFactory $ do + me <- self px + void $ sendShutdown px me + raiseError px "sendShutdown must not return" + , testCase "... self low-level" $ scheduleAndAssert schedulerFactory - $ \assertEff -> - do me <- self px - other <- - spawn (do untilShutdown (SendMessage @r 666 (toDyn "test")) - void (sendMessage px me (toDyn "OK"))) + $ \assertEff -> do + me <- self px + r <- send (SendShutdown @r me) + assertEff + "ShutdownRequested must be returned" + (case r of + ShutdownRequested -> True + _ -> False + ) + , testGroup + "... other process" + [ testCase "while it is sending" + $ scheduleAndAssert schedulerFactory + $ \assertEff -> do + me <- self px + other <- spawn + (do + untilShutdown (SendMessage @r 666 (toDyn "test")) + void (sendMessage px me (toDyn "OK")) + ) void (sendShutdown px other) a <- receiveMessageAs px assertEff "" (a == "OK") - , testCase "while it is receiving" - $ scheduleAndAssert schedulerFactory - $ \assertEff -> - do me <- self px - other <- - spawn (do untilShutdown (ReceiveMessage @r) - void (sendMessage px me (toDyn "OK"))) + , testCase "while it is receiving" + $ scheduleAndAssert schedulerFactory + $ \assertEff -> do + me <- self px + other <- spawn + (do + untilShutdown (ReceiveMessage @r) + void (sendMessage px me (toDyn "OK")) + ) void (sendShutdown px other) a <- receiveMessageAs px assertEff "" (a == "OK") - , testCase "while it is self'ing" - $ scheduleAndAssert schedulerFactory - $ \assertEff -> - do me <- self px - other <- - spawn (do untilShutdown (SelfPid @r) - void (sendMessage px me (toDyn "OK"))) + , testCase "while it is self'ing" + $ scheduleAndAssert schedulerFactory + $ \assertEff -> do + me <- self px + other <- spawn + (do + untilShutdown (SelfPid @r) + void (sendMessage px me (toDyn "OK")) + ) void (sendShutdown px other) a <- receiveMessageAs px assertEff "" (a == "OK") - , testCase "while it is spawning" - $ scheduleAndAssert schedulerFactory - $ \assertEff -> - do me <- self px - other <- - spawn (do untilShutdown (Spawn @r (return ())) - void (sendMessage px me (toDyn "OK"))) + , testCase "while it is spawning" + $ scheduleAndAssert schedulerFactory + $ \assertEff -> do + me <- self px + other <- spawn + (do + untilShutdown (Spawn @r (return ())) + void (sendMessage px me (toDyn "OK")) + ) void (sendShutdown px other) a <- receiveMessageAs px assertEff "" (a == "OK") - , testCase "while it is sending shutdown messages" - $ scheduleAndAssert schedulerFactory - $ \assertEff -> - do me <- self px - other <- - spawn (do untilShutdown (SendShutdown @r 777) - void (sendMessage px me (toDyn "OK"))) + , testCase "while it is sending shutdown messages" + $ scheduleAndAssert schedulerFactory + $ \assertEff -> do + me <- self px + other <- spawn + (do + untilShutdown (SendShutdown @r 777) + void (sendMessage px me (toDyn "OK")) + ) void (sendShutdown px other) a <- receiveMessageAs px assertEff "" (a == "OK") - ] - ] + ] + ] diff --git a/test/SingleThreadedScheduler.hs b/test/SingleThreadedScheduler.hs --- a/test/SingleThreadedScheduler.hs +++ b/test/SingleThreadedScheduler.hs @@ -1,37 +1,48 @@ -module SingleThreadedScheduler where - -import Control.Eff.Concurrent.Process -import Control.Eff.Concurrent.Process.SingleThreadedScheduler as Scheduler -import Control.Monad (void) -import Test.Tasty -import Test.Tasty.HUnit -import Data.Dynamic -import Common +module SingleThreadedScheduler +where -timeoutSeconds :: Integer -> Timeout -timeoutSeconds seconds = Timeout (seconds * 1000000) (show seconds ++ "s") +import Control.Eff.Concurrent.Process +import Control.Eff.Concurrent.Process.SingleThreadedScheduler + as Scheduler +import Control.Monad ( void + ) +import Test.Tasty +import Test.Tasty.HUnit +import Data.Dynamic +import Common test_mainProcessSpawnsAChildAndExitsNormally :: TestTree -test_mainProcessSpawnsAChildAndExitsNormally = - setTravisTestOptions - (testCase "spawn a child and exit normally" - (Scheduler.defaultMain - (do void (spawn (void (receiveMessage singleThreadedIoScheduler))) - void (exitNormally singleThreadedIoScheduler) - fail "This should not happen!!" - ))) +test_mainProcessSpawnsAChildAndExitsNormally = setTravisTestOptions + (testCase + "spawn a child and exit normally" + (Scheduler.defaultMain + (do + void (spawn (void (receiveMessage singleThreadedIoScheduler))) + void (exitNormally singleThreadedIoScheduler) + fail "This should not happen!!" + ) + ) + ) + test_mainProcessSpawnsAChildBothExitNormally :: TestTree -test_mainProcessSpawnsAChildBothExitNormally = - setTravisTestOptions - (testCase "spawn a child and let it exit and exit" - (Scheduler.defaultMain - (do child <- spawn - (do void (receiveMessageAs @String singleThreadedIoScheduler) - void (exitNormally singleThreadedIoScheduler) - error "This should not happen (child)!!" - ) - True <- sendMessage singleThreadedIoScheduler child (toDyn "test") - void (exitNormally singleThreadedIoScheduler) - error "This should not happen!!" - ))) +test_mainProcessSpawnsAChildBothExitNormally = setTravisTestOptions + (testCase + "spawn a child and let it exit and exit" + (Scheduler.defaultMain + (do + child <- spawn + (do + void + (receiveMessageAs @String singleThreadedIoScheduler) + void (exitNormally singleThreadedIoScheduler) + error "This should not happen (child)!!" + ) + True <- sendMessageChecked singleThreadedIoScheduler + child + (toDyn "test") + void (exitNormally singleThreadedIoScheduler) + error "This should not happen!!" + ) + ) + )