extensible-effects-concurrent 0.7.3 → 0.8
raw patch · 13 files changed
+841/−571 lines, 13 filesnew-component:exe:extensible-effects-concurrent-example-1new-component:exe:extensible-effects-concurrent-example-2
Files
- ChangeLog.md +7/−0
- examples/example-1/Main.hs +129/−0
- examples/example-2/Main.hs +166/−0
- extensible-effects-concurrent.cabal +53/−4
- src/Control/Eff/Concurrent.hs +5/−9
- src/Control/Eff/Concurrent/Api/Observer.hs +14/−12
- src/Control/Eff/Concurrent/Api/Server.hs +175/−174
- src/Control/Eff/Concurrent/Examples.hs +0/−129
- src/Control/Eff/Concurrent/Examples2.hs +0/−167
- src/Control/Eff/Concurrent/Process.hs +93/−32
- src/Control/Eff/Concurrent/Process/ForkIOScheduler.hs +51/−15
- src/Control/Eff/Concurrent/Process/SingleThreadedScheduler.hs +41/−29
- test/ProcessBehaviourTestCases.hs +107/−0
ChangeLog.md view
@@ -1,5 +1,12 @@ # Changelog for extensible-effects-concurrent +## 0.8.0++- Add selective receive+- Complete `Api.Server` rewrite (simplification)+- Move examples to `./examples/` and add executables to the+ cabal file+ ## 0.7.3 - Add `withFrozenCallStack` to exposed functions
+ examples/example-1/Main.hs view
@@ -0,0 +1,129 @@+-- | A complete example for the library+module Main where++import GHC.Stack+import Control.Eff+import Control.Eff.Lift+import Control.Monad+import Data.Dynamic+import Control.Eff.Concurrent+import qualified Control.Exception as Exc++data TestApi+ deriving Typeable++data instance Api TestApi x where+ SayHello :: String -> Api TestApi ('Synchronous Bool)+ Shout :: String -> Api TestApi 'Asynchronous+ Terminate :: Api TestApi ('Synchronous ())+ TerminateError :: String -> Api TestApi ('Synchronous ())+ deriving (Typeable)++data MyException = MyException+ deriving Show++instance Exc.Exception MyException++deriving instance Show (Api TestApi x)++main :: IO ()+main = defaultMain (example forkIoScheduler)++mainProcessSpawnsAChildAndReturns+ :: (HasCallStack, SetMember Process (Process q) r)+ => SchedulerProxy q+ -> Eff r ()+mainProcessSpawnsAChildAndReturns px = void (spawn (void (receiveMessage px)))++example+ :: ( HasCallStack+ , SetMember Process (Process q) r+ , Member (Logs LogMessage) r+ , Member (Logs LogMessage) q+ , SetMember Lift (Lift IO) q+ , SetMember Lift (Lift IO) r+ )+ => SchedulerProxy q+ -> Eff r ()+example px = do+ me <- self px+ logInfo ("I am " ++ show me)+ server <- asServer @TestApi <$> spawn testServerLoop+ logInfo ("Started server " ++ show server)+ let go = do+ x <- lift getLine+ case x of+ ('k' : rest) -> do+ callRegistered px (TerminateError rest)+ go+ ('s' : _) -> do+ callRegistered px Terminate+ go+ ('c' : _) -> do+ castRegistered px (Shout x)+ go+ ('r' : rest) -> do+ void (replicateM (read rest) (castRegistered px (Shout x)))+ go+ ('q' : _) -> logInfo "Done."+ _ -> do+ res <- ignoreProcessError px (callRegistered px (SayHello x))+ logInfo ("Result: " ++ show res)+ go+ registerServer server go++testServerLoop+ :: forall r+ . (HasCallStack, Member (Logs LogMessage) r, SetMember Lift (Lift IO) r)+ => Eff (Process r ': r) ()+testServerLoop = serve px $ ApiHandler handleCast handleCall handleTerminate+ where+ px :: SchedulerProxy r+ px = SchedulerProxy+ handleCast :: Api TestApi 'Asynchronous -> Eff (Process r ': r) ()+ handleCast (Shout x) = do+ me <- self px+ logInfo (show me ++ " Shouting: " ++ x)+ handleCall+ :: Api TestApi ( 'Synchronous x)+ -> (x -> Eff (Process r ': r) ())+ -> Eff (Process r ': r) ()+ handleCall (SayHello "e1") _reply = do+ me <- self px+ logInfo (show me ++ " raising an error")+ raiseError px "No body loves me... :,("+ handleCall (SayHello "e2") _reply = do+ me <- self px+ logInfo (show me ++ " throwing a MyException ")+ lift (Exc.throw MyException)+ handleCall (SayHello "self") reply = do+ me <- self px+ logInfo (show me ++ " casting to self")+ cast px (asServer @TestApi me) (Shout "from me")+ void (reply False)+ handleCall (SayHello "die") reply = do+ me <- self px+ logInfo (show me ++ " throwing and catching ")+ catchRaisedError+ px+ (\er -> logInfo ("WOW: " ++ show er ++ " - No. This is wrong!"))+ (raiseError px "No body loves me... :,(")+ void (reply True)+ handleCall (SayHello x) reply = do+ me <- self px+ logInfo (show me ++ " Got Hello: " ++ x)+ void (reply (length x > 3))+ handleCall Terminate reply = do+ me <- self px+ logInfo (show me ++ " exiting")+ void (reply ())+ exitNormally px+ handleCall (TerminateError msg) reply = do+ me <- self px+ logInfo (show me ++ " exiting with error: " ++ msg)+ void (reply ())+ exitWithError px msg+ handleTerminate msg = do+ me <- self px+ logInfo (show me ++ " is exiting: " ++ show msg)+ maybe (exitNormally px) (exitWithError px) msg
+ examples/example-2/Main.hs view
@@ -0,0 +1,166 @@+-- | Another complete example for the library+module Main where++import Data.Dynamic+import Control.Eff+import qualified Control.Eff.Concurrent.Process.ForkIOScheduler+ as ForkIO+import qualified Control.Eff.Concurrent.Process.SingleThreadedScheduler+ as Pure+import Control.Eff.Concurrent+import Control.Eff.State.Strict+import Control.Monad++main :: IO ()+main = do+ ForkIO.defaultMain (counterExample SchedulerProxy)+ Pure.defaultMain (counterExample SchedulerProxy)++-- * First API++data Counter deriving Typeable++data instance Api Counter x where+ Inc :: Api Counter 'Asynchronous+ Cnt :: Api Counter ('Synchronous Integer)+ ObserveCounter :: SomeObserver Counter -> Api Counter 'Asynchronous+ UnobserveCounter :: SomeObserver Counter -> Api Counter 'Asynchronous++deriving instance Show (Api Counter x)++instance Observable Counter where+ data Observation Counter where+ CountChanged :: Integer -> Observation Counter+ deriving (Show, Typeable)+ registerObserverMessage = ObserveCounter+ forgetObserverMessage = UnobserveCounter++logCounterObservations+ :: (SetMember Process (Process q) r, Member (Logs LogMessage) q)+ => SchedulerProxy q+ -> Eff r (Server (CallbackObserver Counter))+logCounterObservations px = spawnCallbackObserver+ px+ (\fromSvr msg -> do+ me <- self px+ logInfo (show me ++ " observed on: " ++ show fromSvr ++ ": " ++ show msg)+ return True+ )++counterHandler+ :: forall r q+ . ( Member (State (Observers Counter)) r+ , Member (State Integer) r+ , Member (Logs LogMessage) r+ , SetMember Process (Process q) r+ )+ => SchedulerProxy q+ -> ApiHandler Counter r+counterHandler px = ApiHandler @Counter handleCast+ handleCall+ (defaultTermination px)+ where+ handleCast :: Api Counter 'Asynchronous -> Eff r ()+ handleCast (ObserveCounter o) = addObserver o+ handleCast (UnobserveCounter o) = removeObserver o+ handleCast Inc = do+ logInfo "Inc"+ modify (+ (1 :: Integer))+ currentCount <- get+ notifyObservers px (CountChanged currentCount)+ handleCall :: Api Counter ( 'Synchronous x) -> (x -> Eff r ()) -> Eff r ()+ handleCall Cnt reply = do+ c <- get+ logInfo ("Cnt is " ++ show c)+ reply c++-- * Second API++data PocketCalc deriving Typeable++data instance Api PocketCalc x where+ Add :: Integer -> Api PocketCalc ('Synchronous Integer)+ AAdd :: Integer -> Api PocketCalc 'Asynchronous++deriving instance Show (Api PocketCalc x)++pocketCalcHandler+ :: forall r q+ . ( Member (State Integer) r+ , Member (Logs LogMessage) r+ , SetMember Process (Process q) r+ )+ => SchedulerProxy q+ -> ApiHandler PocketCalc r+pocketCalcHandler px = ApiHandler @PocketCalc handleCast+ handleCall+ (defaultTermination px)+ where+ handleCast :: Api PocketCalc 'Asynchronous -> Eff r ()+ handleCast (AAdd x) = do+ logInfo ("AsyncAdd " ++ show x)+ modify (+ x)+ c <- get @Integer+ logInfo ("Accumulator is " ++ show c)+ handleCall :: Api PocketCalc ( 'Synchronous x) -> (x -> Eff r ()) -> Eff r ()+ handleCall (Add x) reply = do+ logInfo ("Add " ++ show x)+ modify (+ x)+ c <- get+ logInfo ("Accumulator is " ++ show c)+ reply c++serverLoop+ :: forall r q+ . (Member (Logs LogMessage) r, SetMember Process (Process q) r)+ => SchedulerProxy q+ -> Eff r ()+serverLoop px = evalState @Integer+ 0+ (manageObservers @Counter (serve px (counterHandler px, pocketCalcHandler px))+ )+++-- ** Counter client+counterExample+ :: ( Member (Logs LogMessage) r+ , Member (Logs LogMessage) q+ , SetMember Process (Process q) r+ )+ => SchedulerProxy q+ -> Eff r ()+counterExample px = do+ let cnt sv = do+ r <- call px sv Cnt+ logInfo (show sv ++ " " ++ show r)+ pid1 <- spawn (serverLoop px)+ pid2 <- spawn (serverLoop px)+ let cntServer1 = asServer @Counter pid1+ cntServer2 = asServer @Counter pid2+ calcServer1 = asServer @PocketCalc pid1+ calcServer2 = asServer @PocketCalc pid2+ cast px cntServer1 Inc+ cnt cntServer1+ cnt cntServer2+ co1 <- logCounterObservations px+ co2 <- logCounterObservations px+ registerObserver px co1 cntServer1+ registerObserver px co2 cntServer2+ cast px calcServer1 (AAdd 12313)+ cast px cntServer1 Inc+ cnt cntServer1+ cast px cntServer2 Inc+ cnt cntServer2+ registerObserver px co2 cntServer1+ registerObserver px co1 cntServer2+ cast px cntServer1 Inc+ void (call px calcServer2 (Add 878))+ cnt cntServer1+ cast px cntServer2 Inc+ cnt cntServer2+ forgetObserver px co2 cntServer1+ cast px cntServer1 Inc+ cnt cntServer1+ cast px cntServer2 Inc+ cnt cntServer2+ void $ sendShutdown px pid2
extensible-effects-concurrent.cabal view
@@ -1,5 +1,5 @@ name: extensible-effects-concurrent-version: 0.7.3+version: 0.8 description: Please see the README on GitHub at <https://github.com/sheyll/extensible-effects-concurrent#readme> synopsis: Message passing concurrency as extensible-effect homepage: https://github.com/sheyll/extensible-effects-concurrent#readme@@ -72,9 +72,7 @@ Control.Eff.Concurrent.Api.Observer.Queue other-modules: Control.Eff.Concurrent.Api.Internal,- Paths_extensible_effects_concurrent,- Control.Eff.Concurrent.Examples,- Control.Eff.Concurrent.Examples2+ Paths_extensible_effects_concurrent default-extensions: BangPatterns, ConstraintKinds,@@ -85,6 +83,7 @@ DeriveTraversable, FlexibleContexts, FlexibleInstances,+ FunctionalDependencies, GADTs, GeneralizedNewtypeDeriving, LambdaCase,@@ -102,7 +101,54 @@ default-language: Haskell2010 ghc-options: -Wall -fno-full-laziness +executable extensible-effects-concurrent-example-1+ main-is: Main.hs+ hs-source-dirs: examples/example-1+ build-depends:+ base >=4.9 && <5+ , extensible-effects-concurrent+ , extensible-effects >= 3+ ghc-options: -Wall -threaded -fno-full-laziness -rtsopts -with-rtsopts=-N+ default-language: Haskell2010+ default-extensions: BangPatterns+ , DataKinds+ , FlexibleContexts+ , FlexibleInstances+ , FunctionalDependencies+ , GADTs+ , GeneralizedNewtypeDeriving+ , RankNTypes+ , ScopedTypeVariables+ , StandaloneDeriving+ , TemplateHaskell+ , TypeApplications+ , TypeFamilies+ , TypeOperators +executable extensible-effects-concurrent-example-2+ main-is: Main.hs+ hs-source-dirs: examples/example-2+ build-depends:+ base >=4.9 && <5+ , extensible-effects-concurrent+ , extensible-effects >= 3+ ghc-options: -Wall -threaded -fno-full-laziness -rtsopts -with-rtsopts=-N+ default-language: Haskell2010+ default-extensions: BangPatterns+ , DataKinds+ , FlexibleContexts+ , FlexibleInstances+ , FunctionalDependencies+ , GADTs+ , GeneralizedNewtypeDeriving+ , RankNTypes+ , ScopedTypeVariables+ , StandaloneDeriving+ , TemplateHaskell+ , TypeApplications+ , TypeFamilies+ , TypeOperators+ test-suite extensible-effects-concurrent-test type: exitcode-stdio-1.0 main-is: Driver.hs@@ -134,10 +180,13 @@ default-extensions: BangPatterns , DataKinds , FlexibleContexts+ , FlexibleInstances+ , FunctionalDependencies , GADTs , GeneralizedNewtypeDeriving , RankNTypes , ScopedTypeVariables+ , StandaloneDeriving , TemplateHaskell , TypeApplications , TypeFamilies
src/Control/Eff/Concurrent.hs view
@@ -84,20 +84,16 @@ ) import Control.Eff.Concurrent.Api.Server ( serve+ , spawnServer , ApiHandler(..) , unhandledCallError , unhandledCastError , defaultTermination- , serveBoth- , serve3- , tryApiHandler- , UnhandledRequest()- , catchUnhandled- , ensureAllHandled- , requestFromDynamic- , exitUnhandled+ , Servable(..)+ , ServerCallback(..)+ , requestHandlerSelector+ , terminationHandler )- import Control.Eff.Concurrent.Api.Observer ( Observer(..) , Observable(..)
src/Control/Eff/Concurrent/Api/Observer.hs view
@@ -35,6 +35,8 @@ import Control.Eff.Concurrent.Api import Control.Eff.Concurrent.Api.Client import Control.Eff.Concurrent.Api.Server+import Control.Eff.Exception+import Control.Eff.Extend import Control.Eff.Log import Control.Eff.State.Strict import Control.Lens@@ -191,19 +193,19 @@ => SchedulerProxy q -> (Server o -> Observation o -> Eff (Process q ': q) Bool) -> Eff r (Server (CallbackObserver o))-spawnCallbackObserver px onObserve =- withFrozenCallStack $ asServer @(CallbackObserver o) <$> spawn @r @q- (do- let loopUntil = serve- px- (ApiHandler @(CallbackObserver o) (handleCast loopUntil)- (unhandledCallError px)- (defaultTermination px)- )- loopUntil- )+spawnCallbackObserver px onObserve = withFrozenCallStack $ spawnServer+ px+ (ApiHandler @(CallbackObserver o) handleCast+ (unhandledCallError px)+ (defaultTermination px)+ )+ (runError @ExitCallbackObserver >=> return . const ()) where- handleCast k (CbObserved fromSvr v) = onObserve fromSvr v >>= flip when k+ handleCast (CbObserved fromSvr v) =+ raise (onObserve fromSvr v) >>= flip when (throwError ExitCallbackObserver)++data ExitCallbackObserver = ExitCallbackObserver+ -- | Use 'spawnCallbackObserver' to create a universal logging observer, -- using the 'Show' instance of the 'Observation'.
src/Control/Eff/Concurrent/Api/Server.hs view
@@ -3,94 +3,220 @@ ( -- * Api Server serve+ , spawnServer+ -- * Api Callbacks , ApiHandler(..)- -- * ApiHandler default callbacks , unhandledCallError , unhandledCastError , defaultTermination- -- * Multi Api Server- , serveBoth- , serve3- , tryApiHandler- , UnhandledRequest()- , catchUnhandled- , ensureAllHandled- , requestFromDynamic- , exitUnhandled+ -- * Callback Composition+ , Servable(..)+ , ServerCallback(..)+ , requestHandlerSelector+ , terminationHandler ) where import Control.Eff-import qualified Control.Eff.Exception as Exc-import Control.Eff.Extend import Control.Eff.Concurrent.Api import Control.Eff.Concurrent.Api.Internal import Control.Eff.Concurrent.Process+import Control.Lens import Data.Proxy import Data.Typeable ( Typeable , typeRep ) import Data.Dynamic+import Control.Applicative+import Data.Kind import GHC.Stack --- | Receive and process incoming requests until the process exits, using an 'ApiHandler'.-serve- :: forall r q p- . (Typeable p, SetMember Process (Process q) r, HasCallStack)- => SchedulerProxy q- -> ApiHandler p r- -> Eff r ()-serve px handlers = receiveLoop px $ \case- 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, handling requests sent to a /server/ 'Process', all -- belonging to a specific 'Api' family instance.-data ApiHandler p r where+-- The values of this type can be 'serve'ed or combined via 'Servable' or+-- 'ServerCallback's.+data ApiHandler api eff where ApiHandler :: { -- | A cast will not return a result directly. This is used for async -- methods. _handleCast :: HasCallStack- => Api p 'Asynchronous -> Eff r ()+ => Api api 'Asynchronous -> Eff eff () -- | A call is a blocking operation, the caller is blocked until this -- handler calls the reply continuation. , _handleCall- :: forall x . HasCallStack- => Api p ('Synchronous x) -> (x -> Eff r ()) -> Eff r ()+ :: forall reply . HasCallStack+ => Api api ('Synchronous reply) -> (reply -> Eff eff ()) -> Eff eff () -- | This callback is called with @Nothing@ if the process exits -- peacefully, or @Just "error message..."@ if the process exits with an -- error. This function is responsible to exit the process if necessary. -- The default behavior is defined in 'defaultTermination'. , _handleTerminate :: HasCallStack- => Maybe String -> Eff r ()- } -> ApiHandler p r+ => Maybe String -> Eff eff ()+ } -> ApiHandler api eff ++-- | Building block for composition of 'ApiHandler'.+-- A wrapper for 'ApiHandler'. Use this to combine 'ApiHandler', allowing a+-- process to implement several 'Api' instances. The termination will be evenly+-- propagated.+-- Create this via e.g. 'Servable' instances+-- To serve multiple apis use '<>' to combine server callbacks, e.g.+--+-- @@@+-- let f = apiHandlerServerCallback px $ ApiHandler ...+-- g = apiHandlerServerCallback px $ ApiHandler ...+-- h = f <> g+-- in serve px h+-- @@@+--+data ServerCallback eff =+ ServerCallback { _requestHandlerSelector :: MessageSelector (Eff eff ())+ , _terminationHandler :: Maybe String -> Eff eff ()+ }++makeLenses ''ServerCallback++instance Semigroup (ServerCallback eff) where+ l <> r = l & requestHandlerSelector .~+ MessageSelector (\x ->+ runMessageSelector (view requestHandlerSelector l) x <|>+ runMessageSelector (view requestHandlerSelector r) x)+ & terminationHandler .~+ (\errorMessage ->+ do (l^.terminationHandler) errorMessage+ (r^.terminationHandler) errorMessage)++instance Monoid (ServerCallback eff) where+ mappend = (<>)+ mempty = ServerCallback+ { _requestHandlerSelector = MessageSelector (const Nothing)+ , _terminationHandler = const (return ())+ }++-- | Helper type class to allow composition of 'ApiHandler'.+class Servable a where+ -- | The effect of the callbacks+ type ServerEff a :: [Type -> Type]+ -- | The is used to let the spawn function return multiple 'Server' 'ProcessId's+ -- in a type safe way, e.g. for a tuple instance of this class+ -- @(Server a, Server b)@+ type ServerPids a+ -- | The is used to let the spawn function return multiple 'Server' 'ProcessId's+ -- in a type safe way.+ toServerPids :: proxy a -> ProcessId -> ServerPids a+ -- | Convert the value to a 'ServerCallback'+ toServerCallback :: (SetMember Process (Process effScheduler) (ServerEff a))+ => SchedulerProxy effScheduler -> a -> ServerCallback (ServerEff a)++instance Servable (ServerCallback eff) where+ type ServerEff (ServerCallback eff) = eff+ type ServerPids (ServerCallback eff) = ProcessId+ toServerCallback = const id+ toServerPids = const id++instance Typeable a => Servable (ApiHandler a eff) where+ type ServerEff (ApiHandler a eff) = eff+ type ServerPids (ApiHandler a eff) = Server a+ toServerCallback = apiHandlerServerCallback+ toServerPids _ = asServer++instance (ServerEff a ~ ServerEff b, Servable a, Servable b) => Servable (a, b) where+ type ServerPids (a, b) = (ServerPids a, ServerPids b)+ type ServerEff (a, b) = ServerEff a+ toServerCallback px (a, b) = toServerCallback px a <> toServerCallback px b+ toServerPids _ pid =+ ( toServerPids (Proxy :: Proxy a) pid+ , toServerPids (Proxy :: Proxy b) pid+ )++-- | Receive and process incoming requests until the process exits.+serve+ :: forall a effScheduler+ . ( Servable a+ , SetMember Process (Process effScheduler) (ServerEff a)+ , HasCallStack+ )+ => SchedulerProxy effScheduler+ -> a+ -> Eff (ServerEff a) ()+serve px a =+ let serverCb = toServerCallback px a+ in receiveLoopSuchThat px (serverCb ^. requestHandlerSelector) $ \case+ Left reason -> (serverCb ^. terminationHandler) reason+ Right handleIt -> handleIt++-- | Spawn a new process, that will receive and process incoming requests+-- until the process exits. Also handle all internal effects.+spawnServer+ :: forall a effScheduler eff+ . ( Servable a+ , SetMember Process (Process effScheduler) (ServerEff a)+ , SetMember Process (Process effScheduler) eff+ , HasCallStack+ )+ => SchedulerProxy effScheduler+ -> a+ -> ( Eff (ServerEff a) ()+ -> Eff (Process effScheduler ': effScheduler) ()+ )+ -> Eff eff (ServerPids a)+spawnServer px a handleEff = do+ pid <- spawn (handleEff (serve px a))+ return (toServerPids (Proxy @a) pid)++-- | Wrap an 'ApiHandler' into a composable 'ServerCallback' value.+apiHandlerServerCallback+ :: forall eff effScheduler api+ . ( HasCallStack+ , Typeable api+ , SetMember Process (Process effScheduler) eff+ )+ => SchedulerProxy effScheduler+ -> ApiHandler api eff+ -> ServerCallback eff+apiHandlerServerCallback px handlers = mempty+ { _requestHandlerSelector = selectHandlerMethod px handlers+ , _terminationHandler = _handleTerminate handlers+ }++-- | Try to parse an incoming message to an API request, and apply either+-- the 'handleCall' method or the 'handleCast' method to it.+selectHandlerMethod+ :: forall eff effScheduler api+ . ( HasCallStack+ , Typeable api+ , SetMember Process (Process effScheduler) eff+ )+ => SchedulerProxy effScheduler+ -> ApiHandler api eff+ -> MessageSelector (Eff eff ())+selectHandlerMethod px handlers =+ MessageSelector (fmap (applyHandlerMethod px handlers) . fromDynamic)+ -- | 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+-- callback to an incoming request.+applyHandlerMethod+ :: forall eff effScheduler api+ . ( Typeable api+ , SetMember Process (Process effScheduler) eff+ , HasCallStack+ )+ => SchedulerProxy effScheduler+ -> ApiHandler api eff+ -> Request api+ -> Eff eff ()+applyHandlerMethod _px handlers (Terminate e) = _handleTerminate handlers e+applyHandlerMethod _ handlers (Cast request) = _handleCast handlers request+applyHandlerMethod px handlers (Call fromPid request) = _handleCall handlers+ request+ sendReply where- sendReply :: Typeable x => x -> Eff r ()+ sendReply :: Typeable reply => reply -> Eff eff () sendReply reply =- sendMessage px fromPid (toDyn $! (Response (Proxy @p) $! reply))+ sendMessage px fromPid (toDyn $! (Response (Proxy @api) $! reply)) -- | A default handler to use in '_handleCall' in 'ApiHandler'. It will call -- 'raiseError' with a nice error message.@@ -137,128 +263,3 @@ -> Eff r () defaultTermination px = withFrozenCallStack $ maybe (exitNormally px) (exitWithError px)----- | '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 $ \case- Left Nothing -> applyApiHandler px h1 (Terminate Nothing)- Left (Just reason) -> applyApiHandler px h1 (Terminate (Just reason))- Right dyn -> ensureAllHandled- px- (tryApiHandler px h1 dyn `catchUnhandled` tryApiHandler px h2)---- | 'serve' three 'ApiHandler's at once. The first handler is used for--- termination handling.-serve3- :: forall r q p1 p2 p3- . ( Typeable p1- , Typeable p2- , Typeable p3- , SetMember Process (Process q) r- , HasCallStack- )- => SchedulerProxy q- -> ApiHandler p1 r- -> ApiHandler p2 r- -> ApiHandler p3 r- -> Eff r ()-serve3 px h1 h2 h3 = receiveLoop px $ \mReq -> case mReq of- Left Nothing -> applyApiHandler px h1 (Terminate Nothing)- Left (Just reason) -> applyApiHandler px h1 (Terminate (Just reason))- Right dyn -> ensureAllHandled- px- ( tryApiHandler px h1 dyn- `catchUnhandled` tryApiHandler px h2- `catchUnhandled` tryApiHandler px h3- )---- | The basic building block of the combination of 'ApiHandler's is this--- function, which can not only be passed to 'receiveLoop', but also throws an--- 'UnhandledRequest' exception if 'requestFromDynamic' failed, such that multiple--- invokation of this function for different 'ApiHandler's can be tried, by using 'catchUnhandled'.------ > tryApiHandler px handlers1 message--- > `catchUnhandled`--- > tryApiHandler px handlers2--- > `catchUnhandled`--- > tryApiHandler px handlers3----tryApiHandler- :: forall r q p- . (Typeable p, SetMember Process (Process q) r, HasCallStack)- => SchedulerProxy q- -> ApiHandler p r- -> Dynamic- -> Eff (Exc.Exc UnhandledRequest ': r) ()-tryApiHandler px handlers message = do- request <- 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 '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 '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)- => Eff r a- -> (Dynamic -> Eff r a)- -> Eff r a-catchUnhandled effect handler =- withFrozenCallStack $ effect `Exc.catchError` (handler . fromUnhandledRequest)---- | Catch 'UnhandledRequest's and terminate the process with an error, if necessary.-ensureAllHandled- :: forall r q- . (HasCallStack, SetMember Process (Process q) r)- => SchedulerProxy q- -> Eff (Exc.Exc UnhandledRequest ': r) ()- -> Eff r ()-ensureAllHandled px effect = withFrozenCallStack $ do- result <- Exc.runError effect- either (exitUnhandled px . fromUnhandledRequest) return result---- | Cast a 'Dynamic' value, and if that fails, throw an 'UnhandledRequest'--- error.-requestFromDynamic- :: forall r a- . (HasCallStack, Typeable a, Member (Exc.Exc UnhandledRequest) r)- => Dynamic- -> Eff r a-requestFromDynamic message = withFrozenCallStack $ 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 'requestFromDynamic') one should use this function to--- exit the process with a corresponding error.-exitUnhandled- :: forall r q- . (SetMember Process (Process q) r, HasCallStack)- => SchedulerProxy q- -> Dynamic- -> Eff r ()-exitUnhandled px message = withFrozenCallStack $ do- let reason = "unhandled message: " ++ show message- exitWithError px reason
− src/Control/Eff/Concurrent/Examples.hs
@@ -1,129 +0,0 @@--- | A complete example for the library-module Control.Eff.Concurrent.Examples where--import GHC.Stack-import Control.Eff-import Control.Eff.Lift-import Control.Monad-import Data.Dynamic-import Control.Eff.Concurrent-import qualified Control.Exception as Exc--data TestApi- deriving Typeable--data instance Api TestApi x where- SayHello :: String -> Api TestApi ('Synchronous Bool)- Shout :: String -> Api TestApi 'Asynchronous- Terminate :: Api TestApi ('Synchronous ())- TerminateError :: String -> Api TestApi ('Synchronous ())- deriving (Typeable)--data MyException = MyException- deriving Show--instance Exc.Exception MyException--deriving instance Show (Api TestApi x)--main :: IO ()-main = defaultMain (example forkIoScheduler)--mainProcessSpawnsAChildAndReturns- :: (HasCallStack, SetMember Process (Process q) r)- => SchedulerProxy q- -> Eff r ()-mainProcessSpawnsAChildAndReturns px = void (spawn (void (receiveMessage px)))--example- :: ( HasCallStack- , SetMember Process (Process q) r- , Member (Logs LogMessage) r- , Member (Logs LogMessage) q- , SetMember Lift (Lift IO) q- , SetMember Lift (Lift IO) r- )- => SchedulerProxy q- -> Eff r ()-example px = do- me <- self px- logInfo ("I am " ++ show me)- server <- asServer @TestApi <$> spawn testServerLoop- logInfo ("Started server " ++ show server)- let go = do- x <- lift getLine- case x of- ('k' : rest) -> do- callRegistered px (TerminateError rest)- go- ('s' : _) -> do- callRegistered px Terminate- go- ('c' : _) -> do- castRegistered px (Shout x)- go- ('r' : rest) -> do- void (replicateM (read rest) (castRegistered px (Shout x)))- go- ('q' : _) -> logInfo "Done."- _ -> do- res <- ignoreProcessError px (callRegistered px (SayHello x))- logInfo ("Result: " ++ show res)- go- registerServer server go--testServerLoop- :: forall r- . (HasCallStack, Member (Logs LogMessage) r, SetMember Lift (Lift IO) r)- => Eff (Process r ': r) ()-testServerLoop = serve px $ ApiHandler handleCast handleCall handleTerminate- where- px :: SchedulerProxy r- px = SchedulerProxy- handleCast :: Api TestApi 'Asynchronous -> Eff (Process r ': r) ()- handleCast (Shout x) = do- me <- self px- logInfo (show me ++ " Shouting: " ++ x)- handleCall- :: Api TestApi ( 'Synchronous x)- -> (x -> Eff (Process r ': r) ())- -> Eff (Process r ': r) ()- handleCall (SayHello "e1") _reply = do- me <- self px- logInfo (show me ++ " raising an error")- raiseError px "No body loves me... :,("- handleCall (SayHello "e2") _reply = do- me <- self px- logInfo (show me ++ " throwing a MyException ")- lift (Exc.throw MyException)- handleCall (SayHello "self") reply = do- me <- self px- logInfo (show me ++ " casting to self")- cast px (asServer @TestApi me) (Shout "from me")- void (reply False)- handleCall (SayHello "die") reply = do- me <- self px- logInfo (show me ++ " throwing and catching ")- catchRaisedError- px- (\er -> logInfo ("WOW: " ++ show er ++ " - No. This is wrong!"))- (raiseError px "No body loves me... :,(")- void (reply True)- handleCall (SayHello x) reply = do- me <- self px- logInfo (show me ++ " Got Hello: " ++ x)- void (reply (length x > 3))- handleCall Terminate reply = do- me <- self px- logInfo (show me ++ " exiting")- void (reply ())- exitNormally px- handleCall (TerminateError msg) reply = do- me <- self px- logInfo (show me ++ " exiting with error: " ++ msg)- void (reply ())- exitWithError px msg- handleTerminate msg = do- me <- self px- logInfo (show me ++ " is exiting: " ++ show msg)- maybe (exitNormally px) (exitWithError px) msg
− src/Control/Eff/Concurrent/Examples2.hs
@@ -1,167 +0,0 @@--- | Another complete example for the library-module Control.Eff.Concurrent.Examples2 where--import Data.Dynamic-import Control.Eff-import qualified Control.Eff.Concurrent.Process.ForkIOScheduler- as ForkIO-import qualified Control.Eff.Concurrent.Process.SingleThreadedScheduler- as Pure-import Control.Eff.Concurrent-import Control.Eff.State.Strict-import Control.Monad--main :: IO ()-main = do- ForkIO.defaultMain (counterExample SchedulerProxy)- Pure.defaultMain (counterExample SchedulerProxy)---- * First API--data Counter deriving Typeable--data instance Api Counter x where- Inc :: Api Counter 'Asynchronous- Cnt :: Api Counter ('Synchronous Integer)- ObserveCounter :: SomeObserver Counter -> Api Counter 'Asynchronous- UnobserveCounter :: SomeObserver Counter -> Api Counter 'Asynchronous--deriving instance Show (Api Counter x)--instance Observable Counter where- data Observation Counter where- CountChanged :: Integer -> Observation Counter- deriving (Show, Typeable)- registerObserverMessage = ObserveCounter- forgetObserverMessage = UnobserveCounter--logCounterObservations- :: (SetMember Process (Process q) r, Member (Logs LogMessage) q)- => SchedulerProxy q- -> Eff r (Server (CallbackObserver Counter))-logCounterObservations px = spawnCallbackObserver- px- (\fromSvr msg -> do- me <- self px- logInfo (show me ++ " observed on: " ++ show fromSvr ++ ": " ++ show msg)- return True- )--counterHandler- :: forall r q- . ( Member (State (Observers Counter)) r- , Member (State Integer) r- , Member (Logs LogMessage) r- , SetMember Process (Process q) r- )- => SchedulerProxy q- -> ApiHandler Counter r-counterHandler px = ApiHandler @Counter handleCast- handleCall- (defaultTermination px)- where- handleCast :: Api Counter 'Asynchronous -> Eff r ()- handleCast (ObserveCounter o) = addObserver o- handleCast (UnobserveCounter o) = removeObserver o- handleCast Inc = do- logInfo "Inc"- modify (+ (1 :: Integer))- currentCount <- get- notifyObservers px (CountChanged currentCount)- handleCall :: Api Counter ( 'Synchronous x) -> (x -> Eff r ()) -> Eff r ()- handleCall Cnt reply = do- c <- get- logInfo ("Cnt is " ++ show c)- reply c---- * Second API--data PocketCalc deriving Typeable--data instance Api PocketCalc x where- Add :: Integer -> Api PocketCalc ('Synchronous Integer)- AAdd :: Integer -> Api PocketCalc 'Asynchronous--deriving instance Show (Api PocketCalc x)--pocketCalcHandler- :: forall r q- . ( Member (State Integer) r- , Member (Logs LogMessage) r- , SetMember Process (Process q) r- )- => SchedulerProxy q- -> ApiHandler PocketCalc r-pocketCalcHandler px = ApiHandler @PocketCalc handleCast- handleCall- (defaultTermination px)- where- handleCast :: Api PocketCalc 'Asynchronous -> Eff r ()- handleCast (AAdd x) = do- logInfo ("AsyncAdd " ++ show x)- modify (+ x)- c <- get @Integer- logInfo ("Accumulator is " ++ show c)- handleCall :: Api PocketCalc ( 'Synchronous x) -> (x -> Eff r ()) -> Eff r ()- handleCall (Add x) reply = do- logInfo ("Add " ++ show x)- modify (+ x)- c <- get- logInfo ("Accumulator is " ++ show c)- reply c--serverLoop- :: forall r q- . (Member (Logs LogMessage) r, SetMember Process (Process q) r)- => SchedulerProxy q- -> Eff r ()-serverLoop px = evalState @Integer- 0- (manageObservers @Counter- (serveBoth px (counterHandler px) (pocketCalcHandler px))- )----- ** Counter client-counterExample- :: ( Member (Logs LogMessage) r- , Member (Logs LogMessage) q- , SetMember Process (Process q) r- )- => SchedulerProxy q- -> Eff r ()-counterExample px = do- let cnt sv = do- r <- call px sv Cnt- logInfo (show sv ++ " " ++ show r)- pid1 <- spawn (serverLoop px)- pid2 <- spawn (serverLoop px)- let cntServer1 = asServer @Counter pid1- cntServer2 = asServer @Counter pid2- calcServer1 = asServer @PocketCalc pid1- calcServer2 = asServer @PocketCalc pid2- cast px cntServer1 Inc- cnt cntServer1- cnt cntServer2- co1 <- logCounterObservations px- co2 <- logCounterObservations px- registerObserver px co1 cntServer1- registerObserver px co2 cntServer2- cast px calcServer1 (AAdd 12313)- cast px cntServer1 Inc- cnt cntServer1- cast px cntServer2 Inc- cnt cntServer2- registerObserver px co2 cntServer1- registerObserver px co1 cntServer2- cast px cntServer1 Inc- void (call px calcServer2 (Add 878))- cnt cntServer1- cast px cntServer2 Inc- cnt cntServer2- forgetObserver px co2 cntServer1- cast px cntServer1 Inc- cnt cntServer1- cast px cntServer2 Inc- cnt cntServer2- void $ sendShutdown px pid2
src/Control/Eff/Concurrent/Process.hs view
@@ -19,6 +19,7 @@ , ConsProcess , ResumeProcess(..) , SchedulerProxy(..)+ , MessageSelector(..) , thisSchedulerProxy , executeAndCatch , executeAndResume@@ -30,7 +31,10 @@ , spawn_ , receiveMessage , receiveMessageAs+ , receiveMessageSuchThat , receiveLoop+ , receiveLoopAs+ , receiveLoopSuchThat , self , sendShutdown , sendShutdownChecked@@ -105,12 +109,11 @@ -- given type. SendMessage :: ProcessId -> Dynamic -> Process r (ResumeProcess Bool) -- | Receive a message. This should block until an a message was received. The- -- pure function may convert the incoming message into something, and the- -- result is returned as 'ProcessMessage' value. Another reason why this function- -- returns, is if a process control message was sent to the process. This can- -- only occur from inside the runtime system, aka the effect handler- -- implementation. (Currently there is one in 'Control.Eff.Concurrent.Process.ForkIOScheduler'.)+ -- message is returned as a 'ProcessMessage' value. The function should also+ -- return if an exception was caught or a shutdown was requested. ReceiveMessage :: Process r (ResumeProcess Dynamic)+ -- | Wait for the next message that matches a criterium. Similar to 'ReceiveMessage'.+ ReceiveMessageSuchThat :: MessageSelector a -> Process r (ResumeProcess a) -- | Every 'Process' action returns it's actual result wrapped in this type. It -- will allow to signal errors as well as pass on normal results such as@@ -132,9 +135,32 @@ instance NFData1 ResumeProcess +-- | A function that deciced if the next message will be received by+-- 'ReceiveMessageSuchThat'. It conveniently is an instance of 'Monoid'+-- with first come, first serve bais.+newtype MessageSelector a =+ MessageSelector {runMessageSelector :: Dynamic -> Maybe a }+ deriving (Semigroup, Monoid, Functor)++-- | 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+ -- | Like 'SP' but different+ Scheduler :: SchedulerProxy q+ -- | /Cons/ 'Process' onto a list of effects. type ConsProcess r = Process r ': r +-- | Return a 'SchedulerProxy' for a 'Process' effect.+thisSchedulerProxy :: Eff (Process r ': r) (SchedulerProxy r)+thisSchedulerProxy = return SchedulerProxy+ -- | Execute a 'Process' action and resume the process, retry the action or exit -- the process when a shutdown was requested. executeAndResume@@ -165,20 +191,6 @@ ShutdownRequested -> send (Shutdown @q) OnError e -> return (Left e) --- | 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@@ -222,7 +234,7 @@ -> ProcessId -> o -> Eff r ()-sendMessageAs px pid = withFrozenCallStack $ sendMessage px pid . toDyn+sendMessageAs px pid = withFrozenCallStack (sendMessage px pid . toDyn) -- | Exit a process addressed by the 'ProcessId'. -- See 'SendShutdown'.@@ -232,7 +244,7 @@ => SchedulerProxy q -> ProcessId -> Eff r ()-sendShutdown px pid = withFrozenCallStack $ void (sendShutdownChecked px pid)+sendShutdown px pid = withFrozenCallStack (void (sendShutdownChecked px pid)) -- | Like 'sendShutdown', but also return @True@ iff the process to exit exists. sendShutdownChecked@@ -242,7 +254,7 @@ -> ProcessId -> Eff r Bool sendShutdownChecked _ pid =- withFrozenCallStack $ executeAndResume (SendShutdown pid)+ withFrozenCallStack (executeAndResume (SendShutdown pid)) -- | Start a new process, the new process will execute an effect, the function -- will return immediately with a 'ProcessId'.@@ -251,7 +263,7 @@ . (HasCallStack, SetMember Process (Process q) r) => Eff (Process q ': q) () -> Eff r ProcessId-spawn child = withFrozenCallStack $ executeAndResume (Spawn @q child)+spawn child = withFrozenCallStack (executeAndResume (Spawn @q child)) -- | Like 'spawn' but return @()@. spawn_@@ -259,32 +271,43 @@ . (HasCallStack, SetMember Process (Process q) r) => Eff (Process q ': q) () -> Eff r ()-spawn_ = withFrozenCallStack $ void . spawn+spawn_ = withFrozenCallStack (void . spawn) -- | Block until a message was received.+-- See 'ReceiveMessage' for more documentation. receiveMessage :: forall r q . (HasCallStack, SetMember Process (Process q) r) => SchedulerProxy q -> Eff r Dynamic-receiveMessage _ = withFrozenCallStack $ executeAndResume ReceiveMessage+receiveMessage _ = withFrozenCallStack executeAndResume ReceiveMessage +-- | Block until a message was received, that is not 'Nothing' after applying+-- a callback to it.+-- See 'ReceiveMessageSuchThat' for more documentation.+receiveMessageSuchThat+ :: forall r q a+ . (HasCallStack, Typeable a, SetMember Process (Process q) r)+ => SchedulerProxy q+ -> MessageSelector a+ -> Eff r a+receiveMessageSuchThat _ f =+ withFrozenCallStack (executeAndResume (ReceiveMessageSuchThat f))+ -- | Receive and cast the message to some 'Typeable' instance.+-- See 'ReceiveMessageSuchThat' for more documentation.+-- This will wait for a message of the return type using 'receiveMessageSuchThat' receiveMessageAs :: forall a r q . (HasCallStack, Typeable a, SetMember Process (Process q) r) => SchedulerProxy q -> Eff r a-receiveMessageAs px = withFrozenCallStack $ 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+receiveMessageAs px =+ withFrozenCallStack (receiveMessageSuchThat px (MessageSelector fromDynamic)) -- | Enter a loop to receive messages and pass them to a callback, until the -- function returns 'False'.+-- See 'receiveMesage' or 'ReceiveMessage' for more documentation. receiveLoop :: forall r q . (SetMember Process (Process q) r, HasCallStack)@@ -298,6 +321,44 @@ ShutdownRequested -> handlers (Left Nothing) >> receiveLoop px handlers OnError reason -> handlers (Left (Just reason)) >> receiveLoop px handlers ResumeWith message -> handlers (Right message) >> receiveLoop px handlers++-- | Run receive message of a certain type in an endless loop.+-- Like 'receiveLoopSuchThat' applied to 'fromDynamic'+receiveLoopAs+ :: forall r q a+ . (SetMember Process (Process q) r, HasCallStack, Typeable a)+ => SchedulerProxy q+ -> (Either (Maybe String) a -> Eff r ())+ -> Eff r ()+receiveLoopAs px handlers = do+ mReq <- send (ReceiveMessageSuchThat @a @q (MessageSelector fromDynamic))+ case mReq of+ RetryLastAction -> receiveLoopAs px handlers+ ShutdownRequested -> handlers (Left Nothing) >> receiveLoopAs px handlers+ OnError reason ->+ handlers (Left (Just reason)) >> receiveLoopAs px handlers+ ResumeWith message -> handlers (Right message) >> receiveLoopAs px handlers++-- | Like 'receiveLoop' but /selective/: Only the messages of the given type will be+-- received.+receiveLoopSuchThat+ :: forall r q a+ . (SetMember Process (Process q) r, HasCallStack)+ => SchedulerProxy q+ -> MessageSelector a+ -> (Either (Maybe String) a -> Eff r ())+ -> Eff r ()+receiveLoopSuchThat px selectMesage handlers = do+ mReq <- send (ReceiveMessageSuchThat @a @q selectMesage)+ case mReq of+ RetryLastAction -> receiveLoopSuchThat px selectMesage handlers+ ShutdownRequested ->+ handlers (Left Nothing) >> receiveLoopSuchThat px selectMesage handlers+ OnError reason ->+ handlers (Left (Just reason))+ >> receiveLoopSuchThat px selectMesage handlers+ ResumeWith message ->+ handlers (Right message) >> receiveLoopSuchThat px selectMesage handlers -- | Returns the 'ProcessId' of the current process. self
src/Control/Eff/Concurrent/Process/ForkIOScheduler.hs view
@@ -70,19 +70,19 @@ -- | Contains all process info'elements, as well as the state needed to -- implement inter process communication. It contains also a 'LogChannel' to -- which the logs of all processes are forwarded to.-data Scheduler =- Scheduler { _nextPid :: ProcessId- , _processTable :: Map ProcessId ProcessInfo- , _threadIdTable :: Map ProcessId ThreadId- , _schedulerShuttingDown :: Bool- , _logChannel :: LogChannel LogMessage- }+data SchedulerState =+ SchedulerState { _nextPid :: ProcessId+ , _processTable :: Map ProcessId ProcessInfo+ , _threadIdTable :: Map ProcessId ThreadId+ , _schedulerShuttingDown :: Bool+ , _logChannel :: LogChannel LogMessage+ } -makeLenses ''Scheduler+makeLenses ''SchedulerState -- | A newtype wrapper around an 'STM.TVar' holding the scheduler state. -- This is needed by 'spawn' and provided by 'runScheduler'.-newtype SchedulerVar = SchedulerVar { fromSchedulerVar :: STM.TVar Scheduler }+newtype SchedulerVar = SchedulerVar { fromSchedulerVar :: STM.TVar SchedulerState } deriving Typeable -- | A sum-type with errors that can occur when scheduleing messages.@@ -145,9 +145,10 @@ withNewSchedulerState :: HasCallStack => (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)+ Exc.bracket+ (newTVarIO (SchedulerState myPid Map.empty Map.empty False logC))+ (tearDownScheduler myTId)+ (mainProcessAction . SchedulerVar) where myPid = 1 tearDownScheduler myTId v = do@@ -427,6 +428,40 @@ emdynMsg ) + go pid (ReceiveMessageSuchThat selectMsg) k = shutdownOrGo pid k $ do+ emq <- overProcessInfo pid (use messageQ)+ case emq of+ Left e -> k (OnError (show @SchedulerError e))+ Right mq -> do+ let readTQueueUntilMatches =+ let enqueueAgain = traverse_ (unGetTQueue mq)+ untilMessageMatches unmatched = do+ msg <- readTQueue mq+ maybe+ (do+ -- msg is 'Nothing', this means the receive call+ -- must be interrupted (e.g. because the process was killed, etc)+ enqueueAgain unmatched+ return Nothing+ )+ ( maybe+ (untilMessageMatches (msg : unmatched))+ (\result -> do+ enqueueAgain unmatched+ return (Just result)+ )+ . runMessageSelector selectMsg+ )+ msg+ in untilMessageMatches []++ emdynMsg <- lift (Right <$> atomically readTQueueUntilMatches)+ k+ (either (OnError . show @SchedulerError)+ (maybe RetryLastAction ResumeWith)+ emdynMsg+ )+ go pid SelfPid k = shutdownOrGo pid k $ do lift Concurrent.yield k (ResumeWith pid)@@ -517,13 +552,14 @@ overScheduler :: HasSchedulerIO r- => Mtl.StateT Scheduler STM.STM (Either SchedulerError a)+ => Mtl.StateT SchedulerState STM.STM (Either SchedulerError a) -> Eff r (Either SchedulerError a) overScheduler stAction = do psVar <- getSchedulerTVar lift (overSchedulerIO psVar stAction) -overSchedulerIO :: STM.TVar Scheduler -> Mtl.StateT Scheduler STM.STM a -> IO a+overSchedulerIO+ :: STM.TVar SchedulerState -> Mtl.StateT SchedulerState STM.STM a -> IO a overSchedulerIO psVar stAction = STM.atomically (do ps <- STM.readTVar psVar@@ -532,7 +568,7 @@ return result ) -getSchedulerTVar :: HasSchedulerIO r => Eff r (TVar Scheduler)+getSchedulerTVar :: HasSchedulerIO r => Eff r (TVar SchedulerState) getSchedulerTVar = fromSchedulerVar <$> ask getSchedulerVar :: HasSchedulerIO r => Eff r SchedulerVar
src/Control/Eff/Concurrent/Process/SingleThreadedScheduler.hs view
@@ -123,7 +123,7 @@ OnSend :: !ProcessId -> !Dynamic -> (ResumeProcess Bool -> Eff r (OnYield r a)) -> OnYield r a- OnRecv :: (ResumeProcess Dynamic -> Eff r (OnYield r a))+ OnRecv :: MessageSelector b -> (ResumeProcess b -> Eff r (OnYield r a)) -> OnYield r a OnSendShutdown :: !ProcessId -> (ResumeProcess Bool -> Eff r (OnYield r a)) -> OnYield r a @@ -174,7 +174,7 @@ OnYield tk -> tk ShutdownRequested OnSelf tk -> tk ShutdownRequested OnSend _ _ tk -> tk ShutdownRequested- OnRecv tk -> tk ShutdownRequested+ OnRecv _ tk -> tk ShutdownRequested OnSpawn _ tk -> tk ShutdownRequested OnDone x -> return (OnDone x) OnShutdown -> return OnShutdown@@ -208,7 +208,16 @@ (msgQs & at toPid . _Just %~ (:|> msg)) (rest :|> (nextK, pid)) - recv@(OnRecv k) -> case msgQs ^. at pid of+ OnSpawn f k -> do+ nextK <- runEff $ k (ResumeWith newPid)+ fk <- runAsCoroutinePure runEff (f >> exitNormally SP)+ handleProcess runEff+ yieldEff+ (newPid + 1)+ (msgQs & at newPid ?~ Seq.empty)+ (rest :|> (nextK, pid) :|> (fk, newPid))++ recv@(OnRecv selectMessage k) -> case msgQs ^. at pid of Nothing -> do nextK <- runEff $ k (OnError (show pid ++ " has no message queue!")) handleProcess runEff yieldEff newPid msgQs (rest :|> (nextK, pid))@@ -222,23 +231,25 @@ newPid msgQs (rest :|> (recv, pid))-- Just (nextMessage :<| restMessages) -> do- nextK <- runEff $ k (ResumeWith nextMessage)- handleProcess runEff- yieldEff- newPid- (msgQs & at pid . _Just .~ restMessages)- (rest :|> (nextK, pid))-- OnSpawn f k -> do- nextK <- runEff $ k (ResumeWith newPid)- fk <- runAsCoroutinePure runEff (f >> exitNormally SP)- handleProcess runEff- yieldEff- (newPid + 1)- (msgQs & at newPid ?~ Seq.empty)- (rest :|> (nextK, pid) :|> (fk, newPid))+ Just messages ->+ let partitionMessages Empty _acc = Nothing+ partitionMessages (m :<| msgRest) acc = maybe+ (partitionMessages msgRest (acc :|> m))+ (\res -> Just (res, acc Seq.>< msgRest))+ (runMessageSelector selectMessage m)+ in case partitionMessages messages Empty of+ Nothing -> handleProcess runEff+ yieldEff+ newPid+ msgQs+ (rest :|> (recv, pid))+ Just (result, otherMessages) -> do+ nextK <- runEff $ k (ResumeWith result)+ handleProcess runEff+ yieldEff+ newPid+ (msgQs & at pid . _Just .~ otherMessages)+ (rest :|> (nextK, pid)) runAsCoroutinePure :: forall v r m@@ -249,15 +260,16 @@ runAsCoroutinePure runEff = runEff . handle_relay (return . OnDone) cont where cont :: Process r x -> (x -> Eff r (OnYield r v)) -> Eff r (OnYield r v)- cont YieldProcess k = return (OnYield k)- cont SelfPid k = return (OnSelf k)- cont (Spawn e) k = return (OnSpawn e k)- cont Shutdown _k = return OnShutdown- 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)+ cont YieldProcess k = return (OnYield k)+ cont SelfPid k = return (OnSelf k)+ cont (Spawn e) k = return (OnSpawn e k)+ cont Shutdown _k = return OnShutdown+ 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 (MessageSelector Just) k)+ cont (ReceiveMessageSuchThat f ) k = return (OnRecv f 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.
test/ProcessBehaviourTestCases.hs view
@@ -3,10 +3,14 @@ import Data.List ( sort ) import Data.Dynamic import Data.Foldable ( traverse_ )+import Data.Dynamic ( fromDynamic ) import Control.Exception import Control.Concurrent import Control.Concurrent.STM import Control.Eff.Concurrent.Process+import Control.Eff.Concurrent.Api+import Control.Eff.Concurrent.Api.Client+import Control.Eff.Concurrent.Api.Server import qualified Control.Eff.Concurrent.Process.ForkIOScheduler as ForkIO import qualified Control.Eff.Concurrent.Process.SingleThreadedScheduler@@ -55,8 +59,111 @@ , exitTests schedulerFactory , pingPongTests schedulerFactory , yieldLoopTests schedulerFactory+ , selectiveReceiveTests schedulerFactory ] )+++data ReturnToSender+ deriving Typeable++data instance Api ReturnToSender r where+ ReturnToSender :: ProcessId -> String -> Api ReturnToSender ('Synchronous Bool)+ StopReturnToSender :: Api ReturnToSender ('Synchronous ())++deriving instance Show (Api ReturnToSender x)++deriving instance Typeable (Api ReturnToSender x)++returnToSender+ :: forall q r+ . (HasCallStack, SetMember Process (Process q) r)+ => SchedulerProxy q+ -> Server ReturnToSender+ -> String+ -> Eff r Bool+returnToSender px toP msg = do+ me <- self px+ call px toP (ReturnToSender me msg)+ msgEcho <- receiveMessageAs @String px+ return (msgEcho == msg)++stopReturnToSender+ :: forall q r+ . (HasCallStack, SetMember Process (Process q) r)+ => SchedulerProxy q+ -> Server ReturnToSender+ -> Eff r ()+stopReturnToSender px toP = do+ me <- self px+ call px toP StopReturnToSender++returnToSenderServer+ :: forall q r+ . ( HasCallStack+ , SetMember Process (Process q) r+ , Member (Logs LogMessage) q+ )+ => SchedulerProxy q+ -> Eff r (Server ReturnToSender)+returnToSenderServer px = asServer <$> spawn+ (serve px $ ApiHandler+ { _handleCall = \m k -> case m of+ StopReturnToSender -> k () >> exitNormally px+ ReturnToSender fromP echoMsg ->+ sendMessageChecked px fromP (toDyn echoMsg)+ >>= (\res -> yieldProcess px >> k res)+ , _handleCast = logWarning . show+ , _handleTerminate = logWarning . show+ }+ )++selectiveReceiveTests+ :: forall r+ . (Member (Logs LogMessage) r, SetMember Lift (Lift IO) r)+ => IO (Eff (Process r ': r) () -> IO ())+ -> TestTree+selectiveReceiveTests schedulerFactory = setTravisTestOptions+ (testGroup+ "selective receive tests"+ [ testCase "send 10 messages (from 1..10) and receive messages from 10 to 1"+ $ applySchedulerFactory schedulerFactory+ $ do+ let+ nMax = 10+ receiverLoop donePid = go nMax+ where+ go :: Int -> Eff (Process r ': r) ()+ go 0 = sendMessageAs SP donePid True+ go n = do+ void $ receiveMessageSuchThat+ SP+ (MessageSelector+ (\m -> do+ i <- fromDynamic m+ if i == n then Just i else Nothing+ )+ )+ go (n - 1)++ senderLoop receviverPid =+ traverse_ (sendMessageAs SP receviverPid) [1 .. nMax]++ me <- self SP+ receiverPid <- spawn (receiverLoop me)+ spawn_ (senderLoop receiverPid)+ ok <- receiveMessageAs @Bool SP+ lift (ok @? "selective receive failed")+ , testCase "receive a message while waiting for a call reply"+ $ applySchedulerFactory schedulerFactory+ $ do+ srv <- returnToSenderServer SP+ ok <- returnToSender SP srv "test"+ () <- stopReturnToSender SP srv+ lift (ok @? "selective receive failed")+ ]+ )+ yieldLoopTests :: forall r