diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,30 @@ [1]: http://semver.org/spec/v2.0.0.html [2]: https://github.com/roman/capataz/releases +## v0.2.0.0 Bumblebee release + +**BREAKING CHANGES** + +* Allow the creation of Supervision Trees on Monad Transformer stacks 🎉 +* Move to latest stable stackage snapshot (lts-11.10) +* Remove `protolude` in favor of `rio` (closes #9) +* Add convinience module `Capataz` +* Add `Control.Concurrent.Capataz.Util` which provide `ProcessSpec` for normal use-cases +* Add `buildLogWorkerSpec` and `buildLogWorkerOptions` which create a dedicated supervised + thread that logs messages (sits on top of `rio` logger utilities) +* Add `terminateCapataz` and `terminateCapataz_` , these functions replace the + re-exported `runTeardown` +* Add re-export of `TeardownResult` from the `teardown` package +* Update the `capataz-simple-example` project to use `typed-process` and `rio` +* Add export for `WorkerName` and `SupervisorName` aliases +* Add new dependency `prettyprinter` +* Add new dependency `pretty-show` to prettify exceptions +* Add `Pretty` instance for `CapatazEvent` +* Add `Display` instance for `CapatazEvent` +* Add `ProcessThreadId` newtype to better render `ThreadId` values +* Add `buildWorkerOptions1` and `workerSpec1` (closes #17) +* Move to CircleCI + ## v0.1.0.1 * Bump bounds of `async` dependency diff --git a/README.md b/README.md --- a/README.md +++ b/README.md @@ -4,68 +4,10 @@ ## Table Of Contents -* [Raison d'etre](#raison-detre) +* [Installation](#installation) * [Documentation](#documentation) * [Development](#development) -## Raison d'être - -As time progresses, I've come to love developing concurrent applications in -Haskell, its API (STM, MVars, etc.) and light threading RTS bring a lot to the -table. There is another technology that is more famous than Haskell in -regards to concurrency, and that is Erlang, more specifically its OTP library. - -If you wonder why that is, you may need to look into the OTP library design, -actors systems (in general) provide an architecture that enables applications to -be tolerant to failure through the enforcement of communication via message -passing and by making use of a critical infrastructure piece called a *Supervisor*. - -After trying to replicate Erlang's behavior on Haskell applications by using the -[distributed-process](https://hackage.haskell.org/package/distributed-process) -library (a clone of OTP), and after implementing several (disposable) iterations -of actor systems in Haskell, I've settled with just this library, one that -provides a simple Supervisor API. - -This library is intended to be a drop-in replacement to `forkIO` invocations -throughout your codebase, the difference being, you'll need to do a bit more of -setup specifying supervision rules, and also pass along a reference of a -supervisor for every thread you fork. - -### Why not [distributed-process](https://hackage.haskell.org/package/distributed-process)? - -`distributed-process` is an impressive library, and brings many great utilities -if you need to develop applications that need to be distributed and reliable. -However, it is a heavyweight solution that will enforce serious changes to your -application. This library is intended to provide the reliability benefits of -`distributed-process`, without the constraints imposed by the *distributed* -part. - -### Why not a complete actor system? - -Actor systems are very pervasive, they impose specific design constraints on -your application which can be rather expensive. This library attempts to bring -some of the reliability benefits of actor systems without the "change all your -application to work with actors" part of the equation. - -That said, this library can serve as a basis for a more prominent library that -provides an opinionated Inter-Process communication scheme. If you happen to -attempt at doing exactly that, please let me know, I would love to learn about -such initiatives. - -### Why not [async](https://hackage.haskell.org/package/async)? - -`async` is a fabulous library that allows Applicative composition of small -asynchronous sub-routines into bigger ones and link errors between them. Given -this, `async` fits the bill perfectly for small operations that happen -concurrently, not necessarily for long living threads that need to be restarted. -This library attempts not to replace async's forte, but rather provides other -benefits not found in `async`, like automatically restarting threads with a given -recipe under error situations. - -## Documentation - -Documentation can be found [here](https://romanandreg.gitbooks.io/capataz/content/) - ## Installation [![Hackage](https://img.shields.io/hackage/v/capataz.svg)](https://img.shields.io/hackage/v/capataz.svg) @@ -88,18 +30,24 @@ - capataz ``` +## Documentation + +The documentation website can be found +[here](https://romanandreg.gitbooks.io/capataz/content/). Some relevant links: + +* [Purpose](https://romanandreg.gitbooks.io/capataz/content/purpose.html) +* [FAQ](https://romanandreg.gitbooks.io/capataz/content/FAQ.html) + ## Development -[![Build Status](https://travis-ci.org/roman/Haskell-capataz.svg?branch=master)](https://travis-ci.org/roman/Haskell-capataz) -[![Github](https://img.shields.io/github/commits-since/roman/haskell-capataz/v0.1.0.0.svg)](https://img.shields.io/github/commits-since/roman/haskell-capataz/v0.1.0.0.svg) +![Build Status](https://circleci.com/gh/roman/Haskell-teardown.svg?style=svg) +[![Github](https://img.shields.io/github/commits-since/roman/haskell-capataz/v0.2.0.0.svg)](https://img.shields.io/github/commits-since/roman/haskell-capataz/v0.2.0.0.svg) [![Hackage Dependencies](https://img.shields.io/hackage-deps/v/capataz.svg)](http://packdeps.haskellers.com/feed?needle=capataz) -Follow the [developer guidelines](https://romanandreg.gitbooks.io/capataz/content/developer-guidelines.html) +Follow the [developer guidelines](https://romanandreg.gitbooks.io/capataz/content/CONTRIBUTING.html) ## In future releases * Replace Protolude in favor of RIO * Documentation of performance analysis -* Documentation improvements -* capataz-dashboard package that provides web-ui with Supervisor statistics * Ensure unit tests always finish on all concurrent scenarios (dejafu experiment) diff --git a/capataz.cabal b/capataz.cabal --- a/capataz.cabal +++ b/capataz.cabal @@ -1,12 +1,13 @@ -name: capataz -version: 0.1.0.1 cabal-version: >=1.10 -build-type: Simple +name: capataz +version: 0.2.0.0 license: MIT license-file: LICENSE copyright: © 2018 Roman Gonzalez maintainer: open-source@roman-gonzalez.info +author: Roman Gonzalez stability: alpha (experimental) +tested-with: ghc ==8.0.1 ghc ==8.0.2 ghc ==8.2.1 homepage: https://github.com/roman/Haskell-capataz#readme bug-reports: https://github.com/roman/Haskell-capataz/issues synopsis: OTP-like supervision trees in Haskell @@ -28,8 +29,7 @@ * Complete telemetry on the sub-routine lifecycle of your application (start, error, restarts, shutdown) category: Control, Concurrency -author: Roman Gonzalez -tested-with: GHC ==8.0.1 GHC ==8.0.2 GHC ==8.2.1 +build-type: Simple extra-source-files: CHANGELOG.md README.md @@ -40,6 +40,7 @@ library exposed-modules: + Capataz Control.Concurrent.Capataz Control.Concurrent.Capataz.Event Control.Concurrent.Capataz.Lens @@ -49,57 +50,48 @@ Control.Concurrent.Capataz.Internal.Supervisor Control.Concurrent.Capataz.Internal.Util Control.Concurrent.Capataz.Internal.Worker - build-depends: - async >=2.1.1.1 && <2.3, - base >=4.10.1.0 && <4.11, - bytestring >=0.10.8.2 && <0.11, - data-default >=0.7.1.1 && <0.8, - microlens >=0.4.8.3 && <0.5, - protolude >=0.2.1 && <0.3, - safe-exceptions >=0.1.7.0 && <0.2, - stm >=2.4.4.1 && <2.5, - teardown >=0.3.0.0 && <0.4, - text >=1.2.3.0 && <1.3, - time >=1.8.0.2 && <1.9, - unordered-containers >=0.2.8.0 && <0.3, - uuid >=1.3.13 && <1.4, - vector >=0.12.0.1 && <0.13 - default-language: Haskell2010 hs-source-dirs: src other-modules: Control.Concurrent.Capataz.Internal.Process + Control.Concurrent.Capataz.Util Paths_capataz - ghc-options: -Wall -Wincomplete-uni-patterns -Wincomplete-record-updates + default-language: Haskell2010 + ghc-options: -Wall -Wincomplete-uni-patterns + -Wincomplete-record-updates + build-depends: + async >=2.1.1.1 && <2.3, + base ==4.*, + bytestring >=0.10.8, + pretty-show >=1.6.13, + prettyprinter >=1.1, + rio >=0.1.2.0, + teardown >=0.5.0.0, + time >=1.6.0, + uuid >=1.3 -test-suite capataz-test +test-suite capataz-test type: exitcode-stdio-1.0 main-is: Main.hs - build-depends: - async >=2.1.1.1 && <2.3, - base >=4.10.1.0 && <4.11, - bytestring >=0.10.8.2 && <0.11, - capataz -any, - data-default >=0.7.1.1 && <0.8, - microlens >=0.4.8.3 && <0.5, - pretty-show >=1.6.16 && <1.7, - protolude >=0.2.1 && <0.3, - safe-exceptions >=0.1.7.0 && <0.2, - stm >=2.4.4.1 && <2.5, - tasty >=0.12.0.1 && <0.13, - tasty-hunit >=0.10.0.1 && <0.11, - tasty-rerun >=1.1.9 && <1.2, - tasty-smallcheck >=0.8.1 && <0.9, - teardown >=0.3.0.0 && <0.4, - text >=1.2.3.0 && <1.3, - time >=1.8.0.2 && <1.9, - unordered-containers >=0.2.8.0 && <0.3, - uuid >=1.3.13 && <1.4, - vector >=0.12.0.1 && <0.13 - default-language: Haskell2010 hs-source-dirs: test/testsuite other-modules: Control.Concurrent.Capataz.SupervisorTest Control.Concurrent.CapatazTest Test.Util Paths_capataz - ghc-options: -Wall -Wincomplete-uni-patterns -Wincomplete-record-updates -threaded + default-language: Haskell2010 + ghc-options: -Wall -Wincomplete-uni-patterns + -Wincomplete-record-updates -threaded + build-depends: + async >=2.1.1.1 && <2.3, + base ==4.*, + bytestring >=0.10.8, + capataz -any, + pretty-show >=1.7, + prettyprinter >=1.1, + rio >=0.1.2.0, + tasty >=1.0.1.1, + tasty-hunit >=0.10.0.1, + tasty-smallcheck >=0.8.1, + teardown >=0.5.0.0, + time >=1.6.0, + uuid >=1.3 diff --git a/src/Capataz.hs b/src/Capataz.hs new file mode 100644 --- /dev/null +++ b/src/Capataz.hs @@ -0,0 +1,13 @@ +-- | Convinience module that re-exports modules: +-- +-- * "Control.Concurrent.Capataz" +-- * "Control.Concurrent.Capataz.Util" +-- +-- @since 0.2.0.0 +module Capataz + ( + module X + ) where + +import Control.Concurrent.Capataz as X +import Control.Concurrent.Capataz.Util as X diff --git a/src/Control/Concurrent/Capataz.hs b/src/Control/Concurrent/Capataz.hs --- a/src/Control/Concurrent/Capataz.hs +++ b/src/Control/Concurrent/Capataz.hs @@ -12,6 +12,7 @@ , Control.Concurrent.Capataz.Internal.Types.CallbackType (..) , Control.Concurrent.Capataz.Internal.Types.WorkerId +, Control.Concurrent.Capataz.Internal.Types.WorkerName , Control.Concurrent.Capataz.Internal.Types.WorkerRestartStrategy (..) , Control.Concurrent.Capataz.Internal.Types.WorkerTerminationPolicy (..) , Control.Concurrent.Capataz.Internal.Types.WorkerOptions @@ -23,6 +24,7 @@ , Control.Concurrent.Capataz.Internal.Types.ProcessError (..) , Control.Concurrent.Capataz.Internal.Types.SupervisorId +, Control.Concurrent.Capataz.Internal.Types.SupervisorName , Control.Concurrent.Capataz.Internal.Types.Supervisor , Control.Concurrent.Capataz.Internal.Types.SupervisorRestartStrategy (..) , Control.Concurrent.Capataz.Internal.Types.SupervisorStatus (..) @@ -61,6 +63,8 @@ , Control.Concurrent.Capataz.Internal.Core.forkWorker , Control.Concurrent.Capataz.Internal.Core.forkSupervisor , Control.Concurrent.Capataz.Internal.Core.forkCapataz +, Control.Concurrent.Capataz.Internal.Core.terminateCapataz +, Control.Concurrent.Capataz.Internal.Core.terminateCapataz_ , Control.Concurrent.Capataz.Internal.Core.terminateProcess -- * Utility functions @@ -70,10 +74,9 @@ , Control.Concurrent.Capataz.Internal.Core.getCapatazTeardown -- * Teardown (re-exported) -, Control.Teardown.teardown +, Control.Teardown.TeardownResult -- * Lens (re-exported) -, (.~) , (&) , Control.Concurrent.Capataz.Lens.set ) @@ -81,6 +84,6 @@ import qualified Control.Concurrent.Capataz.Internal.Core import qualified Control.Concurrent.Capataz.Internal.Types -import Control.Concurrent.Capataz.Lens ((&), (.~)) +import Control.Concurrent.Capataz.Lens ((&)) import qualified Control.Concurrent.Capataz.Lens import qualified Control.Teardown diff --git a/src/Control/Concurrent/Capataz/Internal/Core.hs b/src/Control/Concurrent/Capataz/Internal/Core.hs --- a/src/Control/Concurrent/Capataz/Internal/Core.hs +++ b/src/Control/Concurrent/Capataz/Internal/Core.hs @@ -17,6 +17,8 @@ , forkSupervisor , forkCapataz , terminateProcess +, terminateCapataz +, terminateCapataz_ , joinCapatazThread , getSupervisorProcessId , getSupervisorAsync @@ -24,11 +26,10 @@ ) where -import Protolude +import RIO -import Control.Concurrent.MVar (newEmptyMVar, takeMVar) -import Control.Teardown (Teardown, newTeardown) -import Data.Time.Clock (getCurrentTime) +import Control.Teardown (Teardown, TeardownResult, newTeardown, runTeardown, runTeardown_) +import RIO.Time (getCurrentTime) import qualified Data.UUID.V4 as UUID (nextRandom) @@ -43,22 +44,26 @@ -- that contain a supervisor (e.g. Capataz record). class HasSupervisor a where -- | Fetches a supervisor from a record internals. - getSupervisor :: a -> Supervisor + getSupervisor :: a m -> Supervisor m instance HasSupervisor Capataz where getSupervisor Capataz {capatazSupervisor} = capatazSupervisor instance HasSupervisor Supervisor where - getSupervisor = identity + getSupervisor = id -- | Creates a Capataz record, which holds both a root supervisor and a --- "Teardown" to shut down the system. The root supervisor monitors failures on --- process threads defined with "supervisorProcessSpecList" or created --- dynamically using "forkWorker" or "forkSupervisor". -forkCapataz :: Text -> (CapatazOptions -> CapatazOptions) -> IO Capataz +-- 'Teardown' to shut down the system. The root supervisor monitors failures on +-- process threads defined with 'supervisorProcessSpecList' or created +-- dynamically using 'forkWorker' or 'forkSupervisor'. +forkCapataz + :: (MonadUnliftIO m, MonadIO m) + => Text + -> (CapatazOptions m -> CapatazOptions m) + -> m (Capataz m) forkCapataz capatazName modOptionsFn = do - capatazId <- UUID.nextRandom - supervisorId <- UUID.nextRandom + capatazId <- liftIO UUID.nextRandom + supervisorId <- liftIO UUID.nextRandom let capatazOptions@CapatazOptions { notifyEvent } = defCapatazOptions capatazName modOptionsFn @@ -85,14 +90,14 @@ } MonitorEvent ProcessCompleted'{} -> - panic "Capataz completed; this should never happen" + error "Capataz completed; this should never happen" MonitorEvent ProcessForcedRestart{} -> - panic + error "Capataz was restarted from a OneForAll strategy; this should never happen" ControlAction{} -> - panic "Capataz received a ControlAction message; bad implementation" + error "Capataz received a ControlAction message; bad implementation" , notifyEvent } @@ -100,11 +105,11 @@ parentSupervisorEnv supervisorOptions supervisorId - 0 + 0 -- initial restart count - capatazTeardown <- newTeardown + capatazTeardown <- withRunInIO $ \run -> newTeardown "capataz" - ( do + (run $ do Supervisor.haltSupervisor "capataz system shutdown" supervisorEnv eventTime <- getCurrentTime notifyEvent CapatazTerminated {supervisorId , supervisorName , eventTime } @@ -112,25 +117,25 @@ return Capataz {capatazSupervisor , capatazTeardown } --- | Creates a green thread from an "IO ()" sub-routine. Depending in options --- defined in the "WorkerOptions" record, it will automatically restart this +-- | Creates a green thread from an @IO ()@ sub-routine. Depending in options +-- defined in the 'WorkerOptions' record, it will automatically restart this -- sub-routine in case of failures. -- -- See documentation of related functions: -- --- * "buildWorkerOptionsWithDefault" --- * "buildWorkerOptions" +-- * 'buildWorkerOptionsWithDefaults' +-- * 'buildWorkerOptions' -- forkWorker - :: HasSupervisor supervisor - => WorkerOptions -- ^ Worker options (restart, name, callbacks, etc) - -> supervisor -- ^ "Supervisor" that supervises the worker - -> IO WorkerId -- ^ An identifier that can be used to terminate the "Worker" + :: (MonadIO m, HasSupervisor supervisor) + => WorkerOptions m -- ^ Worker options (restart, name, callbacks, etc) + -> supervisor m -- ^ 'Supervisor' that supervises the worker + -> m WorkerId -- ^ An identifier that can be used to terminate the 'Worker' forkWorker workerOptions sup = do let Supervisor { supervisorNotify } = getSupervisor sup workerIdVar <- newEmptyMVar supervisorNotify - ( ControlAction ForkWorker + (ControlAction ForkWorker { workerOptions , returnWorkerId = putMVar workerIdVar } @@ -138,24 +143,24 @@ takeMVar workerIdVar -- | Creates a green thread which monitors other green threads for failures and --- restarts them using settings defined on "SupervisorOptions". +-- restarts them using settings defined on 'SupervisorOptions'. -- -- See documentation of related functions: -- --- * "buildSupervisorOptionsWithDefault" --- * "buildSupervisorOptions" +-- * 'buildSupervisorOptionsWithDefault' +-- * 'buildSupervisorOptions' -- forkSupervisor - :: HasSupervisor parentSupervisor - => SupervisorOptions -- ^ Supervisor options - -> parentSupervisor -- ^ Parent supervisor instance that supervises new supervisor - -> IO Supervisor -- ^ A record used to dynamically create and supervise + :: (MonadIO m, HasSupervisor parentSupervisor) + => SupervisorOptions m -- ^ Supervisor options + -> parentSupervisor m -- ^ Parent supervisor instance that supervises new supervisor + -> m (Supervisor m) -- ^ A record used to dynamically create and supervise -- other processes forkSupervisor supervisorOptions parentSup = do let Supervisor { supervisorNotify } = getSupervisor parentSup supervisorVar <- newEmptyMVar supervisorNotify - ( ControlAction ForkSupervisor + (ControlAction ForkSupervisor { supervisorOptions , returnSupervisor = putMVar supervisorVar } @@ -165,16 +170,21 @@ -- | Stops the execution of a green thread being supervised by the given -- supervisor. -- --- NOTE: If "ProcessId" maps to a worker that is configured with a "Permanent" --- worker restart strategy, the worker green thread __will be restarted again__. +-- __IMPORTANT__ If 'ProcessId' maps to a worker that is configured with a +-- 'Permanent' worker restart strategy, the worker green thread __will be +-- restarted again__. -- terminateProcess - :: HasSupervisor supervisor => Text -> ProcessId -> supervisor -> IO Bool + :: (MonadIO m, HasSupervisor supervisor) + => Text + -> ProcessId + -> supervisor m + -> m Bool terminateProcess processTerminationReason processId supervisor = do let Supervisor { supervisorNotify } = getSupervisor supervisor result <- newEmptyMVar supervisorNotify - ( ControlAction TerminateProcess + (ControlAction TerminateProcess { processId , processTerminationReason , notifyProcessTermination = putMVar result @@ -184,22 +194,34 @@ -- | Joins the thread of the root supervisor of the given capataz system to the -- current thread. -joinCapatazThread :: Capataz -> IO () +joinCapatazThread :: MonadIO m => Capataz m -> m () joinCapatazThread Capataz { capatazSupervisor } = let Supervisor { supervisorAsync } = capatazSupervisor in wait supervisorAsync --- | Gets "Teardown" record of this capataz system. -getCapatazTeardown :: Capataz -> Teardown +-- | Terminates a 'Capataz' system (all supervised threads) and returns a 'TeardownResult' +-- +-- @since 0.2.0.0 +terminateCapataz :: MonadIO m => Capataz m -> m TeardownResult +terminateCapataz = liftIO . runTeardown + +-- | Terminates a 'Capataz' system (all supervised threads) +-- +-- @since 0.2.0.0 +terminateCapataz_ :: MonadIO m => Capataz m -> m () +terminateCapataz_ = liftIO . runTeardown_ + +-- | Gets 'Teardown' record of this capataz system. +getCapatazTeardown :: Capataz m -> Teardown getCapatazTeardown Capataz { capatazTeardown } = capatazTeardown --- | Gets the "Async" of a Supervisor thread. +-- | Gets the 'Async' of a Supervisor thread. -- --- NOTE: There is no way to get the "Async" value of the root supervisor; this --- is to avoid error scenarios. -getSupervisorAsync :: Supervisor -> Async () +-- NOTE: There is no way to get the 'Async' value of the root supervisor; this +-- is done on-purpose to avoid error scenarios. +getSupervisorAsync :: Supervisor m -> Async () getSupervisorAsync Supervisor { supervisorAsync } = supervisorAsync --- | Gets the process identifier of a Supervisor; normally used for termination. -getSupervisorProcessId :: Supervisor -> ProcessId +-- | Gets the process identifier of a 'Supervisor'; normally used for termination. +getSupervisorProcessId :: Supervisor m -> ProcessId getSupervisorProcessId Supervisor { supervisorId } = supervisorId diff --git a/src/Control/Concurrent/Capataz/Internal/Process.hs b/src/Control/Concurrent/Capataz/Internal/Process.hs --- a/src/Control/Concurrent/Capataz/Internal/Process.hs +++ b/src/Control/Concurrent/Capataz/Internal/Process.hs @@ -6,43 +6,43 @@ -- Worker process types. module Control.Concurrent.Capataz.Internal.Process where -import Protolude +import RIO import Control.Concurrent.Capataz.Internal.Types import Control.Concurrent.Capataz.Internal.Util (readProcessMap, sortProcessesByTerminationOrder) -import Data.Time.Clock (UTCTime, getCurrentTime) +import RIO.Time (UTCTime, getCurrentTime) -- | Gets "Async" from a given Process. -getProcessAsync :: Process -> Async () +getProcessAsync :: Process m -> Async () getProcessAsync process = case process of WorkerProcess Worker { workerAsync } -> workerAsync SupervisorProcess Supervisor { supervisorAsync } -> supervisorAsync -- | Gets "ThreadId" from a given Process. -getProcessThreadId :: Process -> ThreadId -getProcessThreadId = asyncThreadId . getProcessAsync +getProcessThreadId :: Process m -> ProcessThreadId +getProcessThreadId = PTID . asyncThreadId . getProcessAsync -- | Gets "ProcessId" from a given Process. -getProcessId :: Process -> ProcessId +getProcessId :: Process m -> ProcessId getProcessId process = case process of WorkerProcess Worker { workerId } -> workerId SupervisorProcess Supervisor { supervisorId } -> supervisorId -- | Gets "ProcessName" from a given "ProcessSpec". -getProcessName :: ProcessSpec -> ProcessName +getProcessName :: ProcessSpec m -> ProcessName getProcessName procSpec = case procSpec of WorkerSpec WorkerOptions { workerName } -> workerName SupervisorSpec SupervisorOptions { supervisorName } -> supervisorName -- | Gets "ProcessType" from a given "ProcessSpec". -getProcessType :: ProcessSpec -> ProcessType +getProcessType :: ProcessSpec m -> ProcessType getProcessType processSpec = case processSpec of WorkerSpec{} -> WorkerType SupervisorSpec{} -> SupervisorType -- | Gets "ProcessSpec" of a given "Process". -getProcessSpec :: Process -> ProcessSpec +getProcessSpec :: Process m -> ProcessSpec m getProcessSpec process = case process of WorkerProcess Worker { workerOptions } -> WorkerSpec workerOptions SupervisorProcess Supervisor { supervisorOptions } -> @@ -50,7 +50,8 @@ -- | Utility function to send notifications to the capataz system callback when -- a Process fails. -notifyProcessFailed :: SupervisorEnv -> Process -> SomeException -> IO () +notifyProcessFailed + :: MonadIO m => SupervisorEnv m -> Process m -> SomeException -> m () notifyProcessFailed SupervisorEnv { supervisorId, supervisorName, notifyEvent } process processError = do eventTime <- getCurrentTime @@ -67,7 +68,8 @@ -- | Utility function to send notifications to the capataz system callback when -- a Process is terminated. -notifyProcessTerminated :: SupervisorEnv -> Process -> Text -> IO () +notifyProcessTerminated + :: MonadIO m => SupervisorEnv m -> Process m -> Text -> m () notifyProcessTerminated SupervisorEnv { supervisorId, supervisorName, notifyEvent } process terminationReason = do eventTime <- getCurrentTime @@ -85,7 +87,11 @@ -- | Utility function to send notifications to the capataz system callback when -- a Process is started or restarted. notifyProcessStarted - :: Maybe (ProcessId, RestartCount) -> ParentSupervisorEnv -> Process -> IO () + :: MonadIO m + => Maybe (ProcessId, RestartCount) + -> ParentSupervisorEnv m + -> Process m + -> m () notifyProcessStarted mRestartInfo ParentSupervisorEnv { supervisorId, supervisorName, notifyEvent } process = do eventTime <- getCurrentTime @@ -112,7 +118,7 @@ -- | Utility function to send notifications when a Process sub-routine completes -- without errors. -notifyProcessCompleted :: SupervisorEnv -> Process -> UTCTime -> IO () +notifyProcessCompleted :: SupervisorEnv m -> Process m -> UTCTime -> m () notifyProcessCompleted SupervisorEnv { supervisorId, supervisorName, notifyEvent } process eventTime = notifyEvent ProcessCompleted { supervisorId @@ -126,39 +132,41 @@ -- | Utility function to execute a Process onCompletion sub-routine. -callProcessOnCompletion :: ProcessSpec -> IO () +callProcessOnCompletion :: Monad m => ProcessSpec m -> m () callProcessOnCompletion procSpec = case procSpec of WorkerSpec WorkerOptions { workerOnCompletion } -> workerOnCompletion _ -> return () -- | Utility function to execute a Process onFailure sub-routine. -callProcessOnFailure :: ProcessSpec -> SomeException -> IO () +callProcessOnFailure :: ProcessSpec m -> SomeException -> m () callProcessOnFailure procSpec err = case procSpec of WorkerSpec WorkerOptions { workerOnFailure } -> workerOnFailure err SupervisorSpec SupervisorOptions { supervisorOnFailure } -> supervisorOnFailure err -- | Utility function to execute a Process onTermination sub-routine. -callProcessOnTermination :: ProcessSpec -> IO () +callProcessOnTermination :: Monad m => ProcessSpec m -> m () callProcessOnTermination procSpec = case procSpec of WorkerSpec WorkerOptions { workerOnTermination } -> workerOnTermination _ -> return () -- | Handles errors produced - or thrown to - a process thread. handleProcessException - :: (IO () -> IO a) - -> ParentSupervisorEnv - -> ProcessSpec + :: (MonadUnliftIO m) + => (m () -> m a) + -> ParentSupervisorEnv m + -> ProcessSpec m -> ProcessId -> RestartCount -> SomeException - -> IO MonitorEvent + -> m MonitorEvent handleProcessException unmask ParentSupervisorEnv { supervisorId, supervisorName, notifyEvent } procSpec processId restartCount err = do let processName = getProcessName procSpec - processThreadId <- myThreadId + processThreadId <- PTID <$> myThreadId monitorEventTime <- getCurrentTime - case fromException err of + + case fromAnyException err of Just RestartProcessException -> return ProcessForcedRestart { processId , processName @@ -166,7 +174,7 @@ } Just TerminateProcessException { processTerminationReason } -> do - eErrResult <- try $ unmask $ callProcessOnTermination procSpec + eErrResult <- unsafeTry $ unmask $ callProcessOnTermination procSpec notifyEvent ProcessCallbackExecuted { supervisorId @@ -180,6 +188,7 @@ , eventTime = monitorEventTime } + case eErrResult of Left processCallbackError -> return ProcessFailed' { processId @@ -210,9 +219,8 @@ , processRestartCount = restartCount } - -- This exception was an error from the given sub-routine _ -> do - eErrResult <- try $ unmask $ callProcessOnFailure procSpec err + eErrResult <- unsafeTry $ unmask $ callProcessOnFailure procSpec err notifyEvent ProcessCallbackExecuted { supervisorId @@ -249,18 +257,19 @@ -- | Handles completion of a Process sub-routine. handleProcessCompletion - :: (IO () -> IO a) - -> ParentSupervisorEnv - -> ProcessSpec + :: (MonadUnliftIO m) + => (m () -> m a) + -> ParentSupervisorEnv m + -> ProcessSpec m -> ProcessId -> RestartCount - -> IO MonitorEvent + -> m MonitorEvent handleProcessCompletion unmask ParentSupervisorEnv { supervisorId, supervisorName, notifyEvent } procSpec processId restartCount = do let processName = getProcessName procSpec - processThreadId <- myThreadId + processThreadId <- PTID <$> myThreadId monitorEventTime <- getCurrentTime - eCompResult <- try $ unmask $ callProcessOnCompletion procSpec + eCompResult <- unsafeTry $ unmask $ callProcessOnCompletion procSpec notifyEvent ProcessCallbackExecuted { supervisorId @@ -296,10 +305,11 @@ -- function gets executed on the supervisor's thread. -- terminateProcess - :: Text -- ^ Description that indicates _why_ there is a termination - -> SupervisorEnv - -> Process - -> IO () + :: (MonadUnliftIO m) + => Text -- ^ Description that indicates _why_ there is a termination + -> SupervisorEnv m + -> Process m + -> m () terminateProcess processTerminationReason env process = do case process of WorkerProcess worker -> terminateWorker processTerminationReason worker @@ -310,7 +320,7 @@ -- | Internal utility function that manages execution of a termination policy -- for a Worker. -terminateWorker :: Text -> Worker -> IO () +terminateWorker :: (MonadUnliftIO m) => Text -> Worker m -> m () terminateWorker processTerminationReason Worker { workerId, workerOptions, workerAsync } = do let processId = workerId @@ -327,24 +337,29 @@ , processTerminationReason } - TimeoutMillis millis -> race_ - ( do - threadDelay (millis * 1000) - cancelWith + TimeoutMillis millis -> do + -- NOTE: Given Teardown executes teardown operations in an uninterruptible mask + -- we need to run asyncWithUnmask to come back to a unmasked state, sadly, race + -- doesn't use asyncWithUnmask_, so we need to use it here + result <- asyncWithUnmask $ \unmask -> unmask $ race_ + (do + threadDelay (millis * 1000) + cancelWith + workerAsync + BrutallyTerminateProcessException + { processId + , processTerminationReason + } + ) + (cancelWith workerAsync - BrutallyTerminateProcessException - { processId - , processTerminationReason - } - ) - ( cancelWith - workerAsync - TerminateProcessException {processId , processTerminationReason } - ) + TerminateProcessException {processId , processTerminationReason } + ) + wait result -- | Internal utility function that manages execution of a termination policy -- for a Supervisor. -terminateSupervisor :: Text -> Supervisor -> IO () +terminateSupervisor :: MonadIO m => Text -> Supervisor m -> m () terminateSupervisor processTerminationReason Supervisor { supervisorId = processId, supervisorAsync } = cancelWith supervisorAsync @@ -352,7 +367,7 @@ -- | Internal sub-routine that terminates workers of a supervisor, used when a -- supervisor instance is terminated. -terminateProcessMap :: Text -> SupervisorEnv -> IO () +terminateProcessMap :: (MonadUnliftIO m) => Text -> SupervisorEnv m -> m () terminateProcessMap terminationReason env@SupervisorEnv { supervisorId, supervisorName, supervisorProcessTerminationOrder, notifyEvent } = do eventTime <- getCurrentTime diff --git a/src/Control/Concurrent/Capataz/Internal/Supervisor.hs b/src/Control/Concurrent/Capataz/Internal/Supervisor.hs --- a/src/Control/Concurrent/Capataz/Internal/Supervisor.hs +++ b/src/Control/Concurrent/Capataz/Internal/Supervisor.hs @@ -7,16 +7,13 @@ {-# LANGUAGE ScopedTypeVariables #-} module Control.Concurrent.Capataz.Internal.Supervisor where -import Control.Concurrent.Async (asyncWithUnmask) -import Control.Concurrent.STM.TQueue (newTQueueIO, readTQueue, writeTQueue) -import Control.Concurrent.STM.TVar (newTVarIO) -import Data.IORef (newIORef) -import Data.Time.Clock (NominalDiffTime, UTCTime, diffUTCTime, getCurrentTime) -import Protolude +import RIO +import qualified RIO.HashMap as HashMap -import qualified Data.HashMap.Strict as HashMap -import qualified Data.UUID.V4 as UUID +import RIO.Time (NominalDiffTime, UTCTime, diffUTCTime, getCurrentTime) +import qualified Data.UUID.V4 as UUID + import Control.Concurrent.Capataz.Internal.Types import qualified Control.Concurrent.Capataz.Internal.Process as Process @@ -27,14 +24,15 @@ -- from the public @forkSupervisor@ function which sends a message to the -- supervisor loop. forkSupervisor - :: ParentSupervisorEnv - -> SupervisorOptions + :: (MonadUnliftIO m) + => ParentSupervisorEnv m + -> SupervisorOptions m -> Maybe (ProcessId, RestartCount) - -> IO Supervisor + -> m (Supervisor m) forkSupervisor parentEnv supervisorOptions mRestartInfo = do (supervisorId, restartCount) <- case mRestartInfo of Just (supervisorId, restartCount) -> pure (supervisorId, restartCount) - Nothing -> (,) <$> UUID.nextRandom <*> pure 0 + Nothing -> (,) <$> liftIO UUID.nextRandom <*> pure 0 supervisor <- supervisorMain parentEnv @@ -50,12 +48,13 @@ -- | Utility function that builds an utility record which is used on all -- internal APIs of the supervision logic. buildSupervisorEnv - :: (CapatazEvent -> IO ()) - -> (SupervisorMessage -> IO ()) - -> STM SupervisorMessage + :: MonadIO m + => (CapatazEvent -> m ()) + -> (SupervisorMessage m -> m ()) + -> STM (SupervisorMessage m) -> SupervisorId - -> SupervisorOptions - -> IO SupervisorEnv + -> SupervisorOptions m + -> m (SupervisorEnv m) buildSupervisorEnv notifyEvent supervisorNotify supervisorGetNotification supervisorId supervisorOptions@SupervisorOptions {..} = do supervisorProcessMap <- newIORef mempty @@ -63,7 +62,8 @@ return SupervisorEnv {..} -- | Handles an event produced by one of the processes this supervisor monitors. -handleMonitorEvent :: SupervisorEnv -> MonitorEvent -> IO Bool +handleMonitorEvent + :: (MonadUnliftIO m) => SupervisorEnv m -> MonitorEvent -> m Bool handleMonitorEvent env monitorEv = do case monitorEv of ProcessForcedRestart{} -> @@ -87,7 +87,8 @@ return True -- | Handles an action triggered by the public Capataz API. -handleControlAction :: SupervisorEnv -> ControlAction -> IO Bool +handleControlAction + :: (MonadUnliftIO m) => SupervisorEnv m -> ControlAction m -> m Bool handleControlAction env controlAction = case controlAction of ForkWorker { workerOptions, returnWorkerId } -> do worker@Worker { workerId } <- Worker.forkWorker @@ -120,7 +121,7 @@ -- | Executes the shutdown operation of a Supervisor, including the termination -- of its supervised processes. -haltSupervisor :: Text -> SupervisorEnv -> IO () +haltSupervisor :: (MonadUnliftIO m) => Text -> SupervisorEnv m -> m () haltSupervisor reason env = do Util.writeSupervisorStatus env Halting Process.terminateProcessMap reason env @@ -130,40 +131,33 @@ -- | Handles all messages that a Supervisor can receive from its monitored -- processes or from the public API. -handleSupervisorMessage :: SupervisorEnv -> SupervisorMessage -> IO Bool +handleSupervisorMessage + :: (MonadUnliftIO m) => SupervisorEnv m -> SupervisorMessage m -> m Bool handleSupervisorMessage env message = case message of ControlAction controlAction -> handleControlAction env controlAction MonitorEvent monitorEvent -> handleMonitorEvent env monitorEvent -- | This sub-routine executes the main thread loop of a "Supervisor" instance. supervisorLoop - :: (forall b . IO b -> IO b) - -> ParentSupervisorEnv - -> SupervisorEnv + :: (MonadUnliftIO m) + => (forall b . m b -> m b) + -> ParentSupervisorEnv m + -> SupervisorEnv m -> RestartCount - -> IO () + -> m () supervisorLoop unmask parentEnv@ParentSupervisorEnv { supervisorId, supervisorName, supervisorNotify = notifyParentSupervisor } env@SupervisorEnv { supervisorId = processId, supervisorName = processName, supervisorOptions, supervisorStatusVar, supervisorGetNotification, notifyEvent } restartCount = do - processThreadId <- myThreadId + processThreadId <- PTID <$> myThreadId loopResult <- - unmask - $ try + unsafeTry + $ unmask $ atomically $ (,) <$> Util.readSupervisorStatusSTM supervisorStatusVar <*> supervisorGetNotification case loopResult of - Left supervisorError -> do - haltSupervisor (show supervisorError) env - result <- Process.handleProcessException - unmask - parentEnv - (SupervisorSpec supervisorOptions) - processId - restartCount - supervisorError - notifyParentSupervisor (MonitorEvent result) + Left supervisorError -> handleSupervisorLoopError supervisorError Right (status, message) -> case status of Initializing -> do @@ -176,18 +170,11 @@ supervisorLoop unmask parentEnv env restartCount Running -> do - eContinueLoop <- try $ unmask $ handleSupervisorMessage env message + eContinueLoop <- unsafeTry $ unmask $ handleSupervisorMessage + env + message case eContinueLoop of - Left supervisorError -> do - haltSupervisor (show supervisorError) env - result <- Process.handleProcessException - unmask - parentEnv - (SupervisorSpec supervisorOptions) - processId - restartCount - supervisorError - notifyParentSupervisor (MonitorEvent result) + Left supervisorError -> handleSupervisorLoopError supervisorError Right continueLoop | continueLoop -> supervisorLoop unmask parentEnv env restartCount @@ -211,15 +198,27 @@ Halted -> -- Discard messages when halted return () + where + handleSupervisorLoopError err = do + haltSupervisor (tshow err) env + result <- Process.handleProcessException + unmask + parentEnv + (SupervisorSpec supervisorOptions) + processId + restartCount + err + notifyParentSupervisor (MonitorEvent result) -- | This sub-routine starts a Supervisor thread and initializes its -- processList. supervisorMain - :: ParentSupervisorEnv - -> SupervisorOptions + :: (MonadUnliftIO m) + => ParentSupervisorEnv m + -> SupervisorOptions m -> SupervisorId -> RestartCount - -> IO Supervisor + -> m (Supervisor m) supervisorMain parentEnv@ParentSupervisorEnv { notifyEvent } supervisorOptions@SupervisorOptions { supervisorName, supervisorProcessSpecList } supervisorId restartCount = do supervisorCreationTime <- getCurrentTime @@ -241,7 +240,7 @@ forM_ supervisorProcessSpecList - ( \processSpec -> case processSpec of + (\processSpec -> case processSpec of WorkerSpec workerOptions -> do worker <- Worker.forkWorker (Util.toParentSupervisorEnv supervisorEnv) @@ -273,7 +272,7 @@ -- | Tracks difference between two timestamps so that we keep track of a -- Supervisor error intensity. -calcDiffSeconds :: UTCTime -> IO NominalDiffTime +calcDiffSeconds :: MonadIO m => UTCTime -> m NominalDiffTime calcDiffSeconds creationTime = do currentTime <- getCurrentTime return $ diffUTCTime currentTime creationTime @@ -282,7 +281,7 @@ -- error intensity has been breached, see "ProcessRestartAction" for possible -- outcomes. calcRestartAction - :: SupervisorEnv -> Int -> NominalDiffTime -> ProcessRestartAction + :: SupervisorEnv m -> Int -> NominalDiffTime -> ProcessRestartAction calcRestartAction SupervisorEnv { supervisorIntensity, supervisorPeriodSeconds } restartCount diffSeconds = case () of _ @@ -298,7 +297,12 @@ -- | Sub-routine responsible of executing a "SupervisorRestartStrategy". execCapatazRestartStrategy - :: SupervisorEnv -> ProcessId -> ProcessSpec -> Int -> IO () + :: (MonadUnliftIO m) + => SupervisorEnv m + -> ProcessId + -> ProcessSpec m + -> Int + -> m () execCapatazRestartStrategy supervisorEnv@SupervisorEnv { supervisorRestartStrategy } processId processSpec processRestartCount = case supervisorRestartStrategy of AllForOne -> do @@ -330,13 +334,14 @@ -- | Executes a restart action returned from the invokation of -- "calcRestartAction". execRestartAction - :: SupervisorEnv + :: (MonadUnliftIO m) + => SupervisorEnv m -> ProcessId - -> ProcessSpec + -> ProcessSpec m -> Text -> UTCTime -> Int - -> IO () + -> m () execRestartAction supervisorEnv@SupervisorEnv { supervisorOnIntensityReached } processId processSpec processName processCreationTime processRestartCount = do restartAction <- calcRestartAction supervisorEnv processRestartCount @@ -345,7 +350,7 @@ case restartAction of HaltSupervisor -> do -- skip exceptions on callback - (_ :: Either SomeException ()) <- try supervisorOnIntensityReached + (_ :: Either SomeException ()) <- unsafeTry supervisorOnIntensityReached throwIO SupervisorIntensityReached { processId , processName @@ -359,7 +364,7 @@ supervisorEnv processId processSpec - (succ processRestartCount) + (processRestartCount + 1) -------------------------------------------------------------------------------- @@ -367,7 +372,12 @@ -- | Restarts _all_ processes that are supervised by Supervisor, invoked when -- one worker green thread fails and causes sibling process threads to get -- restarted as well (e.g. "AllForOne" supervisor restart strategy). -restartProcessList :: SupervisorEnv -> WorkerId -> RestartCount -> IO [Process] +restartProcessList + :: (MonadUnliftIO m) + => SupervisorEnv m + -> WorkerId + -> RestartCount + -> m [Process m] restartProcessList supervisorEnv@SupervisorEnv { supervisorProcessTerminationOrder } failingProcessId restartCount = do processMap <- Util.readProcessMap supervisorEnv @@ -404,7 +414,7 @@ -- | Sub-routine that is used when there is a restart request sent to a Process -- caused by an "AllForOne" restart from a failing sibling process. -forceRestartProcess :: SupervisorEnv -> Process -> IO () +forceRestartProcess :: MonadIO m => SupervisorEnv m -> Process m -> m () forceRestartProcess env process = do Process.notifyProcessTerminated env process "forced restart" cancelWith (Process.getProcessAsync process) RestartProcessException @@ -412,7 +422,12 @@ -- | Starts a new worker thread taking into account an existing "WorkerId" and -- keeping a "RestartCount" to manage the Supervisor error intensity. restartWorker - :: SupervisorEnv -> WorkerOptions -> WorkerId -> RestartCount -> IO Process + :: (MonadUnliftIO m) + => SupervisorEnv m + -> WorkerOptions m + -> WorkerId + -> RestartCount + -> m (Process m) restartWorker supervisorEnv workerOptions workerId restartCount = WorkerProcess <$> Worker.forkWorker (Util.toParentSupervisorEnv supervisorEnv) @@ -423,11 +438,12 @@ -- "SupervisorId" and keeping a "RestartCount" to manage the parent Supervisor -- error intensity. restartSupervisor - :: ParentSupervisorEnv - -> SupervisorOptions + :: (MonadUnliftIO m) + => ParentSupervisorEnv m + -> SupervisorOptions m -> ProcessId -> RestartCount - -> IO Process + -> m (Process m) restartSupervisor parentEnv supervisorOptions processId restartCount = SupervisorProcess <$> forkSupervisor parentEnv supervisorOptions @@ -438,7 +454,8 @@ -- | Executes restart strategy for when a worker finishes it execution because -- of a completion (e.g. worker sub-routine finished without any errors). -handleWorkerCompleted :: SupervisorEnv -> Worker -> IO () +handleWorkerCompleted + :: (MonadUnliftIO m) => SupervisorEnv m -> Worker m -> m () handleWorkerCompleted env worker = do let Worker { workerId, workerOptions, workerCreationTime } = worker WorkerOptions { workerName, workerRestartStrategy } = workerOptions @@ -461,7 +478,8 @@ -- | Executes restart strategy for when a process finishes it execution because -- of a completion (e.g. worker sub-routine finished without any errors). -handleProcessCompleted :: SupervisorEnv -> ProcessId -> UTCTime -> IO () +handleProcessCompleted + :: (MonadUnliftIO m) => SupervisorEnv m -> ProcessId -> UTCTime -> m () handleProcessCompleted env processId completionTime = do mProcess <- Util.fetchProcess env processId case mProcess of @@ -472,14 +490,15 @@ case process of WorkerProcess worker -> handleWorkerCompleted env worker _ -> - panic + error $ "ERROR: Supervisor (" <> show (Process.getProcessId process) <> ") should never complete" -- | Executes restart strategy for when a worker finishes it execution because -- of a failure. -handleWorkerFailed :: SupervisorEnv -> Worker -> Int -> IO () +handleWorkerFailed + :: (MonadUnliftIO m) => SupervisorEnv m -> Worker m -> Int -> m () handleWorkerFailed env worker restartCount = do let Worker { workerId, workerCreationTime, workerOptions } = worker WorkerOptions { workerName, workerRestartStrategy } = workerOptions @@ -494,7 +513,8 @@ -- | Executes restart strategy for when a supervisor finishes it execution because -- of a failure. -handleSupervisorFailed :: SupervisorEnv -> Supervisor -> Int -> IO () +handleSupervisorFailed + :: (MonadUnliftIO m) => SupervisorEnv m -> Supervisor m -> Int -> m () handleSupervisorFailed env supervisor restartCount = do let Supervisor { supervisorId, supervisorCreationTime, supervisorOptions } = supervisor @@ -509,7 +529,12 @@ -- | Executes restart strategy for when a process finishes it execution because -- of a failure. handleProcessFailed - :: SupervisorEnv -> WorkerId -> SomeException -> Int -> IO () + :: (MonadUnliftIO m) + => SupervisorEnv m + -> WorkerId + -> SomeException + -> Int + -> m () handleProcessFailed env processId processError restartCount = do mProcess <- Util.fetchProcess env processId case mProcess of @@ -524,7 +549,8 @@ -- | Executes restart strategy for when a worker finishes it execution because -- of a termination from its supervisor. -handleWorkerTerminated :: SupervisorEnv -> Worker -> Int -> IO () +handleWorkerTerminated + :: (MonadUnliftIO m) => SupervisorEnv m -> Worker m -> Int -> m () handleWorkerTerminated env worker restartCount = do let Worker { workerId, workerCreationTime, workerOptions } = worker WorkerOptions { workerName, workerRestartStrategy } = workerOptions @@ -541,7 +567,8 @@ -- | Executes restart strategy for when a supervisor finishes it execution -- because of a termination from its parent supervisor. -handleSupervisorTerminated :: SupervisorEnv -> Supervisor -> Int -> IO () +handleSupervisorTerminated + :: (MonadUnliftIO m) => SupervisorEnv m -> Supervisor m -> Int -> m () handleSupervisorTerminated env supervisor restartCount = do let Supervisor { supervisorId, supervisorCreationTime, supervisorOptions } = supervisor @@ -555,7 +582,8 @@ -- | Executes restart strategy for when a process finishes it execution because -- of a termination from its supervisor. -handleProcessTerminated :: SupervisorEnv -> ProcessId -> Text -> Int -> IO () +handleProcessTerminated + :: (MonadUnliftIO m) => SupervisorEnv m -> ProcessId -> Text -> Int -> m () handleProcessTerminated env processId terminationReason restartCount = do mProcess <- Util.fetchProcess env processId case mProcess of diff --git a/src/Control/Concurrent/Capataz/Internal/Types.hs b/src/Control/Concurrent/Capataz/Internal/Types.hs --- a/src/Control/Concurrent/Capataz/Internal/Types.hs +++ b/src/Control/Concurrent/Capataz/Internal/Types.hs @@ -2,38 +2,57 @@ {-# LANGUAGE DuplicateRecordFields #-} {-# LANGUAGE NamedFieldPuns #-} {-# LANGUAGE NoImplicitPrelude #-} +{-# LANGUAGE OverloadedStrings #-} {-| This module contains all the types used across all the other modules -} module Control.Concurrent.Capataz.Internal.Types where -import Protolude +import RIO +import RIO.Time (NominalDiffTime, UTCTime) -import Control.Concurrent.STM.TVar (TVar) -import Control.Teardown (ITeardown (..), Teardown) -import Data.Default (Default (..)) -import Data.HashMap.Strict (HashMap) -import Data.IORef (IORef) -import Data.Time.Clock (NominalDiffTime, UTCTime) -import Data.UUID (UUID) +import Control.Teardown (HasTeardown (..), Teardown) +import Data.UUID (UUID) +import qualified Control.Exception as UnsafeE +import Data.Typeable (cast) + +import Data.Text.Prettyprint.Doc (Pretty (..), (<+>)) +import qualified Data.Text.Prettyprint.Doc as Pretty +import Text.Show.Pretty (ppShow) + type CapatazId = UUID type WorkerId = UUID type SupervisorId = UUID type ProcessId = UUID -type WorkerAction = IO () -type ProcessThreadId = ThreadId +type WorkerAction m = WorkerId -> m () type ProcessName = Text type CapatazName = Text type SupervisorName = Text type WorkerName = Text type RestartCount = Int -type ProcessMap = HashMap ProcessId Process +type ProcessMap m = HashMap ProcessId (Process m) type ParentSupervisor = Supervisor --- | Event delivered to the "notifyEvent" callback sub-routine; these events can +-- | Wrapper for 'ThreadId' +-- +-- @since 0.2.0.0 +newtype ProcessThreadId + = PTID ThreadId + deriving (Generic, Eq, Show) + +-- | @since 0.2.0.0 +instance Pretty ProcessThreadId where + pretty (PTID tid) = + case words (show tid) of + (_:threadNumber:_) -> pretty threadNumber + _ -> "unknown" + +-- | Event delivered to the 'notifyEvent' callback sub-routine; these events can -- be used to monitor the capataz system and track what is doing, providing high -- levels of telemetry for all supervisors and workers of a capataz system, -- ergo, should be used for logging, monitoring and testing purposes. +-- +-- @since 0.1.0.0 data CapatazEvent = InvalidSupervisorStatusReached { supervisorId :: !SupervisorId @@ -131,8 +150,212 @@ } deriving (Generic, Show) --- | Defines how a "Worker" process termination should be handled by its +-- | @since 0.2.0.0 +instance Pretty CapatazEvent where + pretty ev = + case ev of + InvalidSupervisorStatusReached {supervisorId, supervisorName} -> + "Supervisor got into an state that should never have happened, please" + <+> "report a ticket to" + <+> "https://github.com/roman/Haskell-capataz/issues/new with title:" + <+> Pretty.dquotes "InvalidSupervisorStatusReached error" + <> prettyAttributes [("Supervisor ID" + , prettySupervisorId supervisorId supervisorName)] + + SupervisorStatusChanged { + supervisorId + , supervisorName + , prevSupervisorStatus + , newSupervisorStatus + } -> + "Supervisor changed state" + <> prettyAttributes [ ("Supervisor ID" + , prettySupervisorId supervisorId supervisorName) + , ("Previous State" + , Pretty.dquotes $ Pretty.pretty prevSupervisorStatus) + , ("New State" + , Pretty.dquotes (Pretty.pretty newSupervisorStatus)) + ] + + ProcessStarted { + supervisorId + , supervisorName + , processId + , processThreadId + , processName + , processType + } -> + "Supervisor spawned new process" + <> prettyAttributes [ ("Supervisor ID" + , prettySupervisorId supervisorId supervisorName) + , ( "Process ID" + , prettyProcessId processId processName processThreadId) + , ( "Process Type", + pretty processType) + ] + + ProcessFailed { + processId + , processThreadId + , processName + , processType + , processError + } -> + "Process failed with error" + <> prettyAttributes [ ( "Process ID" + , prettyProcessId processId processName processThreadId) + , ( "Process Type", pretty processType) + , ( "Error" + , Pretty.nest 2 (Pretty.hardline + <> "|" <+> pretty (ppShow processError))) + ] + + ProcessRestarted { + supervisorId + , supervisorName + , processId + , processThreadId + , processName + , processType + } -> + "Supervisor restarted failed process" + <> prettyAttributes [ ("Supervisor ID" + , prettySupervisorId supervisorId supervisorName) + , ( "Process ID" + , prettyProcessId processId processName processThreadId) + , ( "Process Type", pretty processType) + ] + + ProcessTerminated { + supervisorId + , supervisorName + , processId + , processThreadId + , processName + , processType + , terminationReason + } -> + "Supervisor terminated process" + <> prettyAttributes [ ("Supervisor ID" + , prettySupervisorId supervisorId supervisorName) + , ( "Process ID" + , prettyProcessId processId processName processThreadId) + , ( "Process Type", pretty processType) + , ( "Reason" + , Pretty.nest 2 + (Pretty.fillBreak 10 $ pretty terminationReason)) + ] + + + ProcessCompleted { + processId + , processThreadId + , processName + , processType + } -> + "Process completed execution with no errors" + <> prettyAttributes [ ( "Process ID" + , prettyProcessId processId processName processThreadId) + , ( "Process Type", pretty processType) + ] + + ProcessCallbackExecuted { + processId + , processThreadId + , processName + , processType + , processCallbackError + , processCallbackType + } -> + case processCallbackError of + Nothing -> + "Process executed callback" + <> prettyAttributes [ ( "Process ID" + , prettyProcessId processId processName processThreadId) + , ( "Process Type", pretty processType) + , ( "Callback", pretty processCallbackType) + ] + Just err -> + "Process executed callback and it failed" + <> prettyAttributes [ ( "Process ID" + , prettyProcessId processId processName processThreadId) + , ( "Process Type", pretty processType) + , ( "Callback", pretty processCallbackType) + , ( "Error" + , Pretty.nest 2 + (Pretty.hardline + <> "|" + <+> pretty (ppShow err))) + ] + + ProcessTerminationStarted { + supervisorId + , supervisorName + , terminationReason + } -> + "Supervisor started termination of its children" + <> prettyAttributes [ ("Supervisor ID" + , prettySupervisorId supervisorId supervisorName) + , ( "Reason" + , Pretty.nest 2 + (Pretty.fillBreak 10 $ pretty terminationReason)) + ] + + ProcessTerminationFinished { + supervisorId + , supervisorName + } -> + "Supervisor finished termination of its children" + <> prettyAttributes [ ("Supervisor ID" + , prettySupervisorId supervisorId supervisorName) + ] + + CapatazFailed { + supervisorId + , supervisorName + , supervisorError + } -> + "Root Supervisor had a fatal failure" + <> prettyAttributes [ ("Supervisor ID" + , prettySupervisorId supervisorId supervisorName) + , ( "Error" + , Pretty.nest 2 + (Pretty.hardline + <> "|" + <+> pretty (ppShow supervisorError))) + ] + + CapatazTerminated { + supervisorId + , supervisorName + } -> + "Root supervisor was terminated" + <> prettyAttributes [ ("Supervisor ID" + , prettySupervisorId supervisorId supervisorName) + ] + where + prettyAttributes attrList = + Pretty.nest 4 + (Pretty.hardline + <> Pretty.vsep (map (\(k, v) -> Pretty.fill 20 (k <> ":") <+> v) + attrList)) + + prettySupervisorId supId supName = + Pretty.angles + $ pretty (show supId) <> "/" <> pretty supName + + prettyProcessId procId procName procTid = + Pretty.angles + $ pretty (show procId) <> "/" <> pretty procName <> "/" <> pretty procTid + +-- | @since 0.2.0.0 +instance Display CapatazEvent where + display = displayShow . pretty + +-- | Defines how a 'Worker' process termination should be handled by its -- supervisor. +-- +-- @since 0.0.0.0 data WorkerTerminationPolicy -- | Supervisor waits until infinity for the worker termination callback to -- finish execution. @@ -148,9 +371,11 @@ | TimeoutMillis !Int deriving (Generic, Show, Eq, Ord) -instance Default WorkerTerminationPolicy where - -- | Default worker termination is a timeout of three (3) seconds. - def = TimeoutMillis 3000 +-- | Default worker termination is a timeout of three (3) seconds. +-- +-- @since 0.2.0.0 +defWorkerTerminationPolicy :: WorkerTerminationPolicy +defWorkerTerminationPolicy = TimeoutMillis 3000 instance NFData WorkerTerminationPolicy @@ -174,6 +399,8 @@ -- | Specifies the order in which supervised process should be terminated by a -- Supervisor in case of a restart or shutdown. +-- +-- @since 0.0.0.0 data ProcessTerminationOrder -- | Supervisor terminates supervised process from most recent to oldest. = NewestFirst @@ -181,13 +408,17 @@ | OldestFirst deriving (Generic, Show, Eq, Ord) -instance Default ProcessTerminationOrder where - -- | default is "OldestFirst". - def = OldestFirst +-- | Default termination order is 'OldestFirst'. +-- +-- @since 0.2.0.0 +defProcessTerminationOrder :: ProcessTerminationOrder +defProcessTerminationOrder = OldestFirst instance NFData ProcessTerminationOrder -- | Specifies how a Supervisor restarts a failing process. +-- +-- @since 0.0.0.0 data SupervisorRestartStrategy -- | Supervisor terminates all sibling supervised processes that didn't fail, -- and then restarts all of them together. This strategy serves best when all @@ -198,31 +429,34 @@ | OneForOne deriving (Generic, Show, Eq, Ord) -instance Default SupervisorRestartStrategy where - -- | Default restart strategy is "OneForOne". - def = OneForOne +-- | Default restart strategy is 'OneForOne'. +-- +-- @since 0.2.0.0 +defSupervisorRestartStategy :: SupervisorRestartStrategy +defSupervisorRestartStategy = OneForOne instance NFData SupervisorRestartStrategy -- | Allows to: -- --- * Specify options for The root supervisor of a capataz system. +-- * Specify options for the root 'Supervisor' of a capataz system. -- --- * Provie a "notifyEvent" callback to monitor or log a capataz system. +-- * Provide a 'notifyEvent' callback to monitor or log a capataz system. -- -data CapatazOptions +-- @since 0.1.0.0 +data CapatazOptions m = CapatazOptions { supervisorName :: !SupervisorName , supervisorIntensity :: !Int , supervisorPeriodSeconds :: !NominalDiffTime , supervisorRestartStrategy :: !SupervisorRestartStrategy - , supervisorProcessSpecList :: ![ProcessSpec] + , supervisorProcessSpecList :: ![ProcessSpec m] , supervisorProcessTerminationOrder :: !ProcessTerminationOrder - , supervisorOnIntensityReached :: !(IO ()) + , supervisorOnIntensityReached :: !(m ()) -- | Callback sub-routine that gets executed when the root supervisor fails. - , supervisorOnFailure :: !(SomeException -> IO ()) + , supervisorOnFailure :: !(SomeException -> m ()) -- | Callback used for telemetry purposes. - , notifyEvent :: !(CapatazEvent -> IO ()) + , notifyEvent :: !(CapatazEvent -> m ()) } @@ -243,26 +477,29 @@ deriving (Generic, Show, Eq) instance NFData WorkerRestartStrategy -instance Default WorkerRestartStrategy where - -- | A worker default restart strategy is "Transient". - def = Transient +-- | A worker default restart strategy is "Transient". +defWorkerRestartStrategy :: WorkerRestartStrategy +defWorkerRestartStrategy = Transient + -- | Specifies all options that can be used to create a Worker Process. You may --- create a record of this type via the smart constructor "buildWorkerOptions". -data WorkerOptions +-- create a record of this type via the smart constructor 'buildWorkerOptions'. +-- +-- @since 0.1.0.0 +data WorkerOptions m = WorkerOptions { -- | An @IO ()@ sub-routine that will be executed when the worker -- thread is created, this attribute is lazy given we want to this -- value on a worker thread environment. - workerAction :: WorkerAction + workerAction :: WorkerAction m -- | Name of the Worker (present on "CapatazEvent" records) , workerName :: !WorkerName -- | Callback used when the worker fails with an error - , workerOnFailure :: !(SomeException -> IO ()) + , workerOnFailure :: !(SomeException -> m ()) -- | Callback used when the worker completes execution without error - , workerOnCompletion :: !(IO ()) + , workerOnCompletion :: !(m ()) -- | Callback used when the worker is terminated - , workerOnTermination :: !(IO ()) + , workerOnTermination :: !(m ()) -- | Indicates how a worker should be terminated , workerTerminationPolicy :: !WorkerTerminationPolicy -- | Indicates how a worker should be restarted @@ -271,7 +508,9 @@ deriving (Generic) -- | Record that contains the "Async" record (thread reference) of a worker -data Worker +-- +-- @since 0.1.0.0 +data Worker m = Worker { -- | Unique identifier for a worker that is executing workerId :: !WorkerId @@ -283,7 +522,7 @@ , workerName :: !WorkerName -- | "WorkerOptions" contains all the options around restart and termination -- policies - , workerOptions :: !WorkerOptions + , workerOptions :: !(WorkerOptions m) } data ProcessEnv @@ -295,7 +534,7 @@ , processRestartStrategy :: !WorkerRestartStrategy } -data SupervisorOptions +data SupervisorOptions m = SupervisorOptions { -- | Name of the Supervisor (present on "CapatazEvent" records) supervisorName :: Text @@ -308,40 +547,40 @@ -- | What is the "SupervisorRestartStrategy" for this Capataz , supervisorRestartStrategy :: !SupervisorRestartStrategy -- | Static set of workers that start as soon as the "Capataz" is created - , supervisorProcessSpecList :: ![ProcessSpec] + , supervisorProcessSpecList :: ![ProcessSpec m] -- | In which order the "Supervisor" record is going to terminate it's workers , supervisorProcessTerminationOrder :: !ProcessTerminationOrder -- | Callback used when the error intensity is reached - , supervisorOnIntensityReached :: !(IO ()) - , supervisorOnFailure :: !(SomeException -> IO ()) + , supervisorOnIntensityReached :: !(m ()) + , supervisorOnFailure :: !(SomeException -> m ()) } -data Supervisor +data Supervisor m = Supervisor { supervisorId :: !SupervisorId , supervisorName :: !SupervisorName - , supervisorOptions :: !SupervisorOptions + , supervisorOptions :: !(SupervisorOptions m) , supervisorCreationTime :: !UTCTime , supervisorAsync :: !(Async ()) - , supervisorNotify :: SupervisorMessage -> IO () - , supervisorEnv :: !SupervisorEnv + , supervisorNotify :: SupervisorMessage m -> m () + , supervisorEnv :: !(SupervisorEnv m) } -- | Internal record that represents an action being sent from threads using -- the Capataz public API. -data ControlAction +data ControlAction m = ForkWorker { - workerOptions :: !WorkerOptions - , returnWorkerId :: !(WorkerId -> IO ()) + workerOptions :: !(WorkerOptions m) + , returnWorkerId :: !(WorkerId -> m ()) } | ForkSupervisor { - supervisorOptions :: !SupervisorOptions - , returnSupervisor :: !(Supervisor -> IO ()) + supervisorOptions :: !(SupervisorOptions m) + , returnSupervisor :: !(Supervisor m -> m ()) } | TerminateProcess { processId :: !ProcessId , processTerminationReason :: !Text - , notifyProcessTermination :: !(Bool -> IO ()) + , notifyProcessTermination :: !(Bool -> m ()) } deriving (Generic) @@ -384,11 +623,20 @@ | OnTermination deriving (Generic, Show, Eq) +instance Pretty CallbackType where + pretty = Pretty.pretty . show + data ProcessType = SupervisorType | WorkerType deriving (Show, Eq) +instance Pretty ProcessType where + pretty ty = + case ty of + SupervisorType -> "Supervisor" + WorkerType -> "Worker" + -- | Internal exception triggered when a callback of a Worker fails data ProcessError = ProcessCallbackFailed { @@ -448,97 +696,108 @@ instance NFData SupervisorStatus +instance Pretty SupervisorStatus where + pretty = + pretty . show + -- | Internal message delivered to a supervisor process that can either be a -- call from public API or an event from its monitored worker process. -data SupervisorMessage +data SupervisorMessage m -- | Represents a request from done to the supervisor thread from another -- thread using the public API - = ControlAction !ControlAction + = ControlAction !(ControlAction m) -- | Represents an event (failure, completion, etc) from a monitored worker -- process to the supervisor | MonitorEvent !MonitorEvent deriving (Generic) --- | Internal Type to manage both Worker and Supervisor processes -data Process - = WorkerProcess Worker - | SupervisorProcess Supervisor +-- | Internal Type to manage both 'Worker' and 'Supervisor' processes +-- +-- @since 0.1.0.0 +data Process m + = WorkerProcess !(Worker m) + | SupervisorProcess !(Supervisor m) --- | Record used to specify how to __build__ a runtime "Process" in a static +-- | Record used to specify how to __build__ a runtime 'Process' in a static -- supervision tree; to create values of this type, you must use: -- --- * "workerSpec" or "workerSpecWithDefaults" to build a worker process +-- * 'workerSpec' or 'workerSpecWithDefaults' to build a worker process -- --- * "supervisorSpec" or "supervisorSpecWithDefaults" to build a supervisor +-- * 'supervisorSpec' or 'supervisorSpecWithDefaults' to build a supervisor -- process -- -data ProcessSpec - = WorkerSpec WorkerOptions - | SupervisorSpec SupervisorOptions +-- @since 0.1.0.0 +data ProcessSpec m + = WorkerSpec (WorkerOptions m) + | SupervisorSpec (SupervisorOptions m) -- | Record that contains the environment of a capataz monitor, this is used as -- the main record to create workers and to stop the supervisor thread. -data Capataz +-- +-- @since 0.0.0.0 +data Capataz m = Capataz { - capatazSupervisor :: !Supervisor + capatazSupervisor :: !(Supervisor m) , capatazTeardown :: !Teardown } -instance ITeardown Capataz where - teardown Capataz {capatazTeardown} = - teardown capatazTeardown +instance HasTeardown (Capataz m) where + getTeardown Capataz {capatazTeardown} = + capatazTeardown -- | Internal utility record used to hold part of the runtime information of a -- supervisor that acts as a parent of another supervisor. -data ParentSupervisorEnv +data ParentSupervisorEnv m = ParentSupervisorEnv { supervisorId :: !SupervisorId , supervisorName :: !SupervisorName - , supervisorNotify :: !(SupervisorMessage -> IO ()) - , notifyEvent :: !(CapatazEvent -> IO ()) + , supervisorNotify :: !(SupervisorMessage m -> m ()) + , notifyEvent :: !(CapatazEvent -> m ()) } -- | Convenience internal utility record that contains all values related to a -- supervisor process. -data SupervisorEnv +data SupervisorEnv m = SupervisorEnv { supervisorId :: !SupervisorId , supervisorName :: !SupervisorName - , supervisorNotify :: !(SupervisorMessage -> IO ()) - , supervisorGetNotification :: !(STM SupervisorMessage) - , supervisorProcessMap :: !(IORef ProcessMap) + , supervisorNotify :: !(SupervisorMessage m -> m ()) + , supervisorGetNotification :: !(STM (SupervisorMessage m)) + , supervisorProcessMap :: !(IORef (ProcessMap m)) , supervisorStatusVar :: !(TVar SupervisorStatus) - , supervisorOptions :: !SupervisorOptions + , supervisorOptions :: !(SupervisorOptions m) , supervisorIntensity :: !Int -- ^ http://erlang.org/doc/design_principles/sup_princ.html#max_intensity , supervisorPeriodSeconds :: !NominalDiffTime , supervisorRestartStrategy :: !SupervisorRestartStrategy , supervisorProcessTerminationOrder :: !ProcessTerminationOrder - , supervisorOnIntensityReached :: !(IO ()) - , supervisorOnIntensityReached :: !(SomeException -> IO ()) - , notifyEvent :: !(CapatazEvent -> IO ()) + , supervisorOnIntensityReached :: !(m ()) + , supervisorOnIntensityReached :: !(SomeException -> m ()) + , notifyEvent :: !(CapatazEvent -> m ()) } --- | Builds a "CapatazOptions" record with defaults on how to create a capataz +-- | Builds a 'CapatazOptions' record with defaults on how to create a capataz -- root supervisor, these defaults are: -- -- * Intensity error tolerance is set to 1 error every 5 seconds -- --- * A "SupervisorRestartStrategy" of "OneForOne" +-- * A 'SupervisorRestartStrategy' of 'OneForOne' -- --- * A "ProcessTerminationOrder" of "OldestFirst" +-- * A 'ProcessTerminationOrder' of 'OldestFirst' -- --- This function is intended to be used in combination with "forkCapataz". +-- This function is intended to be used in combination with 'forkCapataz'. -- +-- @since 0.1.0.0 defCapatazOptions - :: Text - -> (CapatazOptions -> CapatazOptions) -- ^ Function to modify root supervisor - -> CapatazOptions + :: Monad m + => Text + -> (CapatazOptions m -> CapatazOptions m) -- ^ Function to modify root supervisor + -> CapatazOptions m defCapatazOptions supervisorName modFn = modFn CapatazOptions { supervisorName , supervisorIntensity = 2 , supervisorPeriodSeconds = 5 - , supervisorRestartStrategy = def + , supervisorRestartStrategy = defSupervisorRestartStategy , supervisorProcessSpecList = [] , supervisorProcessTerminationOrder = OldestFirst , supervisorOnIntensityReached = return () @@ -546,78 +805,106 @@ , notifyEvent = const $ return () } --- | Builds a "ProcessSpec" record for a supervisor process with defaults from --- "supervisorSpecWithDefaults". This function allows overrides of these +-- | Builds a 'ProcessSpec' record for a supervisor process with defaults from +-- 'supervisorSpecWithDefaults'. This function allows overrides of these -- defaults using lenses. -- -- This function is used when building a supervisor branch in a static -- supervision trees. -- +-- @since 0.1.0.0 supervisorSpec - :: SupervisorName -- ^ Name used for telemetry purposes - -> (SupervisorOptions -> SupervisorOptions) -- ^ Function to modify default - -- supervisor options - -> ProcessSpec + :: Monad m + => SupervisorName -- ^ Name used for telemetry purposes + -> (SupervisorOptions m -> SupervisorOptions m) -- ^ 'SupervisorOptions' modifier + -> ProcessSpec m supervisorSpec sName modFn = SupervisorSpec (buildSupervisorOptions sName modFn) {-# INLINE supervisorSpec #-} --- | Builds a "ProcessSpec" record for a supervisor process with defaults from --- "buildSupervisorOptionsWithDefaults". +-- | Builds a 'ProcessSpec' record for a supervisor process with defaults from +-- 'buildSupervisorOptionsWithDefaults'. -- -- This function is used when building a supervisor branch in a static -- supervision trees. -- +-- @since 0.1.0.0 supervisorSpecWithDefaults - :: SupervisorName -- ^ Name used for telemetry purposes - -> ProcessSpec -supervisorSpecWithDefaults sName = supervisorSpec sName identity + :: Monad m + => SupervisorName -- ^ Name used for telemetry purposes + -> ProcessSpec m +supervisorSpecWithDefaults sName = supervisorSpec sName id {-# INLINE supervisorSpecWithDefaults #-} --- | Builds a "ProcessSpec" record for a worker process with defaults from --- "workerSpecWithDefaults". This function allows overrides of these +-- | Builds a 'ProcessSpec' record for a worker process with defaults from +-- 'workerSpecWithDefaults'. This function allows overrides of these -- defaults using lenses. -- -- This function is used when building a worker in a static supervision tree. -- +-- @since 0.1.0.0 workerSpec - :: WorkerName -- ^ Name used for telemetry purposes - -> IO () -- ^ IO sub-routine to be supervised - -> (WorkerOptions -> WorkerOptions) -- ^ Function to modify default worker + :: Monad m + => WorkerName -- ^ Name used for telemetry purposes + -> m () -- ^ 'IO' sub-routine to be supervised + -> (WorkerOptions m -> WorkerOptions m) -- ^ Function to modify default worker -- options - -> ProcessSpec + -> ProcessSpec m workerSpec wName wAction modFn = WorkerSpec (buildWorkerOptions wName wAction modFn) {-# INLINE workerSpec #-} --- | Builds a "ProcessSpec" record for a worker process with defaults from --- "buildSupervisorOptionsWithDefaults". +-- | Builds a 'ProcessSpec' record for a worker process with defaults from +-- 'workerSpecWithDefaults'. This function allows overrides of these +-- defaults using lenses. -- -- This function is used when building a worker in a static supervision tree. +-- The given sub-routine will receive the 'WorkerId' as a parameter -- +-- @since 0.2.0.0 +workerSpec1 + :: Monad m + => WorkerName -- ^ Name used for telemetry purposes + -> (WorkerId -> m ()) -- ^ sub-routine to be supervised + -> (WorkerOptions m -> WorkerOptions m) -- ^ Function to modify default worker + -- options + -> ProcessSpec m +workerSpec1 wName wAction modFn = + WorkerSpec (buildWorkerOptions1 wName wAction modFn) +{-# INLINE workerSpec1 #-} + +-- | Builds a 'ProcessSpec' record for a worker process with defaults from +-- 'buildSupervisorOptionsWithDefaults'. +-- +-- This function is used when building a worker in a static supervision tree. +-- +-- @since 0.1.0.0 workerSpecWithDefaults - :: WorkerName -- ^ Name used for telemetry purposes - -> IO () -- ^ IO sub-routine to be supervised - -> ProcessSpec -workerSpecWithDefaults wName wAction = workerSpec wName wAction identity + :: Monad m + => WorkerName -- ^ Name used for telemetry purposes + -> m () -- ^ IO sub-routine to be supervised + -> ProcessSpec m +workerSpecWithDefaults wName wAction = workerSpec wName wAction id {-# INLINE workerSpecWithDefaults #-} --- | Builds a "SupervisorOptions" record with defaults from --- "buildSupervisorOptionsWithDefaults". This function allows overrides of these +-- | Builds a 'SupervisorOptions' record with defaults from +-- 'buildSupervisorOptionsWithDefaults'. This function allows overrides of these -- defaults using lenses. -- --- This function is intended to be used in combination with "forkSupervisor". +-- This function is intended to be used in combination with 'forkSupervisor'. -- +-- @since 0.1.0.0 buildSupervisorOptions - :: SupervisorName -- ^ Name used for telemetry purposes - -> (SupervisorOptions -> SupervisorOptions) -- ^ Function to modify default + :: Monad m + => SupervisorName -- ^ Name used for telemetry purposes + -> (SupervisorOptions m -> SupervisorOptions m) -- ^ Function to modify default -- supervisor options - -> SupervisorOptions + -> SupervisorOptions m buildSupervisorOptions supervisorName modFn = modFn SupervisorOptions { supervisorName , supervisorIntensity = 2 , supervisorPeriodSeconds = 5 - , supervisorRestartStrategy = def + , supervisorRestartStrategy = defSupervisorRestartStategy , supervisorProcessSpecList = [] , supervisorProcessTerminationOrder = OldestFirst , supervisorOnIntensityReached = return () @@ -625,52 +912,77 @@ } {-# INLINE buildSupervisorOptions #-} --- | Builds a "SupervisorOptions" record with defaults to create a supervisor +-- | Builds a 'SupervisorOptions' record with defaults to create a supervisor -- process, these defaults are: -- -- * Intensity error tolerance is set to 1 error every 5 seconds -- --- * A "SupervisorRestartStrategy" of "OneForOne" +-- * A 'SupervisorRestartStrategy' of 'OneForOne' -- --- * A "ProcessTerminationOrder" of "OldestFirst" +-- * A 'ProcessTerminationOrder' of 'OldestFirst' -- --- This function is intended to be used in combination with "forkSupervisor". +-- This function is intended to be used in combination with 'forkSupervisor'. -- +-- @since 0.1.0.0 buildSupervisorOptionsWithDefaults - :: SupervisorName -- ^ Name used for telemetry purposes - -> SupervisorOptions -buildSupervisorOptionsWithDefaults = flip buildSupervisorOptions identity + :: Monad m + => SupervisorName -- ^ Name used for telemetry purposes + -> SupervisorOptions m +buildSupervisorOptionsWithDefaults = flip buildSupervisorOptions id {-# INLINE buildSupervisorOptionsWithDefaults #-} --- | Builds a "WorkerOptions" record, keeps the defaults from --- "buildWorkerOptionsWithDefaults" but allows overrides using lenses. +-- | Builds a 'WorkerOptions' record, keeps the defaults from +-- 'buildWorkerOptionsWithDefaults' but allows overrides using lenses. -- --- This function is intended to be used in combination with "forkWorker". See --- the ... example in the examples directory for a demonstration. +-- This function is intended to be used in combination with 'forkWorker'. See the +-- capataz-simple-example project in the examples directory for a demonstration. -- +-- @since 0.1.0.0 buildWorkerOptions - :: WorkerName -- ^ Name used for telemetry purposes - -> IO () -- ^ IO sub-routine to be supervised - -> (WorkerOptions -> WorkerOptions) -- ^ Function to modify default worker + :: Monad m + => WorkerName -- ^ Name used for telemetry purposes + -> m () -- ^ Process sub-routine to be supervised + -> (WorkerOptions m -> WorkerOptions m) -- ^ Function to modify default worker -- options - -> WorkerOptions -buildWorkerOptions workerName workerAction f = f WorkerOptions + -> WorkerOptions m +buildWorkerOptions workerName workerAction f = + buildWorkerOptions1 workerName (const workerAction) f +{-# INLINE buildWorkerOptions #-} + +-- | Builds a 'WorkerOptions' record, keeps the defaults from +-- 'buildWorkerOptionsWithDefaults' but allows overrides using lenses. +-- +-- This function is intended to be used in combination with 'forkWorker'. See the +-- capataz-simple-example project in the examples directory for a demonstration. +-- +-- The given sub-routine will receive the 'WorkerId' as a parameter. +-- +-- @since 0.2.0.0 +buildWorkerOptions1 + :: Monad m + => WorkerName -- ^ Name used for telemetry purposes + -> (WorkerId -> m ()) -- ^ Process sub-routine to be supervised + -> (WorkerOptions m -> WorkerOptions m) -- ^ Function to modify default worker + -- options + -> WorkerOptions m +buildWorkerOptions1 workerName workerAction f = f WorkerOptions { workerName - , workerAction + , workerAction = workerAction , workerOnFailure = const $ return () , workerOnCompletion = return () , workerOnTermination = return () - , workerTerminationPolicy = def - , workerRestartStrategy = def + , workerTerminationPolicy = defWorkerTerminationPolicy + , workerRestartStrategy = defWorkerRestartStrategy } -{-# INLINE buildWorkerOptions #-} +{-# INLINE buildWorkerOptions1 #-} --- | Builds a "WorkerOptions" record with defaults to create a worker process, + +-- | Builds a 'WorkerOptions' record with defaults to create a worker process, -- the defaults are: -- --- * A "Transient" "WorkerRestartStrategy" +-- * A 'Transient' 'WorkerRestartStrategy' -- --- * A "WorkerTerminationPolicy" of a 3 seconds timeout +-- * A 'WorkerTerminationPolicy' of a 3 seconds timeout -- -- * A _completion_ callback that just returns unit -- @@ -678,14 +990,36 @@ -- -- * A _failure_ callback that just returns unit -- --- This function is intended to be used in combination with "forkWorker", for --- creating a worker in an static supervision tree, use "workerSpecWithDefaults" --- instead. See the ... example for a demonstration. +-- This function is intended to be used in combination with 'forkWorker', for creating a +-- worker in an static supervision tree, use 'workerSpecWithDefaults' instead. See the +-- capataz-simple-example project in the examples directory for a demonstration. -- +-- @since 0.1.0.0 buildWorkerOptionsWithDefaults - :: WorkerName -- ^ Name used for telemetry purposes - -> IO () -- ^ IO sub-routine to be supervised - -> WorkerOptions + :: Monad m + => WorkerName -- ^ Name used for telemetry purposes + -> m () -- ^ 'IO' sub-routine to be supervised + -> WorkerOptions m buildWorkerOptionsWithDefaults wName wAction = - buildWorkerOptions wName wAction identity + buildWorkerOptions wName wAction id {-# INLINE buildWorkerOptionsWithDefaults #-} + +-- | Used for debugging purposes +getMaskingState :: MonadIO m => m UnsafeE.MaskingState +getMaskingState = liftIO UnsafeE.getMaskingState + +-- | Given we want to capture async exceptions to send them back to a supervisor +-- and we are running on masked states, we need to have a try that catches all +-- kinds of exceptions +unsafeTry :: (Exception e, MonadUnliftIO m) => m a -> m (Either e a) +unsafeTry action = withRunInIO $ \run -> UnsafeE.try (run action) + +-- | Given unliftio wraps exceptions in 3 layers of Exceptions, and we are using +-- vanilla exceptions, we need to make sure that we account for all different +-- exception types +fromAnyException :: Exception e => SomeException -> Maybe e +fromAnyException ex = case UnsafeE.fromException ex of + Just (UnsafeE.SomeAsyncException innerEx1) -> case cast innerEx1 of + Just (AsyncExceptionWrapper innerEx2) -> cast innerEx2 + Nothing -> cast innerEx1 + Nothing -> fromException ex diff --git a/src/Control/Concurrent/Capataz/Internal/Types/Lens.hs b/src/Control/Concurrent/Capataz/Internal/Types/Lens.hs --- a/src/Control/Concurrent/Capataz/Internal/Types/Lens.hs +++ b/src/Control/Concurrent/Capataz/Internal/Types/Lens.hs @@ -3,9 +3,10 @@ {-# LANGUAGE RecordWildCards #-} module Control.Concurrent.Capataz.Internal.Types.Lens where +import RIO + import Control.Concurrent.Capataz.Internal.Types -import Data.Time.Clock (NominalDiffTime) -import Protolude +import RIO.Time (NominalDiffTime) -------------------------------------------------------------------------------- @@ -40,9 +41,9 @@ -- supervisor. supervisorProcessSpecListL :: Functor f - => ([ProcessSpec] -> f [ProcessSpec]) - -> s - -> f s + => ([ProcessSpec m] -> f [ProcessSpec m]) + -> s m + -> f (s m) class HasSupervisorProcessTerminationOrder s where -- | Specifies order in which a supervisor is going to terminate its @@ -58,31 +59,31 @@ -- breach in a supervisor's error intensity. supervisorOnIntensityReachedL :: Functor f - => (IO () -> f (IO ())) - -> s - -> f s + => (m () -> f (m ())) + -> s m + -> f (s m) class HasSupervisorFailureCallback s where -- | Specifies callback sub-routine that gets executed when a supervisor -- fails. supervisorOnFailureL :: Functor f - => ((SomeException -> IO ()) -> f (SomeException -> IO ())) - -> s - -> f s + => ((SomeException -> m ()) -> f (SomeException -> m ())) + -> s m + -> f (s m) -instance HasSupervisorIntensity SupervisorOptions where +instance HasSupervisorIntensity (SupervisorOptions m) where supervisorIntensityL k SupervisorOptions {supervisorIntensity, ..} = fmap (\newSupIntensity -> SupervisorOptions { supervisorIntensity = newSupIntensity, .. }) (k supervisorIntensity) -instance HasSupervisorPeriodSeconds SupervisorOptions where +instance HasSupervisorPeriodSeconds (SupervisorOptions m) where supervisorPeriodSecondsL k SupervisorOptions {supervisorPeriodSeconds, ..} = fmap (\newSupPeriodSeconds -> SupervisorOptions { supervisorPeriodSeconds = newSupPeriodSeconds, .. }) (k supervisorPeriodSeconds) -instance HasSupervisorRestartStrategy SupervisorOptions where +instance HasSupervisorRestartStrategy (SupervisorOptions m) where supervisorRestartStrategyL k SupervisorOptions {supervisorRestartStrategy, ..} = fmap (\newSupRestartStrategy -> SupervisorOptions { supervisorRestartStrategy = newSupRestartStrategy, .. }) @@ -94,7 +95,7 @@ SupervisorOptions { supervisorProcessSpecList = newSupProcessSpecList, .. }) (k supervisorProcessSpecList) -instance HasSupervisorProcessTerminationOrder SupervisorOptions where +instance HasSupervisorProcessTerminationOrder (SupervisorOptions m) where supervisorProcessTerminationOrderL k SupervisorOptions {supervisorProcessTerminationOrder, ..} = fmap (\newSupProcessTerminationOrder -> SupervisorOptions { supervisorProcessTerminationOrder = newSupProcessTerminationOrder, .. }) @@ -112,17 +113,17 @@ -------------------- -instance HasSupervisorIntensity CapatazOptions where +instance HasSupervisorIntensity (CapatazOptions m) where supervisorIntensityL k CapatazOptions {supervisorIntensity, ..} = fmap (\newSupIntensity -> CapatazOptions { supervisorIntensity = newSupIntensity, .. }) (k supervisorIntensity) -instance HasSupervisorPeriodSeconds CapatazOptions where +instance HasSupervisorPeriodSeconds (CapatazOptions m) where supervisorPeriodSecondsL k CapatazOptions {supervisorPeriodSeconds, ..} = fmap (\newSupPeriodSeconds -> CapatazOptions { supervisorPeriodSeconds = newSupPeriodSeconds, .. }) (k supervisorPeriodSeconds) -instance HasSupervisorRestartStrategy CapatazOptions where +instance HasSupervisorRestartStrategy (CapatazOptions m) where supervisorRestartStrategyL k CapatazOptions {supervisorRestartStrategy, ..} = fmap (\newSupRestartStrategy -> CapatazOptions { supervisorRestartStrategy = newSupRestartStrategy, .. }) @@ -134,7 +135,7 @@ CapatazOptions { supervisorProcessSpecList = newSupProcessSpecList, .. }) (k supervisorProcessSpecList) -instance HasSupervisorProcessTerminationOrder CapatazOptions where +instance HasSupervisorProcessTerminationOrder (CapatazOptions m) where supervisorProcessTerminationOrderL k CapatazOptions {supervisorProcessTerminationOrder, ..} = fmap (\newSupProcessTerminationOrder -> CapatazOptions { supervisorProcessTerminationOrder = newSupProcessTerminationOrder, .. }) @@ -155,9 +156,9 @@ -- telemetry purposes (e.g. logging, monitoring, etc). onSystemEventL :: Functor f - => ((CapatazEvent -> IO ()) -> f (CapatazEvent -> IO ())) - -> CapatazOptions - -> f CapatazOptions + => ((CapatazEvent -> m ()) -> f (CapatazEvent -> m ())) + -> CapatazOptions m + -> f (CapatazOptions m) onSystemEventL k CapatazOptions { notifyEvent, ..} = fmap (\newNotifyEvent -> CapatazOptions {notifyEvent = newNotifyEvent, ..}) (k notifyEvent) @@ -171,10 +172,10 @@ -- worker "WorkerTerminationPolicy". -- workerOnFailureL - :: Functor f - => ((SomeException -> IO ()) -> f (SomeException -> IO ())) - -> WorkerOptions - -> f WorkerOptions + :: (Functor f) + => ((SomeException -> m ()) -> f (SomeException -> m ())) + -> WorkerOptions m + -> f (WorkerOptions m) workerOnFailureL k WorkerOptions { workerOnFailure, ..} = fmap (\newWorkerAction -> WorkerOptions {workerOnFailure = newWorkerAction, ..}) (k workerOnFailure) @@ -186,7 +187,7 @@ -- worker "WorkerTerminationPolicy". -- workerOnCompletionL - :: Functor f => (IO () -> f (IO ())) -> WorkerOptions -> f WorkerOptions + :: (Functor f) => (m () -> f (m ())) -> WorkerOptions m -> f (WorkerOptions m) workerOnCompletionL k WorkerOptions { workerOnCompletion, ..} = fmap (\newWorkerAction -> WorkerOptions {workerOnCompletion = newWorkerAction, ..}) (k workerOnCompletion) @@ -199,10 +200,9 @@ -- worker "WorkerTerminationPolicy". -- workerOnTerminationL - :: Functor f => (IO () -> f (IO ())) -> WorkerOptions -> f WorkerOptions + :: Functor f => (m () -> f (m ())) -> WorkerOptions m -> f (WorkerOptions m) workerOnTerminationL k WorkerOptions { workerOnTermination, ..} = fmap - ( \newWorkerAction -> - WorkerOptions {workerOnTermination = newWorkerAction, ..} + (\newWorkerAction -> WorkerOptions {workerOnTermination = newWorkerAction, ..} ) (k workerOnTermination) @@ -211,10 +211,10 @@ workerTerminationPolicyL :: Functor f => (WorkerTerminationPolicy -> f WorkerTerminationPolicy) - -> WorkerOptions - -> f WorkerOptions + -> WorkerOptions m + -> f (WorkerOptions m) workerTerminationPolicyL k WorkerOptions { workerTerminationPolicy, ..} = fmap - ( \newWorkerAction -> + (\newWorkerAction -> WorkerOptions {workerTerminationPolicy = newWorkerAction, ..} ) (k workerTerminationPolicy) @@ -224,10 +224,10 @@ workerRestartStrategyL :: Functor f => (WorkerRestartStrategy -> f WorkerRestartStrategy) - -> WorkerOptions - -> f WorkerOptions + -> WorkerOptions m + -> f (WorkerOptions m) workerRestartStrategyL k WorkerOptions { workerRestartStrategy, ..} = fmap - ( \newWorkerAction -> + (\newWorkerAction -> WorkerOptions {workerRestartStrategy = newWorkerAction, ..} ) (k workerRestartStrategy) diff --git a/src/Control/Concurrent/Capataz/Internal/Util.hs b/src/Control/Concurrent/Capataz/Internal/Util.hs --- a/src/Control/Concurrent/Capataz/Internal/Util.hs +++ b/src/Control/Concurrent/Capataz/Internal/Util.hs @@ -12,30 +12,28 @@ -} module Control.Concurrent.Capataz.Internal.Util where -import Protolude +import RIO +import qualified RIO.HashMap as HashMap +import qualified RIO.List as List +import qualified RIO.Text as T -import Control.Concurrent.STM (STM, atomically, retry) -import Control.Concurrent.STM.TVar (TVar, readTVar, writeTVar) -import Data.IORef (atomicModifyIORef', readIORef) -import qualified Data.Text as T -import Data.Time.Clock (getCurrentTime) +import RIO.Time (getCurrentTime) -import qualified Data.HashMap.Strict as HashMap -import GHC.Conc (labelThread) +import GHC.Conc (labelThread) import Control.Concurrent.Capataz.Internal.Types -- | Returns only the number of the ThreadId. getTidNumber :: ThreadId -> Maybe Text -getTidNumber tid = case T.words $ show tid of - (_:tidNumber:_) -> Just tidNumber - _ -> Nothing +getTidNumber tid = case T.words $ tshow tid of + (_ : tidNumber : _) -> Just tidNumber + _ -> Nothing -------------------------------------------------------------------------------- -- | Internal functions that overwrites the GHC thread name, for increasing -- traceability on GHC internals. -setProcessThreadName :: WorkerId -> WorkerName -> IO () +setProcessThreadName :: MonadIO m => WorkerId -> WorkerName -> m () setProcessThreadName workerId workerName = do tid <- myThreadId let workerIdentifier = @@ -43,55 +41,57 @@ "" T.unpack (getTidNumber tid) - labelThread tid workerIdentifier + liftIO $ labelThread tid workerIdentifier --- | Gets the "ProcessId" of both a Worker or Supervisor process. -getProcessId :: Process -> ProcessId +-- | Gets the 'ProcessId' of both a Worker or Supervisor process. +getProcessId :: Process m -> ProcessId getProcessId process = case process of WorkerProcess Worker { workerId } -> workerId SupervisorProcess Supervisor { supervisorId } -> supervisorId --- | Gets a supervised "Process" from a "Supervisor" instance. -fetchProcess :: SupervisorEnv -> ProcessId -> IO (Maybe Process) +-- | Gets a supervised 'Process' from a 'Supervisor' instance. +fetchProcess + :: MonadIO m => SupervisorEnv m -> ProcessId -> m (Maybe (Process m)) fetchProcess SupervisorEnv { supervisorProcessMap } processId = do processMap <- readIORef supervisorProcessMap case HashMap.lookup processId processMap of Just process -> return $ Just process _ -> return Nothing --- | Appends a new "Process" to the "Supervisor" existing process map. -appendProcessToMap :: SupervisorEnv -> Process -> IO () +-- | Appends a new 'Process' to the 'Supervisor' existing process map. +appendProcessToMap :: MonadIO m => SupervisorEnv m -> Process m -> m () appendProcessToMap SupervisorEnv { supervisorProcessMap } process = atomicModifyIORef' supervisorProcessMap (\processMap -> (appendProcess processMap, ())) where appendProcess = HashMap.alter (const $ Just process) (getProcessId process) --- | Removes a "Process" from a "Supervisor" existing process map. -removeProcessFromMap :: SupervisorEnv -> ProcessId -> IO () +-- | Removes a 'Process' from a 'Supervisor' existing process map. +removeProcessFromMap :: MonadIO m => SupervisorEnv m -> ProcessId -> m () removeProcessFromMap SupervisorEnv { supervisorProcessMap } processId = atomicModifyIORef' supervisorProcessMap - ( \processMap -> maybe (processMap, ()) - (const (HashMap.delete processId processMap, ())) - (HashMap.lookup processId processMap) + (\processMap -> maybe (processMap, ()) + (const (HashMap.delete processId processMap, ())) + (HashMap.lookup processId processMap) ) --- | Function to modify a "Supervisor" process map using a pure function. -resetProcessMap :: SupervisorEnv -> (ProcessMap -> ProcessMap) -> IO () +-- | Function to modify a 'Supervisor' process map using a pure function. +resetProcessMap + :: MonadIO m => SupervisorEnv m -> (ProcessMap m -> ProcessMap m) -> m () resetProcessMap SupervisorEnv { supervisorProcessMap } processMapFn = atomicModifyIORef' supervisorProcessMap (\processMap -> (processMapFn processMap, ())) --- | Function to get a snapshot of a "Supervisor" process map. -readProcessMap :: SupervisorEnv -> IO ProcessMap +-- | Function to get a snapshot of a 'Supervisor' process map. +readProcessMap :: MonadIO m => SupervisorEnv m -> m (ProcessMap m) readProcessMap SupervisorEnv { supervisorProcessMap } = readIORef supervisorProcessMap --- | Returns all processes of a "Supervisor" by "ProcessTerminationOrder". This --- is used on "AllForOne" restarts and shutdown operations. +-- | Returns all processes of a 'Supervisor' by 'ProcessTerminationOrder'. This +-- is used on 'AllForOne' restarts and shutdown operations. sortProcessesByTerminationOrder - :: ProcessTerminationOrder -> ProcessMap -> [Process] + :: ProcessTerminationOrder -> ProcessMap m -> [Process m] sortProcessesByTerminationOrder terminationOrder processMap = case terminationOrder of OldestFirst -> workers @@ -103,28 +103,30 @@ processCreationTime (SupervisorProcess Supervisor { supervisorCreationTime }) = supervisorCreationTime - workers = sortBy (comparing processCreationTime) (HashMap.elems processMap) + workers = + List.sortBy (comparing processCreationTime) (HashMap.elems processMap) -------------------------------------------------------------------------------- --- | Returns the "SupervisorStatus", this sub-routine will retry transaction --- until its associated "Supervisor" has a status different from "Initializing". +-- | Returns the 'SupervisorStatus', this sub-routine will retry transaction +-- until its associated 'Supervisor' has a status different from 'Initializing'. readSupervisorStatusSTM :: TVar SupervisorStatus -> STM SupervisorStatus readSupervisorStatusSTM statusVar = do status <- readTVar statusVar - if status == Initializing then retry else return status + if status == Initializing then retrySTM else return status --- | Executes transaction that returns the "SupervisorStatus". -readSupervisorStatus :: SupervisorEnv -> IO SupervisorStatus +-- | Executes transaction that returns the 'SupervisorStatus'. +readSupervisorStatus :: MonadIO m => SupervisorEnv m -> m SupervisorStatus readSupervisorStatus SupervisorEnv { supervisorStatusVar } = atomically $ readTVar supervisorStatusVar --- | Modifes the "Supervisor" status. +-- | Modifes the 'Supervisor' status. -- --- IMPORTANT: this is the only function that should be used for this purpose +-- __IMPORTANT__ This is the only function that should be used for this purpose -- given it has the side-effect of notifying a status change via the --- "notifyEvent" sub-routine, orginally given in the "CapatazOption" record. -writeSupervisorStatus :: SupervisorEnv -> SupervisorStatus -> IO () +-- 'notifyEvent' sub-routine, orginally given in the 'CapatazOption' record. +writeSupervisorStatus + :: MonadIO m => SupervisorEnv m -> SupervisorStatus -> m () writeSupervisorStatus SupervisorEnv { supervisorId, supervisorName, supervisorStatusVar, notifyEvent } newSupervisorStatus = do @@ -142,9 +144,9 @@ , eventTime } --- | Used from public API functions to send "ControlAction" messages to a +-- | Used from public API functions to send 'ControlAction' messages to a -- Supervisor thread loop. -sendControlMsg :: SupervisorEnv -> ControlAction -> IO () +sendControlMsg :: SupervisorEnv m -> ControlAction m -> m () sendControlMsg SupervisorEnv { supervisorNotify } ctrlMsg = supervisorNotify (ControlAction ctrlMsg) @@ -152,22 +154,23 @@ -- Supervisor thread loop, it receives an IO sub-routine that expects an IO -- operation that blocks a thread until the message is done. sendSyncControlMsg - :: SupervisorEnv - -> (IO () -> ControlAction) -- ^ Blocking sub-routine used from the caller - -> IO () + :: MonadIO m + => SupervisorEnv m + -> (m () -> ControlAction m) -- ^ Blocking sub-routine used from the caller + -> m () sendSyncControlMsg SupervisorEnv { supervisorNotify } mkCtrlMsg = do result <- newEmptyMVar supervisorNotify (ControlAction $ mkCtrlMsg (putMVar result ())) takeMVar result --- | Utility function to transform a "CapatazOptions" record to a --- "SupervisorOptions" record. -capatazOptionsToSupervisorOptions :: CapatazOptions -> SupervisorOptions +-- | Utility function to transform a 'CapatazOptions' record to a +-- 'SupervisorOptions' record. +capatazOptionsToSupervisorOptions :: CapatazOptions m -> SupervisorOptions m capatazOptionsToSupervisorOptions CapatazOptions {..} = SupervisorOptions {..} --- | Utility function to transform a "SupervisorEnv" record to a --- "ParentSupervisorEnv" record; used on functions where supervision of +-- | Utility function to transform a 'SupervisorEnv' record to a +-- 'ParentSupervisorEnv' record; used on functions where supervision of -- supervisors is managed. -toParentSupervisorEnv :: SupervisorEnv -> ParentSupervisorEnv +toParentSupervisorEnv :: SupervisorEnv m -> ParentSupervisorEnv m toParentSupervisorEnv SupervisorEnv { supervisorId, supervisorName, supervisorNotify, notifyEvent } = ParentSupervisorEnv {..} diff --git a/src/Control/Concurrent/Capataz/Internal/Worker.hs b/src/Control/Concurrent/Capataz/Internal/Worker.hs --- a/src/Control/Concurrent/Capataz/Internal/Worker.hs +++ b/src/Control/Concurrent/Capataz/Internal/Worker.hs @@ -7,10 +7,9 @@ -} module Control.Concurrent.Capataz.Internal.Worker where -import Protolude +import RIO -import Control.Concurrent.Async (asyncWithUnmask) -import Data.Time.Clock (getCurrentTime) +import RIO.Time (getCurrentTime) import qualified Data.UUID.V4 as UUID @@ -21,19 +20,18 @@ -- | Decorates the given @IO ()@ sub-routine with failure handling workerMain - :: ParentSupervisorEnv - -> WorkerOptions + :: (MonadUnliftIO m) + => ParentSupervisorEnv m + -> WorkerOptions m -> WorkerId -> RestartCount - -> IO Worker + -> m (Worker m) workerMain env@ParentSupervisorEnv { supervisorNotify } workerOptions@WorkerOptions { workerName, workerAction } workerId restartCount = do workerCreationTime <- getCurrentTime workerAsync <- asyncWithUnmask $ \unmask -> do - - eResult <- try $ do - Util.setProcessThreadName workerId workerName - unmask workerAction + Util.setProcessThreadName workerId workerName + eResult <- unsafeTry $ unmask (workerAction workerId) resultEvent <- case eResult of Left err -> Process.handleProcessException unmask @@ -62,14 +60,15 @@ -- this is different from the public @forkWorker@ function which sends a message -- to the capataz loop forkWorker - :: ParentSupervisorEnv - -> WorkerOptions + :: (MonadUnliftIO m) + => ParentSupervisorEnv m + -> WorkerOptions m -> Maybe (WorkerId, RestartCount) - -> IO Worker + -> m (Worker m) forkWorker env workerOptions mRestartInfo = do (workerId, restartCount) <- case mRestartInfo of Just (workerId, restartCount) -> pure (workerId, restartCount) - Nothing -> (,) <$> UUID.nextRandom <*> pure 0 + Nothing -> (,) <$> liftIO UUID.nextRandom <*> pure 0 worker <- workerMain env workerOptions workerId restartCount Process.notifyProcessStarted mRestartInfo env (WorkerProcess worker) diff --git a/src/Control/Concurrent/Capataz/Lens.hs b/src/Control/Concurrent/Capataz/Lens.hs --- a/src/Control/Concurrent/Capataz/Lens.hs +++ b/src/Control/Concurrent/Capataz/Lens.hs @@ -17,5 +17,4 @@ , workerRestartStrategyL , workerTerminationPolicyL ) -import Lens.Micro as X (set, (&), (.~), (^.)) -import Lens.Micro.Extras as X (view) +import RIO as X (set, view, (&), (^.)) diff --git a/src/Control/Concurrent/Capataz/Util.hs b/src/Control/Concurrent/Capataz/Util.hs new file mode 100644 --- /dev/null +++ b/src/Control/Concurrent/Capataz/Util.hs @@ -0,0 +1,141 @@ +{-# LANGUAGE NamedFieldPuns #-} + +{-# LANGUAGE NoImplicitPrelude #-} +module Control.Concurrent.Capataz.Util ( + buildLogWorkerSpec + , buildLogWorkerOptions + ) where + +import Control.Concurrent.Capataz + ( ProcessSpec + , WorkerName + , WorkerOptions + , WorkerRestartStrategy (Permanent) + , buildWorkerOptions + , workerRestartStrategyL + , workerSpec + ) +import RIO + +data LogMsg + = LogMsg + { + lmCallStack :: !CallStack + , lmLogSource :: !LogSource + , lmLogLevel :: !LogLevel + , lmPayload :: !Utf8Builder + } + + +runLoggerThread :: MonadUnliftIO m => LogOptions -> TBQueue LogMsg -> m a +runLoggerThread logOptions inputQueue = withLogFunc logOptions $ \logFunc -> + runRIO logFunc $ forever $ do + logMsg <- atomically $ readTBQueue inputQueue + let LogMsg { lmLogSource, lmLogLevel, lmPayload } = logMsg + -- TODO: create a ticket for a logGenericWithStack function in rio + logGeneric lmLogSource lmLogLevel lmPayload + +-- | Builds a 'ProcessSpec' that spawns a thread that logs messages written with +-- the returned 'LogFunc'. Use this function when your want your logger to be +-- part of a static supervision tree. +-- +-- +-- __IMPORTANT__ If you use the returned 'LogFunc' to log functions and the +-- 'ProcessSpec' is not used in a supervision tree, your logging won't work and +-- your application will eventually block the current thread when logging. +-- +-- A minimal example: +-- +-- @ +-- {-# LANGUAGE OverloadedStrings #-} +-- {-# LANGUAGE NoImplicitPrelude #-} +-- import RIO +-- import Capataz +-- +-- main :: IO () +-- main = do +-- logOptions <- logOptionsHandle stdout True +-- (loggerSpec, logFunc) <- buildLogWorkerSpec logOptions "app-logger" 100 id +-- runRIO logFunc $ do +-- bracket (forkCapataz "application" (set supervisorProcessSpecListL [loggerSpec])) +-- terminateCapataz_ $ \_capataz -> do +-- logInfo "this log message is written by a dedicated supervised thread" +-- threadDelay 1000100 +-- @ +-- +-- @since 0.2.0.0 +buildLogWorkerSpec + :: (MonadUnliftIO m, MonadIO m0) + => LogOptions -- ^ options for the 'LogFunc' instance + -> WorkerName -- ^ name of the logger worker process + -> Int -- ^ how many log messages can be in-flight when writer is slow? + -> (WorkerOptions m -> WorkerOptions m) -- ^ worker process modifier + -> m0 (ProcessSpec m, LogFunc) +buildLogWorkerSpec logOptions procName bufferSize modOptions = do + inputQueue <- newTBQueueIO bufferSize + let myLogFunc = + mkLogFunc $ \lmCallStack lmLogSource lmLogLevel lmPayload -> atomically + (writeTBQueue + inputQueue + LogMsg {lmCallStack , lmLogSource , lmLogLevel , lmPayload } + ) + + loggerSpec = workerSpec + procName + (runLoggerThread logOptions inputQueue) + (modOptions . set workerRestartStrategyL Permanent) + + return (loggerSpec, myLogFunc) + +-- | Builds a 'WorkerOptions' record that spawns a thread that logs messages +-- written with the returned 'LogFunc'. Use this function if you want to build a +-- logger thread dynamically via 'forkWorker'. +-- +-- +-- __IMPORTANT__ If you use the returned 'LogFunc' to log functions and the +-- 'WorkerOptions' is not used in a 'forkWorker' call, your logging won't work +-- and your application will eventually block the current thread when logging. +-- +-- A minimal example: +-- +-- @ +-- {-# LANGUAGE OverloadedStrings #-} +-- {-# LANGUAGE NoImplicitPrelude #-} +-- import RIO +-- import Capataz +-- +-- main :: IO () +-- main = do +-- logOptions <- logOptionsHandle stdout True +-- (loggerOptions, logFunc) <- buildLogWorkerOptions logOptions "app-logger" 100 id +-- runRIO logFunc $ do +-- bracket (forkCapataz "application" id) +-- terminateCapataz_ $ \capataz -> do +-- _workerId <- forkWorker loggerOptions capataz +-- logInfo "this log message is written by a dedicated supervised thread" +-- threadDelay 1000100 +-- @ +-- +-- @since 0.2.0.0 +buildLogWorkerOptions + :: (MonadUnliftIO m, MonadIO m0) + => LogOptions + -> WorkerName + -> Int + -> (WorkerOptions m -> WorkerOptions m) + -> m0 (WorkerOptions m, LogFunc) +buildLogWorkerOptions logOptions procName bufferSize modOptions = do + inputQueue <- newTBQueueIO bufferSize + let myLogFunc = + mkLogFunc $ \lmCallStack lmLogSource lmLogLevel lmPayload -> atomically + (writeTBQueue + inputQueue + LogMsg {lmCallStack , lmLogSource , lmLogLevel , lmPayload } + ) + + loggerSpec = buildWorkerOptions + procName + (runLoggerThread logOptions inputQueue) + (modOptions . set workerRestartStrategyL Permanent) + + return (loggerSpec, myLogFunc) diff --git a/test/testsuite/Control/Concurrent/Capataz/SupervisorTest.hs b/test/testsuite/Control/Concurrent/Capataz/SupervisorTest.hs --- a/test/testsuite/Control/Concurrent/Capataz/SupervisorTest.hs +++ b/test/testsuite/Control/Concurrent/Capataz/SupervisorTest.hs @@ -3,7 +3,7 @@ {-# LANGUAGE OverloadedStrings #-} module Control.Concurrent.Capataz.SupervisorTest where -import Protolude +import RIO import Control.Concurrent.Capataz (set) import qualified Control.Concurrent.Capataz as SUT @@ -17,11 +17,11 @@ "supervision trees" [ testCase "initialize and teardown of supervision tree works as expected" $ testCapatazStreamWithOptions - ( & set + (& set SUT.supervisorProcessSpecListL [ SUT.supervisorSpec "tree-1" - ( set + (set SUT.supervisorProcessSpecListL [ SUT.workerSpecWithDefaults "1-A" (forever $ threadDelay 10001000) @@ -31,7 +31,7 @@ ) , SUT.supervisorSpec "tree-2" - ( set + (set SUT.supervisorProcessSpecListL [ SUT.workerSpecWithDefaults "2-A" (forever $ threadDelay 10001000) @@ -85,7 +85,7 @@ , testCase "supervision sub-tree gets restarted on failure" $ do failingAction <- mkFailingSubRoutine 2 testCapatazStreamWithOptions - ( set SUT.supervisorIntensityL 3 . set + (set SUT.supervisorIntensityL 3 . set SUT.supervisorProcessSpecListL [ SUT.supervisorSpec "tree-1" @@ -123,11 +123,11 @@ ) , SUT.supervisorSpec "tree-2" - ( set + (set SUT.supervisorProcessSpecListL [ SUT.workerSpec "stable-worker" (forever $ threadDelay 1000100) - identity + id ] ) ] diff --git a/test/testsuite/Control/Concurrent/CapatazTest.hs b/test/testsuite/Control/Concurrent/CapatazTest.hs --- a/test/testsuite/Control/Concurrent/CapatazTest.hs +++ b/test/testsuite/Control/Concurrent/CapatazTest.hs @@ -20,13 +20,13 @@ -} module Control.Concurrent.CapatazTest (tests) where -import Protolude +import RIO +import Control.Exception (ErrorCall (..)) + import Test.Tasty (TestTree, testGroup) import Test.Tasty.HUnit (testCase) -import Control.Concurrent.STM.TVar (modifyTVar', newTVarIO, readTVar) - import Control.Concurrent.Capataz (set) import qualified Control.Concurrent.Capataz as SUT @@ -63,10 +63,10 @@ "capataz with processSpecList" [ testCase "initialize and teardown of workers works as expected" $ testCapatazStreamWithOptions - ( set + (set SUT.supervisorProcessSpecListL - [ SUT.workerSpec "A" (forever $ threadDelay 10001000) identity - , SUT.workerSpec "B" (forever $ threadDelay 10001000) identity + [ SUT.workerSpec "A" (forever $ threadDelay 10001000) id + , SUT.workerSpec "B" (forever $ threadDelay 10001000) id ] ) [ assertWorkerStarted "A" @@ -100,11 +100,11 @@ testCapatazStreamWithOptions (set SUT.supervisorOnIntensityReachedL signalIntensityReached) [] - ( \capataz -> do + (\capataz -> do _workerId <- SUT.forkWorker - ( SUT.buildWorkerOptions "test-worker" - (throwIO RestartingWorkerError) - identity + (SUT.buildWorkerOptions "test-worker" + (throwIO RestartingWorkerError) + id ) capataz waitTillIntensityReached @@ -126,9 +126,9 @@ "does execute callback when worker sub-routine is completed" $ testCapatazStream [] - ( \capataz -> do + (\capataz -> do _workerId <- SUT.forkWorker - ( SUT.buildWorkerOptions + (SUT.buildWorkerOptions "test-worker" (return ()) (set SUT.workerRestartStrategyL SUT.Temporary) @@ -147,11 +147,11 @@ , testCase "does not execute callback when worker sub-routine fails" $ testCapatazStream [] - ( \capataz -> do + (\capataz -> do _workerId <- SUT.forkWorker - ( SUT.buildWorkerOptions "test-worker" - (throwIO RestartingWorkerError) - identity + (SUT.buildWorkerOptions "test-worker" + (throwIO RestartingWorkerError) + id ) capataz return () @@ -163,7 +163,7 @@ , assertEventType ProcessFailed ] [assertEventType CapatazTerminated] - ( Just $ not . andP + (Just $ not . andP [ assertEventType ProcessCallbackExecuted , assertCallbackType SUT.OnCompletion ] @@ -172,9 +172,9 @@ "does not execute callback when worker sub-routine is terminated" $ testCapatazStream [] - ( \capataz -> do + (\capataz -> do workerId <- SUT.forkWorker - ( SUT.buildWorkerOptions + (SUT.buildWorkerOptions "test-worker" (forever $ threadDelay 1000100) (set SUT.workerRestartStrategyL SUT.Temporary) @@ -189,7 +189,7 @@ ) [assertEventType ProcessTerminated] [assertEventType CapatazTerminated] - ( Just $ not . andP + (Just $ not . andP [ assertEventType ProcessCallbackExecuted , assertCallbackType SUT.OnCompletion ] @@ -197,9 +197,9 @@ , testCase "treats as worker sub-routine failed if callback fails" $ testCapatazStream [] - ( \capataz -> do + (\capataz -> do _workerId <- SUT.forkWorker - ( SUT.buildWorkerOptions + (SUT.buildWorkerOptions "test-worker" (return ()) ( set SUT.workerRestartStrategyL SUT.Temporary @@ -228,9 +228,9 @@ [ testCase "does execute callback when worker sub-routine fails" $ testCapatazStream [] - ( \capataz -> do + (\capataz -> do _workerId <- SUT.forkWorker - ( SUT.buildWorkerOptions + (SUT.buildWorkerOptions "test-worker" (throwIO RestartingWorkerError) (set SUT.workerRestartStrategyL SUT.Temporary) @@ -250,9 +250,9 @@ "does not execute callback when worker sub-routine is completed" $ testCapatazStream [] - ( \capataz -> do + (\capataz -> do _workerId <- SUT.forkWorker - ( SUT.buildWorkerOptions + (SUT.buildWorkerOptions "test-worker" (return ()) (set SUT.workerRestartStrategyL SUT.Temporary) @@ -278,9 +278,9 @@ "does not execute callback when worker sub-routine is terminated" $ testCapatazStream [] - ( \capataz -> do + (\capataz -> do workerId <- SUT.forkWorker - ( SUT.buildWorkerOptions + (SUT.buildWorkerOptions "test-worker" (forever $ threadDelay 1000100) (set SUT.workerRestartStrategyL SUT.Temporary) @@ -303,9 +303,9 @@ , testCase "treats as worker sub-routine failed if callback fails" $ testCapatazStream [] - ( \capataz -> do + (\capataz -> do _workerId <- SUT.forkWorker - ( SUT.buildWorkerOptions + (SUT.buildWorkerOptions "test-worker" (throwIO RestartingWorkerError) ( set SUT.workerRestartStrategyL SUT.Temporary @@ -337,9 +337,9 @@ "gets brutally killed when TimeoutSeconds termination policy is not met" $ testCapatazStream [] - ( \capataz -> do + (\capataz -> do workerId <- SUT.forkWorker - ( SUT.buildWorkerOptions + (SUT.buildWorkerOptions "test-worker" (forever $ threadDelay 10001000) ( set SUT.workerRestartStrategyL SUT.Temporary @@ -372,9 +372,9 @@ "does execute callback when worker sub-routine is terminated" $ testCapatazStream [] - ( \capataz -> do + (\capataz -> do workerId <- SUT.forkWorker - ( SUT.buildWorkerOptions + (SUT.buildWorkerOptions "test-worker" (forever $ threadDelay 1000100) (set SUT.workerRestartStrategyL SUT.Temporary) @@ -398,9 +398,9 @@ "does not execute callback when worker sub-routine is completed" $ testCapatazStream [] - ( \capataz -> do + (\capataz -> do _workerId <- SUT.forkWorker - ( SUT.buildWorkerOptions + (SUT.buildWorkerOptions "test-worker" (return ()) (set SUT.workerRestartStrategyL SUT.Temporary) @@ -410,7 +410,7 @@ ) [assertEventType ProcessCompleted] [] - ( Just $ not . andP + (Just $ not . andP [ assertEventType ProcessCallbackExecuted , assertCallbackType SUT.OnTermination ] @@ -418,9 +418,9 @@ , testCase "does not execute callback when worker sub-routine fails" $ testCapatazStream [] - ( \capataz -> do + (\capataz -> do _workerId <- SUT.forkWorker - ( SUT.buildWorkerOptions + (SUT.buildWorkerOptions "test-worker" (throwIO (ErrorCall "surprise!")) (set SUT.workerRestartStrategyL SUT.Temporary) @@ -430,7 +430,7 @@ ) [assertEventType ProcessFailed] [] - ( Just $ not . andP + (Just $ not . andP [ assertEventType ProcessCallbackExecuted , assertCallbackType SUT.OnTermination ] @@ -438,9 +438,9 @@ , testCase "treats as worker sub-routine failed if callback fails" $ testCapatazStream [] - ( \capataz -> do + (\capataz -> do workerId <- SUT.forkWorker - ( SUT.buildWorkerOptions + (SUT.buildWorkerOptions "test-worker" (forever $ threadDelay 10001000) ( set SUT.workerRestartStrategyL SUT.Temporary @@ -472,9 +472,9 @@ "with transient strategy" [ testCase "does not restart worker on completion" $ testCapatazStream [] - ( \capataz -> do + (\capataz -> do _workerId <- SUT.forkWorker - ( SUT.buildWorkerOptions + (SUT.buildWorkerOptions "test-worker" (return ()) (set SUT.workerRestartStrategyL SUT.Transient) @@ -487,9 +487,9 @@ (Just $ not . assertEventType ProcessRestarted) , testCase "does not restart worker on termination" $ testCapatazStream [] - ( \capataz -> do + (\capataz -> do workerId <- SUT.forkWorker - ( SUT.buildWorkerOptions + (SUT.buildWorkerOptions "test-worker" (forever $ threadDelay 1000100) (set SUT.workerRestartStrategyL SUT.Transient) @@ -502,10 +502,10 @@ (Just $ not . assertEventType ProcessRestarted) , testCase "does restart on failure" $ testCapatazStream [] - ( \capataz -> do + (\capataz -> do subRoutineAction <- mkFailingSubRoutine 1 _workerId <- SUT.forkWorker - ( SUT.buildWorkerOptions + (SUT.buildWorkerOptions "test-worker" subRoutineAction (set SUT.workerRestartStrategyL SUT.Transient) @@ -522,10 +522,10 @@ , testCase "does increase restart count on multiple worker failures" $ testCapatazStream [] - ( \capataz -> do + (\capataz -> do subRoutineAction <- mkFailingSubRoutine 2 _workerId <- SUT.forkWorker - ( SUT.buildWorkerOptions + (SUT.buildWorkerOptions "test-worker" subRoutineAction (set SUT.workerRestartStrategyL SUT.Transient) @@ -545,10 +545,10 @@ "with permanent strategy" [ testCase "does restart worker on completion" $ testCapatazStream [] - ( \capataz -> do + (\capataz -> do subRoutineAction <- mkCompletingOnceSubRoutine _workerId <- SUT.forkWorker - ( SUT.buildWorkerOptions + (SUT.buildWorkerOptions "test-worker" subRoutineAction (set SUT.workerRestartStrategyL SUT.Permanent) @@ -566,13 +566,13 @@ "does not increase restart count on multiple worker completions" $ testCapatazStream [] - ( \capataz -> do + (\capataz -> do -- Note the number is two (2) given the assertion list has two `ProcessRestarted` assertions let expectedRestartCount = 2 subRoutineAction <- mkCompletingBeforeNRestartsSubRoutine expectedRestartCount _workerId <- SUT.forkWorker - ( SUT.buildWorkerOptions + (SUT.buildWorkerOptions "test-worker" subRoutineAction (set SUT.workerRestartStrategyL SUT.Permanent) @@ -589,9 +589,9 @@ Nothing , testCase "does restart on worker termination" $ testCapatazStream [] - ( \capataz -> do + (\capataz -> do workerId <- SUT.forkWorker - ( SUT.buildWorkerOptions + (SUT.buildWorkerOptions "test-worker" (forever $ threadDelay 10001000) (set SUT.workerRestartStrategyL SUT.Permanent) @@ -611,12 +611,12 @@ atomically (modifyTVar' terminationCountVar (+ 1)) waitWorkerTermination i = atomically $ do n <- readTVar terminationCountVar - when (n /= i) retry + when (n /= i) retrySTM testCapatazStream [] - ( \capataz -> do + (\capataz -> do workerId <- SUT.forkWorker - ( SUT.buildWorkerOptions + (SUT.buildWorkerOptions "test-worker" (forever $ threadDelay 10001000) ( set SUT.workerRestartStrategyL SUT.Permanent @@ -645,10 +645,10 @@ Nothing , testCase "does restart on worker failure" $ testCapatazStream [] - ( \capataz -> do + (\capataz -> do subRoutineAction <- mkFailingSubRoutine 1 _workerId <- SUT.forkWorker - ( SUT.buildWorkerOptions + (SUT.buildWorkerOptions "test-worker" subRoutineAction (set SUT.workerRestartStrategyL SUT.Permanent) @@ -665,10 +665,10 @@ , testCase "does increase restart count on multiple worker failures" $ testCapatazStream [] - ( \capataz -> do + (\capataz -> do subRoutineAction <- mkFailingSubRoutine 2 _workerId <- SUT.forkWorker - ( SUT.buildWorkerOptions + (SUT.buildWorkerOptions "test-worker" subRoutineAction (set SUT.workerRestartStrategyL SUT.Permanent) @@ -688,9 +688,9 @@ "with temporary strategy" [ testCase "does not restart on worker completion" $ testCapatazStream [] - ( \capataz -> do + (\capataz -> do _workerId <- SUT.forkWorker - ( SUT.buildWorkerOptions + (SUT.buildWorkerOptions "test-worker" (return ()) (set SUT.workerRestartStrategyL SUT.Temporary) @@ -703,9 +703,9 @@ (Just $ not . assertEventType ProcessRestarted) , testCase "does not restart on worker termination" $ testCapatazStream [] - ( \capataz -> do + (\capataz -> do workerId <- SUT.forkWorker - ( SUT.buildWorkerOptions + (SUT.buildWorkerOptions "test-worker" (forever $ threadDelay 1000100) (set SUT.workerRestartStrategyL SUT.Temporary) @@ -719,11 +719,11 @@ (Just $ not . assertEventType ProcessRestarted) , testCase "does not restart on worker failure" $ testCapatazStream [] - ( \capataz -> do + (\capataz -> do _workerId <- SUT.forkWorker - ( SUT.buildWorkerOptions + (SUT.buildWorkerOptions "failing-worker" - (panic "worker failed!") + (error "worker failed!") (set SUT.workerRestartStrategyL SUT.Temporary) ) capataz @@ -739,9 +739,9 @@ [ testCase "terminates all supervised worker sub-routines on teardown" $ testCapatazStream [] - ( \capataz -> do + (\capataz -> do _workerA <- SUT.forkWorker - ( SUT.buildWorkerOptions + (SUT.buildWorkerOptions "A" (forever $ threadDelay 1000100) (set SUT.workerRestartStrategyL SUT.Permanent) @@ -750,7 +750,7 @@ _workerB <- SUT.forkWorker - ( SUT.buildWorkerOptions + (SUT.buildWorkerOptions "B" (forever $ threadDelay 1000100) (set SUT.workerRestartStrategyL SUT.Permanent) @@ -773,9 +773,9 @@ $ testCapatazStreamWithOptions (set SUT.supervisorRestartStrategyL SUT.OneForOne) [] - ( \capataz -> do + (\capataz -> do _workerA <- SUT.forkWorker - ( SUT.buildWorkerOptions + (SUT.buildWorkerOptions "A" (forever $ threadDelay 1000100) (set SUT.workerRestartStrategyL SUT.Temporary) @@ -785,7 +785,7 @@ ioB <- mkFailingSubRoutine 1 _workerB <- SUT.forkWorker - ( SUT.buildWorkerOptions + (SUT.buildWorkerOptions "B" (forever $ ioB >> threadDelay 1000100) (set SUT.workerRestartStrategyL SUT.Permanent) @@ -796,7 +796,7 @@ ) [andP [assertEventType ProcessRestarted, assertProcessName "B"]] [] - ( Just $ not . andP + (Just $ not . andP [assertEventType ProcessRestarted, assertProcessName "A"] ) ] @@ -804,20 +804,20 @@ "with all for one supervisor restart strategy with newest first order" [ testCase "does terminate all other workers that did not fail" $ testCapatazStreamWithOptions - ( \supOptions -> + (\supOptions -> supOptions & set SUT.supervisorRestartStrategyL SUT.AllForOne & set SUT.supervisorProcessTerminationOrderL SUT.OldestFirst ) [] - ( \capataz -> do + (\capataz -> do -- This lockVar guarantees that workerB executes before workerA lockVar <- newEmptyMVar ioA <- mkFailingSubRoutine 1 _workerA <- SUT.forkWorker - ( SUT.buildWorkerOptions + (SUT.buildWorkerOptions "A" (forever $ readMVar lockVar >> ioA) (set SUT.workerRestartStrategyL SUT.Permanent) @@ -825,7 +825,7 @@ capataz _workerB <- SUT.forkWorker - ( SUT.buildWorkerOptions + (SUT.buildWorkerOptions "B" (putMVar lockVar () >> forever (threadDelay 10)) (set SUT.workerRestartStrategyL SUT.Permanent) @@ -845,19 +845,19 @@ Nothing , testCase "does not restart workers that are temporary" $ testCapatazStreamWithOptions - ( \supOptions -> + (\supOptions -> supOptions & set SUT.supervisorRestartStrategyL SUT.AllForOne & set SUT.supervisorProcessTerminationOrderL SUT.OldestFirst ) [] - ( \capataz -> do + (\capataz -> do lockVar <- newEmptyMVar ioA <- mkFailingSubRoutine 1 _workerA <- SUT.forkWorker - ( SUT.buildWorkerOptions + (SUT.buildWorkerOptions "A" (forever $ readMVar lockVar >> ioA) (set SUT.workerRestartStrategyL SUT.Permanent) @@ -865,7 +865,7 @@ capataz _workerB <- SUT.forkWorker - ( SUT.buildWorkerOptions + (SUT.buildWorkerOptions "B" (putMVar lockVar () >> forever (threadDelay 10)) (set SUT.workerRestartStrategyL SUT.Temporary) @@ -881,24 +881,24 @@ , andP [assertEventType ProcessTerminated, assertProcessName "B"] ] [] - ( Just $ not . andP + (Just $ not . andP [assertEventType ProcessRestarted, assertProcessName "B"] ) , testCase "restarts workers that are not temporary" $ testCapatazStreamWithOptions - ( \supOptions -> + (\supOptions -> supOptions & set SUT.supervisorRestartStrategyL SUT.AllForOne & set SUT.supervisorProcessTerminationOrderL SUT.NewestFirst ) [] - ( \capataz -> do + (\capataz -> do ioA <- mkFailingSubRoutine 1 lockVar <- newEmptyMVar _workerA <- SUT.forkWorker - ( SUT.buildWorkerOptions + (SUT.buildWorkerOptions "A" (forever $ readMVar lockVar >> ioA) (set SUT.workerRestartStrategyL SUT.Permanent) @@ -906,7 +906,7 @@ capataz _workerB <- SUT.forkWorker - ( SUT.buildWorkerOptions + (SUT.buildWorkerOptions "B" (putMVar lockVar () >> forever (threadDelay 10)) (set SUT.workerRestartStrategyL SUT.Transient) @@ -925,20 +925,20 @@ "with all for one capataz restart strategy with oldest first order" [ testCase "does not restart workers that are temporary" $ testCapatazStreamWithOptions - ( \supOptions -> + (\supOptions -> supOptions & set SUT.supervisorRestartStrategyL SUT.AllForOne & set SUT.supervisorProcessTerminationOrderL SUT.OldestFirst ) [] - ( \capataz -> do + (\capataz -> do ioA <- mkFailingSubRoutine 1 -- This lockVar guarantees that workerB executes before workerA lockVar <- newEmptyMVar _workerA <- SUT.forkWorker - ( SUT.buildWorkerOptions + (SUT.buildWorkerOptions "A" (forever $ readMVar lockVar >> ioA) (set SUT.workerRestartStrategyL SUT.Permanent) @@ -946,7 +946,7 @@ capataz _workerB <- SUT.forkWorker - ( SUT.buildWorkerOptions + (SUT.buildWorkerOptions "B" (putMVar lockVar () >> forever (threadDelay 10)) (set SUT.workerRestartStrategyL SUT.Temporary) @@ -957,25 +957,25 @@ ) [andP [assertEventType ProcessRestarted, assertProcessName "A"]] [] - ( Just $ not . andP + (Just $ not . andP [assertEventType ProcessRestarted, assertProcessName "B"] ) , testCase "restarts workers that are not temporary" $ testCapatazStreamWithOptions - ( \supOptions -> + (\supOptions -> supOptions & set SUT.supervisorRestartStrategyL SUT.AllForOne & set SUT.supervisorProcessTerminationOrderL SUT.OldestFirst ) [] - ( \capataz -> do + (\capataz -> do ioA <- mkFailingSubRoutine 1 -- This lockVar guarantees that workerB executes before workerA lockVar <- newEmptyMVar _workerA <- SUT.forkWorker - ( SUT.buildWorkerOptions + (SUT.buildWorkerOptions "A" (forever $ readMVar lockVar >> ioA) (set SUT.workerRestartStrategyL SUT.Permanent) @@ -983,7 +983,7 @@ capataz _workerB <- SUT.forkWorker - ( SUT.buildWorkerOptions + (SUT.buildWorkerOptions "B" (putMVar lockVar () >> forever (threadDelay 10)) (set SUT.workerRestartStrategyL SUT.Transient) diff --git a/test/testsuite/Main.hs b/test/testsuite/Main.hs --- a/test/testsuite/Main.hs +++ b/test/testsuite/Main.hs @@ -5,17 +5,16 @@ {-# LANGUAGE ScopedTypeVariables #-} module Main where -import Protolude +import RIO import qualified Control.Concurrent.Capataz.SupervisorTest as Supervisor import qualified Control.Concurrent.CapatazTest as Capataz import Test.Tasty (defaultMainWithIngredients, testGroup) -import Test.Tasty.Ingredients.Rerun (rerunningTests) import Test.Tasty.Runners (consoleTestReporter, listingTests) main :: IO () main = defaultMainWithIngredients - [rerunningTests [listingTests, consoleTestReporter]] + [listingTests, consoleTestReporter] (testGroup "capataz" [Capataz.tests, Supervisor.tests]) diff --git a/test/testsuite/Test/Util.hs b/test/testsuite/Test/Util.hs --- a/test/testsuite/Test/Util.hs +++ b/test/testsuite/Test/Util.hs @@ -4,14 +4,12 @@ {-# LANGUAGE OverloadedStrings #-} module Test.Util where -import Protolude +import RIO +import qualified RIO.Text as T -import Control.Concurrent.STM.TQueue (newTQueueIO, readTQueue, writeTQueue) -import Data.IORef (atomicModifyIORef', newIORef, readIORef, writeIORef) -import Test.Tasty.HUnit (assertBool, assertFailure) -import Text.Show.Pretty (ppShow) +import Test.Tasty.HUnit (assertBool, assertFailure) +import Text.Show.Pretty (ppShow) -import qualified Data.Text as T import qualified Control.Concurrent.Capataz as SUT import qualified Control.Concurrent.Capataz.Event as SUT @@ -22,7 +20,7 @@ -- | Utility function that gets the type name of a Record through it's Show -- output. fetchRecordName :: Show a => a -> Text -fetchRecordName = T.takeWhile (/= ' ') . show +fetchRecordName = T.takeWhile (/= ' ') . tshow -- | Composes two predicate functions together with a boolean AND andP :: [a -> Bool] -> a -> Bool @@ -235,14 +233,15 @@ -- | Utility function to create a Worker sub-routine that fails at least a -- number of times mkFailingSubRoutine - :: Int -- ^ Number of times the Worker sub-routine will fail - -> IO (IO ()) -- ^ Sub-routine used on worker creation + :: (MonadIO m, MonadThrow m1, MonadIO m1) + => Int -- ^ Number of times the Worker sub-routine will fail + -> m (m1 ()) -- ^ Sub-routine used on worker creation mkFailingSubRoutine failCount = do countRef <- newIORef failCount let subRoutine = do shouldFail <- atomicModifyIORef' countRef - (\count -> (pred count, count > 0)) - when shouldFail (throwIO RestartingWorkerError) + (\count -> (count - 1, count > 0)) + when shouldFail (throwM RestartingWorkerError) return subRoutine @@ -250,19 +249,19 @@ -- function works great when testing `Permanent` strategies, as you would like -- to assert restart events once (if it keeps completing it will fill up the log -- with restart events) -mkCompletingBeforeNRestartsSubRoutine :: Int -> IO (IO ()) +mkCompletingBeforeNRestartsSubRoutine :: MonadIO m => Int -> m (m ()) mkCompletingBeforeNRestartsSubRoutine initCount = do countRef <- newIORef initCount let subRoutine = do shouldStop <- atomicModifyIORef' countRef - (\count -> (pred count, count > 0)) + (\count -> (count - 1, count > 0)) if shouldStop then return () else forever $ threadDelay 1000100 return subRoutine -- | A sub-routine that will complete once. This function works great when -- testing `Permanent` strategies, as you would like to assert restart events -- once (if it keeps completing it will fill up the log with restart events) -mkCompletingOnceSubRoutine :: IO (IO ()) +mkCompletingOnceSubRoutine :: MonadIO m => m (m ()) mkCompletingOnceSubRoutine = mkCompletingBeforeNRestartsSubRoutine 1 -- | Utility function to build a test environment for a Capataz execution. @@ -274,11 +273,12 @@ -- * A function to modify the default "CapatazOptions", this utility function injects -- a special @notifyEvent@ callback to execute given assertions. testCapatazStreamWithOptions - :: (SUT.CapatazOptions -> SUT.CapatazOptions) -- ^ Function to modify default + :: (MonadUnliftIO m) + => (SUT.CapatazOptions (RIO LogFunc) -> SUT.CapatazOptions (RIO LogFunc)) -- ^ Function to modify default -- @CapatazOptions@ -> [SUT.CapatazEvent -> Bool] -- ^ Assertions happening before setup function -- is called - -> (SUT.Capataz -> IO ()) -- ^ Function used to test public the supervisor + -> (SUT.Capataz (RIO LogFunc) -> RIO LogFunc ()) -- ^ Function used to test public the supervisor -- public API (a.k.a setup function) -> [SUT.CapatazEvent -> Bool] -- ^ Assertions happening after the setup -- function @@ -288,69 +288,75 @@ -- @CapatazEvents@ that happened in a -- test, great when testing that an event -- __did not__ happen - -> IO () + -> m () testCapatazStreamWithOptions optionModFn preSetupAssertion setupFn postSetupAssertions postTeardownAssertions mAllEventsAssertion = do eventStream <- newTQueueIO accRef <- newIORef [] pendingCountVar <- newIORef - ( sum $ fmap + (sum $ fmap length [preSetupAssertion, postSetupAssertions, postTeardownAssertions] ) - capataz <- SUT.forkCapataz - rootSupervisorName - (SUT.set SUT.onSystemEventL (trackEvent accRef eventStream) . optionModFn) + logOptions <- logOptionsHandle stdout False + withLogFunc logOptions $ \logFunc -> runRIO logFunc $ do + capataz <- SUT.forkCapataz + rootSupervisorName + ( set + SUT.onSystemEventL + (\ev -> logDebug (display ev) >> trackEvent accRef eventStream ev) + . optionModFn + ) - -- We check preSetup assertions are met before we execute the setup - -- function. This serves to test initialization of capataz instance - runAssertions "PRE-SETUP" - (eventStream, accRef) - pendingCountVar - preSetupAssertion - capataz + -- We check preSetup assertions are met before we execute the setup + -- function. This serves to test initialization of capataz instance + runAssertions "PRE-SETUP" + (eventStream, accRef) + pendingCountVar + preSetupAssertion + capataz - -- We execute the setup sub-routine, which is going to use the Capataz public - -- API to assert events - setupResult <- try (setupFn capataz) + -- We execute the setup sub-routine, which is going to use the Capataz public + -- API to assert events + setupResult <- try (setupFn capataz) - case setupResult of - -- If the sub-routine fails, show exception - Left err -> assertFailure (show (err :: SomeException)) - Right _ -> do - -- We now run post-setup assertions - runAssertions "POST-SETUP" - (eventStream, accRef) - pendingCountVar - postSetupAssertions - capataz + case setupResult of + -- If the sub-routine fails, show exception + Left err -> liftIO $ assertFailure (show (err :: SomeException)) + Right _ -> do + -- We now run post-setup assertions + runAssertions "POST-SETUP" + (eventStream, accRef) + pendingCountVar + postSetupAssertions + capataz - -- We now shutdown the capataz instance - void $ SUT.teardown capataz + -- We now shutdown the capataz instance + SUT.terminateCapataz_ capataz - -- We run assertions for after the capataz has been shut down - runAssertions "POST-TEARDOWN" - (eventStream, accRef) - pendingCountVar - postTeardownAssertions - capataz + -- We run assertions for after the capataz has been shut down + runAssertions "POST-TEARDOWN" + (eventStream, accRef) + pendingCountVar + postTeardownAssertions + capataz - -- Lastly, we check if there is a function that we want to execute - -- across all events that happened in the test, this is to assert the - -- absence of an event - case mAllEventsAssertion of - Nothing -> return () - Just allEventsAssertion -> do - events <- reverse <$> readIORef accRef - assertBool - ( "On AFTER-TEST, expected all events to match predicate, but didn't (" - <> show (length events) - <> " events tried)\n" - <> ppShow (zip ([0 ..] :: [Int]) events) - ) - (all allEventsAssertion events) + -- Lastly, we check if there is a function that we want to execute + -- across all events that happened in the test, this is to assert the + -- absence of an event + case mAllEventsAssertion of + Nothing -> return () + Just allEventsAssertion -> do + events <- reverse <$> readIORef accRef + liftIO $ assertBool + ("On AFTER-TEST, expected all events to match predicate, but didn't (" + <> show (length events) + <> " events tried)\n" + <> ppShow (zip ([0 ..] :: [Int]) events) + ) + (all allEventsAssertion events) where -- Utility functions that runs the readEventLoop function with a timeout -- of a second, this way we can guarantee assertions are met without having @@ -358,14 +364,14 @@ runAssertions stageName (eventStream, accRef) pendingCountVar assertions capataz = do raceResult <- race - (threadDelay 1000100) + (liftIO $ threadDelay 1000100) (readEventLoop eventStream pendingCountVar assertions) case raceResult of Left _ -> do events <- reverse <$> readIORef accRef pendingCount <- readIORef pendingCountVar - void $ SUT.teardown capataz - assertFailure + SUT.terminateCapataz_ capataz + liftIO $ assertFailure ( "On " <> stageName <> " stage, expected all assertions to match, but didn't (" @@ -389,8 +395,8 @@ readEventLoop eventStream pendingCount assertions = do writeIORef pendingCount (length assertions) case assertions of - [] -> return () - (assertionFn:assertions1) -> do + [] -> return () + (assertionFn : assertions1) -> do event <- atomically $ readTQueue eventStream if assertionFn event then readEventLoop eventStream pendingCount assertions1 @@ -400,9 +406,10 @@ -- | A version of "testCapatazStreamWithOptions" that does not receive the -- function that modifies a "CapatazOptions" record. testCapatazStream - :: [SUT.CapatazEvent -> Bool] -- ^ Assertions happening before setup function + :: (MonadUnliftIO m) + => [SUT.CapatazEvent -> Bool] -- ^ Assertions happening before setup function -- is called - -> (SUT.Capataz -> IO ()) -- ^ Function used to test public the supervisor + -> (SUT.Capataz (RIO LogFunc) -> RIO LogFunc ()) -- ^ Function used to test public the supervisor -- public API (a.k.a setup function) -> [SUT.CapatazEvent -> Bool] -- ^ Assertions happening after the setup -- function @@ -412,6 +419,5 @@ -- @CapatazEvents@ that happened in a -- test, great when testing that an event -- __did not__ happen - -> IO () -testCapatazStream preSetupAssertions = - testCapatazStreamWithOptions identity preSetupAssertions + -> m () +testCapatazStream = testCapatazStreamWithOptions id