diff --git a/ChangeLog.md b/ChangeLog.md --- a/ChangeLog.md +++ b/ChangeLog.md @@ -1,6 +1,6 @@ # Changelog for extensible-effects-concurrent -## Plan for future 0.23.0 +## Plan for future 0.24.0 - Add `gen_server` behaviour clone: - Disallow non-`Api`-messages @@ -18,6 +18,15 @@ - Introduce `ToLogText` - Remove `ToLogMessage` - Remove `logXXX'` users have to use `logXXX` and `ToLogText` + +## 0.23.0 + +- Include the process id in the console and trace log renderer +- Add a **process supervisor** similar to Erlang/OTPs simple_one_for_one supervisor. +- Fix `SingleThreadedScheduler` process linking bug: A process shall not be interrupted + when a linked process exits normally. +- Rename **ExitReason** to **Interrupt** and make the interrupt and exit handling + API more robust. ## 0.22.1 diff --git a/README.md b/README.md --- a/README.md +++ b/README.md @@ -18,11 +18,13 @@ ### Modeling an Application with Processes The fundamental approach to modelling applications in Erlang is -based on the concept of concurrent, communicating processes, without -shared state. +based on the concept of concurrent, communicating processes. +The mental model of the programming framework regards objects as **processes** +with an isolated internal state. + **`Processes`** are at the center of that contraption. All *actions* -happens in processes, and all *interactions* happen via messages sent +happen in processes, and all *interactions* happen via messages sent between processes. This is called **Message Passing Concurrency**; @@ -40,34 +42,34 @@ All processes except the first process are **`spawned`** by existing processes. -When a process **`spawns`** a new process, both are mutually **linked**, and -the former is called *parent* and the other *child*. - -Process links form a trees. - -When a parent process dies, the child processes dies as well. +When a process **`spawns`** a new process they are independent apart from the fact that +the parent knows the process-id of the spawend child process. + +Processes can **monitor** each other to be notified when a communication partner exits, +potentially in unforseen ways. -If on the other hand a child dies, the parent will not die unless the -child *crashed*. +Similarily processes may choose to mutually **link** each other. -A parent might also react by *restarting* the child from a defined starting -state. +That allows to model **trees** in which processes watch and start or +restart each other. Because processes never share memory, the internal - possibly broken - state of a process is gone, when a process exits; hence restarting a process will not be bothered by left-over, possibly inconsistent, state. -Erlang such parent processes are call *supervisor* processes in Erlang. +### Higher Level Abstractions -In order to build **supervision trees** the `Process` effect allows: +Processes can receive only message of type `Dynamic`. -- Interrupting and killing Processes -- Process Monitoring -- Process Linking -- Timers and Timeouts +In order to leverage Haskells type-safety, a bit of support code is available. +There is a **data family** called **`Api`** allowing to model **calls** and +**casts**, as well as event management and process supervision. + These facilities are very important to build **non-defensive**, **let-it-crash** applications, resilient to runtime errors. + +### Additional services Currently a custom **logging effect** is also part of the code base. diff --git a/examples/example-1/Main.hs b/examples/example-1/Main.hs --- a/examples/example-1/Main.hs +++ b/examples/example-1/Main.hs @@ -9,10 +9,13 @@ import qualified Control.Exception as Exc import qualified Data.Text as T import Control.DeepSeq +import Data.Type.Pretty data TestApi deriving Typeable +type instance ToPretty TestApi = PutStr "test" + data instance Api TestApi x where SayHello :: String -> Api TestApi ('Synchronous Bool) Shout :: String -> Api TestApi 'Asynchronous @@ -83,11 +86,11 @@ me <- self logInfo (T.pack (show me ++ " Shouting: " ++ x)) return AwaitNext - handleCallTest :: Api TestApi ('Synchronous r) -> (Eff (InterruptableProcess q) (Maybe r, CallbackResult) -> xxx) -> xxx + handleCallTest :: Api TestApi ('Synchronous r) -> (Eff (InterruptableProcess q) (Maybe r, CallbackResult 'Recoverable) -> xxx) -> xxx handleCallTest (SayHello "e1") k = k $ do me <- self logInfo (T.pack (show me ++ " raising an error")) - interrupt (ProcessError "No body loves me... :,(") + interrupt (ErrorInterrupt "No body loves me... :,(") handleCallTest (SayHello "e2") k = k $ do me <- self logInfo (T.pack (show me ++ " throwing a MyException ")) @@ -101,7 +104,7 @@ handleCallTest (SayHello "stop") k = k $ do me <- self logInfo (T.pack (show me ++ " stopping me")) - return (Just False, StopServer (ProcessError "test error")) + return (Just False, StopServer (ErrorInterrupt "test error")) handleCallTest (SayHello x) k = k $ do me <- self logInfo (T.pack (show me ++ " Got Hello: " ++ x)) @@ -109,13 +112,13 @@ handleCallTest Terminate k = k $ do me <- self logInfo (T.pack (show me ++ " exiting")) - pure (Just (), StopServer ProcessFinished) + pure (Just (), StopServer NormalExitRequested) handleCallTest (TerminateError msg) k = k $ do me <- self logInfo (T.pack (show me ++ " exiting with error: " ++ msg)) - pure (Just (), StopServer (ProcessError msg)) + pure (Just (), StopServer (ErrorInterrupt msg)) handleTerminateTest = InterruptCallback $ \msg -> do me <- self logInfo (T.pack (show me ++ " is exiting: " ++ show msg)) logProcessExit msg - pure (StopServer msg) + pure (StopServer (interruptToExit msg)) diff --git a/examples/example-2/Main.hs b/examples/example-2/Main.hs --- a/examples/example-2/Main.hs +++ b/examples/example-2/Main.hs @@ -9,6 +9,7 @@ import Data.Foldable import Control.Concurrent import Control.DeepSeq +import Data.Type.Pretty main :: IO () main = defaultMain (void counterExample) @@ -17,6 +18,8 @@ data Counter deriving Typeable +type instance ToPretty Counter = PutStr "counter" + data instance Api Counter x where Inc :: Api Counter 'Asynchronous Cnt :: Api Counter ('Synchronous Integer) @@ -58,6 +61,8 @@ data SupiDupi deriving Typeable +type instance ToPretty SupiDupi = PutStr "supi dupi" + data instance Api SupiDupi r where Whoopediedoo :: Bool -> Api SupiDupi ('Synchronous (Maybe ())) deriving Typeable @@ -67,6 +72,8 @@ newtype CounterChanged = CounterChanged Integer deriving (Show, Typeable, NFData) + +type instance ToPretty CounterChanged = PutStr "counter changed" spawnCounter :: (Member Logs q) diff --git a/extensible-effects-concurrent.cabal b/extensible-effects-concurrent.cabal --- a/extensible-effects-concurrent.cabal +++ b/extensible-effects-concurrent.cabal @@ -1,6 +1,6 @@ cabal-version: 2.0 name: extensible-effects-concurrent -version: 0.22.1 +version: 0.23.0 description: Please see the README on GitHub at synopsis: Message passing concurrency as extensible-effect homepage: https://github.com/sheyll/extensible-effects-concurrent#readme @@ -60,7 +60,8 @@ stm >= 2.4.5 && <2.6, transformers-base >= 0.4 && < 0.5, text >= 1.2 && < 1.3, - network >=2 && <4 + network >= 2 && < 4, + pretty-types >= 0.2.3.1 && < 0.4 autogen-modules: Paths_extensible_effects_concurrent exposed-modules: Control.Eff.Loop, @@ -82,8 +83,9 @@ Control.Eff.Concurrent, Control.Eff.Concurrent.Api, Control.Eff.Concurrent.Api.Client, - Control.Eff.Concurrent.Api.Server, Control.Eff.Concurrent.Api.GenServer, + Control.Eff.Concurrent.Api.Server, + Control.Eff.Concurrent.Api.Supervisor, Control.Eff.Concurrent.Process, Control.Eff.Concurrent.Process.Timer, Control.Eff.Concurrent.Process.ForkIOScheduler, @@ -93,6 +95,7 @@ Control.Eff.Concurrent.Api.Observer.Queue other-modules: Control.Eff.Concurrent.Api.Request, + Control.Eff.Concurrent.Api.Supervisor.InternalState, Paths_extensible_effects_concurrent default-extensions: AllowAmbiguousTypes, @@ -136,6 +139,7 @@ , lens , text , deepseq + , pretty-types >= 0.2.3.1 && < 0.4 ghc-options: -Wall -threaded -fno-full-laziness default-language: Haskell2010 default-extensions: AllowAmbiguousTypes @@ -168,6 +172,7 @@ , lens , text , deepseq + , pretty-types >= 0.2.3.1 && < 0.4 ghc-options: -Wall -threaded -fno-full-laziness default-language: Haskell2010 default-extensions: AllowAmbiguousTypes @@ -200,6 +205,7 @@ , filepath , lens , text + , pretty-types >= 0.2.3.1 && < 0.4 ghc-options: -Wall -threaded -fno-full-laziness default-language: Haskell2010 default-extensions: AllowAmbiguousTypes @@ -230,6 +236,7 @@ , deepseq , text , deepseq + , pretty-types >= 0.2.3.1 && < 0.4 ghc-options: -Wall -threaded -fno-full-laziness default-language: Haskell2010 default-extensions: AllowAmbiguousTypes @@ -263,6 +270,7 @@ , LogMessageIdeaTest , LoopTests , ProcessBehaviourTestCases + , SupervisorTests , ServerForkIO , SingleThreadedScheduler ghc-options: -Wall -threaded -fno-full-laziness @@ -286,6 +294,7 @@ , time , filepath , hostname + , pretty-types >= 0.2.3.1 && < 0.4 default-language: Haskell2010 default-extensions: AllowAmbiguousTypes , BangPatterns 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 @@ -133,9 +133,10 @@ , executeAndResume , executeAndResumeOrExit , executeAndResumeOrThrow - , ExitReason(..) + , Interrupt(..) + , interruptToExit , ExitRecovery(..) - , InterruptReason + , RecoverableInterrupt , Interrupts , InterruptableProcess , ExitSeverity(..) @@ -143,7 +144,7 @@ , toExitRecovery , isRecoverable , toExitSeverity - , isBecauseDown + , isProcessDownInterrupt , isCrash , toCrashReason , fromSomeExitReason @@ -165,6 +166,7 @@ import Control.Eff.Concurrent.Api ( Api , Synchronicity(..) , Server(..) + , Tangible , fromServer , proxyAsServer , asServer @@ -187,31 +189,6 @@ , RequestOrigin(..) , sendReply ) -import Control.Eff.Concurrent.Api.Server - ( spawnApiServer - , spawnLinkApiServer - , spawnApiServerStateful - , spawnApiServerEffectful - , spawnLinkApiServerEffectful - , CallbackResult(..) - , MessageCallback(..) - , handleCasts - , handleCalls - , handleCastsAndCalls - , handleCallsDeferred - , handleMessages - , handleSelectedMessages - , handleAnyMessages - , handleProcessDowns - , dropUnhandledMessages - , exitOnUnhandled - , logUnhandledMessages - , (^:) - , fallbackHandler - , ToServerPids(..) - , InterruptCallback(..) - , stopServerOnInterrupt - ) import Control.Eff.Concurrent.Api.Observer ( Observer(..) , Api @@ -238,6 +215,47 @@ , flushObservationQueue , withObservationQueue , spawnLinkObservationQueueWriter + ) +import Control.Eff.Concurrent.Api.Server + ( spawnApiServer + , spawnLinkApiServer + , spawnApiServerStateful + , spawnApiServerEffectful + , spawnLinkApiServerEffectful + , CallbackResult(..) + , MessageCallback(..) + , handleCasts + , handleCalls + , handleCastsAndCalls + , handleCallsDeferred + , handleMessages + , handleSelectedMessages + , handleAnyMessages + , handleProcessDowns + , dropUnhandledMessages + , exitOnUnhandled + , logUnhandledMessages + , (^:) + , fallbackHandler + , ToServerPids(..) + , InterruptCallback(..) + , stopServerOnInterrupt + ) +import Control.Eff.Concurrent.Api.Supervisor + ( Sup() + , SpawnFun + , SupConfig(MkSupConfig) + , supConfigChildStopTimeout + , supConfigSpawnFun + , SpawnErr(AlreadyStarted) + , startSupervisor + , stopSupervisor + , isSupervisorAlive + , monitorSupervisor + , getDiagnosticInfo + , spawnChild + , lookupChild + , stopChild ) import Control.Eff.Concurrent.Process.ForkIOScheduler ( schedule diff --git a/src/Control/Eff/Concurrent/Api.hs b/src/Control/Eff/Concurrent/Api.hs --- a/src/Control/Eff/Concurrent/Api.hs +++ b/src/Control/Eff/Concurrent/Api.hs @@ -1,3 +1,4 @@ +{-# LANGUAGE UndecidableInstances #-} -- | This module contains a mechanism to specify what kind of messages (aka -- /requests/) a 'Server' ('Process') can handle, and if the caller blocks and -- waits for an answer, which the server process provides. @@ -17,6 +18,7 @@ module Control.Eff.Concurrent.Api ( Api , Synchronicity(..) + , Tangible , Server(..) , fromServer , proxyAsServer @@ -24,13 +26,12 @@ ) where +import Control.DeepSeq ( NFData ) import Control.Eff.Concurrent.Process import Control.Lens import Data.Kind -import Data.Typeable ( Typeable - , typeRep - ) -import Control.DeepSeq (NFData) +import Data.Typeable ( Typeable ) +import Data.Type.Pretty -- | This data family defines an API, a communication interface description -- between at least two processes. The processes act as __servers__ or @@ -43,6 +44,9 @@ -- @Api@ instance is 'Synchronous', i.e. returns a result and blocks the caller -- or if it is 'Asynchronous' -- +-- Also, for better logging, the an instance of 'ToPretty' for the 'Api' index +-- type must be given. +-- -- Example: -- -- > @@ -52,12 +56,27 @@ -- > RentBook :: BookId -> Api BookShop ('Synchronous (Either RentalError RentalId)) -- > BringBack :: RentalId -> Api BookShop 'Asynchronous -- > +-- > type instance ToPretty BookShop = PutStr "book shop" +-- > -- > type BookId = Int -- > type RentalId = Int -- > type RentalError = String -- > data family Api (api :: Type) (reply :: Synchronicity) +type instance ToPretty (Api x y) = + PrettySurrounded (PutStr "<") (PutStr ">") ("API" <:> ToPretty x <+> ToPretty y) + +-- | A set of constraints for types that can evaluated via 'NFData', compared via 'Ord' and presented +-- dynamically via 'Typeable', and represented both as values +-- via 'Show', as well as on the type level via 'ToPretty'. +type Tangible i = + ( NFData i + , Typeable i + , Show i + , PrettyTypeShow (ToPretty i) + ) + -- | The (promoted) constructors of this type specify (at the type level) the -- reply behavior of a specific constructor of an @Api@ instance. data Synchronicity = @@ -72,9 +91,10 @@ newtype Server api = Server { _fromServer :: ProcessId } deriving (Eq,Ord,Typeable, NFData) -instance Typeable api => Show (Server api) where - showsPrec d s@(Server c) = - showParen (d >= 10) (showsPrec 11 (typeRep s) . showsPrec 11 c) +instance (PrettyTypeShow (ToPretty api)) => Show (Server api) where + showsPrec _ s@(Server c) = showString (showPretty s) . showsPrec 10 c + +type instance ToPretty (Server a) = ToPretty a <+> PutStr "server" makeLenses ''Server diff --git a/src/Control/Eff/Concurrent/Api/Client.hs b/src/Control/Eff/Concurrent/Api/Client.hs --- a/src/Control/Eff/Concurrent/Api/Client.hs +++ b/src/Control/Eff/Concurrent/Api/Client.hs @@ -24,6 +24,7 @@ import Control.Eff.Concurrent.Process.Timer import Control.Eff.Log import Data.Typeable ( Typeable ) +import Data.Type.Pretty import Control.DeepSeq import GHC.Stack @@ -37,6 +38,7 @@ :: forall r q o . ( HasCallStack , SetMember Process (Process q) r + , PrettyTypeShow (ToPretty o) , Member Interrupts r , Typeable o , Typeable (Api o 'Asynchronous) @@ -58,6 +60,7 @@ . ( SetMember Process (Process q) r , Member Interrupts r , Typeable api + , PrettyTypeShow (ToPretty api) , Typeable (Api api ( 'Synchronous result)) , NFData (Api api ( 'Synchronous result)) , Typeable result @@ -92,7 +95,7 @@ -- If the server that was called dies, this function interrupts the -- process with 'ProcessDown'. -- If the server takes longer to reply than the given timeout, this --- function interrupts the process with 'ProcessTimeout'. +-- function interrupts the process with 'TimeoutInterrupt'. -- -- __Always prefer this function over 'call'__ -- @@ -111,6 +114,7 @@ , Lifted IO q , Lifted IO r , HasCallStack + , PrettyTypeShow (ToPretty api) ) => Server api -> Api api ( 'Synchronous result) @@ -135,7 +139,7 @@ ++ show fromPid ++ " " ++ show timerRef logWarning' msg - interrupt (ProcessTimeout msg) + interrupt (TimeoutInterrupt msg) onProcDown p = do logWarning' ("call to dead server: "++ show serverP ++ " from " ++ show fromPid) interrupt (becauseProcessIsDown p) @@ -148,6 +152,7 @@ -- 'Server'. type ServesApi o r q = ( Typeable o + , PrettyTypeShow (ToPretty o) , SetMember Process (Process q) r , Member (ServerReader o) r ) diff --git a/src/Control/Eff/Concurrent/Api/GenServer.hs b/src/Control/Eff/Concurrent/Api/GenServer.hs --- a/src/Control/Eff/Concurrent/Api/GenServer.hs +++ b/src/Control/Eff/Concurrent/Api/GenServer.hs @@ -1,6 +1,6 @@ -- | A better, more safe implementation of the Erlang/OTP gen_server behaviour. -- **PLANNED TODO** --- @since 0.23.0 +-- @since 0.24.0 module Control.Eff.Concurrent.Api.GenServer where @@ -28,7 +28,7 @@ -> Eff (State (GenServerState a) ': eff) (ApiReply v) genServerInterrupt :: ('[Interrupts, Logs] <:: eff, SetMember Process (Process q) eff) - => InterruptReason + => Interrupt 'Recoverable -> Eff (State (GenServerState a) ': eff) () genServerInfoCommand :: Api a ('Synchronous Text) 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 @@ -20,6 +20,7 @@ ) where +import Control.DeepSeq (NFData(rnf)) import Control.Eff import Control.Eff.Concurrent.Process import Control.Eff.Concurrent.Api @@ -36,8 +37,8 @@ import qualified Data.Set as Set import Data.Text ( pack ) import Data.Typeable ( typeRep ) +import Data.Type.Pretty import GHC.Stack -import Control.DeepSeq (NFData(rnf)) -- * Observers @@ -49,9 +50,12 @@ -- @since 0.16.0 data Observer o where Observer - :: (Show (Server p), Typeable p, Typeable o, NFData o, NFData (Api p 'Asynchronous)) + :: (PrettyTypeShow (ToPretty p), Show (Server p), Typeable p, Typeable o, NFData o, NFData (Api p 'Asynchronous)) => (o -> Maybe (Api p 'Asynchronous)) -> Server p -> Observer o +type instance ToPretty (Observer o) = + PrettyParens ("observing" <:> ToPretty o) + instance (NFData o) => NFData (Observer o) where rnf (Observer k s) = rnf k `seq` rnf s @@ -80,6 +84,7 @@ , Member Interrupts r , Typeable o , NFData o + , PrettyTypeShow (ToPretty o) ) => Observer o -> Server (ObserverRegistry o) @@ -96,6 +101,7 @@ , Member Interrupts r , Typeable o , NFData o + , PrettyTypeShow (ToPretty o) ) => Observer o -> Server (ObserverRegistry o) @@ -127,7 +133,7 @@ -- @since 0.16.0 handleObservations :: (HasCallStack, Typeable o, SetMember Process (Process q) r, NFData (Observer o)) - => (o -> Eff r CallbackResult) + => (o -> Eff r (CallbackResult 'Recoverable)) -> MessageCallback (Observer o) r handleObservations k = handleCasts (\case @@ -137,7 +143,9 @@ -- | Use a 'Server' as an 'Observer' for 'handleObservations'. -- -- @since 0.16.0 -toObserver :: (NFData o, Typeable o, NFData (Api (Observer o) 'Asynchronous)) => Server (Observer o) -> Observer o +toObserver + :: (NFData o, Typeable o, NFData (Api (Observer o) 'Asynchronous), PrettyTypeShow (ToPretty o)) + => Server (Observer o) -> Observer o toObserver = toObserverFor Observed -- | Create an 'Observer' that conditionally accepts all observations of the @@ -146,7 +154,7 @@ -- -- @since 0.16.0 toObserverFor - :: (Typeable a, NFData (Api a 'Asynchronous), Typeable o, NFData o) + :: (Typeable a, PrettyTypeShow (ToPretty a), NFData (Api a 'Asynchronous), Typeable o, NFData o) => (o -> Api a 'Asynchronous) -> Server a -> Observer o @@ -160,6 +168,9 @@ -- @since 0.16.0 data ObserverRegistry o +type instance ToPretty (ObserverRegistry o) = + PrettyParens ("observer registry" <:> ToPretty o) + -- | Api for managing observers. This can be added to any server for any number of different observation types. -- The functions 'manageObservers' and 'handleObserverRegistration' are used to include observer handling; -- @@ -241,6 +252,7 @@ observed :: forall o r q . ( SetMember Process (Process q) r + , PrettyTypeShow (ToPretty o) , Member (ObserverState o) r , Member Interrupts r -- , NFData (Api o 'Asynchronous) diff --git a/src/Control/Eff/Concurrent/Api/Observer/Queue.hs b/src/Control/Eff/Concurrent/Api/Observer/Queue.hs --- a/src/Control/Eff/Concurrent/Api/Observer/Queue.hs +++ b/src/Control/Eff/Concurrent/Api/Observer/Queue.hs @@ -23,8 +23,8 @@ import Control.Exception.Safe as Safe import Control.Monad.IO.Class import Control.Monad ( unless ) -import Data.Typeable import qualified Data.Text as T +import Data.Typeable import GHC.Stack -- | Contains a 'TBQueue' capturing observations. @@ -136,9 +136,7 @@ -- @since 0.18.0 spawnLinkObservationQueueWriter :: forall o q - . ( Typeable o - , Show o - , NFData o + . ( Tangible o , NFData (Api (Observer o) 'Asynchronous) , Member Logs q , Lifted IO q 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 @@ -1,3 +1,4 @@ +{-# LANGUAGE UndecidableInstances #-} -- | Support code to implement 'Api' _server_ processes. -- -- @since 0.16.0 @@ -31,7 +32,7 @@ , (^:) , fallbackHandler , ToServerPids(..) - -- ** Interrupt handler + -- ** RecoverableInterrupt handler , InterruptCallback(..) , stopServerOnInterrupt ) @@ -53,6 +54,7 @@ import Data.Kind import Data.Proxy import Data.Text as T +import Data.Type.Pretty import GHC.Stack @@ -61,7 +63,11 @@ -- @since 0.13.2 spawnApiServer :: forall api eff - . (ToServerPids api, HasCallStack) + . ( ToServerPids api + , HasCallStack + , Member Logs eff + , PrettyTypeShow (PrettyServerPids api) + ) => MessageCallback api (InterruptableProcess eff) -> InterruptCallback (ConsProcess eff) -> Eff (InterruptableProcess eff) (ServerPids api) @@ -73,7 +79,12 @@ -- @since 0.14.2 spawnLinkApiServer :: forall api eff - . (ToServerPids api, HasCallStack) + . ( ToServerPids api + , HasCallStack + , Member Logs eff + , Typeable api + , PrettyTypeShow (PrettyServerPids api) + ) => MessageCallback api (InterruptableProcess eff) -> InterruptCallback (ConsProcess eff) -> Eff (InterruptableProcess eff) (ServerPids api) @@ -81,12 +92,19 @@ <$> spawnLink (apiServerLoop scb (InterruptCallback (raise . icb))) -- | /Server/ an 'Api' in a newly spawned process; the callbacks have access --- to some state initialed by the function in the first parameter. +-- to a state value. -- +-- The initial state value is returned by the action given as parameter, from +-- within the new process. +-- -- @since 0.13.2 spawnApiServerStateful :: forall api eff state - . (HasCallStack, ToServerPids api, NFData state) + . ( HasCallStack + , ToServerPids api + , NFData state + , PrettyTypeShow (PrettyServerPids api) + ) => Eff (InterruptableProcess eff) state -> MessageCallback api (State state ': InterruptableProcess eff) -> InterruptCallback (State state ': ConsProcess eff) @@ -110,9 +128,9 @@ invokeIntCb j = do l <- intCb j case l of - AwaitNext -> return Nothing - StopServer ProcessFinished -> return (Just ()) - StopServer k -> exitBecause (NotRecovered k) + AwaitNext -> return Nothing + StopServer ExitNormally -> return (Just ()) + StopServer k -> exitBecause k -- | /Server/ an 'Api' in a newly spawned process; The caller provides an -- effect handler for arbitrary effects used by the server callbacks. @@ -124,6 +142,8 @@ , ToServerPids api , Member Interrupts serverEff , SetMember Process (Process eff) serverEff + , Member Logs serverEff + , PrettyTypeShow (PrettyServerPids api) ) => (forall b . Eff serverEff b -> Eff (InterruptableProcess eff) b) -> MessageCallback api serverEff @@ -145,6 +165,8 @@ , ToServerPids api , Member Interrupts serverEff , SetMember Process (Process eff) serverEff + , Member Logs serverEff + , PrettyTypeShow (PrettyServerPids api) ) => (forall b . Eff serverEff b -> Eff (InterruptableProcess eff) b) -> MessageCallback api serverEff @@ -165,22 +187,30 @@ , ToServerPids api , Member Interrupts serverEff , SetMember Process (Process eff) serverEff + , Member Logs serverEff + , PrettyTypeShow (PrettyServerPids api) ) => MessageCallback api serverEff -> InterruptCallback serverEff -> Eff serverEff () -apiServerLoop (MessageCallback sel cb) (InterruptCallback intCb) = +apiServerLoop (MessageCallback sel cb) (InterruptCallback intCb) = do + logInfo ( "enter server loop: " + <> pack (showPretty (Proxy @(PrettySurrounded (PutStr "[") (PutStr "]") + (PrettyServerPids api)))) + ) receiveSelectedLoop sel - ( either (fmap Left . intCb) (fmap Right . tryUninterrupted . cb) - >=> handleCallbackResult + ( ( either + (fmap Left . intCb) + (fmap Right . tryUninterrupted . cb)) + >=> handleCallbackResult ) where handleCallbackResult - :: Either CallbackResult (Either InterruptReason CallbackResult) + :: Either (CallbackResult 'NoRecovery) (Either (Interrupt 'Recoverable) (CallbackResult 'Recoverable)) -> Eff serverEff (Maybe ()) handleCallbackResult (Left AwaitNext) = return Nothing - handleCallbackResult (Left (StopServer r)) = exitBecause (NotRecovered r) + handleCallbackResult (Left (StopServer r)) = exitBecause r handleCallbackResult (Right (Right AwaitNext)) = return Nothing handleCallbackResult (Right (Right (StopServer r))) = intCb r >>= handleCallbackResult . Left @@ -192,12 +222,12 @@ -- should continue or stop. -- -- @since 0.13.2 -data CallbackResult where +data CallbackResult (r :: ExitRecovery) where -- | Tell the server to keep the server loop running - AwaitNext :: CallbackResult + AwaitNext :: CallbackResult r -- | Tell the server to exit, this will cause 'apiServerLoop' to stop handling requests without -- exiting the process. - StopServer :: InterruptReason -> CallbackResult + StopServer :: Interrupt r -> CallbackResult r -- SendReply :: reply -> CallbackResult () -> CallbackResult (reply -> Eff eff ()) deriving ( Typeable ) @@ -209,7 +239,7 @@ -- -- @since 0.13.2 data MessageCallback api eff where - MessageCallback :: MessageSelector a -> (a -> Eff eff CallbackResult) -> MessageCallback api eff + MessageCallback :: MessageSelector a -> (a -> Eff eff (CallbackResult 'Recoverable)) -> MessageCallback api eff instance Semigroup (MessageCallback api eff) where (MessageCallback selL runL) <> (MessageCallback selR runR) = @@ -228,7 +258,7 @@ handleMessages :: forall eff a . (HasCallStack, NFData a, Typeable a) - => (a -> Eff eff CallbackResult) + => (a -> Eff eff (CallbackResult 'Recoverable)) -> MessageCallback '[] eff handleMessages = MessageCallback selectMessage @@ -239,7 +269,7 @@ :: forall eff a . HasCallStack => MessageSelector a - -> (a -> Eff eff CallbackResult) + -> (a -> Eff eff (CallbackResult 'Recoverable)) -> MessageCallback '[] eff handleSelectedMessages = MessageCallback @@ -249,7 +279,7 @@ handleAnyMessages :: forall eff . HasCallStack - => (StrictDynamic -> Eff eff CallbackResult) + => (StrictDynamic -> Eff eff (CallbackResult 'Recoverable)) -> MessageCallback '[] eff handleAnyMessages = MessageCallback selectAnyMessage @@ -260,7 +290,7 @@ :: forall api eff . ( HasCallStack, Typeable api, Typeable (Api api 'Asynchronous) , NFData (Request api)) - => (Api api 'Asynchronous -> Eff eff CallbackResult) + => (Api api 'Asynchronous -> Eff eff (CallbackResult 'Recoverable)) -> MessageCallback api eff handleCasts h = MessageCallback (selectMessageWith @@ -298,7 +328,7 @@ => ( forall secret reply . (NFData reply, Typeable reply, Typeable (Api api ( 'Synchronous reply))) => Api api ( 'Synchronous reply) - -> (Eff eff (Maybe reply, CallbackResult) -> secret) + -> (Eff eff (Maybe reply, CallbackResult 'Recoverable) -> secret) -> secret ) -> MessageCallback api eff @@ -331,11 +361,11 @@ , SetMember Process (Process effScheduler) eff , Member Interrupts eff ) - => (Api api 'Asynchronous -> Eff eff CallbackResult) + => (Api api 'Asynchronous -> Eff eff (CallbackResult 'Recoverable)) -> ( forall secret reply . (Typeable reply, Typeable (Api api ( 'Synchronous reply)), NFData reply) => Api api ( 'Synchronous reply) - -> (Eff eff (Maybe reply, CallbackResult) -> secret) + -> (Eff eff (Maybe reply, CallbackResult 'Recoverable) -> secret) -> secret ) -> MessageCallback api eff @@ -356,7 +386,7 @@ . (Typeable reply, NFData reply, Typeable (Api api ( 'Synchronous reply))) => RequestOrigin (Api api ( 'Synchronous reply)) -> Api api ( 'Synchronous reply) - -> Eff eff CallbackResult + -> Eff eff (CallbackResult 'Recoverable) ) -> MessageCallback api eff handleCallsDeferred h = MessageCallback @@ -380,7 +410,7 @@ handleProcessDowns :: forall eff . HasCallStack - => (MonitorReference -> Eff eff CallbackResult) + => (MonitorReference -> Eff eff (CallbackResult 'Recoverable)) -> MessageCallback '[] eff handleProcessDowns k = MessageCallback selectMessage (k . downReference) @@ -423,7 +453,7 @@ -- @since 0.13.2 exitOnUnhandled :: forall eff . HasCallStack => MessageCallback '[] eff exitOnUnhandled = MessageCallback selectAnyMessage $ \msg -> - return (StopServer (ProcessError ("unhandled message " <> show msg))) + return (StopServer (ErrorInterrupt ("unhandled message " <> show msg))) -- | A 'fallbackHandler' that drops the left-over messages. -- @@ -441,10 +471,12 @@ -- -- @since 0.13.2 class ToServerPids (t :: k) where + type PrettyServerPids t :: PrettyType type ServerPids t toServerPids :: proxy t -> ProcessId -> ServerPids t instance ToServerPids '[] where + type PrettyServerPids '[] = PutStr "any message" type ServerPids '[] = ProcessId toServerPids _ = id @@ -452,6 +484,7 @@ forall (api1 :: Type) (api2 :: [Type]) . (ToServerPids api1, ToServerPids api2) => ToServerPids (api1 ': api2) where + type PrettyServerPids (api1 ': api2) = PrettyServerPids api1 <++> PutStr ", " <++> PrettyServerPids api2 type ServerPids (api1 ': api2) = (ServerPids api1, ServerPids api2) toServerPids _ p = (toServerPids (Proxy @api1) p, toServerPids (Proxy @api2) p) @@ -460,17 +493,19 @@ forall (api1 :: Type) . (ToServerPids api1) => ToServerPids api1 where + type PrettyServerPids api1 = ToPretty api1 type ServerPids api1 = Server api1 toServerPids _ = asServer + -- | Just a wrapper around a function that will be applied to the result of --- a 'MessageCallback's 'StopServer' clause, or an 'InterruptReason' caught during +-- a 'MessageCallback's 'StopServer' clause, or an 'Interrupt' caught during -- the execution of @receive@ or a 'MessageCallback' -- -- @since 0.13.2 data InterruptCallback eff where InterruptCallback :: - (InterruptReason -> Eff eff CallbackResult) -> InterruptCallback eff + (Interrupt 'Recoverable -> Eff eff (CallbackResult 'NoRecovery)) -> InterruptCallback eff instance Default (InterruptCallback eff) where def = stopServerOnInterrupt @@ -479,4 +514,4 @@ -- -- @since 0.13.2 stopServerOnInterrupt :: forall eff . HasCallStack => InterruptCallback eff -stopServerOnInterrupt = InterruptCallback (pure . StopServer) +stopServerOnInterrupt = InterruptCallback (pure . StopServer . interruptToExit) diff --git a/src/Control/Eff/Concurrent/Api/Supervisor.hs b/src/Control/Eff/Concurrent/Api/Supervisor.hs new file mode 100644 --- /dev/null +++ b/src/Control/Eff/Concurrent/Api/Supervisor.hs @@ -0,0 +1,460 @@ +{-# LANGUAGE UndecidableInstances #-} +-- | A process supervisor spawns and monitors child processes. +-- +-- The child processes are mapped to symbolic identifier values: Child-IDs. +-- +-- This is the barest, most minimal version of a supervisor. Children +-- can be started, but not restarted. +-- +-- Children can efficiently be looked-up by an id-value, and when +-- the supervisor is shutdown, all children will be shutdown these +-- are actually all the features of this supervisor implementation. +-- +-- Also, this minimalist supervisor only knows how to spawn a +-- single kind of child process. +-- +-- When a supervisor spawns a new child process, it expects the +-- child process to return a 'ProcessId'. The supervisor will +-- 'monitor' the child process. +-- +-- This is in stark contrast to how Erlang/OTP handles things; +-- In the OTP Supervisor, the child has to link to the parent. +-- This allows the child spec to be more flexible in that no +-- @pid@ has to be passed from the child start function to the +-- supervisor process, and also, a child may break free from +-- the supervisor by unlinking. +-- +-- Now while this seems nice at first, this might actually +-- cause surprising results, since it is usually expected that +-- stopping a supervisor also stops the children, or that a child +-- exit shows up in the logging originating from the former +-- supervisor. +-- +-- The approach here is to allow any child to link to the +-- supervisor to realize when the supervisor was violently killed, +-- and otherwise give the child no chance to unlink itself from +-- its supervisor. +-- +-- This module is far simpler than the Erlang/OTP counter part, +-- of a @simple_one_for_one@ supervisor. +-- +-- The future of this supervisor might not be a-lot more than +-- it currently is. The ability to restart processes might be +-- implemented outside of this supervisor module. +-- +-- One way to do that is to implement the restart logic in +-- a seperate module, since the child-id can be reused when a child +-- exits. +-- +-- @since 0.23.0 +module Control.Eff.Concurrent.Api.Supervisor + ( Sup() + , SpawnFun + , SupConfig(MkSupConfig) + , supConfigChildStopTimeout + , supConfigSpawnFun + , SpawnErr(AlreadyStarted) + , startSupervisor + , stopSupervisor + , isSupervisorAlive + , monitorSupervisor + , getDiagnosticInfo + , spawnChild + , lookupChild + , stopChild + ) where + +import Control.DeepSeq (NFData(rnf)) +import Control.Eff as Eff +import Control.Eff.Concurrent.Api +import Control.Eff.Concurrent.Api.Client +import Control.Eff.Concurrent.Api.Server +import Control.Eff.Concurrent.Api.Supervisor.InternalState +import Control.Eff.Concurrent.Process +import Control.Eff.Concurrent.Process.Timer +import Control.Eff.Extend (raise) +import Control.Eff.Log +import Control.Eff.State.Strict as Eff +import Control.Lens hiding ((.=), use) +import Data.Default +import Data.Dynamic +import Data.Foldable +import qualified Data.Map as Map +import Data.Text (Text, pack) +import Data.Type.Pretty +import GHC.Generics (Generic) +import GHC.Stack +import Control.Applicative ((<|>)) + +-- * Supervisor Server API +-- ** Types + + +-- | A function that will initialize the child process. +-- +-- The process-id returned from this function is kept private, but +-- it is possible to return '(ProcessId, ProcessId)' for example. +-- +-- The function is expected to return a value - usually a tuple of 'Server' values for +-- the different aspects of a process, such as sending API requests, managing +-- observers and receiving other auxiliary API messages. +-- +-- @since 0.23.0 +type SpawnFun i e o = i -> Eff e (o, ProcessId) + +-- | Options that control the 'Sup i o' process. +-- +-- This contains: +-- +-- * a 'SpawnFun' +-- * the 'Timeout' after requesting a normal child exit before brutally killing the child. +-- +-- @since 0.23.0 +data SupConfig i e o = + MkSupConfig + { _supConfigSpawnFun :: SpawnFun i e o + , _supConfigChildStopTimeout :: Timeout + } + +makeLenses ''SupConfig + +-- | 'Api' type for supervisor processes. +-- +-- The /supervisor/ process contains a 'SpawnFun' from which it can spawn new child processes. +-- +-- The supervisor maps an identifier value of type @childId@ to a 'ProcessId' and a +-- 'spawnResult' type. +-- +-- This 'spawnResult' is likely a tuple of 'Server' process ids that allow type-safe interaction with +-- the process. +-- +-- Also, this serves as handle or reference for interacting with a supervisor process. +-- +-- A value of this type is returned by 'startSupervisor' +-- +-- @since 0.23.0 +newtype Sup childId spawnResult = + MkSup (Server (Sup childId spawnResult)) + deriving (Ord, Eq, Typeable, NFData) + +instance (PrettyTypeShow (ToPretty childId), PrettyTypeShow (ToPretty spawnResult)) + => Show (Sup childId spawnResult) where + showsPrec d (MkSup svr) = showsPrec d svr + +-- | The 'Api' instance contains methods to start, stop and lookup a child +-- process, as well as a diagnostic callback. +-- +-- @since 0.23.0 +data instance Api (Sup i o) r where + StartC :: i -> Api (Sup i o) ('Synchronous (Either (SpawnErr i o) o)) + StopC :: i -> Timeout -> Api (Sup i o) ('Synchronous Bool) + LookupC :: i -> Api (Sup i o) ('Synchronous (Maybe o)) + GetDiagnosticInfo :: Api (Sup i o) ('Synchronous Text) + deriving Typeable + +type instance ToPretty (Sup i o) = + PutStr "supervisor{" <++> ToPretty i <+> PutStr "=>" <+> ToPretty o <++> PutStr "}" + +instance (Show i) => Show (Api (Sup i o) ('Synchronous r)) where + showsPrec d (StartC c) = showParen (d >= 10) (showString "StartC " . showsPrec 10 c) + showsPrec d (StopC c t) = showParen (d >= 10) (showString "StopC " . showsPrec 10 c . showChar ' ' . showsPrec 10 t) + showsPrec d (LookupC c) = showParen (d >= 10) (showString "LookupC " . showsPrec 10 c) + showsPrec _ GetDiagnosticInfo = showString "GetDiagnosticInfo" + +instance (NFData i) => NFData (Api (Sup i o) ('Synchronous r)) where + rnf (StartC ci) = rnf ci + rnf (StopC ci t) = rnf ci `seq` rnf t + rnf (LookupC ci) = rnf ci + rnf GetDiagnosticInfo = () + +-- | Runtime-Errors occurring when spawning child-processes. +-- +-- @since 0.23.0 +data SpawnErr i o + = AlreadyStarted i + o + deriving (Eq, Ord, Show, Typeable, Generic) + +instance (NFData i, NFData o) => NFData (SpawnErr i o) + + +-- ** Functions +-- | Start and link a new supervisor process with the given 'SpawnFun'unction. +-- +-- To spawn new child processes use 'spawnChild'. +-- +-- @since 0.23.0 +startSupervisor + :: forall i e o + . ( HasCallStack + , Member Logs e + , Lifted IO e + , Ord i + , Tangible i + , Tangible o + ) + => SupConfig i (InterruptableProcess e) o + -> Eff (InterruptableProcess e) (Sup i o) +startSupervisor supConfig = do + (_sup, supPid) <- spawnApiServerStateful initChildren (onRequest ^: onMessage) onInterrupt + return (mkSup supPid) + where + mkSup supPid = MkSup (asServer supPid) + initChildren :: Eff (InterruptableProcess e) (Children i o) + initChildren = return def + onRequest :: MessageCallback (Sup i o) (State (Children i o) ': InterruptableProcess e) + onRequest = handleCalls onCall + where + onCall :: + Api (Sup i o) ('Synchronous reply) + -> (Eff (State (Children i o) ': InterruptableProcess e) (Maybe reply, CallbackResult 'Recoverable) -> st) + -> st + onCall GetDiagnosticInfo k = k ((, AwaitNext) . Just . pack . show <$> getChildren @i @o) + onCall (LookupC i) k = k ((, AwaitNext) . Just . fmap _childOutput <$> lookupChildById @i @o i) + onCall (StopC i t) k = + k $ do + mExisting <- lookupAndRemoveChildById @i @o i + case mExisting of + Nothing -> do + return (Just False, AwaitNext) + Just existingChild -> do + stopOrKillChild i existingChild t + return (Just True, AwaitNext) + + onCall (StartC i) k = + k $ do + (o, cPid) <- raise ((supConfig ^. supConfigSpawnFun) i) + cMon <- monitor cPid + mExisting <- lookupChildById i + case mExisting of + Nothing -> do + putChild i (MkChild o cPid cMon) + return (Just (Right o), AwaitNext) + Just existingChild -> + return (Just (Left (AlreadyStarted i (existingChild ^. childOutput))), AwaitNext) + onMessage :: MessageCallback '[] (State (Children i o) ': InterruptableProcess e) + onMessage = handleProcessDowns onDown <> handleAnyMessages onInfo + where + onDown mrChild = do + oldEntry <- lookupAndRemoveChildByMonitor @i @o mrChild + case oldEntry of + Nothing -> + logWarning ("unexpected process down: " <> pack (show mrChild)) + Just (i, c) -> do + logInfo ( "process down: " + <> pack (show mrChild) + <> " for child " + <> pack (show i) + <> " => " + <> pack (show (_childOutput c)) + ) + return AwaitNext + + onInfo d = do + logInfo ("unexpected message received: " <> pack (show d)) + return AwaitNext + onInterrupt :: InterruptCallback (State (Children i o) ': ConsProcess e) + onInterrupt = + InterruptCallback $ \e -> do + let (logSev, exitReason) = + case e of + NormalExitRequested -> + (debugSeverity, ExitNormally) + _ -> + (warningSeverity, interruptToExit (ErrorInterrupt ("supervisor interrupted: " <> show e))) + stopAllChildren @i @o (supConfig ^. supConfigChildStopTimeout) + logWithSeverity logSev ("supervisor stopping: " <> pack (show e)) + + pure (StopServer exitReason) + +-- | Stop the supervisor and shutdown all processes. +-- +-- Block until the supervisor has finished. +-- +-- @since 0.23.0 +stopSupervisor + :: ( HasCallStack + , Member Interrupts e + , SetMember Process (Process q0) e + , Member Logs e + , Lifted IO e + , Ord i + , Tangible i + , Tangible o + ) + => Sup i o + -> Eff e () +stopSupervisor (MkSup svr) = do + logInfo ("stopping supervisor: " <> pack (show svr)) + mr <- monitor (_fromServer svr) + sendInterrupt (_fromServer svr) NormalExitRequested + r <- receiveSelectedMessage (selectProcessDown mr) + logInfo ("supervisor stopped: " <> pack (show svr) <> " " <> pack (show r)) + +-- | Check if a supervisor process is still alive. +-- +-- @since 0.23.0 +isSupervisorAlive + :: ( HasCallStack + , Member Interrupts e + , Member Logs e + , Typeable i + , Typeable o + , NFData i + , NFData o + , Show i + , Show o + , SetMember Process (Process q0) e + ) + => Sup i o + -> Eff e Bool +isSupervisorAlive (MkSup x) = isProcessAlive (_fromServer x) + +-- | Monitor a supervisor process. +-- +-- @since 0.23.0 +monitorSupervisor + :: ( HasCallStack + , Member Interrupts e + , Member Logs e + , Typeable i + , Typeable o + , NFData i + , NFData o + , Show i + , Show o + , SetMember Process (Process q0) e + ) + => Sup i o + -> Eff e MonitorReference +monitorSupervisor (MkSup x) = monitor (_fromServer x) + +-- | Start, link and monitor a new child process using the 'SpawnFun' passed +-- to 'startSupervisor'. +-- +-- @since 0.23.0 +spawnChild :: + ( HasCallStack + , Member Interrupts e + , Member Logs e + , Ord i + , Tangible i + , Tangible o + , SetMember Process (Process q0) e + ) + => Sup i o + -> i + -> Eff e (Either (SpawnErr i o) o) +spawnChild (MkSup svr) cId = call svr (StartC cId) + +-- | Lookup the given child-id and return the output value of the 'SpawnFun' +-- if the client process exists. +-- +-- @since 0.23.0 +lookupChild :: + ( HasCallStack + , Member Interrupts e + , Member Logs e + , Ord i + , Tangible i + , Tangible o + , SetMember Process (Process q0) e + ) + => Sup i o + -> i + -> Eff e (Maybe o) +lookupChild (MkSup svr) cId = call svr (LookupC cId) + +-- | Stop a child process, and block until the child has exited. +-- +-- Return 'True' if a process with that ID was found, 'False' if no process +-- with the given ID was running. +-- +-- @since 0.23.0 +stopChild :: + ( HasCallStack + , Member Interrupts e + , Member Logs e + , Ord i + , Tangible i + , Tangible o + , SetMember Process (Process q0) e + ) + => Sup i o + -> i + -> Eff e Bool +stopChild (MkSup svr) cId = call svr (StopC cId (TimeoutMicros 4000000)) + +-- | Return a 'Text' describing the current state of the supervisor. +-- +-- @since 0.23.0 +getDiagnosticInfo + :: ( Ord i + , Tangible i + , Tangible o + , Typeable e + , HasCallStack + , Member Interrupts e + , SetMember Process (Process q0) e + ) + => Sup i o + -> Eff e Text +getDiagnosticInfo (MkSup s) = call s GetDiagnosticInfo + +-- Internal Functions + +stopOrKillChild + :: forall i o e q0 . + ( Ord i + , Tangible i + , Tangible o + , HasCallStack + , SetMember Process (Process q0) e + , Member Interrupts e + , Lifted IO e + , Lifted IO q0 + , Member Logs e + , Member (State (Children i o)) e + ) + => i + -> Child o + -> Timeout + -> Eff e () +stopOrKillChild cId c stopTimeout = + do + sup <- MkSup . asServer @(Sup i o) <$> self + t <- startTimer stopTimeout + sendInterrupt (c^.childProcessId) NormalExitRequested + r1 <- receiveSelectedMessage ( Right <$> selectProcessDown (c^.childMonitoring) + <|> Left <$> selectTimerElapsed t ) + case r1 of + Left timerElapsed -> do + logWarning (pack (show timerElapsed) <> ": child "<> pack (show cId) <>" => " <> pack(show (c^.childOutput)) <>" did not shutdown in time") + sendShutdown + (c^.childProcessId) + (interruptToExit + (TimeoutInterrupt + ("child did not shut down in time and was terminated by the " + ++ show sup))) + Right downMsg -> + logInfo ("child "<> pack (show cId) <>" => " <> pack(show (c^.childOutput)) <>" terminated: " <> pack (show (downReason downMsg))) + +stopAllChildren + :: forall i o e q0 . + ( Ord i + , Tangible i + , Tangible o + , HasCallStack + , SetMember Process (Process q0) e + , Lifted IO e + , Lifted IO q0 + , Member Logs e + , Member (State (Children i o)) e + ) + => Timeout -> Eff e () +stopAllChildren stopTimeout = removeAllChildren @i @o >>= pure . Map.assocs >>= traverse_ xxx + where + xxx (cId, c) = provideInterrupts (stopOrKillChild cId c stopTimeout) >>= either crash return + where + crash e = do + logError (pack (show e) <> " while stopping child: " <> pack (show cId) <> " " <> pack (show c)) diff --git a/src/Control/Eff/Concurrent/Api/Supervisor/InternalState.hs b/src/Control/Eff/Concurrent/Api/Supervisor/InternalState.hs new file mode 100644 --- /dev/null +++ b/src/Control/Eff/Concurrent/Api/Supervisor/InternalState.hs @@ -0,0 +1,101 @@ +module Control.Eff.Concurrent.Api.Supervisor.InternalState where + +import Control.DeepSeq +import Control.Eff as Eff +import Control.Eff.Concurrent.Process +import Control.Eff.State.Strict as Eff +import Control.Lens hiding ((.=), use) +import Data.Default +import Data.Dynamic +import Data.Map (Map) +import GHC.Generics (Generic) + + +data Child o = MkChild + { _childOutput :: o + , _childProcessId :: ProcessId + , _childMonitoring :: MonitorReference + } + deriving (Show, Generic, Typeable, Eq, Ord) + +instance (NFData o) => NFData (Child o) + +makeLenses ''Child + + +-- | Internal state. +data Children i o = MkChildren + { _childrenById :: Map i (Child o) + , _childrenByMonitor :: Map MonitorReference (i, Child o) + } deriving (Show, Generic, Typeable) + +instance Default (Children i o) where + def = MkChildren def def + +instance (NFData i, NFData o) => NFData (Children i o) + +makeLenses ''Children + +-- | State accessor +getChildren + :: (Ord i, Member (State (Children i o)) e) + => Eff e (Children i o) +getChildren = Eff.get + +putChild + :: (Ord i, Member (State (Children i o)) e) + => i + -> Child o + -> Eff e () +putChild cId c = modify ( (childrenById . at cId .~ Just c) + . (childrenByMonitor . at (_childMonitoring c) .~ Just (cId, c)) + ) + +lookupChildById + :: (Ord i, Member (State (Children i o)) e) + => i + -> Eff e (Maybe (Child o)) +lookupChildById i = view (childrenById . at i) <$> get + +lookupChildByMonitor + :: (Ord i, Member (State (Children i o)) e) + => MonitorReference + -> Eff e (Maybe (i, Child o)) +lookupChildByMonitor m = view (childrenByMonitor . at m) <$> get + +lookupAndRemoveChildById + :: forall i o e. (Ord i, Member (State (Children i o)) e) + => i + -> Eff e (Maybe (Child o)) +lookupAndRemoveChildById i = + traverse go =<< lookupChildById i + where + go c = pure c <* removeChild i c + +removeChild + :: forall i o e. (Ord i, Member (State (Children i o)) e) + => i + -> Child o + -> Eff e () +removeChild i c = do + modify @(Children i o) ( (childrenById . at i .~ Nothing) + . (childrenByMonitor . at (_childMonitoring c) .~ Nothing) + ) + +lookupAndRemoveChildByMonitor + :: forall i o e. (Ord i, Member (State (Children i o)) e) + => MonitorReference + -> Eff e (Maybe (i, Child o)) +lookupAndRemoveChildByMonitor r = do + traverse go =<< lookupChildByMonitor r + where + go (i, c) = pure (i, c) <* removeChild i c + +removeAllChildren + :: forall i o e. (Ord i, Member (State (Children i o)) e) + => Eff e (Map i (Child o)) +removeAllChildren = do + cm <- view childrenById <$> getChildren @i + modify @(Children i o) (childrenById .~ mempty) + modify @(Children i o) (childrenByMonitor .~ mempty) + return cm 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 @@ -60,7 +60,7 @@ , spawnLink , spawnRaw , spawnRaw_ - -- ** Process Exit or Interrupt + -- ** Process Exit or Interrupt Recoverable , exitBecause , exitNormally , exitWithError @@ -76,7 +76,7 @@ , MonitorReference(..) , withMonitor , receiveWithMonitor - -- ** Process Interrupt Handling + -- ** Process Interrupt Recoverable Handling , provideInterruptsShutdown , handleInterrupts , tryUninterrupted @@ -89,10 +89,11 @@ , executeAndResume , executeAndResumeOrExit , executeAndResumeOrThrow - -- ** Exit Or Interrupt Reasons - , ExitReason(..) + -- ** Exit Or Interrupt Recoverable Reasons + , Interrupt(..) + , interruptToExit , ExitRecovery(..) - , InterruptReason + , RecoverableInterrupt , Interrupts , InterruptableProcess , ExitSeverity(..) @@ -100,7 +101,7 @@ , toExitRecovery , isRecoverable , toExitSeverity - , isBecauseDown + , isProcessDownInterrupt , isCrash , toCrashReason , fromSomeExitReason @@ -128,6 +129,7 @@ import Control.Applicative import Data.Maybe import Data.String (fromString) +import Data.Text (Text, pack, unpack) import qualified Data.Text as T import qualified Control.Exception as Exc @@ -157,32 +159,32 @@ -- * when the first process exists, all process should be killed immediately data Process (r :: [Type -> Type]) b where -- | Remove all messages from the process' message queue - FlushMessages ::Process r (ResumeProcess [StrictDynamic]) + FlushMessages :: Process r (ResumeProcess [StrictDynamic]) -- | In cooperative schedulers, this will give processing time to the -- scheduler. Every other operation implicitly serves the same purpose. -- -- @since 0.12.0 - YieldProcess ::Process r (ResumeProcess ()) + YieldProcess :: Process r (ResumeProcess ()) -- | Return the current 'ProcessId' - SelfPid ::Process r (ResumeProcess ProcessId) + SelfPid :: Process r (ResumeProcess ProcessId) -- | Start a new process, the new process will execute an effect, the function -- will return immediately with a 'ProcessId'. - Spawn ::Eff (Process r ': r) () -> Process r (ResumeProcess ProcessId) + Spawn :: Eff (Process r ': r) () -> Process r (ResumeProcess ProcessId) -- | Start a new process, and 'Link' to it . -- -- @since 0.12.0 - SpawnLink ::Eff (Process r ': r) () -> Process r (ResumeProcess ProcessId) + SpawnLink :: Eff (Process r ': r) () -> Process r (ResumeProcess ProcessId) -- | Get the process state (or 'Nothing' if the process is dead) - GetProcessState ::ProcessId -> Process r (ResumeProcess (Maybe ProcessState)) + GetProcessState :: ProcessId -> Process r (ResumeProcess (Maybe ProcessState)) -- | Shutdown the process; irregardless of the exit reason, this function never -- returns, - Shutdown ::ExitReason 'NoRecovery -> Process r a - -- | Raise an error, that can be handled. - SendShutdown ::ProcessId -> ExitReason 'NoRecovery -> Process r (ResumeProcess ()) + Shutdown :: Interrupt 'NoRecovery -> Process r a + -- | Shutdown another process immediately, the other process has no way of handling this! + SendShutdown ::ProcessId -> Interrupt 'NoRecovery -> Process r (ResumeProcess ()) -- | Request that another a process interrupts. The targeted process is interrupted -- and gets an 'Interrupted', the target process may decide to ignore the -- interrupt and continue as if nothing happened. - SendInterrupt ::ProcessId -> InterruptReason -> Process r (ResumeProcess ()) + SendInterrupt :: ProcessId -> Interrupt 'Recoverable -> Process r (ResumeProcess ()) -- | Send a message to a process addressed by the 'ProcessId'. Sending a -- message should __always succeed__ and return __immediately__, even if the -- destination process does not exist, or does not accept messages of the @@ -194,7 +196,7 @@ -- was caught or a shutdown was requested. ReceiveSelectedMessage :: forall r a . MessageSelector a -> Process r (ResumeProcess a) -- | Generate a unique 'Int' for the current process. - MakeReference ::Process r (ResumeProcess Int) + MakeReference :: Process r (ResumeProcess Int) -- | Monitor another process. When the monitored process exits a -- 'ProcessDown' is sent to the calling process. -- The return value is a unique identifier for that monitor. @@ -204,21 +206,21 @@ -- will be sent immediately, without exit reason -- -- @since 0.12.0 - Monitor ::ProcessId -> Process r (ResumeProcess MonitorReference) + Monitor :: ProcessId -> Process r (ResumeProcess MonitorReference) -- | Remove a monitor. -- -- @since 0.12.0 - Demonitor ::MonitorReference -> Process r (ResumeProcess ()) + Demonitor :: MonitorReference -> Process r (ResumeProcess ()) -- | Connect the calling process to another process, such that -- if one of the processes crashes (i.e. 'isCrash' returns 'True'), the other - -- is shutdown with the 'ExitReason' 'LinkedProcessCrashed'. + -- is shutdown with the 'Interrupt' 'LinkedProcessCrashed'. -- -- @since 0.12.0 - Link ::ProcessId -> Process r (ResumeProcess ()) + Link :: ProcessId -> Process r (ResumeProcess ()) -- | Unlink the calling process from the other process. -- -- @since 0.12.0 - Unlink ::ProcessId -> Process r (ResumeProcess ()) + Unlink :: ProcessId -> Process r (ResumeProcess ()) instance Show (Process r b) where showsPrec d = \case @@ -297,9 +299,9 @@ -- incoming messages. data ResumeProcess v where -- | The current operation of the process was interrupted with a - -- 'ExitReason'. If 'isRecoverable' holds for the given reason, + -- 'Interrupt'. If 'isRecoverable' holds for the given reason, -- the process may choose to continue. - Interrupted ::InterruptReason -> ResumeProcess v + Interrupted :: Interrupt 'Recoverable -> ResumeProcess v -- | The process may resume to do work, using the given result. ResumeWith ::a -> ResumeProcess a deriving ( Typeable, Generic, Generic1, Show ) @@ -400,7 +402,7 @@ instance Default ProcessState where def = ProcessBooting --- | This kind is used to indicate if a 'ExitReason' can be treated like +-- | This kind is used to indicate if a 'Interrupt' can be treated like -- a short interrupt which can be handled or ignored. data ExitRecovery = Recoverable | NoRecovery deriving (Typeable, Ord, Eq, Generic) @@ -416,17 +418,16 @@ ) -- | Get the 'ExitRecovery' -toExitRecovery :: ExitReason r -> ExitRecovery +toExitRecovery :: Interrupt r -> ExitRecovery toExitRecovery = \case - ProcessFinished -> Recoverable - (ProcessNotRunning _) -> Recoverable - (ProcessTimeout _) -> Recoverable + NormalExitRequested -> Recoverable + (OtherProcessNotRunning _) -> Recoverable + (TimeoutInterrupt _) -> Recoverable (LinkedProcessCrashed _) -> Recoverable - (ProcessError _) -> Recoverable + (ErrorInterrupt _) -> Recoverable ExitNormally -> NoRecovery - (NotRecovered _ ) -> NoRecovery - (UnexpectedException _ _) -> NoRecovery - Killed -> NoRecovery + (ExitUnhandledError _) -> NoRecovery + ExitProcessCancelled -> NoRecovery -- | This value indicates whether a process exited in way consistent with -- the planned behaviour or not. @@ -443,176 +444,173 @@ instance NFData ExitSeverity --- | Get the 'ExitSeverity' of a 'ExitReason'. -toExitSeverity :: ExitReason e -> ExitSeverity +-- | Get the 'ExitSeverity' of a 'Interrupt'. +toExitSeverity :: Interrupt e -> ExitSeverity toExitSeverity = \case - ExitNormally -> NormalExit - ProcessFinished -> NormalExit - NotRecovered ProcessFinished -> NormalExit - _ -> Crash + ExitNormally -> NormalExit + NormalExitRequested -> NormalExit + _ -> Crash --- | A sum-type with reasons for why a process exists the scheduling loop, --- this includes errors, that can occur when scheduling messages. -data ExitReason (t :: ExitRecovery) where +-- | A sum-type with reasons for why a process operation, such as receiving messages, +-- is interrupted in the scheduling loop. +-- +-- This includes errors, that can occur when scheduling messages. +-- +-- @since 0.23.0 +data Interrupt (t :: ExitRecovery) where -- | A process has finished a unit of work and might exit or work on -- something else. This is primarily used for interrupting infinite -- server loops, allowing for additional cleanup work before -- exiting (e.g. with 'ExitNormally') -- -- @since 0.13.2 - ProcessFinished - :: ExitReason 'Recoverable + NormalExitRequested + :: Interrupt 'Recoverable -- | A process that should be running was not running. - ProcessNotRunning - :: ProcessId -> ExitReason 'Recoverable + OtherProcessNotRunning + :: ProcessId -> Interrupt 'Recoverable -- | A 'Recoverable' timeout has occurred. - ProcessTimeout - :: String -> ExitReason 'Recoverable + TimeoutInterrupt + :: String -> Interrupt 'Recoverable -- | A linked process is down LinkedProcessCrashed - :: ProcessId -> ExitReason 'Recoverable + :: ProcessId -> Interrupt 'Recoverable -- | An exit reason that has an error message and is 'Recoverable'. - ProcessError - :: String -> ExitReason 'Recoverable + ErrorInterrupt + :: String -> Interrupt 'Recoverable -- | A process function returned or exited without any error. ExitNormally - :: ExitReason 'NoRecovery - -- | An unhandled 'Recoverable' allows 'NoRecovery'. - NotRecovered - :: (ExitReason 'Recoverable) -> ExitReason 'NoRecovery - -- | An unexpected runtime exception was thrown, i.e. an exception - -- derived from 'Control.Exception.Safe.SomeException' - UnexpectedException - :: String -> String -> ExitReason 'NoRecovery - -- | A process was cancelled (e.g. killed, in 'Async.cancel') - Killed - :: ExitReason 'NoRecovery + :: Interrupt 'NoRecovery + -- | An error causes the process to exit immediately. + -- For example an unexpected runtime exception was thrown, i.e. an exception + -- derived from 'Control.Exception.Safe.SomeException' + -- Or a 'Recoverable' Interrupt was not recoverd. + ExitUnhandledError + :: Text -> Interrupt 'NoRecovery + -- | A process shall exit immediately, without any cleanup was cancelled (e.g. killed, in 'Async.cancel') + ExitProcessCancelled + :: Interrupt 'NoRecovery deriving Typeable -instance Show (ExitReason x) where +-- | Return either 'ExitNormally' or 'interruptToExit' from a 'Recoverable' 'Interrupt'; +-- +-- If the 'Interrupt' is 'NormalExitRequested' then return 'ExitNormally' +interruptToExit :: Interrupt 'Recoverable -> Interrupt 'NoRecovery +interruptToExit NormalExitRequested = ExitNormally +interruptToExit x = ExitUnhandledError (pack (show x)) + +instance Show (Interrupt x) where showsPrec d = - showParen (d >= 10) + showParen (d >= 9) . (\case - ProcessFinished -> showString "process finished" - ProcessNotRunning p -> showString "process not running: " . shows p - ProcessTimeout reason -> showString "timeout: " . showString reason + NormalExitRequested -> showString "interrupt: A normal exit was requested" + OtherProcessNotRunning p -> showString "interrupt: Another process is not running: " . showsPrec 10 p + TimeoutInterrupt reason -> showString "interrupt: A timeout occured: " . showString reason LinkedProcessCrashed m -> - showString "linked process " . shows m . showString " crashed" - ProcessError reason -> showString "error: " . showString reason - ExitNormally -> showString "exit normally" - NotRecovered e -> showString "not recovered from: " . shows e - UnexpectedException w m -> - showString "unhandled runtime exception: " - . showString m - . showString " caught here: " - . showString w - Killed -> showString "killed" + showString "interrupt: A linked process " . showsPrec 10 m . showString " crashed" + ErrorInterrupt reason -> showString "interrupt: An error occured: " . showString reason + ExitNormally -> showString "exit: Process finished successfully" + ExitUnhandledError w -> + showString "exit: Unhandled " . showString (unpack w) + ExitProcessCancelled -> showString "exit: The process was cancelled" ) -instance Exc.Exception (ExitReason 'Recoverable) -instance Exc.Exception (ExitReason 'NoRecovery ) +instance Exc.Exception (Interrupt 'Recoverable) +instance Exc.Exception (Interrupt 'NoRecovery ) -instance NFData (ExitReason x) where - rnf ProcessFinished = rnf () - rnf (ProcessNotRunning !l) = rnf l - rnf (ProcessTimeout !l) = rnf l +instance NFData (Interrupt x) where + rnf NormalExitRequested = rnf () + rnf (OtherProcessNotRunning !l) = rnf l + rnf (TimeoutInterrupt !l) = rnf l rnf (LinkedProcessCrashed !l) = rnf l - rnf (ProcessError !l) = rnf l + rnf (ErrorInterrupt !l) = rnf l rnf ExitNormally = rnf () - rnf (NotRecovered !l ) = rnf l - rnf (UnexpectedException !l1 !l2) = rnf l1 `seq` rnf l2 - rnf Killed = rnf () + rnf (ExitUnhandledError !l) = rnf l + rnf ExitProcessCancelled = rnf () -instance Ord (ExitReason x) where - compare ProcessFinished ProcessFinished = EQ - compare ProcessFinished _ = LT - compare _ ProcessFinished = GT - compare (ProcessNotRunning l) (ProcessNotRunning r) = compare l r - compare (ProcessNotRunning _) _ = LT - compare _ (ProcessNotRunning _) = GT - compare (ProcessTimeout l) (ProcessTimeout r) = compare l r - compare (ProcessTimeout _) _ = LT - compare _ (ProcessTimeout _) = GT - compare (LinkedProcessCrashed l) (LinkedProcessCrashed r) = compare l r - compare (LinkedProcessCrashed _) _ = LT - compare _ (LinkedProcessCrashed _) = GT - compare (ProcessError l) (ProcessError r) = compare l r - compare ExitNormally ExitNormally = EQ - compare ExitNormally _ = LT - compare _ ExitNormally = GT - compare (NotRecovered l) (NotRecovered r) = compare l r - compare (NotRecovered _) _ = LT - compare _ (NotRecovered _) = GT - compare (UnexpectedException l1 l2) (UnexpectedException r1 r2) = - compare l1 r1 <> compare l2 r2 - compare (UnexpectedException _ _) _ = LT - compare _ (UnexpectedException _ _) = GT - compare Killed Killed = EQ +instance Ord (Interrupt x) where + compare NormalExitRequested NormalExitRequested = EQ + compare NormalExitRequested _ = LT + compare _ NormalExitRequested = GT + compare (OtherProcessNotRunning l) (OtherProcessNotRunning r) = compare l r + compare (OtherProcessNotRunning _) _ = LT + compare _ (OtherProcessNotRunning _) = GT + compare (TimeoutInterrupt l) (TimeoutInterrupt r) = compare l r + compare (TimeoutInterrupt _) _ = LT + compare _ (TimeoutInterrupt _) = GT + compare (LinkedProcessCrashed l) (LinkedProcessCrashed r) = compare l r + compare (LinkedProcessCrashed _) _ = LT + compare _ (LinkedProcessCrashed _) = GT + compare (ErrorInterrupt l) (ErrorInterrupt r) = compare l r + compare ExitNormally ExitNormally = EQ + compare ExitNormally _ = LT + compare _ ExitNormally = GT + compare (ExitUnhandledError l) (ExitUnhandledError r) = compare l r + compare (ExitUnhandledError _ ) _ = LT + compare _ (ExitUnhandledError _) = GT + compare ExitProcessCancelled ExitProcessCancelled = EQ -instance Eq (ExitReason x) where - (==) ProcessFinished ProcessFinished = True - (==) (ProcessNotRunning l) (ProcessNotRunning r) = (==) l r +instance Eq (Interrupt x) where + (==) NormalExitRequested NormalExitRequested = True + (==) (OtherProcessNotRunning l) (OtherProcessNotRunning r) = (==) l r (==) ExitNormally ExitNormally = True - (==) (ProcessTimeout l) (ProcessTimeout r) = l == r + (==) (TimeoutInterrupt l) (TimeoutInterrupt r) = l == r (==) (LinkedProcessCrashed l) (LinkedProcessCrashed r) = l == r - (==) (ProcessError l) (ProcessError r) = (==) l r - (==) (NotRecovered l) (NotRecovered r) = (==) l r - (==) (UnexpectedException l1 l2) (UnexpectedException r1 r2) = - (==) l1 r1 && (==) l2 r2 - (==) Killed Killed = True + (==) (ErrorInterrupt l) (ErrorInterrupt r) = (==) l r + (==) (ExitUnhandledError l) (ExitUnhandledError r) = (==) l r + (==) ExitProcessCancelled ExitProcessCancelled = True (==) _ _ = False -- | A predicate for linked process __crashes__. -isBecauseDown :: Maybe ProcessId -> ExitReason r -> Bool -isBecauseDown mp = \case - ProcessFinished -> False - ProcessNotRunning _ -> False - ProcessTimeout _ -> False - LinkedProcessCrashed p -> maybe True (== p) mp - ProcessError _ -> False +isProcessDownInterrupt :: Maybe ProcessId -> Interrupt r -> Bool +isProcessDownInterrupt mOtherProcess = \case + NormalExitRequested -> False + OtherProcessNotRunning _ -> False + TimeoutInterrupt _ -> False + LinkedProcessCrashed p -> maybe True (== p) mOtherProcess + ErrorInterrupt _ -> False ExitNormally -> False - NotRecovered e -> isBecauseDown mp e - UnexpectedException _ _ -> False - Killed -> False + ExitUnhandledError _ -> False + ExitProcessCancelled -> False --- | 'ExitReason's which are recoverable are interrupts. -type InterruptReason = ExitReason 'Recoverable +-- | 'Interrupt's which are 'Recoverable'. +type RecoverableInterrupt = Interrupt 'Recoverable --- | 'Exc'eptions containing 'InterruptReason's. +-- | 'Exc'eptions containing 'Interrupt's. -- See 'handleInterrupts', 'exitOnInterrupt' or 'provideInterrupts' -type Interrupts = Exc InterruptReason +type Interrupts = Exc (Interrupt 'Recoverable) -- | This adds a layer of the 'Interrupts' effect on top of 'ConsProcess' type InterruptableProcess e = Interrupts ': ConsProcess e --- | Handle all 'InterruptReason's of an 'InterruptableProcess' by --- wrapping them up in 'NotRecovered' and then do a process 'Shutdown'. +-- | Handle all 'Interrupt's of an 'InterruptableProcess' by +-- wrapping them up in 'interruptToExit' and then do a process 'Shutdown'. provideInterruptsShutdown :: forall e a . Eff (InterruptableProcess e) a -> Eff (ConsProcess e) a provideInterruptsShutdown e = do res <- provideInterrupts e case res of - Left ex -> send (Shutdown @e (NotRecovered ex)) + Left ex -> send (Shutdown @e (interruptToExit ex)) Right a -> return a --- | Handle 'InterruptReason's arising during process operations, e.g. +-- | Handle 'Interrupt's arising during process operations, e.g. -- when a linked process crashes while we wait in a 'receiveSelectedMessage' -- via a call to 'interrupt'. handleInterrupts :: (HasCallStack, Member Interrupts r) - => (InterruptReason -> Eff r a) + => (Interrupt 'Recoverable -> Eff r a) -> Eff r a -> Eff r a handleInterrupts = flip catchError --- | Like 'handleInterrupts', but instead of passing the 'InterruptReason' +-- | Like 'handleInterrupts', but instead of passing the 'Interrupt' -- to a handler function, 'Either' is returned. -- -- @since 0.13.2 tryUninterrupted :: (HasCallStack, Member Interrupts r) => Eff r a - -> Eff r (Either InterruptReason a) + -> Eff r (Either (Interrupt 'Recoverable) a) tryUninterrupted = handleInterrupts (pure . Left) . fmap Right -- | Handle interrupts by logging them with `logProcessExit` and otherwise @@ -624,50 +622,50 @@ -> Eff r () logInterrupts = handleInterrupts logProcessExit --- | Handle 'InterruptReason's arising during process operations, e.g. +-- | Handle 'Interrupt's arising during process operations, e.g. -- when a linked process crashes while we wait in a 'receiveSelectedMessage' -- via a call to 'interrupt'. exitOnInterrupt :: (HasCallStack, Member Interrupts r, SetMember Process (Process q) r) => Eff r a -> Eff r a -exitOnInterrupt = handleInterrupts (exitBecause . NotRecovered) +exitOnInterrupt = handleInterrupts (exitBecause . interruptToExit) --- | Handle 'InterruptReason's arising during process operations, e.g. +-- | Handle 'Interrupt's arising during process operations, e.g. -- when a linked process crashes while we wait in a 'receiveSelectedMessage' -- via a call to 'interrupt'. provideInterrupts - :: HasCallStack => Eff (Interrupts ': r) a -> Eff r (Either InterruptReason a) + :: HasCallStack => Eff (Interrupts ': r) a -> Eff r (Either (Interrupt 'Recoverable) a) provideInterrupts = runError --- | Wrap all (left) 'InterruptReason's into 'NotRecovered' and --- return the (right) 'NoRecovery' 'ExitReason's as is. +-- | Wrap all (left) 'Interrupt's into 'interruptToExit' and +-- return the (right) 'NoRecovery' 'Interrupt's as is. mergeEitherInterruptAndExitReason - :: Either InterruptReason (ExitReason 'NoRecovery) -> ExitReason 'NoRecovery -mergeEitherInterruptAndExitReason = either NotRecovered id + :: Either (Interrupt 'Recoverable) (Interrupt 'NoRecovery) -> Interrupt 'NoRecovery +mergeEitherInterruptAndExitReason = either interruptToExit id --- | Throw an 'InterruptReason', can be handled by 'handleInterrupts' or +-- | Throw an 'Interrupt', can be handled by 'handleInterrupts' or -- 'exitOnInterrupt' or 'provideInterrupts'. -interrupt :: (HasCallStack, Member Interrupts r) => InterruptReason -> Eff r a +interrupt :: (HasCallStack, Member Interrupts r) => Interrupt 'Recoverable -> Eff r a interrupt = throwError -- | A predicate for crashes. A /crash/ happens when a process exits --- with an 'ExitReason' other than 'ExitNormally' -isCrash :: ExitReason x -> Bool -isCrash (NotRecovered !x) = isCrash x +-- with an 'Interrupt' other than 'ExitNormally' +isCrash :: Interrupt x -> Bool +isCrash NormalExitRequested = False isCrash ExitNormally = False isCrash _ = True -- | A predicate for recoverable exit reasons. This predicate defines the -- exit reasons which functions such as 'executeAndResume' -isRecoverable :: ExitReason x -> Bool +isRecoverable :: Interrupt x -> Bool isRecoverable (toExitRecovery -> Recoverable) = True isRecoverable _ = False --- | An existential wrapper around 'ExitReason' +-- | An existential wrapper around 'Interrupt' data SomeExitReason where - SomeExitReason ::ExitReason x -> SomeExitReason + SomeExitReason ::Interrupt x -> SomeExitReason instance Ord SomeExitReason where compare = compare `on` fromSomeExitReason @@ -682,25 +680,24 @@ rnf = rnf . fromSomeExitReason -- | Partition a 'SomeExitReason' back into either a 'NoRecovery' --- or a 'Recoverable' 'ExitReason' +-- or a 'Recoverable' 'Interrupt' fromSomeExitReason - :: SomeExitReason -> Either (ExitReason 'NoRecovery) InterruptReason + :: SomeExitReason -> Either (Interrupt 'NoRecovery) (Interrupt 'Recoverable) fromSomeExitReason (SomeExitReason e) = case e of - recoverable@ProcessFinished -> Right recoverable - recoverable@(ProcessNotRunning _) -> Right recoverable - recoverable@(ProcessTimeout _) -> Right recoverable + recoverable@NormalExitRequested -> Right recoverable + recoverable@(OtherProcessNotRunning _) -> Right recoverable + recoverable@(TimeoutInterrupt _) -> Right recoverable recoverable@(LinkedProcessCrashed _) -> Right recoverable - recoverable@(ProcessError _) -> Right recoverable + recoverable@(ErrorInterrupt _) -> Right recoverable noRecovery@ExitNormally -> Left noRecovery - noRecovery@(NotRecovered _) -> Left noRecovery - noRecovery@(UnexpectedException _ _) -> Left noRecovery - noRecovery@Killed -> Left noRecovery + noRecovery@(ExitUnhandledError _) -> Left noRecovery + noRecovery@ExitProcessCancelled -> Left noRecovery --- | Print a 'ExitReason' to 'Just' a formatted 'String' when 'isCrash' +-- | Print a 'Interrupt' to 'Just' a formatted 'String' when 'isCrash' -- is 'True'. -- This can be useful in combination with view patterns, e.g.: -- --- > logCrash :: ExitReason -> Eff e () +-- > logCrash :: Interrupt -> Eff e () -- > logCrash (toCrashReason -> Just reason) = logError reason -- > logCrash _ = return () -- @@ -708,13 +705,13 @@ -- -- > logCrash = traverse_ logError . toCrashReason -- -toCrashReason :: ExitReason x -> Maybe T.Text +toCrashReason :: Interrupt x -> Maybe T.Text toCrashReason e | isCrash e = Just (T.pack (show e)) | otherwise = Nothing --- | Log the 'ExitReason's +-- | Log the 'Interrupt's logProcessExit - :: forall e x . (Member Logs e, HasCallStack) => ExitReason x -> Eff e () + :: forall e x . (Member Logs e, HasCallStack) => Interrupt x -> Eff e () logProcessExit (toCrashReason -> Just ex) = withFrozenCallStack (logError ex) logProcessExit ex = withFrozenCallStack (logDebug (fromString (show ex))) @@ -727,7 +724,7 @@ :: forall q r v . (SetMember Process (Process q) r, HasCallStack) => Process q (ResumeProcess v) - -> Eff r (Either (ExitReason 'Recoverable) v) + -> Eff r (Either (Interrupt 'Recoverable) v) executeAndResume processAction = do result <- send processAction case result of @@ -746,7 +743,7 @@ result <- send processAction case result of ResumeWith !value -> return value - Interrupted r -> send (Shutdown @q (NotRecovered r)) + Interrupted r -> send (Shutdown @q (interruptToExit r)) -- | Execute a 'Process' action and resume the process, exit -- the process when an 'Interrupts' was raised. Use 'executeAndResume' to catch @@ -809,7 +806,7 @@ :: forall r q . (SetMember Process (Process q) r, HasCallStack, Member Interrupts r) => ProcessId - -> ExitReason 'NoRecovery + -> Interrupt 'NoRecovery -> Eff r () sendShutdown pid s = pid `deepseq` s `deepseq` executeAndResumeOrThrow (SendShutdown pid s) @@ -821,15 +818,15 @@ :: forall r q . (SetMember Process (Process q) r, HasCallStack, Member Interrupts r) => ProcessId - -> InterruptReason + -> Interrupt 'Recoverable -> Eff r () sendInterrupt pid s = pid `deepseq` s `deepseq` executeAndResumeOrThrow (SendInterrupt pid s) -- | Start a new process, the new process will execute an effect, the function -- will return immediately with a 'ProcessId'. If the new process is --- interrupted, the process will 'Shutdown' with the 'InterruptReason' --- wrapped in 'NotRecovered'. For specific use cases it might be better to use +-- interrupted, the process will 'Shutdown' with the 'Interrupt' +-- wrapped in 'interruptToExit'. For specific use cases it might be better to use -- 'spawnRaw'. spawn :: forall r q @@ -940,14 +937,14 @@ -- Only the messages of the given type will be received. -- If the process is interrupted by an exception of by a 'SendShutdown' from -- another process, with an exit reason that satisfies 'isRecoverable', then --- the callback will be invoked with @'Left' 'ExitReason'@, otherwise the +-- the callback will be invoked with @'Left' 'Interrupt'@, otherwise the -- process will be exited with the same reason using 'exitBecause'. -- See also 'ReceiveSelectedMessage' for more documentation. receiveSelectedLoop :: forall r q a endOfLoopResult . (SetMember Process (Process q) r, HasCallStack) => MessageSelector a - -> (Either InterruptReason a -> Eff r (Maybe endOfLoopResult)) + -> (Either (Interrupt 'Recoverable) a -> Eff r (Maybe endOfLoopResult)) -> Eff r endOfLoopResult receiveSelectedLoop selector handlers = do mReq <- send (ReceiveSelectedMessage @q @a selector) @@ -961,7 +958,7 @@ receiveAnyLoop :: forall r q endOfLoopResult . (SetMember Process (Process q) r, HasCallStack) - => (Either InterruptReason StrictDynamic -> Eff r (Maybe endOfLoopResult)) + => (Either (Interrupt 'Recoverable) StrictDynamic -> Eff r (Maybe endOfLoopResult)) -> Eff r endOfLoopResult receiveAnyLoop = receiveSelectedLoop selectAnyMessage @@ -970,7 +967,7 @@ receiveLoop :: forall r q a endOfLoopResult . (SetMember Process (Process q) r, HasCallStack, NFData a, Typeable a) - => (Either InterruptReason a -> Eff r (Maybe endOfLoopResult)) + => (Either (Interrupt 'Recoverable) a -> Eff r (Maybe endOfLoopResult)) -> Eff r endOfLoopResult receiveLoop = receiveSelectedLoop selectMessage @@ -1077,13 +1074,13 @@ } deriving (Typeable, Generic, Eq, Ord) --- | Make an 'InterruptReason' for a 'ProcessDown' message. +-- | Make an 'Interrupt' for a 'ProcessDown' message. -- -- For example: @doSomething >>= either (interrupt . becauseProcessIsDown) return@ -- -- @since 0.12.0 -becauseProcessIsDown :: ProcessDown -> InterruptReason -becauseProcessIsDown = ProcessNotRunning . monitoredProcess . downReference +becauseProcessIsDown :: ProcessDown -> Interrupt 'Recoverable +becauseProcessIsDown = OtherProcessNotRunning . monitoredProcess . downReference instance NFData ProcessDown @@ -1108,7 +1105,7 @@ -- | Connect the calling process to another process, such that -- if one of the processes crashes (i.e. 'isCrash' returns 'True'), the other --- is shutdown with the 'ExitReason' 'LinkedProcessCrashed'. +-- is shutdown with the 'Interrupt' 'LinkedProcessCrashed'. -- -- @since 0.12.0 linkProcess @@ -1128,11 +1125,11 @@ -> Eff r () unlinkProcess = executeAndResumeOrThrow . Unlink . force --- | Exit the process with a 'ExitReason'. +-- | Exit the process with a 'Interrupt'. exitBecause :: forall r q a . (HasCallStack, SetMember Process (Process q) r) - => ExitReason 'NoRecovery + => Interrupt 'NoRecovery -> Eff r a exitBecause = send . Shutdown @q . force @@ -1147,7 +1144,7 @@ . (HasCallStack, SetMember Process (Process q) r) => String -> Eff r a -exitWithError = exitBecause . NotRecovered . ProcessError +exitWithError = exitBecause . interruptToExit . ErrorInterrupt -- | Each process is identified by a single process id, that stays constant -- throughout the life cycle of a process. Also, message sending relies on these 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 @@ -101,7 +101,7 @@ data SchedulerState = SchedulerState { _nextPid :: TVar ProcessId , _processTable :: TVar (Map ProcessId ProcessInfo) - , _processCancellationTable :: TVar (Map ProcessId (Async (ExitReason 'NoRecovery))) + , _processCancellationTable :: TVar (Map ProcessId (Async (Interrupt 'NoRecovery))) , _processMonitors :: TVar (Set (MonitorReference, ProcessId)) -- ^ Set of monitors and monitor owners , _nextMonitorIndex :: TVar Int } @@ -142,7 +142,7 @@ (Set.insert (monitorRef, owner)) else let processDownMessage = - ProcessDown monitorRef (SomeExitReason (ProcessNotRunning target)) + ProcessDown monitorRef (SomeExitReason (OtherProcessNotRunning target)) in enqueueMessageOtherProcess owner (toStrictDynamic processDownMessage) schedulerState @@ -272,18 +272,18 @@ handleProcess :: (HasCallStack) => ProcessInfo - -> Eff ProcEff (ExitReason 'NoRecovery) - -> Eff SchedulerIO (ExitReason 'NoRecovery) + -> Eff ProcEff (Interrupt 'NoRecovery) + -> Eff SchedulerIO (Interrupt 'NoRecovery) handleProcess myProcessInfo actionToRun = fix - (handle_relay' singleStep (const . const (return ExitNormally))) + (handle_relay' singleStep (\er _nextRef -> return er)) actionToRun 0 where singleStep - :: (Eff ProcEff xx -> (Int -> Eff SchedulerIO (ExitReason 'NoRecovery))) + :: (Eff ProcEff xx -> (Int -> Eff SchedulerIO (Interrupt 'NoRecovery))) -> Arrs ProcEff x xx -> Process SchedulerIO x - -> (Int -> Eff SchedulerIO (ExitReason 'NoRecovery)) + -> (Int -> Eff SchedulerIO (Interrupt 'NoRecovery)) singleStep k q p !nextRef = stepProcessInterpreter nextRef p @@ -310,8 +310,8 @@ diskontinueWith :: forall a . HasCallStack - => Arr SchedulerIO (ExitReason 'NoRecovery) a - -> Arr SchedulerIO (ExitReason 'NoRecovery) a + => Arr SchedulerIO (Interrupt 'NoRecovery) a + -> Arr SchedulerIO (Interrupt 'NoRecovery) a diskontinueWith diskontinue !reason = do setMyProcessState ProcessShuttingDown diskontinue reason @@ -321,7 +321,7 @@ => Int -> Process SchedulerIO v -> (Int -> Arr SchedulerIO v a) - -> Arr SchedulerIO (ExitReason 'NoRecovery) a + -> Arr SchedulerIO (Interrupt 'NoRecovery) a -> Eff SchedulerIO a stepProcessInterpreter !nextRef !request kontinue diskontinue = tryTakeNextShutdownRequest >>= maybe @@ -336,13 +336,11 @@ tryTakeNextShutdownRequest = lift (atomically (tryTakeNextShutdownRequestSTM myMessageQVar)) onShutdownRequested shutdownRequest = do - logDebug ("shutdown requested: " <> T.pack (show shutdownRequest)) setMyProcessState ProcessShuttingDown interpretRequestAfterShutdownRequest (diskontinueWith diskontinue) shutdownRequest request onInterruptRequested interruptRequest = do - logDebug ("interrupt requested: " <> T.pack (show interruptRequest)) setMyProcessState ProcessShuttingDown interpretRequestAfterInterruptRequest (kontinueWith kontinue nextRef) (diskontinueWith diskontinue) @@ -361,8 +359,8 @@ interpretRequestAfterShutdownRequest :: forall v a . HasCallStack - => Arr SchedulerIO (ExitReason 'NoRecovery) a - -> ExitReason 'NoRecovery + => Arr SchedulerIO (Interrupt 'NoRecovery) a + -> Interrupt 'NoRecovery -> Process SchedulerIO v -> Eff SchedulerIO a interpretRequestAfterShutdownRequest diskontinue shutdownRequest = \case @@ -387,8 +385,8 @@ :: forall v a . HasCallStack => Arr SchedulerIO v a - -> Arr SchedulerIO (ExitReason 'NoRecovery) a - -> ExitReason 'Recoverable + -> Arr SchedulerIO (Interrupt 'NoRecovery) a + -> Interrupt 'Recoverable -> Process SchedulerIO v -> Eff SchedulerIO a interpretRequestAfterInterruptRequest kontinue diskontinue interruptRequest = @@ -415,7 +413,7 @@ :: forall v a . HasCallStack => (Int -> Arr SchedulerIO v a) - -> Arr SchedulerIO (ExitReason 'NoRecovery) a + -> Arr SchedulerIO (Interrupt 'NoRecovery) a -> Int -> Process SchedulerIO v -> Eff SchedulerIO a @@ -441,8 +439,9 @@ >>= kontinue nextRef . ResumeWith . fst - ReceiveSelectedMessage f -> - interpretReceive f >>= either diskontinue (kontinue nextRef) + ReceiveSelectedMessage f -> do + recvRes <- interpretReceive f + either diskontinue (kontinue nextRef) recvRes FlushMessages -> interpretFlush >>= kontinue nextRef SelfPid -> kontinue nextRef (ResumeWith myPid) MakeReference -> kontinue (nextRef + 1) (ResumeWith nextRef) @@ -505,7 +504,7 @@ >>= lift . atomically . enqueueMessageOtherProcess toPid msg - interpretSendShutdownOrInterrupt !toPid !msg = + interpretSendShutdownOrInterrupt !toPid !msg = do setMyProcessState (either (const ProcessBusySendingShutdown) (const ProcessBusySendingInterrupt) @@ -524,7 +523,7 @@ return (ResumeWith (toList (myMessageQ ^. incomingMessages))) interpretReceive :: MessageSelector b - -> Eff SchedulerIO (Either (ExitReason 'NoRecovery) (ResumeProcess b)) + -> Eff SchedulerIO (Either (Interrupt 'NoRecovery) (ResumeProcess b)) interpretReceive f = do setMyProcessState ProcessBusyReceiving lift $ atomically $ do @@ -575,7 +574,7 @@ :: (HasCallStack) => Maybe ProcessInfo -> Eff ProcEff () - -> Eff SchedulerIO (ProcessId, Async (ExitReason 'NoRecovery)) + -> Eff SchedulerIO (ProcessId, Async (Interrupt 'NoRecovery)) spawnNewProcess mLinkedParent mfa = do schedulerState <- getSchedulerState procInfo <- allocateProcInfo schedulerState @@ -583,12 +582,6 @@ procAsync <- doForkProc procInfo schedulerState return (procInfo ^. processId, procAsync) where - linkToParent toProcInfo parent = do - let toPid = toProcInfo ^. processId - parentPid = parent ^. processId - lift $ atomically $ do - modifyTVar' (toProcInfo ^. processLinks) (Set.insert parentPid) - modifyTVar' (parent ^. processLinks) (Set.insert toPid) allocateProcInfo schedulerState = lift (atomically (do @@ -601,13 +594,20 @@ return procInfo ) ) + linkToParent toProcInfo parent = do + let toPid = toProcInfo ^. processId + parentPid = parent ^. processId + logDebug' ("linked to new child: " <> show toPid) + lift $ atomically $ do + modifyTVar' (toProcInfo ^. processLinks) (Set.insert parentPid) + modifyTVar' (parent ^. processLinks) (Set.insert toPid) logAppendProcInfo pid = let addProcessId = over lmProcessId (maybe (Just (T.pack (printf "% 9s" (show pid)))) Just) in censorLogs @IO addProcessId triggerProcessLinksAndMonitors - :: ProcessId -> ExitReason e -> TVar (Set ProcessId) -> Eff SchedulerIO () + :: ProcessId -> Interrupt e -> TVar (Set ProcessId) -> Eff SchedulerIO () triggerProcessLinksAndMonitors !pid !reason !linkSetVar = do schedulerState <- getSchedulerState lift $ atomically $ triggerAndRemoveMonitor pid @@ -630,12 +630,12 @@ then do writeTVar linkedLinkSetVar (Set.delete pid linkedLinkSet) - when - (exitSeverity == Crash) - (modifyTVar' linkedMsgQVar + if exitSeverity == Crash then do + modifyTVar' linkedMsgQVar (shutdownRequests ?~ msg) - ) - return (Right linkedPid) + return (Right (Left linkedPid)) + else + return (Right (Right linkedPid)) else return (Left linkedPid) linkedPids <- lift (atomically @@ -649,13 +649,16 @@ traverse_ (logDebug . either (("linked process no found: " <>) . T.pack . show) - (("sent shutdown to linked process: " <>) . T.pack . show) + (either + (("linked process crashed, interrupting linked process: " <>) . T.pack . show) + (("linked process exited peacefully, not sending shutdown to linked process: " <>) . T.pack . show) + ) ) res doForkProc :: ProcessInfo -> SchedulerState - -> Eff SchedulerIO (Async (ExitReason 'NoRecovery)) + -> Eff SchedulerIO (Async (Interrupt 'NoRecovery)) doForkProc procInfo schedulerState = control (\inScheduler -> do let cancellationsVar = schedulerState ^. processCancellationTable @@ -696,9 +699,12 @@ ) where exitReasonFromException exc = case Safe.fromException exc of - Just Async.AsyncCancelled -> Killed - Nothing -> UnexpectedException (prettyCallStack callStack) - (Safe.displayException exc) + Just Async.AsyncCancelled -> ExitProcessCancelled + Nothing -> ExitUnhandledError ( "runtime exception: " + <> T.pack (prettyCallStack callStack) + <> " " + <> T.pack (Safe.displayException exc) + ) logExitAndTriggerLinksAndMonitors reason pid = do triggerProcessLinksAndMonitors pid reason (procInfo ^. processLinks) logProcessExit reason diff --git a/src/Control/Eff/Concurrent/Process/Interactive.hs b/src/Control/Eff/Concurrent/Process/Interactive.hs --- a/src/Control/Eff/Concurrent/Process/Interactive.hs +++ b/src/Control/Eff/Concurrent/Process/Interactive.hs @@ -44,6 +44,7 @@ import Control.Monad import Data.Foldable import Data.Typeable ( Typeable ) +import Data.Type.Pretty import Control.DeepSeq import System.Timeout @@ -132,7 +133,11 @@ -- | Combination of 'submit' and 'cast'. submitCast :: forall o r - . (SetMember Lift (Lift IO) r, Typeable o, NFData (Api o 'Asynchronous), Member Interrupts r) + . ( SetMember Lift (Lift IO) r + , Typeable o + , PrettyTypeShow (ToPretty o) + , NFData (Api o 'Asynchronous) + , Member Interrupts r) => SchedulerSession r -> Server o -> Api o 'Asynchronous @@ -144,6 +149,7 @@ :: forall o q r . ( SetMember Lift (Lift IO) r , Typeable o + , PrettyTypeShow (ToPretty o) , Typeable q , NFData q , Show q 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 @@ -133,7 +133,7 @@ pt <- use msgQs if Map.member target pt then monitors %= Set.insert (mref, owner) - else let pdown = ProcessDown mref (SomeExitReason (ProcessNotRunning target)) + else let pdown = ProcessDown mref (SomeExitReason (OtherProcessNotRunning target)) in State.modify' (enqueueMsg owner (toStrictDynamic pdown)) return mref @@ -151,7 +151,7 @@ in State.modify' (enqueueMsg owner (toStrictDynamic pdown) . removeMonitoring mr) ) -addLink :: ProcessId -> ProcessId -> STS m r -> (Maybe InterruptReason, STS m r) +addLink :: ProcessId -> ProcessId -> STS m r -> (Maybe (Interrupt 'Recoverable), STS m r) addLink fromPid toPid = State.runState $ do hasToPid <- use (msgQs . to (Map.member toPid)) if hasToPid @@ -173,7 +173,7 @@ kontinue :: STS r m -> (ResumeProcess a -> Eff r a1) -> a -> m a1 kontinue sts k x = (sts ^. runEff) (k (ResumeWith x)) -diskontinue :: STS r m -> (ResumeProcess v -> Eff r a) -> InterruptReason -> m a +diskontinue :: STS r m -> (ResumeProcess v -> Eff r a) -> Interrupt 'Recoverable -> m a diskontinue sts k e = (sts ^. runEff) (k (Interrupted e)) -- ----------------------------------------------------------------------------- @@ -186,7 +186,7 @@ -- @since 0.3.0.2 schedulePure :: Eff (InterruptableProcess '[Logs, LogWriterReader PureLogWriter]) a - -> Either (ExitReason 'NoRecovery) a + -> Either (Interrupt 'NoRecovery) a schedulePure e = run (scheduleM (withSomeLogging @PureLogWriter) (return ()) e) -- | Invoke 'scheduleM' with @lift 'Control.Concurrent.yield'@ as yield effect. @@ -197,7 +197,7 @@ :: MonadIO m => (forall b . Eff r b -> Eff '[Lift m] b) -> Eff (InterruptableProcess r) a - -> m (Either (ExitReason 'NoRecovery) a) + -> m (Either (Interrupt 'NoRecovery) a) scheduleIO r = scheduleM (runLift . r) (liftIO yield) -- | Invoke 'scheduleM' with @lift 'Control.Concurrent.yield'@ as yield effect. @@ -207,7 +207,7 @@ scheduleMonadIOEff :: MonadIO (Eff r) => Eff (InterruptableProcess r) a - -> Eff r (Either (ExitReason 'NoRecovery) a) + -> Eff r (Either (Interrupt 'NoRecovery) a) scheduleMonadIOEff = -- schedule (lift yield) scheduleM id (liftIO yield) @@ -223,7 +223,7 @@ :: HasCallStack => LogWriter IO -> Eff (InterruptableProcess LoggingAndIo) a - -> IO (Either (ExitReason 'NoRecovery) a) + -> IO (Either (Interrupt 'NoRecovery) a) scheduleIOWithLogging h = scheduleIO (withLogging h) -- | Handle the 'Process' effect, as well as all lower effects using an effect handler function. @@ -250,7 +250,7 @@ -- @r@. E.g. if @Lift IO@ is present, this might be: -- @lift 'Control.Concurrent.yield'. -> Eff (InterruptableProcess r) a - -> m (Either (ExitReason 'NoRecovery) a) + -> m (Either (Interrupt 'NoRecovery) a) scheduleM r y e = do c <- runAsCoroutinePure r (provideInterruptsShutdown e) handleProcess (initStsMainProcess r y) (Seq.singleton (c, 0)) @@ -269,8 +269,8 @@ -> (ResumeProcess ProcessId -> Eff r (OnYield r a)) -> OnYield r a OnDone :: !a -> OnYield r a - OnShutdown :: ExitReason 'NoRecovery -> OnYield r a - OnInterrupt :: ExitReason 'Recoverable + OnShutdown :: Interrupt 'NoRecovery -> OnYield r a + OnInterrupt :: Interrupt 'Recoverable -> (ResumeProcess b -> Eff r (OnYield r a)) -> OnYield r a OnSend :: !ProcessId -> !StrictDynamic @@ -282,9 +282,9 @@ :: ProcessId -> (ResumeProcess (Maybe ProcessState) -> Eff r (OnYield r a)) -> OnYield r a - OnSendShutdown :: !ProcessId -> ExitReason 'NoRecovery + OnSendShutdown :: !ProcessId -> Interrupt 'NoRecovery -> (ResumeProcess () -> Eff r (OnYield r a)) -> OnYield r a - OnSendInterrupt :: !ProcessId -> ExitReason 'Recoverable + OnSendInterrupt :: !ProcessId -> Interrupt 'Recoverable -> (ResumeProcess () -> Eff r (OnYield r a)) -> OnYield r a OnMakeReference :: (ResumeProcess Int -> Eff r (OnYield r a)) -> OnYield r a OnMonitor @@ -359,9 +359,9 @@ :: Monad m => STS r m -> Seq (OnYield r finalResult, ProcessId) - -> m (Either (ExitReason 'NoRecovery) finalResult) + -> m (Either (Interrupt 'NoRecovery) finalResult) handleProcess _sts Empty = - return $ Left (NotRecovered (ProcessError "no main process")) + return $ Left (interruptToExit (ErrorInterrupt "no main process")) handleProcess sts allProcs@((!processState, !pid) :<| rest) = let handleExit res = @@ -371,9 +371,18 @@ let (downPids, stsNew) = removeLinksTo pid sts linkedPids = filter (/= pid) downPids reason = LinkedProcessCrashed pid - unlinkLoop dPidRest ps = foldM (\ps' dPid -> sendInterruptToOtherPid dPid reason ps') ps dPidRest - let allProcsWithoutPid = Seq.filter (\(_, p) -> p /= pid) rest - nextTargets <- unlinkLoop linkedPids allProcsWithoutPid + unlinkLoop dPidRest ps = foldM sendInterruptOrNot ps dPidRest + where + sendInterruptOrNot ps' dPid = + case res of + Right _ -> return ps' + Left er -> + case toExitSeverity er of + NormalExit -> return ps' + Crash -> sendInterruptToOtherPid dPid reason ps' + + let allButMe = Seq.filter (\(_, p) -> p /= pid) rest + nextTargets <- unlinkLoop linkedPids allButMe handleProcess (dropMsgQ pid @@ -451,12 +460,12 @@ recv@(OnRecv messageSelector k) -> case receiveMsg pid messageSelector sts of Nothing -> do - nextK <- diskontinue sts k (ProcessError (show pid ++ " has no message queue")) + nextK <- diskontinue sts k (ErrorInterrupt (show pid ++ " has no message queue")) handleProcess sts (rest :|> (nextK, pid)) Just Nothing -> if Seq.length rest == 0 then do - nextK <- diskontinue sts k (ProcessError (show pid ++ " deadlocked")) + nextK <- diskontinue sts k (ErrorInterrupt (show pid ++ " deadlocked")) handleProcess sts (rest :|> (nextK, pid)) else handleProcess sts (rest :|> (recv, pid)) Just (Just (result, newSts)) -> do diff --git a/src/Control/Eff/Concurrent/Process/Timer.hs b/src/Control/Eff/Concurrent/Process/Timer.hs --- a/src/Control/Eff/Concurrent/Process/Timer.hs +++ b/src/Control/Eff/Concurrent/Process/Timer.hs @@ -5,7 +5,7 @@ -- -- @since 0.12.0 module Control.Eff.Concurrent.Process.Timer - ( Timeout(fromTimeoutMicros) + ( Timeout(TimeoutMicros, fromTimeoutMicros) , TimerReference() , TimerElapsed(fromTimerElapsed) , sendAfter diff --git a/src/Control/Eff/Log/MessageRenderer.hs b/src/Control/Eff/Log/MessageRenderer.hs --- a/src/Control/Eff/Log/MessageRenderer.hs +++ b/src/Control/Eff/Log/MessageRenderer.hs @@ -118,14 +118,13 @@ renderLogMessageSrcLoc :: LogMessageRenderer (Maybe T.Text) renderLogMessageSrcLoc = view ( lmSrcLoc - . (to + . to (fmap (\sl -> T.pack $ printf "at %s:%i" (takeFileName (srcLocFile sl)) (srcLocStartLine sl) ) ) - ) ) @@ -156,10 +155,11 @@ -- | Render a 'LogMessage' human readable, for console logging renderLogMessageConsoleLog :: LogMessageRenderer T.Text -renderLogMessageConsoleLog l@(MkLogMessage _ _ ts _ _ _ _ sd _ _ _) = +renderLogMessageConsoleLog l@(MkLogMessage _ _ ts _ _ pd _ sd _ _ _) = T.unwords $ filter (not . T.null) [ view (lmSeverity . to (T.pack . show)) l + , fromMaybe " no proc " pd , maybe "" (renderLogMessageTime rfc5424Timestamp) ts , renderLogMessageBodyFixWidth l , if null sd then "" else T.concat (renderSdElement <$> sd) diff --git a/test/ForkIOScheduler.hs b/test/ForkIOScheduler.hs --- a/test/ForkIOScheduler.hs +++ b/test/ForkIOScheduler.hs @@ -43,7 +43,7 @@ case resultOrError of Left _down -> lift (atomically (putTMVar aVar False)) Right () -> withMonitor p1 $ \ref -> do - sendShutdown p1 (NotRecovered (ProcessError "test 123")) + sendShutdown p1 (interruptToExit (ErrorInterrupt "test 123")) _down <- receiveSelectedMessage (selectProcessDown ref) lift (atomically (putTMVar aVar True)) ) diff --git a/test/ProcessBehaviourTestCases.hs b/test/ProcessBehaviourTestCases.hs --- a/test/ProcessBehaviourTestCases.hs +++ b/test/ProcessBehaviourTestCases.hs @@ -1,8 +1,6 @@ module ProcessBehaviourTestCases where -import Data.List ( sort ) -import Data.Foldable ( traverse_ ) -import Data.Typeable +import Common import Control.Exception import Control.Concurrent import Control.Concurrent.STM @@ -23,18 +21,22 @@ import Control.Eff.LogWriter.IO import Control.Eff.Loop import Control.Monad -import Test.Tasty -import Test.Tasty.HUnit -import Common import Control.Applicative -import Data.Void import Control.Lens (view) import Control.DeepSeq +import Data.List ( sort ) +import Data.Foldable ( traverse_ ) +import Data.Typeable +import Data.Type.Pretty +import Data.Void +import Test.Tasty +import Test.Tasty.HUnit import GHC.Generics (Generic) -testInterruptReason :: InterruptReason -testInterruptReason = ProcessError "test interrupt" +testInterruptReason :: Interrupt 'Recoverable +testInterruptReason = ErrorInterrupt "test interrupt" + test_forkIo :: TestTree test_forkIo = setTravisTestOptions $ withTestLogC (\c -> @@ -84,6 +86,8 @@ data ReturnToSender deriving Typeable +type instance ToPretty ReturnToSender = PutStr "ReturnToSender" + data instance Api ReturnToSender r where ReturnToSender :: ProcessId -> String -> Api ReturnToSender ('Synchronous Bool) StopReturnToSender :: Api ReturnToSender ('Synchronous ()) @@ -595,7 +599,7 @@ , (howToExit, doExit ) <- [ ("normally" , void exitNormally) , ("simply returning", return ()) - , ("raiseError", void (interrupt (ProcessError "test error raised"))) + , ("raiseError", void (interrupt (ErrorInterrupt "test error raised"))) , ("exitWithError" , void (exitWithError "test error exit")) , ( "sendShutdown to self" , do @@ -617,16 +621,16 @@ [ testCase "... self" $ applySchedulerFactory schedulerFactory $ do me <- self void $ send (SendShutdown @r me ExitNormally) - interrupt (ProcessError "sendShutdown must not return") + interrupt (ErrorInterrupt "sendShutdown must not return") , testCase "sendInterrupt to self" $ scheduleAndAssert schedulerFactory $ \assertEff -> do me <- self - r <- send (SendInterrupt @r me (ProcessError "123")) + r <- send (SendInterrupt @r me (ErrorInterrupt "123")) assertEff "Interrupted must be returned" (case r of - Interrupted (ProcessError "123") -> True + Interrupted (ErrorInterrupt "123") -> True _ -> False ) , testGroup @@ -664,7 +668,7 @@ untilInterrupted (SelfPid @r) void (sendMessage me ("OK" :: String)) ) - void (sendInterrupt other (ProcessError "testError")) + void (sendInterrupt other (ErrorInterrupt "testError")) (a :: String) <- receiveMessage assertEff "" (a == "OK") , testCase "while it is spawning" @@ -694,8 +698,8 @@ , testCase "handleInterrupt handles my own interrupts" $ scheduleAndAssert schedulerFactory $ \assertEff -> - handleInterrupts (\e -> return (ProcessError "test" == e)) - (interrupt (ProcessError "test") >> return False) + handleInterrupts (\e -> return (ErrorInterrupt "test" == e)) + (interrupt (ErrorInterrupt "test") >> return False) >>= assertEff "exception handler not invoked" ] ] @@ -747,7 +751,7 @@ (lift . (@?= LinkedProcessCrashed foo)) (do linkProcess foo - sendShutdown foo Killed + sendShutdown foo ExitProcessCancelled void (receiveMessage @Void) ) , testCase "link multiple times" @@ -764,7 +768,7 @@ linkProcess foo linkProcess foo linkProcess foo - sendShutdown foo Killed + sendShutdown foo ExitProcessCancelled void (receiveMessage @Void) ) , testCase "unlink multiple times" @@ -783,15 +787,62 @@ unlinkProcess foo unlinkProcess foo withMonitor foo $ \ref -> do - sendShutdown foo Killed + sendShutdown foo ExitProcessCancelled void (receiveSelectedMessage (selectProcessDown ref)) ) , testCase "spawnLink" $ applySchedulerFactory schedulerFactory $ do let foo = void (receiveMessage @Void) - handleInterrupts (\er -> lift (isBecauseDown Nothing er @? show er)) $ do + handleInterrupts (\er -> lift (isProcessDownInterrupt Nothing er @? show er)) $ do x <- spawnLink foo - sendShutdown x Killed + sendShutdown x ExitProcessCancelled void (receiveMessage @Void) + , testCase "spawnLink and child exits by returning from spawn" $ applySchedulerFactory schedulerFactory $ do + me <- self + u <- spawn $ do + logCritical "unlinked child started" + l <- spawnLink $ do + logCritical "linked child started" + () <- receiveMessage + logCritical "linked child done" + sendMessage me l + x <- receiveAnyMessage + logCritical' ("got: " <> show x) + l <- receiveMessage + _ <- monitor l + sendMessage l () + pL@(ProcessDown _ _) <- receiveMessage + logCritical' ("linked process down: " <> show pL) + _ <- monitor u + mpU <- receiveAfter (TimeoutMicros 1000) + case mpU of + Just (pU@(ProcessDown _ _)) -> + error ("unlinked process down: " <> show pU) + Nothing -> logInfo "passed" + + , testCase "spawnLink and child exits via exitWithError" $ applySchedulerFactory schedulerFactory $ do + me <- self + u <- spawn $ do + logCritical "unlinked child started" + l <- spawnLink $ do + logCritical "linked child started" + () <- receiveMessage + logCritical "linked child done" + exitWithError "linked process test error" + sendMessage me l + x <- receiveAnyMessage + logCritical' ("got: " <> show x) + l <- receiveMessage + _ <- monitor l + sendMessage l () + pL@(ProcessDown _ _) <- receiveMessage + logCritical' ("linked process down: " <> show pL) + _ <- monitor u + mpU <- receiveAfter (TimeoutMicros 1000) + case mpU of + Just (pU@(ProcessDown _ _)) -> + logInfo' ("unlinked process down: " <> show pU) + Nothing -> error "linked process not exited!" + , testCase "ignore normal exit" $ applySchedulerFactory schedulerFactory $ do @@ -819,7 +870,7 @@ sendShutdown x ExitNormally void (receiveSelectedMessage (selectProcessDown xRef)) handleInterrupts (lift . (LinkedProcessCrashed linker @=?)) $ do - sendShutdown linker Killed + sendShutdown linker ExitProcessCancelled void (receiveMessage @Void) , testCase "unlink" $ applySchedulerFactory schedulerFactory $ do let @@ -842,7 +893,7 @@ (const (return ())) (do linkProcess foo1Pid - sendShutdown foo1Pid Killed + sendShutdown foo1Pid ExitProcessCancelled void (receiveMessage @Void) ) handleInterrupts @@ -881,7 +932,7 @@ let badPid = 132123 ref <- monitor badPid pd <- receiveSelectedMessage (selectProcessDown ref) - lift (downReason pd @?= SomeExitReason (ProcessNotRunning badPid)) + lift (downReason pd @?= SomeExitReason (OtherProcessNotRunning badPid)) lift (threadDelay 10000) , testCase "monitored process exit normally" $ applySchedulerFactory schedulerFactory @@ -916,9 +967,9 @@ $ do target <- spawn (receiveMessage >>= exitBecause) ref <- monitor target - sendMessage target Killed + sendMessage target ExitProcessCancelled pd <- receiveSelectedMessage (selectProcessDown ref) - lift (downReason pd @?= SomeExitReason Killed) + lift (downReason pd @?= SomeExitReason ExitProcessCancelled) lift (threadDelay 10000) , testCase "demonitored process killed" $ applySchedulerFactory schedulerFactory @@ -926,7 +977,7 @@ target <- spawn (receiveMessage >>= exitBecause) ref <- monitor target demonitor ref - sendMessage target Killed + sendMessage target ExitProcessCancelled me <- self spawn_ (lift (threadDelay 10000) >> sendMessage me ()) pd <- receiveSelectedMessage diff --git a/test/ServerForkIO.hs b/test/ServerForkIO.hs --- a/test/ServerForkIO.hs +++ b/test/ServerForkIO.hs @@ -11,9 +11,11 @@ import Test.Tasty.HUnit import Common import Data.Typeable hiding ( cast ) +import Data.Type.Pretty import Control.DeepSeq data TestServer deriving Typeable +type instance ToPretty TestServer = PutStr "test" data instance Api TestServer x where TestGetStringLength :: String -> Api TestServer ('Synchronous Int) @@ -65,7 +67,7 @@ hm :: MessageCallback TestServer InterruptableProcEff hm = handleCastsAndCalls onCast onCall where - onCast :: Api TestServer 'Asynchronous -> Eff InterruptableProcEff CallbackResult + onCast :: Api TestServer 'Asynchronous -> Eff InterruptableProcEff (CallbackResult 'Recoverable) onCast (TestSetNextDelay x) = do logDebug "relaying delay" cast target (TestSetNextDelay x) @@ -73,7 +75,7 @@ onCall :: (NFData l, Typeable l) => Api TestServer ('Synchronous l) - -> (Eff InterruptableProcEff (Maybe l, CallbackResult) -> k) + -> (Eff InterruptableProcEff (Maybe l, CallbackResult 'Recoverable) -> k) -> k onCall (TestGetStringLength s) runHandler = runHandler $ do logDebug "relaying get string length" @@ -91,7 +93,7 @@ hm :: MessageCallback TestServer (State Timeout ': InterruptableProcEff) hm = handleCastsAndCalls onCast onCall where - onCast :: Api TestServer 'Asynchronous -> Eff (State Timeout ': InterruptableProcEff) CallbackResult + onCast :: Api TestServer 'Asynchronous -> Eff (State Timeout ': InterruptableProcEff) (CallbackResult 'Recoverable) onCast (TestSetNextDelay x) = do logDebug' ("setting delay: " ++ show x) put x @@ -99,7 +101,7 @@ onCall :: (NFData l, Typeable l) => Api TestServer ('Synchronous l) - -> (Eff (State Timeout ': InterruptableProcEff) (Maybe l, CallbackResult) -> k) + -> (Eff (State Timeout ': InterruptableProcEff) (Maybe l, CallbackResult 'Recoverable) -> k) -> k onCall (TestGetStringLength s) runHandler = runHandler $ do logDebug' ("calculating string length: " ++ show s) diff --git a/test/SupervisorTests.hs b/test/SupervisorTests.hs new file mode 100644 --- /dev/null +++ b/test/SupervisorTests.hs @@ -0,0 +1,297 @@ +module SupervisorTests + ( test_Supervisor + ) where + +import Common +import Control.Concurrent (threadDelay) +import Control.DeepSeq +import Control.Eff +import Control.Eff.Concurrent +import Control.Eff.Concurrent.Api.Supervisor as Sup +import Control.Eff.Concurrent.Process.ForkIOScheduler as Scheduler +import Control.Eff.Concurrent.Process.Timer +import Data.Either (fromRight, isLeft, isRight) +import Data.Maybe (fromMaybe) +import Data.Text (pack) +import Data.Type.Pretty +import Data.Typeable (Typeable) +import Test.Tasty +import Test.Tasty.HUnit + +test_Supervisor :: HasCallStack => TestTree +test_Supervisor = + setTravisTestOptions $ + testGroup + "Supervisor" + (let startTestSup = startTestSupWith ExitWhenRequested (TimeoutMicros 500000) + startTestSupWith e t = Sup.startSupervisor (MkSupConfig (spawnTestApiProcess e) t) + spawnTestChild sup i = Sup.spawnChild sup i >>= either (lift . assertFailure . show) pure + in [ runTestCase "The supervisor starts and is shut down" $ do + outerSelf <- self + testWorker <- + spawn $ do + sup <- startTestSup + sendMessage outerSelf sup + () <- receiveMessage + sendMessage outerSelf () + () <- receiveMessage + Sup.stopSupervisor sup + unlinkProcess testWorker + sup <- receiveMessage :: Eff InterruptableProcEff (Sup.Sup Int (Server TestApi)) + supAliveAfter1 <- isSupervisorAlive sup + logInfo ("still alive 1: " <> pack (show supAliveAfter1)) + lift (supAliveAfter1 @=? True) + sendMessage testWorker () + () <- receiveMessage + supAliveAfter2 <- isSupervisorAlive sup + logInfo ("still alive 2: " <> pack (show supAliveAfter2)) + lift (supAliveAfter2 @=? True) + sendMessage testWorker () + _ <- monitor testWorker + d1@(ProcessDown _ _) <- receiveMessage + logInfo ("got test worker down: " <> pack (show d1)) + _ <- monitorSupervisor sup + d2@(ProcessDown _ _) <- receiveMessage + logInfo ("got supervisor down: " <> pack (show d2)) + supAliveAfterOwnerExited <- isSupervisorAlive sup + logInfo ("still alive after owner exited: " <> pack (show supAliveAfterOwnerExited)) + lift (supAliveAfterOwnerExited @=? False) + , testGroup + "Diagnostics" + [ runTestCase "When only time passes the diagnostics do not change" $ do + sup <- startTestSup + diag1 <- Sup.getDiagnosticInfo sup + lift (threadDelay 10000) + diag2 <- Sup.getDiagnosticInfo sup + lift (assertEqual "diagnostics should not differ: " diag1 diag2) + , runTestCase "When a child is started the diagnostics change" $ do + sup <- startTestSup + diag1 <- Sup.getDiagnosticInfo sup + logInfo ("got diagnostics: " <> diag1) + let childId = 1 + _child <- fromRight (error "failed to spawn child") <$> Sup.spawnChild sup childId + diag2 <- Sup.getDiagnosticInfo sup + logInfo ("got diagnostics: " <> diag2) + lift $ assertBool ("diagnostics should differ: " ++ show (diag1, diag2)) (diag1 /= diag2) + ] + , let childId = 1 + in testGroup + "Startup and shutdown" + [ runTestCase "When a supervisor is shut down, all children are shutdown" $ do + sup <- startTestSup + child <- spawnTestChild sup childId + let childPid = _fromServer child + supMon <- monitorSupervisor sup + childMon <- monitor childPid + isProcessAlive childPid >>= lift . assertBool "child process not running" + isSupervisorAlive sup >>= lift . assertBool "supervisor process not running" + call child (TestGetStringLength "123") >>= lift . assertEqual "child not working" 3 + stopSupervisor sup + d1@(ProcessDown mon1 er1) <- + fromMaybe (error "receive timeout 1") <$> receiveAfter (TimeoutMicros 1000000) + logInfo ("got process down: " <> pack (show d1)) + d2@(ProcessDown mon2 er2) <- + fromMaybe (error "receive timeout 2") <$> receiveAfter (TimeoutMicros 1000000) + logInfo ("got process down: " <> pack (show d2)) + case if mon1 == supMon && mon2 == childMon + then Right (er1, er2) + else if mon1 == childMon && mon2 == supMon + then Right (er2, er1) + else Left + ("unexpected monitor down: first: " <> show (mon1, er1) <> ", and then: " <> + show (mon2, er2) <> + ", supMon: " <> + show supMon <> + ", childMon: " <> + show childMon) of + Right (supER, childER) -> do + lift (assertEqual "bad supervisor exit reason" (SomeExitReason ExitNormally) supER) + lift (assertEqual "bad child exit reason" (SomeExitReason ExitNormally) childER) + Left x -> lift (assertFailure x) + , runTestCase + "When a supervisor is shut down, children that won't shutdown, are killed after some time" $ do + sup <- startTestSupWith IgnoreNormalExitRequest (TimeoutMicros 10000) + child <- spawnTestChild sup childId + let childPid = _fromServer child + supMon <- monitorSupervisor sup + childMon <- monitor childPid + isProcessAlive childPid >>= lift . assertBool "child process not running" + isSupervisorAlive sup >>= lift . assertBool "supervisor process not running" + call child (TestGetStringLength "123") >>= lift . assertEqual "child not working" 3 + stopSupervisor sup + d1@(ProcessDown mon1 er1) <- + fromMaybe (error "receive timeout 1") <$> receiveAfter (TimeoutMicros 1000000) + logInfo ("got process down: " <> pack (show d1)) + d2@(ProcessDown mon2 er2) <- + fromMaybe (error "receive timeout 2") <$> receiveAfter (TimeoutMicros 1000000) + logInfo ("got process down: " <> pack (show d2)) + case if mon1 == supMon && mon2 == childMon + then Right (er1, er2) + else if mon1 == childMon && mon2 == supMon + then Right (er2, er1) + else Left + ("unexpected monitor down: first: " <> show (mon1, er1) <> ", and then: " <> + show (mon2, er2) <> + ", supMon: " <> + show supMon <> + ", childMon: " <> + show childMon) of + Right (supER, childER) -> do + lift (assertEqual "bad supervisor exit reason" (SomeExitReason ExitNormally) supER) + lift (assertBool "bad child exit reason" (isLeft (fromSomeExitReason childER))) + Left x -> lift (assertFailure x) + ] + , let i = 123 + in testGroup + "Spawning and Using Children" + [ runTestCase + "When a supervisor is requested to start two children with the same id, an already started error is returned" $ do + sup <- startTestSup + c <- spawnTestChild sup i + x <- Sup.spawnChild sup i + case x of + Left (AlreadyStarted i' c') -> + lift $ do + assertEqual "bad pid returned" c c' + assertEqual "bad child id returned" 123 i' + _ -> lift (assertFailure "AlreadyStarted expected!") + , runTestCase "When a child is started it can be lookup up" $ do + sup <- startTestSup + c <- spawnTestChild sup i + c' <- Sup.lookupChild sup i >>= maybe (lift (assertFailure "child not found")) pure + lift (assertEqual "lookupChild returned wrong child" c c') + , runTestCase "When several children are started they can be lookup up and don't crash" $ do + sup <- startTestSup + c <- spawnTestChild sup i + someOtherChild <- spawnTestChild sup (i + 1) + c' <- Sup.lookupChild sup i >>= maybe (lift (assertFailure "child not found")) pure + lift (assertEqual "lookupChild returned wrong child" c c') + childStillRunning <- isProcessAlive (_fromServer c) + lift (assertBool "child not running" childStillRunning) + someOtherChildStillRunning <- isProcessAlive (_fromServer someOtherChild) + lift (assertBool "someOtherChild not running" someOtherChildStillRunning) + , let startTestSupAndChild = do + sup <- startTestSup + c <- spawnTestChild sup i + cm <- monitor (_fromServer c) + return (sup, cm) + in testGroup + "Stopping children" + [ runTestCase "When a child is started it can be stopped" $ do + (sup, cm) <- startTestSupAndChild + Sup.stopChild sup i >>= lift . assertBool "child not found" + (ProcessDown _ r) <- receiveSelectedMessage (selectProcessDown cm) + lift (assertEqual "bad exit reason" (SomeExitReason ExitNormally) r) + , runTestCase + "When a child is stopped but doesn't exit voluntarily, it is kill after some time" $ do + sup <- startTestSupWith IgnoreNormalExitRequest (TimeoutMicros 5000) + c <- spawnTestChild sup i + cm <- monitor (_fromServer c) + Sup.stopChild sup i >>= lift . assertBool "child not found" + (ProcessDown _ r) <- receiveSelectedMessage (selectProcessDown cm) + case r of + SomeExitReason (ExitUnhandledError _) -> return () + _ -> lift (assertFailure ("bad exit reason: " ++ show r)) + , runTestCase "When a stopChild is called with an unallocated ID, False is returned" $ do + (sup, _) <- startTestSupAndChild + Sup.stopChild sup (i + 1) >>= lift . assertBool "child not found" . not + , runTestCase "When a child is stopped, lookup won't find it" $ do + (sup, _) <- startTestSupAndChild + Sup.stopChild sup i >>= lift . assertBool "child not found" + x <- Sup.lookupChild sup i + lift (assertEqual "lookup should not find a child" Nothing x) + , runTestCase "When a child is stopped, the child id can be reused" $ do + (sup, _) <- startTestSupAndChild + Sup.stopChild sup i >>= lift . assertBool "child not found" + Sup.spawnChild sup i >>= lift . assertBool "id could not be reused" . isRight + ] + , let startTestSupAndChild = do + sup <- startTestSup + c <- spawnTestChild sup i + cm <- monitor (_fromServer c) + return (sup, c, cm) + in testGroup + "Child exit handling" + [ runTestCase "When a child exits normally, lookupChild will not find it" $ do + (sup, c, cm) <- startTestSupAndChild + cast c (TestInterruptWith NormalExitRequested) + (ProcessDown _ r) <- receiveSelectedMessage (selectProcessDown cm) + lift (assertEqual "bad exit reason" (SomeExitReason ExitNormally) r) + x <- Sup.lookupChild sup i + lift (assertEqual "lookup should not find a child" Nothing x) + , runTestCase "When a child exits with an error, lookupChild will not find it" $ do + (sup, c, cm) <- startTestSupAndChild + cast c (TestInterruptWith (ErrorInterrupt "test error reason")) + (ProcessDown _ _) <- receiveSelectedMessage (selectProcessDown cm) + x <- Sup.lookupChild sup i + lift (assertEqual "lookup should not find a child" Nothing x) + , runTestCase "When a child is interrupted from another process and dies, lookupChild will not find it" $ do + (sup, c, cm) <- startTestSupAndChild + sendInterrupt (_fromServer c) NormalExitRequested + (ProcessDown _ _) <- receiveSelectedMessage (selectProcessDown cm) + x <- Sup.lookupChild sup i + lift (assertEqual "lookup should not find a child" Nothing x) + , runTestCase "When a child is shutdown from another process and dies, lookupChild will not find it" $ do + (sup, c, cm) <- startTestSupAndChild + sendShutdown (_fromServer c) ExitProcessCancelled + (ProcessDown _ _) <- receiveSelectedMessage (selectProcessDown cm) + x <- Sup.lookupChild sup i + lift (assertEqual "lookup should not find a child" Nothing x) + ] + ] + ]) + +runTestCase :: TestName -> Eff InterruptableProcEff () -> TestTree +runTestCase msg = + testCase msg . + runLift . withTraceLogging "supervisor-test" local0 allLogMessages . Scheduler.schedule . handleInterrupts onInt + where + onInt = lift . assertFailure . show + +data TestApiServerMode + = IgnoreNormalExitRequest + | ExitWhenRequested + deriving (Eq) + +spawnTestApiProcess :: TestApiServerMode -> Sup.SpawnFun Int InterruptableProcEff (Server TestApi) +spawnTestApiProcess testMode tId = + spawnApiServer (handleCasts onCast <> handleCalls onCall ^: handleAnyMessages onInfo) (InterruptCallback onInterrupt) + where + onCast :: + Api TestApi 'Asynchronous -> Eff InterruptableProcEff (CallbackResult 'Recoverable) + onCast (TestInterruptWith i) = do + logInfo (pack (show tId) <> ": stopping with: " <> pack (show i)) + pure (StopServer i) + onCall :: + Api TestApi ('Synchronous r) -> (Eff InterruptableProcEff (Maybe r, CallbackResult 'Recoverable) -> x) -> x + onCall (TestGetStringLength str) runMe = + runMe $ do + logInfo (pack (show tId) <> ": calculating length of: " <> pack str) + pure (Just (length str), AwaitNext) + onInfo :: StrictDynamic -> Eff InterruptableProcEff (CallbackResult 'Recoverable) + onInfo sd = do + logDebug (pack (show tId) <> ": got some info: " <> pack (show sd)) + pure AwaitNext + onInterrupt x = do + logNotice (pack (show tId) <> ": " <> pack (show x)) + if testMode == IgnoreNormalExitRequest + then do + logNotice $ pack (show tId) <> ": ignoring normal exit request" + pure AwaitNext + else do + logNotice $ pack (show tId) <> ": exitting normally" + pure (StopServer (interruptToExit x)) + +data TestApi + deriving (Typeable) + +type instance ToPretty TestApi = PutStr "test" + +data instance Api TestApi x where + TestGetStringLength :: String -> Api TestApi ('Synchronous Int) + TestInterruptWith :: Interrupt 'Recoverable -> Api TestApi 'Asynchronous + deriving Typeable + +instance NFData (Api TestApi x) where + rnf (TestGetStringLength x) = rnf x + rnf (TestInterruptWith x) = rnf x