diff --git a/ChangeLog.md b/ChangeLog.md --- a/ChangeLog.md +++ b/ChangeLog.md @@ -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 diff --git a/examples/example-1/Main.hs b/examples/example-1/Main.hs new file mode 100644 --- /dev/null +++ b/examples/example-1/Main.hs @@ -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 diff --git a/examples/example-2/Main.hs b/examples/example-2/Main.hs new file mode 100644 --- /dev/null +++ b/examples/example-2/Main.hs @@ -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 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.7.3 +version: 0.8 description: Please see the README on GitHub at 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 diff --git a/src/Control/Eff/Concurrent.hs b/src/Control/Eff/Concurrent.hs --- a/src/Control/Eff/Concurrent.hs +++ b/src/Control/Eff/Concurrent.hs @@ -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(..) 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,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'. 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 @@ -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 diff --git a/src/Control/Eff/Concurrent/Examples.hs b/src/Control/Eff/Concurrent/Examples.hs deleted file mode 100644 --- a/src/Control/Eff/Concurrent/Examples.hs +++ /dev/null @@ -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 diff --git a/src/Control/Eff/Concurrent/Examples2.hs b/src/Control/Eff/Concurrent/Examples2.hs deleted file mode 100644 --- a/src/Control/Eff/Concurrent/Examples2.hs +++ /dev/null @@ -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 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 @@ -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 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 @@ -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 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 @@ -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. diff --git a/test/ProcessBehaviourTestCases.hs b/test/ProcessBehaviourTestCases.hs --- a/test/ProcessBehaviourTestCases.hs +++ b/test/ProcessBehaviourTestCases.hs @@ -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