diff --git a/CHANGELOG.md b/CHANGELOG.md new file mode 100644 --- /dev/null +++ b/CHANGELOG.md @@ -0,0 +1,12 @@ +# Change log + +capataz uses [Semantic Versioning][1]. +The change log is available [on GitHub][2]. + +[1]: http://semver.org/spec/v2.0.0.html +[2]: https://github.com/roman/capataz/releases + +## v0.0.0.0 + +* First release of capataz +* Support for supervising simple worker `IO ()` sub-routines diff --git a/LICENSE b/LICENSE new file mode 100644 --- /dev/null +++ b/LICENSE @@ -0,0 +1,7 @@ +Copyright 2018, Roman Gonzalez + +Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. diff --git a/README.md b/README.md new file mode 100644 --- /dev/null +++ b/README.md @@ -0,0 +1,102 @@ +# Capataz + +> Our greatest glory is not in never failing, but in rising every time we fail.– Confucius + +## Table Of Contents + +* [Raison d'etre](#raison-detre) +* [Documentation](#documentation) +* [Development](#development) + +## Raison d'etre + +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 +capataz descriptor to every thread 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 are reliable. However, it is a +heavyweight solution that will enforce serious changes to your application. It +also optimizes its implementation around the *distributed* part of its name. +This library is intended to provide some benefits of `distributed-process` , +without the baggage. + +### 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. This library attempts not +to replace async's forte, but rather provides ways to make threads reliable in +situations where the usage of `async` or `forkIO` would give you the same +outcome. + +## 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) +[![Stackage LTS](https://www.stackage.org/package/capataz/badge/lts)](http://stackage.org/lts/package/capataz) +[![Stackage Nightly](https://www.stackage.org/package/capataz/badge/nightly)](http://stackage.org/nightly/package/capataz) + +Make sure you include the following entry on your [cabal file's +dependecies](https://www.haskell.org/cabal/users-guide/developing-packages.html#build-information) +section. + +```cabal +library: + build-depends: capataz +``` + +Or on your `package.yaml` + +``` +dependencies: +- capataz +``` + +## 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.0.0.0.svg)](https://img.shields.io/github/commits-since/roman/haskell-capataz/v0.0.0.0.svg) +[![Hackage Dependencies](https://img.shields.io/hackage-deps/v/capataz.svg)](https://img.shields.io/hackage/v/capataz.svg) + +Follow the [developer guidelines](https://romanandreg.gitbooks.io/capataz/content/developer-guidelines.html) + +## In next release + +* Add support for supervising supervisors +* Ensure unit tests always finish on all concurrent scenarios (dejafu experiment) diff --git a/Setup.hs b/Setup.hs new file mode 100644 --- /dev/null +++ b/Setup.hs @@ -0,0 +1,2 @@ +import Distribution.Simple +main = defaultMain diff --git a/capataz.cabal b/capataz.cabal new file mode 100644 --- /dev/null +++ b/capataz.cabal @@ -0,0 +1,97 @@ +name: capataz +version: 0.0.0.0 +cabal-version: >=1.10 +build-type: Simple +license: MIT +license-file: LICENSE +copyright: © 2018 Roman Gonzalez +maintainer: capataz@roman-gonzalez.info +stability: alpha (experimental) +homepage: https://github.com/roman/Haskell-capataz#readme +bug-reports: https://github.com/roman/Haskell-capataz/issues +synopsis: OTP-like supervision trees in Haskell +description: + `capataz` enhances the reliability of your concurrent applications by offering + supervision of green threads that run in your application. + . + Advantages over standard library: + . + * Links related long-living processes together under a common capataz + supervisor, with restart/shutdown order + . + * Set restart strategies (Permanent, Transient, Temporary) on `IO` + sub-routines on a granular level + . + * Set restart strategies on a pool of long-living worker threads (AllForOne, + OneForOne) + . + * 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 +extra-source-files: + CHANGELOG.md + README.md + +source-repository head + type: git + location: https://github.com/roman/Haskell-capataz + +library + exposed-modules: + Control.Concurrent.Capataz + build-depends: + async >=2.1.1.1 && <2.2, + base >=4.10.1.0 && <4.11, + bytestring >=0.10.8.2 && <0.11, + data-default >=0.7.1.1 && <0.8, + protolude ==0.2.*, + safe-exceptions >=0.1.6.0 && <0.2, + stm >=2.4.4.1 && <2.5, + teardown >=0.3.0.0 && <0.4, + text >=1.2.2.2 && <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.Internal.Capataz.Core + Control.Concurrent.Internal.Capataz.Restart + Control.Concurrent.Internal.Capataz.Types + Control.Concurrent.Internal.Capataz.Util + Control.Concurrent.Internal.Capataz.Worker + Paths_capataz + ghc-options: -Wall -Wincomplete-uni-patterns -Wincomplete-record-updates + +test-suite capataz-test + type: exitcode-stdio-1.0 + main-is: Main.hs + build-depends: + async >=2.1.1.1 && <2.2, + base >=4.10.1.0 && <4.11, + bytestring >=0.10.8.2 && <0.11, + capataz -any, + data-default >=0.7.1.1 && <0.8, + pretty-show >=1.6.15 && <1.7, + protolude ==0.2.*, + safe-exceptions >=0.1.6.0 && <0.2, + stm >=2.4.4.1 && <2.5, + tasty >=0.11.3 && <0.12, + tasty-hunit >=0.9.2 && <0.10, + tasty-rerun >=1.1.8 && <1.2, + tasty-smallcheck >=0.8.1 && <0.9, + teardown >=0.3.0.0 && <0.4, + text >=1.2.2.2 && <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.CapatazTest + Paths_capataz + ghc-options: -Wall -Wincomplete-uni-patterns -Wincomplete-record-updates -threaded -rtsopts -with-rtsopts=-N diff --git a/src/Control/Concurrent/Capataz.hs b/src/Control/Concurrent/Capataz.hs new file mode 100644 --- /dev/null +++ b/src/Control/Concurrent/Capataz.hs @@ -0,0 +1,58 @@ +{-# LANGUAGE NoImplicitPrelude #-} +{-| Public API for the capataz library + + Capataz is a library that brings an OTP-like supervisor API to the Haskell + concurrency toolset. + +-} +module Control.Concurrent.Capataz +( +-- * Types + CallbackType (..) +, WorkerAction +, WorkerError (..) +, WorkerOptions (..) +, WorkerRestartStrategy (..) +, WorkerSpec (..) +, WorkerTerminationOrder (..) +, WorkerTerminationPolicy (..) +, Capataz (..) +, CapatazEvent (..) +, CapatazOptions (..) +, CapatazRestartStrategy (..) +, CapatazStatus (..) +, defWorkerOptions +, defWorkerSpec +, defCapatazOptions +-- * Core functionality +, forkWorker +, forkCapataz +, terminateWorker +-- * Utility functions +, capatazToAsync +-- * Teardown (re-exported) +, teardown +) +where + +import Control.Concurrent.Internal.Capataz.Core (forkCapataz, forkWorker, terminateWorker) +import Control.Concurrent.Internal.Capataz.Types + ( CallbackType (..) + , Capataz (..) + , CapatazEvent (..) + , CapatazOptions (..) + , CapatazRestartStrategy (..) + , CapatazStatus (..) + , WorkerAction + , WorkerError (..) + , WorkerOptions (..) + , WorkerRestartStrategy (..) + , WorkerSpec (..) + , WorkerTerminationOrder (..) + , WorkerTerminationPolicy (..) + , defCapatazOptions + , defWorkerOptions + , defWorkerSpec + ) +import Control.Concurrent.Internal.Capataz.Util (capatazToAsync) +import Control.Teardown (teardown) diff --git a/src/Control/Concurrent/Internal/Capataz/Core.hs b/src/Control/Concurrent/Internal/Capataz/Core.hs new file mode 100644 --- /dev/null +++ b/src/Control/Concurrent/Internal/Capataz/Core.hs @@ -0,0 +1,259 @@ +{-# LANGUAGE NamedFieldPuns #-} +{-# LANGUAGE NoImplicitPrelude #-} +{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE RankNTypes #-} +{-# LANGUAGE RecordWildCards #-} +{-| This module contains: + +* Functions exported on the public API +* The supervisor thread loop +* High level message handlers of the supervisor thread loop + +-} +module Control.Concurrent.Internal.Capataz.Core where + +import Protolude + +import Control.Concurrent.Async (asyncWithUnmask) +import Control.Concurrent.MVar (newEmptyMVar, takeMVar) +import Control.Concurrent.STM (atomically) +import Control.Concurrent.STM.TQueue (newTQueueIO, readTQueue, writeTQueue) +import Control.Concurrent.STM.TVar (newTVarIO) +import Control.Teardown (newTeardown) +import Data.IORef (newIORef) +import Data.Time.Clock (getCurrentTime) + +import qualified Data.HashMap.Strict as HashMap +import qualified Data.UUID.V4 as UUID (nextRandom) + +import qualified Control.Concurrent.Internal.Capataz.Restart as Restart +import qualified Control.Concurrent.Internal.Capataz.Worker as Worker + +import Control.Concurrent.Internal.Capataz.Types +import Control.Concurrent.Internal.Capataz.Util + ( appendWorkerToMap + , capatazToEnv + , fetchWorker + , readCapatazStatus + , readCapatazStatusSTM + , resetWorkerMap + , sendSyncControlMsg + , workerOptionsToSpec + , writeCapatazStatus + ) + +-------------------------------------------------------------------------------- + +-- | Executes the shutdown operation of a Capataz, including the termination of +-- Workers being supervised by it. +haltCapataz :: CapatazEnv -> IO () +haltCapataz env = do + writeCapatazStatus env Halting + Worker.terminateWorkers "capataz shutdown" env + resetWorkerMap env (const HashMap.empty) + writeCapatazStatus env Halted + +-- | Handles an event produced by one of the workers this capataz monitors +handleMonitorEvent :: CapatazEnv -> MonitorEvent -> IO Bool +handleMonitorEvent env monitorEv = do + case monitorEv of + WorkerForcedRestart{} -> + -- We do nothing, as restart is being handled on restartWorkers + -- sub-routine + return () + + WorkerCompleted' { workerId, monitorEventTime } -> + Restart.handleWorkerCompleted env workerId monitorEventTime + + WorkerFailed' { workerId, workerError, workerRestartCount } -> + Restart.handleWorkerFailed env workerId workerError workerRestartCount + + WorkerTerminated' { workerId, workerRestartCount, workerTerminationReason } + -> Restart.handleWorkerTerminated env + workerId + workerTerminationReason + workerRestartCount + + + return True + +-- | Handles an action triggered by the public API +handleControlAction :: CapatazEnv -> ControlAction -> IO Bool +handleControlAction env controlAction = case controlAction of + ForkWorker { workerSpec, returnWorkerId } -> do + worker@Worker { workerId } <- Worker.forkWorker env workerSpec Nothing + appendWorkerToMap env worker + returnWorkerId workerId + return True + + TerminateWorker { terminationReason, workerId, notifyWorkerTermination } -> + do + mWorker <- fetchWorker env workerId + case mWorker of + Nothing -> return True + Just worker -> do + Worker.terminateWorker terminationReason env worker + -- removeWorkerFromMap env workerId + notifyWorkerTermination + return True + + TerminateCapataz { notifyCapatazTermination } -> do + haltCapataz env + notifyCapatazTermination + return False + +-- | Handles all messages that a capataz instance can receive +handleCapatazMessage :: CapatazEnv -> CapatazMessage -> IO Bool +handleCapatazMessage env message = case message of + ControlAction controlAction -> handleControlAction env controlAction + MonitorEvent monitorEvent -> handleMonitorEvent env monitorEvent + +-- | Handles errors caused by the execution of the "runCapatazLoop" sub-routine +handleCapatazException :: CapatazEnv -> SomeException -> IO () +handleCapatazException env@CapatazEnv { capatazId, capatazName, notifyEvent } capatazError + = do + eventTime <- getCurrentTime + notifyEvent CapatazFailed + { capatazId + , capatazName + , capatazError + , eventTime + } + haltCapataz env + throwIO capatazError + +-- | This is the main thread loop of a "Capataz" instance +runCapatazLoop :: (forall b . IO b -> IO b) -> CapatazEnv -> IO () +runCapatazLoop unmask env@CapatazEnv { capatazId, capatazName, capatazStatusVar, capatazQueue, notifyEvent } + = do + loopResult <- + unmask + $ try + $ atomically + $ (,) + <$> readCapatazStatusSTM capatazStatusVar + <*> readTQueue capatazQueue + + case loopResult of + Left capatazError -> handleCapatazException env capatazError + + Right (status, message) -> case status of + Initializing -> do + eventTime <- getCurrentTime + notifyEvent InvalidCapatazStatusReached + { capatazId + , capatazName + , eventTime + } + runCapatazLoop unmask env + + Running -> do + eContinueLoop <- try $ unmask $ handleCapatazMessage env message + case eContinueLoop of + Left capatazError -> handleCapatazException env capatazError + + Right continueLoop + | continueLoop -> runCapatazLoop unmask env + | otherwise -> do + eventTime <- getCurrentTime + notifyEvent CapatazTerminated + { capatazId + , capatazName + , eventTime + } + + Halting -> + -- Discard messages when halting + return () + + Halted -> panic "TODO: Pending halted state" + +-- | Builds a record that contains runtime values of a "Capataz" (id, queue, status, etc.) +buildCapatazRuntime :: CapatazOptions -> IO CapatazRuntime +buildCapatazRuntime capatazOptions = do + capatazId <- UUID.nextRandom + capatazQueue <- newTQueueIO + capatazStatusVar <- newTVarIO Initializing + capatazWorkerMap <- newIORef HashMap.empty + return CapatazRuntime {..} + +-- | Creates a Capataz record, which represents a supervision thread which +-- monitors failure on worker threads defined in the "CapatazOptions" or worker +-- threads that are created dynamically using "forkWorker". +forkCapataz :: CapatazOptions -> IO Capataz +forkCapataz capatazOptions@CapatazOptions { capatazName, capatazWorkerSpecList, notifyEvent } + = do + capatazRuntime <- buildCapatazRuntime capatazOptions + + let capatazEnv@CapatazEnv { capatazId } = capatazToEnv capatazRuntime + + capatazAsync <- asyncWithUnmask + $ \unmask -> runCapatazLoop unmask capatazEnv + + forM_ + capatazWorkerSpecList + ( \workerSpec -> do + worker <- Worker.forkWorker capatazEnv workerSpec Nothing + appendWorkerToMap capatazEnv worker + ) + + writeCapatazStatus capatazEnv Running + + capatazTeardown <- newTeardown + ("capataz[" <> capatazName <> "]") + ( do + status <- readCapatazStatus capatazEnv + case status of + Halted -> return () + Halting -> return () + _ -> do + eventTime <- getCurrentTime + notifyEvent CapatazShutdownInvoked + { capatazId + , capatazName + , eventTime + } + sendSyncControlMsg capatazEnv TerminateCapataz + ) + + return Capataz {..} + +-- | Creates a worker green thread "IO ()" sub-routine, and depending in options +-- defined in the "WorkerOptions" record, it will restart the Worker sub-routine +-- in case of failures +forkWorker + :: WorkerOptions -- ^ Worker options (restart, name, callbacks, etc) + -> IO () -- ^ IO sub-routine that will be executed on worker thread + -> Capataz -- ^ "Capataz" instance that supervises the worker + -> IO WorkerId -- ^ An identifier that can be used to terminate the "Worker" +forkWorker workerOptions workerAction Capataz { capatazEnv } = do + let workerSpec = workerOptionsToSpec workerOptions workerAction + CapatazEnv { capatazQueue } = capatazEnv + + workerIdVar <- newEmptyMVar + atomically $ writeTQueue + capatazQueue + ( ControlAction ForkWorker + { workerSpec + , returnWorkerId = putMVar workerIdVar + } + ) + takeMVar workerIdVar + +-- | Stops the execution of a worker green thread being supervised by the given +-- "Capataz" instance, if the WorkerId does not belong to the Capataz, the +-- operation does not perform any side-effect. +-- +-- Note: If your worker has a "Permanent" worker restart strategy, the worker +-- thread __will be restarted again__; so use a "Transient" restart strategy +-- instead. +terminateWorker :: Text -> WorkerId -> Capataz -> IO () +terminateWorker terminationReason workerId Capataz { capatazEnv } = + sendSyncControlMsg + capatazEnv + ( \notifyWorkerTermination -> TerminateWorker + { terminationReason + , workerId + , notifyWorkerTermination + } + ) diff --git a/src/Control/Concurrent/Internal/Capataz/Restart.hs b/src/Control/Concurrent/Internal/Capataz/Restart.hs new file mode 100644 --- /dev/null +++ b/src/Control/Concurrent/Internal/Capataz/Restart.hs @@ -0,0 +1,219 @@ +{-# LANGUAGE DuplicateRecordFields #-} +{-# LANGUAGE NamedFieldPuns #-} +{-# LANGUAGE NoImplicitPrelude #-} +{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE ScopedTypeVariables #-} +{-| This module contains all logic related to the restart of workers -} +module Control.Concurrent.Internal.Capataz.Restart where + +import Data.Time.Clock (NominalDiffTime, UTCTime, diffUTCTime, getCurrentTime) + +import Protolude + +import Control.Concurrent.Internal.Capataz.Types +import Control.Concurrent.Internal.Capataz.Util + ( appendWorkerToMap + , fetchWorkerEnv + , readWorkerMap + , removeWorkerFromMap + , resetWorkerMap + , sortWorkersByTerminationOrder + ) +import qualified Control.Concurrent.Internal.Capataz.Worker as Worker +import qualified Data.HashMap.Strict as HashMap + +-------------------------------------------------------------------------------- + +-- | Function used to track difference between two timestamps to track capataz +-- error intensity +calcDiffSeconds :: UTCTime -> IO NominalDiffTime +calcDiffSeconds creationTime = do + currentTime <- getCurrentTime + return $ diffUTCTime currentTime creationTime + +-- | Function that checks restart counts and worker start time to assess if the +-- capataz error intensity has been breached, see "WorkerRestartAction" for +-- possible outcomes. +calcRestartAction :: CapatazEnv -> Int -> NominalDiffTime -> WorkerRestartAction +calcRestartAction CapatazEnv { capatazIntensity, capatazPeriodSeconds } restartCount diffSeconds + = case () of + _ + | diffSeconds < capatazPeriodSeconds && restartCount > capatazIntensity + -> HaltCapataz + | diffSeconds > capatazPeriodSeconds + -> ResetRestartCount + | otherwise + -> IncreaseRestartCount + +-- | Sub-routine responsible of executing a "CapatazRestartStrategy" +execCapatazRestartStrategy :: CapatazEnv -> WorkerEnv -> Int -> IO () +execCapatazRestartStrategy capatazEnv@CapatazEnv { capatazRestartStrategy } WorkerEnv { workerId, workerSpec } workerRestartCount + = case capatazRestartStrategy of + AllForOne -> do + newWorkers <- restartWorkers capatazEnv workerId workerRestartCount + let newWorkersMap = + newWorkers + & fmap (\worker@Worker { workerId = cid } -> (cid, worker)) + & HashMap.fromList + resetWorkerMap capatazEnv (const newWorkersMap) + + OneForOne -> do + removeWorkerFromMap capatazEnv workerId + newWorker <- restartWorker capatazEnv + workerSpec + workerId + workerRestartCount + appendWorkerToMap capatazEnv newWorker + +-- | Executes a restart action returned from the invokation of "calcRestartAction" +execRestartAction :: CapatazEnv -> WorkerEnv -> Int -> IO () +execRestartAction capatazEnv@CapatazEnv { onCapatazIntensityReached } workerEnv@WorkerEnv { workerId, workerName, workerCreationTime } workerRestartCount + = do + restartAction <- calcRestartAction capatazEnv workerRestartCount + <$> calcDiffSeconds workerCreationTime + + case restartAction of + HaltCapataz -> do + -- skip exceptions on callback + (_ :: Either SomeException ()) <- try onCapatazIntensityReached + throwIO CapatazIntensityReached + { workerId + , workerName + , workerRestartCount = succ workerRestartCount + } + + ResetRestartCount -> execCapatazRestartStrategy capatazEnv workerEnv 0 + + IncreaseRestartCount -> execCapatazRestartStrategy + capatazEnv + workerEnv + (succ workerRestartCount) + +-------------------------------------------------------------------------------- + +-- | Restarts _all_ the worker green thread of a Capataz, invoked when one +-- worker green thread fails and causes sibling worker threads to get restarted +-- as well +restartWorkers :: CapatazEnv -> WorkerId -> RestartCount -> IO [Worker] +restartWorkers capatazEnv@CapatazEnv { capatazWorkerTerminationOrder } failingWorkerId restartCount + = do + workerMap <- readWorkerMap capatazEnv + + let workers = + sortWorkersByTerminationOrder capatazWorkerTerminationOrder workerMap + + newWorkers <- forM workers $ \worker@Worker { workerId, workerSpec } -> do + unless (failingWorkerId == workerId) + $ forceRestartWorker capatazEnv worker + + let WorkerSpec { workerRestartStrategy } = workerSpec + case workerRestartStrategy of + Temporary -> return Nothing + _ -> Just <$> restartWorker capatazEnv workerSpec workerId restartCount + + return $ catMaybes newWorkers + +-- | Sub-routine that is used when there is a restart request to a Worker caused +-- by an "AllForOne" restart from a failing sibling worker. +forceRestartWorker :: CapatazEnv -> Worker -> IO () +forceRestartWorker CapatazEnv { capatazName, capatazId, notifyEvent } Worker { workerId, workerName, workerAsync } + = do + eventTime <- getCurrentTime + notifyEvent WorkerTerminated + { capatazName + , capatazId + , workerId + , workerName + , eventTime + , workerThreadId = asyncThreadId workerAsync + , terminationReason = "forced restart" + } + cancelWith workerAsync RestartWorkerException + +-- | Starts a new worker thread taking into account an existing "WorkerId" and +-- keeping a "RestartCount" to manage Capataz error intensity. +restartWorker + :: CapatazEnv -> WorkerSpec -> WorkerId -> RestartCount -> IO Worker +restartWorker capatazEnv workerSpec workerId restartCount = + Worker.forkWorker capatazEnv workerSpec (Just (workerId, restartCount)) + +-------------------------------------------------------------------------------- + +-- | This sub-routine is responsible of the restart strategies execution when a +-- supervised worker finishes it execution because of a completion (e.g. worker +-- sub-routine finished without any errors). +handleWorkerCompleted :: CapatazEnv -> WorkerId -> UTCTime -> IO () +handleWorkerCompleted env@CapatazEnv { capatazName, capatazId, notifyEvent } workerId eventTime + = do + mWorkerEnv <- fetchWorkerEnv env workerId + case mWorkerEnv of + Nothing -> return () + Just workerEnv@WorkerEnv { workerName, workerAsync, workerRestartStrategy } + -> do + notifyEvent WorkerCompleted + { capatazId + , capatazName + , workerId + , workerName + , eventTime + , workerThreadId = asyncThreadId workerAsync + } + case workerRestartStrategy of + Permanent -> do + -- NOTE: Completed workers should never account as errors happening on + -- a supervised thread, ergo, they should be restarted every time. + + -- TODO: Notify a warning around having a workerRestartStrategy different + -- than Temporary on workers that may complete. + let restartCount = 0 + execRestartAction env workerEnv restartCount + + _ -> removeWorkerFromMap env workerId + +-- | This sub-routine is responsible of the restart strategies execution when a +-- supervised worker finishes it execution because of a failure. +handleWorkerFailed :: CapatazEnv -> WorkerId -> SomeException -> Int -> IO () +handleWorkerFailed env@CapatazEnv { capatazName, capatazId, notifyEvent } workerId workerError restartCount + = do + mWorkerEnv <- fetchWorkerEnv env workerId + case mWorkerEnv of + Nothing -> return () + Just workerEnv@WorkerEnv { workerName, workerAsync, workerRestartStrategy } + -> do + eventTime <- getCurrentTime + notifyEvent WorkerFailed + { capatazName + , capatazId + , workerId + , workerName + , workerError + , workerThreadId = asyncThreadId workerAsync + , eventTime + } + case workerRestartStrategy of + Temporary -> removeWorkerFromMap env workerId + _ -> execRestartAction env workerEnv restartCount + +-- | This sub-routine is responsible of the restart strategies execution when a +-- supervised worker finishes it execution because of a termination. +handleWorkerTerminated :: CapatazEnv -> WorkerId -> Text -> Int -> IO () +handleWorkerTerminated env@CapatazEnv { capatazName, capatazId, notifyEvent } workerId terminationReason workerRestartCount + = do + mWorkerEnv <- fetchWorkerEnv env workerId + case mWorkerEnv of + Nothing -> return () + Just workerEnv@WorkerEnv { workerName, workerAsync, workerRestartStrategy } + -> do + eventTime <- getCurrentTime + notifyEvent WorkerTerminated + { capatazName + , capatazId + , workerId + , workerName + , eventTime + , terminationReason + , workerThreadId = asyncThreadId workerAsync + } + case workerRestartStrategy of + Permanent -> execRestartAction env workerEnv workerRestartCount + _ -> removeWorkerFromMap env workerId diff --git a/src/Control/Concurrent/Internal/Capataz/Types.hs b/src/Control/Concurrent/Internal/Capataz/Types.hs new file mode 100644 --- /dev/null +++ b/src/Control/Concurrent/Internal/Capataz/Types.hs @@ -0,0 +1,519 @@ +{-# LANGUAGE DeriveGeneric #-} +{-# LANGUAGE DuplicateRecordFields #-} +{-# LANGUAGE NamedFieldPuns #-} +{-# LANGUAGE NoImplicitPrelude #-} +{-# LANGUAGE OverloadedStrings #-} +{-| This module contains all the types used across all the other modules -} +module Control.Concurrent.Internal.Capataz.Types where + +import Protolude + +import Control.Concurrent.STM.TQueue (TQueue) +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) + +type CapatazId = UUID +type WorkerId = UUID +type WorkerAction = IO () +type WorkerThreadId = ThreadId +type CapatazName = Text +type WorkerName = Text +type RestartCount = Int +type WorkerMap = HashMap WorkerId Worker + +-- | Event passed to the "notifyEvent" callback sub-routine, this events can be +-- used to monitor the capataz system and understanding what is doing. This +-- provides high levels of telemetry for the Capataz instance, so is mainly used +-- for logging, monitoring and testing purposes. +data CapatazEvent + = InvalidCapatazStatusReached { + capatazId :: !CapatazId + , capatazName :: !CapatazName + , eventTime :: !UTCTime + } + | CapatazStatusChanged { + capatazId :: !CapatazId + , capatazName :: !CapatazName + , prevCapatazStatus :: !CapatazStatus + , newCapatazStatus :: !CapatazStatus + , eventTime :: !UTCTime + } + | WorkerTerminated { + capatazName :: !CapatazName + , capatazId :: !CapatazId + , workerThreadId :: !WorkerThreadId + , workerId :: !WorkerId + , workerName :: !WorkerName + , terminationReason :: !Text + , eventTime :: !UTCTime + } + | WorkerStarted { + capatazName :: !CapatazName + , capatazId :: !CapatazId + , workerThreadId :: !WorkerThreadId + , workerId :: !WorkerId + , workerName :: !WorkerName + , eventTime :: !UTCTime + } + | WorkerRestarted { + capatazName :: !CapatazName + , capatazId :: !CapatazId + , workerThreadId :: !WorkerThreadId + , workerId :: !WorkerId + , workerName :: !WorkerName + , workerRestartCount :: !Int + , eventTime :: !UTCTime + } + | WorkerCompleted { + capatazName :: !CapatazName + , capatazId :: !CapatazId + , workerThreadId :: !WorkerThreadId + , workerId :: !WorkerId + , workerName :: !WorkerName + , eventTime :: !UTCTime + } + | WorkerFailed { + capatazName :: !CapatazName + , capatazId :: !CapatazId + , workerThreadId :: !WorkerThreadId + , workerId :: !WorkerId + , workerName :: !WorkerName + , workerError :: !SomeException + , eventTime :: !UTCTime + } + | WorkerCallbackExecuted { + capatazName :: !CapatazName + , capatazId :: !CapatazId + , workerThreadId :: !WorkerThreadId + , workerId :: !WorkerId + , workerName :: !WorkerName + , workerCallbackError :: !(Maybe SomeException) + , callbackType :: !CallbackType + , eventTime :: !UTCTime + } + | WorkersTerminationStarted { + capatazName :: !CapatazName + , capatazId :: !CapatazId + , terminationReason :: !Text + , eventTime :: !UTCTime + } + | WorkersTerminationFinished { + capatazName :: !CapatazName + , capatazId :: !CapatazId + , terminationReason :: !Text + , eventTime :: !UTCTime + } + | CapatazFailed { + capatazName :: !CapatazName + , capatazId :: !CapatazId + , capatazError :: !SomeException + , eventTime :: !UTCTime + } + | CapatazTerminated { + capatazName :: !CapatazName + , capatazId :: !CapatazId + , eventTime :: !UTCTime + } + | CapatazShutdownInvoked { + capatazName :: !CapatazName + , capatazId :: !CapatazId + , eventTime :: !UTCTime + } + deriving (Generic, Show) + +-- | Defines how a "Worker" termination should be handled, default +-- "WorkerTerminationPolicy" is 3 seconds +data WorkerTerminationPolicy + -- | Waits until infinity for the worker to terminate + = Infinity + + -- | Worker is terminated wihtout a chance to call its callback + | BrutalTermination + + -- | Allows n milliseconds for worker termination callback to be + -- executed, otherwise "BrutalTermination occurs" + | TimeoutMillis !Int + deriving (Generic, Show, Eq, Ord) + +instance Default WorkerTerminationPolicy where + def = TimeoutMillis 3000 + +instance NFData WorkerTerminationPolicy + +-- | Helper record to assess if the capataz error intensity has been breached +data WorkerRestartAction + -- | The capataz will restart the failed worker and reset the restart count + -- given intensity period has passed + = ResetRestartCount + + -- | The capataz will restart the failed worker and increase the restart count + | IncreaseRestartCount + + -- | The error intensity has been reached + | HaltCapataz + deriving (Generic, Show, Eq) + +instance NFData WorkerRestartAction + +-- | Specifies how order in which workers should be terminated by a Capataz in +-- case of restart or shutdown; default is "OldestFirst" +data WorkerTerminationOrder + -- | Terminate worker threads from most recent to oldest + = NewestFirst + -- | Terminate worker threads from oldest to most recent + | OldestFirst + deriving (Generic, Show, Eq, Ord) + +instance Default WorkerTerminationOrder where + def = OldestFirst + +instance NFData WorkerTerminationOrder + +-- | Specifies how a Capataz should restart a failing worker. Default is +-- "OneForOne" +data CapatazRestartStrategy + -- | Terminate all workers threads when one fails and restart them all + = AllForOne + + -- | Only restart worker thread that failed + | OneForOne + deriving (Generic, Show, Eq, Ord) + +instance Default CapatazRestartStrategy where + def = OneForOne + +instance NFData CapatazRestartStrategy + +-- | Utility record used to specify options to a "Capataz" instance +data CapatazOptions + = CapatazOptions { + -- | Name of the Capataz (present on "CapatazEvent" records) + capatazName :: Text + -- | How many errors is the Capataz be able to handle; check: + -- http://erlang.org/doc/design_principles/sup_princ.html#max_intensity + , capatazIntensity :: !Int + -- | Period of time where the Capataz can receive "capatazIntensity" amount + -- of errors + , capatazPeriodSeconds :: !NominalDiffTime + -- | What is the "CapatazRestartStrategy" for this Capataz + , capatazRestartStrategy :: !CapatazRestartStrategy + -- | Static set of workers that start as soon as the "Capataz" is created + , capatazWorkerSpecList :: ![WorkerSpec] + -- | In which order the "Capataz" record is going to terminate it's workers + , capatazWorkerTerminationOrder :: !WorkerTerminationOrder + -- | Callback used when the error intensity is reached + , onCapatazIntensityReached :: !(IO ()) + -- | Callback used for telemetry purposes + , notifyEvent :: !(CapatazEvent -> IO ()) + } + + +-- | Utility record used to specify options to a "Worker" instance +data WorkerOptions + = WorkerOptions { + -- | Name of the Worker (present on "CapatazEvent" records) + workerName :: !WorkerName + -- | Callback used when the worker fails with an error + , workerOnFailure :: !(SomeException -> IO ()) + -- | Callback used when the worker completes execution without error + , workerOnCompletion :: !(IO ()) + -- | Callback used when the worker is terminated + , workerOnTermination :: !(IO ()) + -- | Indicates how a worker should be terminated + , workerTerminationPolicy :: !WorkerTerminationPolicy + -- | Indicates how a worker should be restarted + , workerRestartStrategy :: !WorkerRestartStrategy + } + deriving (Generic) + +-- | Specifies how a "Worker" should restart on failure. Default is "Transient" +data WorkerRestartStrategy + -- | Worker thread is __always__ restarted + = Permanent + + -- | Worker thread is restarted only if it failed + | Transient + + -- | Worker thread is __never__ restarted + | Temporary + + deriving (Generic, Show, Eq) + +instance NFData WorkerRestartStrategy +instance Default WorkerRestartStrategy where + def = Transient + +-- | WorkerSpec is a representation of the "WorkerOptions" record that embeds +-- the @"IO" ()@ sub-routine of the worker thread. This record is used when we +-- want to bound worker threads to a "Capataz" instance +data WorkerSpec + = WorkerSpec { + -- | 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 + -- | Name of the Worker (present on "CapatazEvent" records) + , workerName :: !WorkerName + -- | Callback used when the worker fails with an error + , workerOnFailure :: !(SomeException -> IO ()) + -- | Callback used when the worker completes execution without error + , workerOnCompletion :: !(IO ()) + -- | Callback used when the worker is terminated + , workerOnTermination :: !(IO ()) + -- | Indicates how a worker should be terminated + , workerTerminationPolicy :: !WorkerTerminationPolicy + -- | Indicates how a worker should be restarted + , workerRestartStrategy :: !WorkerRestartStrategy + } + deriving (Generic) + +-- | Record that contains the "Async" record (thread reference) of a worker +data Worker + = Worker { + -- | Unique identifier for a worker that is executing + workerId :: !WorkerId + -- | "Async" thread of a worker, this Async executes the @IO ()@ sub-routine + , workerAsync :: !(Async ()) + -- | Time where this worker was created (used for error intensity checks) + , workerCreationTime :: !UTCTime + -- | Name of the Worker (present on "CapatazEvent" records) + , workerName :: !WorkerName + -- | "WorkerSpec" contains all the options around restart and termination + -- policies + , workerSpec :: !WorkerSpec + } + +-- | Convenience utility record that contains all values related to a "Worker"; +-- this is used on internal functions of the Capataz library. +data WorkerEnv + = WorkerEnv { + workerAction :: WorkerAction + , workerId :: !WorkerId + , workerAsync :: !(Async ()) + , workerCreationTime :: !UTCTime + , workerName :: !WorkerName + , workerSpec :: !WorkerSpec + , workerOnFailure :: !(SomeException -> IO ()) + , workerOnCompletion :: !(IO ()) + , workerOnTermination :: !(IO ()) + , workerRestartStrategy :: !WorkerRestartStrategy + } + +-- | Internal record that represents an action being sent from threads using +-- the Capataz public API. +data ControlAction + = ForkWorker { + workerSpec :: !WorkerSpec + , returnWorkerId :: !(WorkerId -> IO ()) + } + | TerminateWorker { + workerId :: !WorkerId + , terminationReason :: !Text + , notifyWorkerTermination :: !(IO ()) + } + | TerminateCapataz { + notifyCapatazTermination :: !(IO ()) + } + deriving (Generic) + +-- | Internal exception thrown to the Capataz loop to indicate termination of +-- execution +data CapatazSignal + = RestartWorkerException + | TerminateWorkerException { + workerId :: !WorkerId + , workerTerminationReason :: !Text + } + | BrutallyTerminateWorkerException { + workerId :: !WorkerId + , workerTerminationReason :: !Text + } + deriving (Generic, Show) + +instance Exception CapatazSignal +instance NFData CapatazSignal + +-- | Internal exception triggered when a Worker violates error intensity +-- specification +data CapatazError + = CapatazIntensityReached { + workerId :: !WorkerId + , workerName :: !WorkerName + , workerRestartCount :: !Int + } + deriving (Generic, Show) + +instance Exception CapatazError +instance NFData CapatazError + +-- | Internal record that indicates what type of callback function is being +-- invoked; this is used for telemetry purposes +data CallbackType + = OnCompletion + | OnFailure + | OnTermination + deriving (Generic, Show, Eq) + +-- | Internal exception triggered when a callback of a Worker fails +data WorkerError + = WorkerCallbackFailed { + workerId :: !WorkerId + , workerActionError :: !(Maybe SomeException) + , callbackType :: !CallbackType + , workerCallbackError :: !SomeException + } + deriving (Generic, Show) + +instance Exception WorkerError + +-- | Internal event delivered from Worker threads to the Capataz thread to +-- indicate completion, failure or termination +data MonitorEvent + = WorkerTerminated' { + workerId :: !WorkerId + , workerName :: !WorkerName + , workerRestartCount :: !RestartCount + , workerTerminationReason :: !Text + , monitorEventTime :: !UTCTime + } + | WorkerFailed' { + workerId :: !WorkerId + , workerName :: !WorkerName + , workerRestartCount :: !RestartCount + , workerError :: !SomeException + , monitorEventTime :: !UTCTime + } + | WorkerCompleted' { + workerId :: !WorkerId + , workerName :: !WorkerName + , monitorEventTime :: !UTCTime + } + | WorkerForcedRestart { + workerId :: !WorkerId + , workerName :: !WorkerName + , monitorEventTime :: !UTCTime + } + deriving (Show) + +-- | Internal state machine record that indicates the state of a Capataz +data CapatazStatus + -- | This state is set when Worker is created and it spawn static worker + -- threads + = Initializing + -- | This state is set when the Capataz thread is listenting to both + -- "ControlAction" and "MonitorEvent" messages + | Running + -- | This state is set when the Capataz thread is terminating it's assigned + -- worker + | Halting + -- | The Capataz thread is done + | Halted + deriving (Generic, Show, Eq) + +instance NFData CapatazStatus + +-- | Internal message delivered to a Capataz thread that can either be a call +-- from public API or an event from a monitored Worker +data CapatazMessage + = ControlAction !ControlAction + | MonitorEvent !MonitorEvent + deriving (Generic) + +-- | 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 + = Capataz { + capatazRuntime :: !CapatazRuntime + , capatazEnv :: !CapatazEnv + , capatazAsync :: !(Async ()) + , capatazTeardown :: !Teardown + } + +instance ITeardown Capataz where + teardown Capataz {capatazTeardown} = + teardown capatazTeardown + +-- | Internal record used to hold part of the runtime information of a "Capataz" +-- record +data CapatazRuntime + = CapatazRuntime { + capatazId :: !CapatazId + , capatazQueue :: !(TQueue CapatazMessage) + , capatazWorkerMap :: !(IORef (HashMap WorkerId Worker)) + , capatazStatusVar :: !(TVar CapatazStatus) + , capatazOptions :: !CapatazOptions + } + +-- | Convenience utility record that contains all values related to a "Capataz"; +-- this is used on internal functions of the Capataz library. +data CapatazEnv + = CapatazEnv { + capatazId :: !CapatazId + , capatazName :: !CapatazName + , capatazQueue :: !(TQueue CapatazMessage) + , capatazWorkerMap :: !(IORef (HashMap WorkerId Worker)) + , capatazStatusVar :: !(TVar CapatazStatus) + , capatazOptions :: !CapatazOptions + , capatazRuntime :: !CapatazRuntime + , capatazIntensity :: !Int + -- ^ http://erlang.org/doc/design_principles/sup_princ.html#max_intensity + , capatazPeriodSeconds :: !NominalDiffTime + , capatazRestartStrategy :: !CapatazRestartStrategy + , capatazWorkerTerminationOrder :: !WorkerTerminationOrder + , onCapatazIntensityReached :: !(IO ()) + , notifyEvent :: !(CapatazEvent -> IO ()) + } + +-- | Default options to easily create capataz instances: +-- * name defaults to \"default-capataz\" +-- * intensity error tolerance is set to 1 error every 5 seconds +-- * has a "OneForOne " capataz restart strategy +-- * has a termination order of "OldestFirst" +defCapatazOptions :: CapatazOptions +defCapatazOptions = CapatazOptions + { capatazName = "default-capataz" + + -- One (1) restart every five (5) seconds + , capatazIntensity = 1 + , capatazPeriodSeconds = 5 + , capatazRestartStrategy = def + , capatazWorkerSpecList = [] + , capatazWorkerTerminationOrder = OldestFirst + , onCapatazIntensityReached = return () + , notifyEvent = const $ return () + } + +-- | Default options to easily create worker instances: +-- * name defaults to \"default-worker\" +-- * has a "Transient" worker restart strategy +-- * has a termination policy of three (3) seconds +defWorkerOptions :: WorkerOptions +defWorkerOptions = WorkerOptions + { workerName = "default-worker" + , workerOnFailure = const $ return () + , workerOnCompletion = return () + , workerOnTermination = return () + , workerTerminationPolicy = def + , workerRestartStrategy = def + } + +-- | Default spec to easily create worker instances: +-- * @IO ()@ sub-routine simply returns unit +-- * name defaults to \"default-worker\" +-- * has a "Transient" worker restart strategy +-- * has a termination policy of three (3) seconds +defWorkerSpec :: WorkerSpec +defWorkerSpec = WorkerSpec + { workerName = "default-worker" + , workerAction = return () + , workerOnFailure = const $ return () + , workerOnCompletion = return () + , workerOnTermination = return () + , workerTerminationPolicy = def + , workerRestartStrategy = def + } diff --git a/src/Control/Concurrent/Internal/Capataz/Util.hs b/src/Control/Concurrent/Internal/Capataz/Util.hs new file mode 100644 --- /dev/null +++ b/src/Control/Concurrent/Internal/Capataz/Util.hs @@ -0,0 +1,165 @@ +{-# LANGUAGE DuplicateRecordFields #-} +{-# LANGUAGE NamedFieldPuns #-} +{-# LANGUAGE NoImplicitPrelude #-} +{-# LANGUAGE RecordWildCards #-} +{-| This module contains: + +* Functions to manipulate the state of the Capataz record +* Utility functions used for communication between threads +* Public API utility functions + +-} +module Control.Concurrent.Internal.Capataz.Util where + +import Protolude + +import Control.Concurrent.STM (STM, atomically, retry) +import Control.Concurrent.STM.TQueue (writeTQueue) +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 qualified Data.HashMap.Strict as HashMap + +import Control.Concurrent.Internal.Capataz.Types + +-- | Returns only the number of the ThreadId +getTidNumber :: ThreadId -> Maybe Text +getTidNumber tid = case T.words $ show tid of + (_:tidNumber:_) -> Just tidNumber + _ -> Nothing + +-------------------------------------------------------------------------------- + +-- | Fetches a "Worker" from the "Capataz" instance environment +fetchWorker :: CapatazEnv -> WorkerId -> IO (Maybe Worker) +fetchWorker CapatazEnv { capatazWorkerMap } workerId = + HashMap.lookup workerId <$> readIORef capatazWorkerMap + +-- | Fetches a "WorkerEnv" from the "Capataz" instance environment +fetchWorkerEnv :: CapatazEnv -> WorkerId -> IO (Maybe WorkerEnv) +fetchWorkerEnv CapatazEnv { capatazWorkerMap } workerId = + ((workerToEnv <$>) . HashMap.lookup workerId) <$> readIORef capatazWorkerMap + +-- | Appends a new "Worker" to the "Capataz" existing worker map. +appendWorkerToMap :: CapatazEnv -> Worker -> IO () +appendWorkerToMap CapatazEnv { capatazWorkerMap } worker@Worker { workerId } = + atomicModifyIORef' capatazWorkerMap + (\workerMap -> (appendWorker workerMap, ())) + where appendWorker = HashMap.alter (const $ Just worker) workerId + +-- | Removes a "Worker" from the "Capataz" existing worker map. +removeWorkerFromMap :: CapatazEnv -> WorkerId -> IO () +removeWorkerFromMap CapatazEnv { capatazWorkerMap } workerId = + atomicModifyIORef' + capatazWorkerMap + ( \workerMap -> maybe (workerMap, ()) + (const (HashMap.delete workerId workerMap, ())) + (HashMap.lookup workerId workerMap) + ) + +-- | Function to modify a "Capataz" worker map using a pure function. +resetWorkerMap :: CapatazEnv -> (WorkerMap -> WorkerMap) -> IO () +resetWorkerMap CapatazEnv { capatazWorkerMap } workerMapFn = atomicModifyIORef' + capatazWorkerMap + (\workerMap -> (workerMapFn workerMap, ())) + +-- | Function to get a snapshot of the "Capataz"' worker map +readWorkerMap :: CapatazEnv -> IO WorkerMap +readWorkerMap CapatazEnv { capatazWorkerMap } = readIORef capatazWorkerMap + +-- | Returns all worker's of a "Capataz" by "WorkerTerminationOrder". This is +-- used "AllForOne" restarts and shutdown operations. +sortWorkersByTerminationOrder :: WorkerTerminationOrder -> WorkerMap -> [Worker] +sortWorkersByTerminationOrder terminationOrder workerMap = + case terminationOrder of + OldestFirst -> workers + NewestFirst -> reverse workers + where + -- NOTE: dissambiguates workerCreationTime field + workerCreationTime' Worker { workerCreationTime } = workerCreationTime + + workers = sortBy (comparing workerCreationTime') (HashMap.elems workerMap) + +-------------------------------------------------------------------------------- + +-- | Sub-routine that returns the "CapatazStatus", this sub-routine will block +-- until the "Capataz" has a status different from "Initializing". +readCapatazStatusSTM :: TVar CapatazStatus -> STM CapatazStatus +readCapatazStatusSTM statusVar = do + status <- readTVar statusVar + if status == Initializing then retry else return status + +-- | Sub-routine that returns the "CapatazStatus" on the IO monad +readCapatazStatus :: CapatazEnv -> IO CapatazStatus +readCapatazStatus CapatazEnv { capatazStatusVar } = + atomically $ readTVar capatazStatusVar + +-- | Modifes the "Capataz" status, this is the only function that should be used +-- to this end given it has the side-effect of notifying a status change via the +-- "notifyEvent" sub-routine, given via an attribute of the "CapatazOption" +-- record. +writeCapatazStatus :: CapatazEnv -> CapatazStatus -> IO () +writeCapatazStatus CapatazEnv { capatazId, capatazName, capatazStatusVar, notifyEvent } newCapatazStatus + = do + + prevCapatazStatus <- atomically $ do + prevStatus <- readTVar capatazStatusVar + writeTVar capatazStatusVar newCapatazStatus + return prevStatus + + eventTime <- getCurrentTime + notifyEvent CapatazStatusChanged + { capatazId + , capatazName + , prevCapatazStatus + , newCapatazStatus + , eventTime + } + + +-- | Used from public API functions to send a ControlAction to the Capataz +-- supervisor thread loop +sendControlMsg :: CapatazEnv -> ControlAction -> IO () +sendControlMsg CapatazEnv { capatazQueue } ctrlMsg = + atomically $ writeTQueue capatazQueue (ControlAction ctrlMsg) + +-- | Used from public API functions to send a ControlAction to the Capataz +-- supervisor thread loop, it receives an IO sub-routine that expects an IO +-- operation that blocks a thread until the message is done. +sendSyncControlMsg + :: CapatazEnv + -> (IO () -> ControlAction) -- ^ Blocking sub-routine used from the caller + -> IO () +sendSyncControlMsg CapatazEnv { capatazQueue } mkCtrlMsg = do + result <- newEmptyMVar + atomically + $ writeTQueue capatazQueue (ControlAction $ mkCtrlMsg (putMVar result ())) + takeMVar result + +-- | Utility function to transform a "CapatazRuntime" into a "CapatazEnv" +capatazToEnv :: CapatazRuntime -> CapatazEnv +capatazToEnv capatazRuntime@CapatazRuntime {..} = + let CapatazOptions {..} = capatazOptions in CapatazEnv {..} + +-- | Utility function to transform a "Worker" into a "WorkerEnv" +workerToEnv :: Worker -> WorkerEnv +workerToEnv Worker {..} = + let + WorkerSpec { workerAction, workerOnFailure, workerOnCompletion, workerOnTermination, workerRestartStrategy } + = workerSpec + in + WorkerEnv {..} + +-- | Utility function to transform a "WorkerEnv" into a "Worker" +envToWorker :: WorkerEnv -> Worker +envToWorker WorkerEnv {..} = Worker {..} + +-- | Utility function to transform a "WorkerOptions" into a "WorkerSpec" +workerOptionsToSpec :: WorkerOptions -> IO () -> WorkerSpec +workerOptionsToSpec WorkerOptions {..} workerAction = WorkerSpec {..} + +-- | Utility function to transform a "Capataz" into an @"Async" ()@ +capatazToAsync :: Capataz -> Async () +capatazToAsync = capatazAsync diff --git a/src/Control/Concurrent/Internal/Capataz/Worker.hs b/src/Control/Concurrent/Internal/Capataz/Worker.hs new file mode 100644 --- /dev/null +++ b/src/Control/Concurrent/Internal/Capataz/Worker.hs @@ -0,0 +1,312 @@ +{-# LANGUAGE DuplicateRecordFields #-} +{-# LANGUAGE NamedFieldPuns #-} +{-# LANGUAGE NoImplicitPrelude #-} +{-# LANGUAGE OverloadedStrings #-} +{-| This module contains all logic related to error handling when spawning threads + to execute Worker sub-routines +-} +module Control.Concurrent.Internal.Capataz.Worker where + +import Protolude + +import Control.Concurrent.Async (asyncWithUnmask) +import Control.Concurrent.STM.TQueue (writeTQueue) +import Data.Time.Clock (getCurrentTime) +import GHC.Conc (labelThread) + +import qualified Data.Text as T +import qualified Data.UUID.V4 as UUID + +import Control.Concurrent.Internal.Capataz.Types +import Control.Concurrent.Internal.Capataz.Util + (getTidNumber, readWorkerMap, sortWorkersByTerminationOrder) + +-- | Internal functions that overwrites the GHC thread name, for increasing +-- traceability on GHC internals +setWorkerThreadName :: WorkerId -> WorkerName -> IO () +setWorkerThreadName workerId workerName = do + tid <- myThreadId + let workerIdentifier = + T.unpack workerName <> "_" <> show workerId <> "_" <> maybe + "" + T.unpack + (getTidNumber tid) + labelThread tid workerIdentifier + +-- | Handles errors caused by the execution of the "workerMain" sub-routine +handleWorkerException + :: (IO () -> IO a) + -> CapatazEnv + -> WorkerSpec + -> WorkerId + -> RestartCount + -> SomeException + -> IO MonitorEvent +handleWorkerException unmask CapatazEnv { capatazId, capatazName, notifyEvent } WorkerSpec { workerName, workerOnFailure, workerOnTermination } workerId restartCount err + = do + workerThreadId <- myThreadId + monitorEventTime <- getCurrentTime + case fromException err of + Just RestartWorkerException -> + return WorkerForcedRestart {workerId , workerName , monitorEventTime } + + Just TerminateWorkerException { workerTerminationReason } -> do + eErrResult <- try $ unmask workerOnTermination + + notifyEvent WorkerCallbackExecuted + { capatazId + , capatazName + , workerId + , workerName + , workerThreadId + , workerCallbackError = either Just (const Nothing) eErrResult + , callbackType = OnTermination + , eventTime = monitorEventTime + } + + case eErrResult of + Left workerCallbackError -> return WorkerFailed' + { workerName + , workerId + , monitorEventTime + , workerError = toException WorkerCallbackFailed + { workerId + , workerCallbackError + , callbackType = OnTermination + , workerActionError = Just err + } + , workerRestartCount = restartCount + } + Right _ -> return WorkerTerminated' + { workerId + , workerName + , monitorEventTime + , workerTerminationReason + , workerRestartCount = restartCount + } + + Just BrutallyTerminateWorkerException { workerTerminationReason } -> + return WorkerTerminated' + { workerId + , workerName + , monitorEventTime + , workerTerminationReason + , workerRestartCount = restartCount + } + + -- This exception was an error from the given sub-routine + Nothing -> do + eErrResult <- try $ unmask $ workerOnFailure err + + notifyEvent WorkerCallbackExecuted + { capatazId + , capatazName + , workerId + , workerName + , workerThreadId + , workerCallbackError = either Just (const Nothing) eErrResult + , callbackType = OnFailure + , eventTime = monitorEventTime + } + + case eErrResult of + Left workerCallbackError -> return WorkerFailed' + { workerName + , workerId + , monitorEventTime + , workerRestartCount = restartCount + , workerError = toException WorkerCallbackFailed + { workerId + , workerCallbackError + , callbackType = OnFailure + , workerActionError = Just err + } + } + Right _ -> return WorkerFailed' + { workerName + , workerId + , monitorEventTime + , workerError = err + , workerRestartCount = restartCount + } + +-- | Handles completion of the "workerMain" sub-routine +handleWorkerCompletion + :: (IO () -> IO a) + -> CapatazEnv + -> WorkerSpec + -> WorkerId + -> RestartCount + -> IO MonitorEvent +handleWorkerCompletion unmask CapatazEnv { capatazId, capatazName, notifyEvent } WorkerSpec { workerName, workerOnCompletion } workerId restartCount + = do + workerThreadId <- myThreadId + monitorEventTime <- getCurrentTime + eCompResult <- try $ unmask workerOnCompletion + + notifyEvent WorkerCallbackExecuted + { capatazId + , capatazName + , workerId + , workerName + , workerThreadId + , workerCallbackError = either Just (const Nothing) eCompResult + , callbackType = OnCompletion + , eventTime = monitorEventTime + } + + case eCompResult of + Left err -> return WorkerFailed' + { workerName + , workerId + , monitorEventTime + , workerError = toException WorkerCallbackFailed + { workerId + , workerCallbackError = err + , callbackType = OnCompletion + , workerActionError = Nothing + } + , workerRestartCount = restartCount + } + Right _ -> + return WorkerCompleted' {workerName , workerId , monitorEventTime } + +-- | Decorates the given @IO ()@ sub-routine with failure handling +workerMain :: CapatazEnv -> WorkerSpec -> WorkerId -> RestartCount -> IO Worker +workerMain env@CapatazEnv { capatazQueue } workerSpec@WorkerSpec { workerName, workerAction } workerId restartCount + = do + workerCreationTime <- getCurrentTime + workerAsync <- asyncWithUnmask $ \unmask -> do + + eResult <- try $ do + setWorkerThreadName workerId workerName + unmask workerAction + + resultEvent <- case eResult of + Left err -> + handleWorkerException unmask env workerSpec workerId restartCount err + Right _ -> + handleWorkerCompletion unmask env workerSpec workerId restartCount + + atomically $ writeTQueue capatazQueue (MonitorEvent resultEvent) + + return Worker + { workerId + , workerName + , workerAsync + , workerCreationTime + , workerSpec + } + +-- | Internal function used to send a proper "CapatazEvent" to the "notifyEvent" +-- callback, this event can either be a @WorkerStarted@ or a @WorkerRestarted@ +notifyWorkerStarted :: Maybe (WorkerId, Int) -> CapatazEnv -> Worker -> IO () +notifyWorkerStarted mRestartInfo CapatazEnv { capatazId, capatazName, notifyEvent } Worker { workerId, workerName, workerAsync } + = do + eventTime <- getCurrentTime + case mRestartInfo of + Just (_workerId, workerRestartCount) -> notifyEvent WorkerRestarted + { capatazId + , capatazName + , workerId + , workerName + , workerRestartCount + , workerThreadId = asyncThreadId workerAsync + , eventTime + } + Nothing -> notifyEvent WorkerStarted + { capatazId + , capatazName + , workerId + , workerName + , eventTime + , workerThreadId = asyncThreadId workerAsync + } + +-- | Internal function that forks a worker thread on the Capataz thread; note +-- this is different from the public @forkWorker@ function which sends a message +-- to the capataz loop +forkWorker + :: CapatazEnv -> WorkerSpec -> Maybe (WorkerId, RestartCount) -> IO Worker +forkWorker env workerSpec mRestartInfo = do + (workerId, restartCount) <- case mRestartInfo of + Just (workerId, restartCount) -> pure (workerId, restartCount) + Nothing -> (,) <$> UUID.nextRandom <*> pure 0 + + worker <- workerMain env workerSpec workerId restartCount + notifyWorkerStarted mRestartInfo env worker + return worker + +-- | Internal function that forks a worker thread on the Capataz thread; note +-- this is different from the public @forkWorker@ function which sends a message +-- to the capataz loop +terminateWorker + :: Text -- ^ Text that indicates why there is a termination + -> CapatazEnv + -> Worker + -> IO () +terminateWorker workerTerminationReason CapatazEnv { capatazId, capatazName, notifyEvent } Worker { workerId, workerName, workerSpec, workerAsync } + = do + let WorkerSpec { workerTerminationPolicy } = workerSpec + case workerTerminationPolicy of + Infinity -> cancelWith + workerAsync + TerminateWorkerException {workerId , workerTerminationReason } + + BrutalTermination -> cancelWith + workerAsync + BrutallyTerminateWorkerException {workerId , workerTerminationReason } + + TimeoutMillis millis -> race_ + ( do + threadDelay (millis * 1000) + cancelWith + workerAsync + BrutallyTerminateWorkerException + { workerId + , workerTerminationReason + } + ) + ( cancelWith + workerAsync + TerminateWorkerException {workerId , workerTerminationReason } + ) + + eventTime <- getCurrentTime + notifyEvent WorkerTerminated + { capatazId + , capatazName + , eventTime + , workerId + , workerName + , workerThreadId = asyncThreadId workerAsync + , terminationReason = workerTerminationReason + } + + +-- | Internal sub-routine that terminates workers of a Capataz, used when a +-- Capataz instance is terminated +terminateWorkers :: Text -> CapatazEnv -> IO () +terminateWorkers terminationReason env@CapatazEnv { capatazName, capatazId, capatazWorkerTerminationOrder, notifyEvent } + = do + eventTime <- getCurrentTime + workerMap <- readWorkerMap env + + let workers = + sortWorkersByTerminationOrder capatazWorkerTerminationOrder workerMap + + notifyEvent WorkersTerminationStarted + { capatazName + , capatazId + , terminationReason + , eventTime + } + + forM_ workers (terminateWorker terminationReason env) + + notifyEvent WorkersTerminationFinished + { capatazName + , capatazId + , terminationReason + , eventTime + } diff --git a/test/testsuite/Control/Concurrent/CapatazTest.hs b/test/testsuite/Control/Concurrent/CapatazTest.hs new file mode 100644 --- /dev/null +++ b/test/testsuite/Control/Concurrent/CapatazTest.hs @@ -0,0 +1,1267 @@ +{-# LANGUAGE DuplicateRecordFields #-} +{-# LANGUAGE NamedFieldPuns #-} +{-# LANGUAGE NoImplicitPrelude #-} +{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE ScopedTypeVariables #-} +{-| This module contains: + +* Assertion functions to get attributes from a `CapatazEvent` + +* Helpers to run the test (reduce boilerplate) + +* Actual tests + +Tests just exercises the __public API__ and asserts all the events delivered via +the @notifyEvent@ callback are what we are expecting. + +NOTE: This tests may be flaky depending on the load of the application, there is +a ticket pending to add dejafu tests to ensure our tests are stable. + +-} +module Control.Concurrent.CapatazTest (tests) where + +import Protolude + +import qualified Data.Text as T + +import Data.IORef (atomicModifyIORef', newIORef, readIORef, writeIORef) +import Text.Show.Pretty (ppShow) + +import Test.Tasty (TestTree, testGroup) +import Test.Tasty.HUnit (assertBool, assertFailure, testCase) + +import Control.Concurrent.STM.TQueue (newTQueueIO, readTQueue, writeTQueue) +import Control.Concurrent.STM.TVar (modifyTVar', newTVarIO, readTVar) + +import qualified Control.Concurrent.Capataz as SUT + +-------------------------------------------------------------------------------- +-- Util + +-- | Utility function that gets the type name of a Record through it's Show +-- output. +fetchRecordName :: Show a => a -> Text +fetchRecordName = T.takeWhile (/= ' ') . show + +-- | Composes two predicate functions together with a boolean AND +andP :: [a -> Bool] -> a -> Bool +andP predList a = all ($ a) predList + +-------------------------------------------------------------------------------- +-- Assertions and Testers + +-- | This record duplicate the same event names as the ones found in the +-- "CapatazEvent" type, we use this to avoid using Text comparisons on assertion +-- helper functions. The "CapatazEvent" record is imported qualified, so there +-- is no conflict happening. +data EventType + = InvalidCapatazStatusReached + | CapatazStatusChanged + | WorkerTerminated + | WorkerStarted + | WorkerRestarted + | WorkerCompleted + | WorkerFailed + | WorkerCallbackExecuted + | WorkersTerminationStarted + | WorkersTerminationFinished + | CapatazFailed + | CapatazTerminated + deriving (Show) + +-- | Predicate function to assert "CapatazEvent" types +assertEventType :: EventType -> SUT.CapatazEvent -> Bool +assertEventType evType ev = fetchRecordName ev == show evType + +-- | Predicate function to assert "CapatazEvent" worker name +assertWorkerName :: Text -> SUT.CapatazEvent -> Bool +assertWorkerName workerName' ev = case ev of + SUT.WorkerRestarted { workerName } -> workerName' == workerName + SUT.WorkerFailed { workerName } -> workerName' == workerName + SUT.WorkerTerminated { workerName } -> workerName' == workerName + SUT.WorkerStarted { workerName } -> workerName' == workerName + _ -> False + +-- | Predicate function to assert type of an error inside a "CapatazEvent" +assertErrorType :: Text -> SUT.CapatazEvent -> Bool +assertErrorType errType ev = case ev of + SUT.WorkerFailed { workerError } -> fetchRecordName workerError == errType + SUT.CapatazFailed { capatazError } -> fetchRecordName capatazError == errType + SUT.WorkerCallbackExecuted { workerCallbackError } -> + case workerCallbackError of + Nothing -> False + Just originalError -> fetchRecordName originalError == errType + _ -> False + +-- | Predicate function to assert type of callback executed inside a +-- "CapatazEvent" +assertCallbackType :: SUT.CallbackType -> SUT.CapatazEvent -> Bool +assertCallbackType cbType ev = case ev of + SUT.WorkerFailed { workerError } -> case fromException workerError of + Just SUT.WorkerCallbackFailed { callbackType } -> cbType == callbackType + _ -> False + SUT.WorkerCallbackExecuted { callbackType } -> cbType == callbackType + _ -> False + +-- | Predicate function to assert restart count inside a "CapatazEvent" +assertRestartCount :: (Int -> Bool) -> SUT.CapatazEvent -> Bool +assertRestartCount predFn ev = case ev of + SUT.WorkerRestarted { workerRestartCount } -> predFn workerRestartCount + _ -> False + +-- | Predicate function to assert a Capataz status change +assertCapatazStatusChanged + :: SUT.CapatazStatus -> SUT.CapatazStatus -> SUT.CapatazEvent -> Bool +assertCapatazStatusChanged fromEv toEv ev = case ev of + SUT.CapatazStatusChanged { prevCapatazStatus, newCapatazStatus } -> + fromEv == prevCapatazStatus && toEv == newCapatazStatus + _ -> False + +-- | Predicate function to assert a worker was started +assertWorkerStarted :: Text -> SUT.CapatazEvent -> Bool +assertWorkerStarted workerName = + andP [assertEventType WorkerStarted, assertWorkerName workerName] + +-- | Predicate function to assert a worker was terminated +assertWorkerTerminated :: Text -> SUT.CapatazEvent -> Bool +assertWorkerTerminated workerName = + andP [assertEventType WorkerTerminated, assertWorkerName workerName] + +-- | Predicate function to assert a capataz thread failed with error type +assertCapatazFailedWith :: Text -> SUT.CapatazEvent -> Bool +assertCapatazFailedWith errorName = + andP [assertEventType CapatazFailed, assertErrorType errorName] + +-------------------------------------------------------------------------------- + +-- | Exception used to test failures inside Worker sub-routines +data RestartingWorkerError + = RestartingWorkerError + deriving (Show) + +instance Exception RestartingWorkerError + +-- | Exception used to test failures inside Worker callback sub-routines +data TimeoutError + = TimeoutError + deriving (Show) + +instance Exception TimeoutError + +-- | 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 +mkFailingSubRoutine failCount = do + countRef <- newIORef failCount + let subRoutine = do + shouldFail <- atomicModifyIORef' countRef + (\count -> (pred count, count > 0)) + when shouldFail (throwIO RestartingWorkerError) + + return subRoutine + +-- | A sub-routine that will complete for `initCount` amount of times. 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) +mkCompletingBeforeNRestartsSubRoutine :: Int -> IO (IO ()) +mkCompletingBeforeNRestartsSubRoutine initCount = do + countRef <- newIORef initCount + let subRoutine = do + shouldStop <- atomicModifyIORef' countRef + (\count -> (pred count, 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 = mkCompletingBeforeNRestartsSubRoutine 1 + +-- | Utility function to build a test environment for a Capataz execution. +-- It is composed by: +-- +-- * List of assertions that represent events that should be triggered by the +-- capataz instance in order +-- +-- * A function to modify the default "CapatazOptions", this utility function injects +-- a special @notifyEvent@ callback to execute given assertions. +testCapatazStreamWithOptions + :: [SUT.CapatazEvent -> Bool] -- ^ Assertions happening before setup function + -- is called + -> (SUT.CapatazOptions -> SUT.CapatazOptions) -- ^ Function to modify default + -- @CapatazOptions@ + -> (SUT.Capataz -> IO ()) -- ^ Function used to test public the supervisor + -- public API (a.k.a setup function) + -> [SUT.CapatazEvent -> Bool] -- ^ Assertions happening after the setup + -- function + -> [SUT.CapatazEvent -> Bool] -- ^ Assertions happening after the capataz + -- record is terminated + -> Maybe (SUT.CapatazEvent -> Bool) -- ^ An assertion checked across all + -- @CapatazEvents@ that happened in a + -- test, great when testing that an event + -- __did not__ happen + -> IO () +testCapatazStreamWithOptions preSetupAssertion optionModFn setupFn postSetupAssertions postTeardownAssertions mAllEventsAssertion + = do + + eventStream <- newTQueueIO + accRef <- newIORef [] + pendingCountVar <- newIORef + ( sum $ fmap + length + [preSetupAssertion, postSetupAssertions, postTeardownAssertions] + ) + + capataz <- SUT.forkCapataz $ (optionModFn SUT.defCapatazOptions) + { SUT.notifyEvent = trackEvent accRef eventStream + } + + -- 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) + + 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 + + -- We now shutdown the capataz instance + void $ SUT.teardown 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) + where + -- Utility functions that runs the readEventLoop function with a timeout + -- of a second, this way we can guarantee assertions are met without having + -- to add @threadDelays@ to the test execution + runAssertions stageName (eventStream, accRef) pendingCountVar assertions capataz + = do + raceResult <- race + (threadDelay 1000100) + (readEventLoop eventStream pendingCountVar assertions) + case raceResult of + Left _ -> do + events <- reverse <$> readIORef accRef + pendingCount <- readIORef pendingCountVar + void $ SUT.teardown capataz + assertFailure + ( "On " + <> stageName + <> " stage, expected all assertions to match, but didn't (" + <> show pendingCount + <> " assertions remaining, " + <> show (length events) + <> " events tried)\n" + <> ppShow (zip ([0 ..] :: [Int]) events) + ) + Right _ -> return () + + + -- Sub-routine that accumulates all events that have happened in the Capataz + -- instance so far + trackEvent accRef eventStream event = do + atomicModifyIORef' accRef (\old -> (event : old, ())) + atomically $ writeTQueue eventStream event + + -- Sub-routine that reads the event stream, and ensures that all assertions + -- are executed, this loop won't stop until all assertions are met + readEventLoop eventStream pendingCount assertions = do + writeIORef pendingCount (length assertions) + case assertions of + [] -> return () + (assertionFn:assertions1) -> do + event <- atomically $ readTQueue eventStream + if assertionFn event + then readEventLoop eventStream pendingCount assertions1 + else readEventLoop eventStream pendingCount assertions + + +-- | A version of "testCapatazStreamWithOptions" that does not receive the +-- function that modifies a "CapatazOptions" record. +testCapatazStream + :: [SUT.CapatazEvent -> Bool] -- ^ Assertions happening before setup function + -- is called + -> (SUT.Capataz -> IO ()) -- ^ Function used to test public the supervisor + -- public API (a.k.a setup function) + -> [SUT.CapatazEvent -> Bool] -- ^ Assertions happening after the setup + -- function + -> [SUT.CapatazEvent -> Bool] -- ^ Assertions happening after the capataz + -- record is terminated + -> Maybe (SUT.CapatazEvent -> Bool) -- ^ An assertion checked across all + -- @CapatazEvents@ that happened in a + -- test, great when testing that an event + -- __did not__ happen + -> IO () +testCapatazStream preSetupAssertions = + testCapatazStreamWithOptions preSetupAssertions identity + +-------------------------------------------------------------------------------- +-- Actual Tests + +tests :: [TestTree] +tests + = [ testGroup + "capataz without workerSpecList" + [ testCase "initialize and teardown works as expected" $ testCapatazStream + [ andP + [ assertEventType CapatazStatusChanged + , assertCapatazStatusChanged SUT.Initializing SUT.Running + ] + ] + (const $ return ()) + [] + [ andP + [ assertEventType CapatazStatusChanged + , assertCapatazStatusChanged SUT.Running SUT.Halting + ] + , andP + [ assertEventType CapatazStatusChanged + , assertCapatazStatusChanged SUT.Halting SUT.Halted + ] + ] + Nothing + ] + , testGroup + "capataz with workerSpecList" + [ testCase "initialize and teardown works as expected" + $ testCapatazStreamWithOptions + [ assertWorkerStarted "A" + , assertWorkerStarted "B" + , andP + [ assertEventType CapatazStatusChanged + , assertCapatazStatusChanged SUT.Initializing SUT.Running + ] + ] + ( \supOptions -> supOptions + { SUT.capatazWorkerSpecList = [ SUT.defWorkerSpec + { SUT.workerName = "A" + , SUT.workerAction = forever + (threadDelay 10001000) + } + , SUT.defWorkerSpec + { SUT.workerName = "B" + , SUT.workerAction = forever + (threadDelay 10001000) + } + ] + } + ) + (const $ return ()) + [] + [ andP + [ assertEventType CapatazStatusChanged + , assertCapatazStatusChanged SUT.Running SUT.Halting + ] + , assertEventType WorkersTerminationStarted + , assertWorkerTerminated "A" + , assertWorkerTerminated "B" + , assertEventType WorkersTerminationFinished + , andP + [ assertEventType CapatazStatusChanged + , assertCapatazStatusChanged SUT.Halting SUT.Halted + ] + ] + Nothing + ] + , testCase "reports error when capataz thread receives async exception" + $ testCapatazStream + [ andP + [ assertEventType CapatazStatusChanged + , assertCapatazStatusChanged SUT.Initializing SUT.Running + ] + ] + ( \SUT.Capataz { capatazAsync } -> do + threadDelay 100 -- leave enough room for capataz to start + cancelWith capatazAsync (ErrorCall "async exception") + ) + [assertEventType CapatazFailed] + [] + Nothing + , testCase "reports error when worker retries violate restart intensity" + $ do + lockVar <- newEmptyMVar + let (signalIntensityReached, waitTillIntensityReached) = + (putMVar lockVar (), takeMVar lockVar) + testCapatazStreamWithOptions + [] + ( \supOptions -> supOptions + { SUT.onCapatazIntensityReached = signalIntensityReached + } + ) + ( \capataz -> do + _workerId <- SUT.forkWorker SUT.defWorkerOptions + (throwIO RestartingWorkerError) + capataz + waitTillIntensityReached + ) + [ assertEventType WorkerFailed + , assertEventType WorkerFailed + , assertEventType WorkerFailed + , assertCapatazFailedWith "CapatazIntensityReached" + ] + [] + Nothing + , testGroup + "single supervised IO sub-routine" + [ testGroup + "callbacks" + [ testGroup + "workerOnCompletion" + [ testCase "does execute callback when sub-routine is completed" + $ testCapatazStream + [] + ( \capataz -> do + _workerId <- SUT.forkWorker + ( SUT.defWorkerOptions + { SUT.workerRestartStrategy = SUT.Temporary + } + ) + (return ()) + capataz + return () + ) + [ andP + [ assertEventType WorkerCallbackExecuted + , assertCallbackType SUT.OnCompletion + ] + , assertEventType WorkerCompleted + ] + [] + Nothing + , testCase "does not execute callback when sub-routine fails" + $ testCapatazStream + [] + ( \capataz -> do + _workerId <- SUT.forkWorker + ( SUT.defWorkerOptions + { SUT.workerRestartStrategy = SUT.Temporary + } + ) + (throwIO RestartingWorkerError) + capataz + return () + ) + [ andP + [ assertEventType WorkerCallbackExecuted + , assertCallbackType SUT.OnFailure + ] + , assertEventType WorkerFailed + ] + [assertEventType CapatazTerminated] + ( Just $ not . andP + [ assertEventType WorkerCallbackExecuted + , assertCallbackType SUT.OnCompletion + ] + ) + , testCase "does not execute callback when sub-routine is terminated" + $ testCapatazStream + [] + ( \capataz -> do + workerId <- SUT.forkWorker + ( SUT.defWorkerOptions + { SUT.workerRestartStrategy = SUT.Temporary + } + ) + (forever $ threadDelay 1000100) + capataz + + _workerId <- SUT.terminateWorker + "testing onCompletion callback" + workerId + capataz + return () + ) + [assertEventType WorkerTerminated] + [assertEventType CapatazTerminated] + ( Just $ not . andP + [ assertEventType WorkerCallbackExecuted + , assertCallbackType SUT.OnCompletion + ] + ) + , testCase "treats as sub-routine failed if callback fails" + $ testCapatazStream + [] + ( \capataz -> do + _workerId <- SUT.forkWorker + ( SUT.defWorkerOptions + { SUT.workerRestartStrategy = SUT.Temporary + , SUT.workerOnCompletion = throwIO TimeoutError + } + ) + (return ()) + capataz + + return () + ) + [ andP + [ assertEventType WorkerCallbackExecuted + , assertCallbackType SUT.OnCompletion + , assertErrorType "TimeoutError" + ] + , andP + [ assertEventType WorkerFailed + , assertErrorType "WorkerCallbackFailed" + ] + ] + [] + Nothing + ] + , testGroup + "workerOnFailure" + [ testCase "does execute callback when sub-routine fails" + $ testCapatazStream + [] + ( \capataz -> do + _workerId <- SUT.forkWorker + ( SUT.defWorkerOptions + { SUT.workerRestartStrategy = SUT.Temporary + } + ) + (throwIO RestartingWorkerError) + capataz + return () + ) + [ andP + [ assertEventType WorkerCallbackExecuted + , assertCallbackType SUT.OnFailure + ] + , assertEventType WorkerFailed + ] + [assertEventType CapatazTerminated] + Nothing + , testCase "does not execute callback when sub-routine is completed" + $ testCapatazStream + [] + ( \capataz -> do + _workerId <- SUT.forkWorker + ( SUT.defWorkerOptions + { SUT.workerRestartStrategy = SUT.Temporary + } + ) + (return ()) + capataz + return () + ) + [ andP + [ assertEventType WorkerCallbackExecuted + , assertCallbackType SUT.OnCompletion + ] + , assertEventType WorkerCompleted + ] + [] + ( Just + $ not + . andP + [ assertEventType WorkerCallbackExecuted + , assertCallbackType SUT.OnFailure + ] + ) + , testCase "does not execute callback when sub-routine is terminated" + $ testCapatazStream + [] + ( \capataz -> do + workerId <- SUT.forkWorker + ( SUT.defWorkerOptions + { SUT.workerRestartStrategy = SUT.Temporary + } + ) + (forever $ threadDelay 1000100) + capataz + + SUT.terminateWorker "testing onFailure callback" + workerId + capataz + ) + [assertEventType WorkerTerminated] + [] + ( Just + $ not + . andP + [ assertEventType WorkerCallbackExecuted + , assertCallbackType SUT.OnFailure + ] + ) + , testCase "treats as sub-routine failed if callback fails" + $ testCapatazStream + [] + ( \capataz -> do + _workerId <- SUT.forkWorker + ( SUT.defWorkerOptions + { SUT.workerRestartStrategy = SUT.Temporary + , SUT.workerOnFailure = const $ throwIO TimeoutError + } + ) + (throwIO RestartingWorkerError) + capataz + + return () + ) + [ andP + [ assertEventType WorkerCallbackExecuted + , assertCallbackType SUT.OnFailure + , assertErrorType "TimeoutError" + ] + , andP + [ assertEventType WorkerFailed + , assertErrorType "WorkerCallbackFailed" + ] + ] + [] + Nothing + ] + , testGroup + "workerOnTermination" + [ testCase + "gets brutally killed when TimeoutSeconds termination policy is not met" + $ testCapatazStream + [] + ( \capataz -> do + workerId <- SUT.forkWorker + ( SUT.defWorkerOptions + { SUT.workerRestartStrategy = SUT.Temporary + , SUT.workerTerminationPolicy = SUT.TimeoutMillis 1 + , SUT.workerOnTermination = forever $ threadDelay 100100 + } + ) + (forever $ threadDelay 10001000) + capataz + + SUT.terminateWorker "testing workerOnTermination callback" + workerId + capataz + ) + [ andP + [ assertEventType WorkerCallbackExecuted + , assertCallbackType SUT.OnTermination + , assertErrorType "BrutallyTerminateWorkerException" + ] + , andP + [ assertEventType WorkerFailed + , assertErrorType "WorkerCallbackFailed" + , assertCallbackType SUT.OnTermination + ] + ] + [] + Nothing + , testCase "does execute callback when sub-routine is terminated" + $ testCapatazStream + [] + ( \capataz -> do + workerId <- SUT.forkWorker + ( SUT.defWorkerOptions + { SUT.workerRestartStrategy = SUT.Temporary + } + ) + (forever $ threadDelay 1000100) + capataz + + SUT.terminateWorker "testing workerOnTermination callback" + workerId + capataz + ) + [ andP + [ assertEventType WorkerCallbackExecuted + , assertCallbackType SUT.OnTermination + ] + , assertEventType WorkerTerminated + ] + [assertEventType CapatazTerminated] + Nothing + , testCase "does not execute callback when sub-routine is completed" + $ testCapatazStream + [] + ( \capataz -> do + _workerId <- SUT.forkWorker + ( SUT.defWorkerOptions + { SUT.workerRestartStrategy = SUT.Temporary + } + ) + (return ()) + capataz + return () + ) + [assertEventType WorkerCompleted] + [] + ( Just $ not . andP + [ assertEventType WorkerCallbackExecuted + , assertCallbackType SUT.OnTermination + ] + ) + , testCase "does not execute callback when sub-routine fails" + $ testCapatazStream + [] + ( \capataz -> do + _workerId <- SUT.forkWorker + ( SUT.defWorkerOptions + { SUT.workerRestartStrategy = SUT.Temporary + } + ) + (throwIO (ErrorCall "surprise!")) + capataz + return () + ) + [assertEventType WorkerFailed] + [] + ( Just $ not . andP + [ assertEventType WorkerCallbackExecuted + , assertCallbackType SUT.OnTermination + ] + ) + , testCase "treats as sub-routine failed if callback fails" + $ testCapatazStream + [] + ( \capataz -> do + workerId <- SUT.forkWorker + ( SUT.defWorkerOptions + { SUT.workerRestartStrategy = SUT.Temporary + , SUT.workerOnTermination = throwIO TimeoutError + } + ) + (forever $ threadDelay 10001000) + capataz + + SUT.terminateWorker "testing workerOnTermination callback" + workerId + capataz + ) + [ andP + [ assertEventType WorkerCallbackExecuted + , assertCallbackType SUT.OnTermination + , assertErrorType "TimeoutError" + ] + , andP + [ assertEventType WorkerFailed + , assertErrorType "WorkerCallbackFailed" + ] + ] + [] + Nothing + ] + ] + , testGroup + "with transient strategy" + [ testCase "does not restart on completion" $ testCapatazStream + [] + ( \capataz -> do + _workerId <- SUT.forkWorker + SUT.defWorkerOptions { SUT.workerRestartStrategy = SUT.Transient } + (return ()) + capataz + return () + ) + [assertEventType WorkerStarted, assertEventType WorkerCompleted] + [assertEventType CapatazTerminated] + (Just $ not . assertEventType WorkerRestarted) + , testCase "does not restart on termination" $ testCapatazStream + [] + ( \capataz -> do + workerId <- SUT.forkWorker + SUT.defWorkerOptions { SUT.workerRestartStrategy = SUT.Transient } + (forever $ threadDelay 1000100) + capataz + SUT.terminateWorker "termination test (1)" workerId capataz + ) + [assertEventType WorkerTerminated] + [assertEventType CapatazTerminated] + (Just $ not . assertEventType WorkerRestarted) + , testCase "does restart on failure" $ testCapatazStream + [] + ( \capataz -> do + subRoutineAction <- mkFailingSubRoutine 1 + _workerId <- SUT.forkWorker + SUT.defWorkerOptions { SUT.workerRestartStrategy = SUT.Transient } + subRoutineAction + capataz + return () + ) + [ assertEventType WorkerStarted + , assertEventType WorkerFailed + , andP [assertEventType WorkerRestarted, assertRestartCount (== 1)] + ] + [] + Nothing + , testCase "does increase restart count on multiple failures" + $ testCapatazStream + [] + ( \capataz -> do + subRoutineAction <- mkFailingSubRoutine 2 + _workerId <- SUT.forkWorker + SUT.defWorkerOptions + { SUT.workerRestartStrategy = SUT.Transient + } + subRoutineAction + capataz + return () + ) + [ andP + [assertEventType WorkerRestarted, assertRestartCount (== 1)] + , andP + [assertEventType WorkerRestarted, assertRestartCount (== 2)] + ] + [] + Nothing + ] + , testGroup + "with permanent strategy" + [ testCase "does restart on completion" $ testCapatazStream + [] + ( \capataz -> do + subRoutineAction <- mkCompletingOnceSubRoutine + _workerId <- SUT.forkWorker + SUT.defWorkerOptions { SUT.workerRestartStrategy = SUT.Permanent } + subRoutineAction + capataz + return () + ) + [ assertEventType WorkerStarted + , assertEventType WorkerCompleted + , assertEventType WorkerRestarted + ] + [assertEventType CapatazTerminated] + Nothing + , testCase "does not increase restart count on multiple completions" + $ testCapatazStream + [] + ( \capataz -> do + -- Note the number is two (2) given the assertion list has two `WorkerRestarted` assertions + let expectedRestartCount = 2 + subRoutineAction <- mkCompletingBeforeNRestartsSubRoutine + expectedRestartCount + _workerId <- SUT.forkWorker + SUT.defWorkerOptions + { SUT.workerRestartStrategy = SUT.Permanent + } + subRoutineAction + capataz + return () + ) + [ andP + [assertEventType WorkerRestarted, assertRestartCount (== 1)] + , andP + [assertEventType WorkerRestarted, assertRestartCount (== 1)] + ] + [] + Nothing + , testCase "does restart on termination" $ testCapatazStream + [] + ( \capataz -> do + workerId <- SUT.forkWorker + SUT.defWorkerOptions { SUT.workerRestartStrategy = SUT.Permanent } + (forever $ threadDelay 10001000) + capataz + SUT.terminateWorker "testing termination (1)" workerId capataz + ) + [assertEventType WorkerTerminated, assertEventType WorkerRestarted] + [] + Nothing + , testCase "does increase restart count on multiple terminations" $ do + terminationCountVar <- newTVarIO (0 :: Int) + let signalWorkerTermination = + atomically (modifyTVar' terminationCountVar (+ 1)) + waitWorkerTermination i = atomically $ do + n <- readTVar terminationCountVar + when (n /= i) retry + testCapatazStream + [] + ( \capataz -> do + workerId <- SUT.forkWorker + SUT.defWorkerOptions + { SUT.workerRestartStrategy = SUT.Permanent + , SUT.workerOnTermination = signalWorkerTermination + } + (forever $ threadDelay 10001000) + capataz + + SUT.terminateWorker "testing termination (1)" workerId capataz + waitWorkerTermination 1 + SUT.terminateWorker "testing termination (2)" workerId capataz + waitWorkerTermination 2 + ) + [ assertEventType WorkerTerminated + , andP [assertEventType WorkerRestarted, assertRestartCount (== 1)] + , assertEventType WorkerTerminated + , andP [assertEventType WorkerRestarted, assertRestartCount (== 2)] + ] + [] + Nothing + , testCase "does restart on failure" $ testCapatazStream + [] + ( \capataz -> do + subRoutineAction <- mkFailingSubRoutine 1 + _workerId <- SUT.forkWorker + SUT.defWorkerOptions { SUT.workerRestartStrategy = SUT.Permanent } + subRoutineAction + capataz + return () + ) + [ assertEventType WorkerStarted + , assertEventType WorkerFailed + , andP [assertEventType WorkerRestarted, assertRestartCount (== 1)] + ] + [] + Nothing + , testCase "does increase restart count on multiple failures" + $ testCapatazStream + [] + ( \capataz -> do + subRoutineAction <- mkFailingSubRoutine 2 + _workerId <- SUT.forkWorker + SUT.defWorkerOptions + { SUT.workerRestartStrategy = SUT.Permanent + } + subRoutineAction + capataz + return () + ) + [ andP + [assertEventType WorkerRestarted, assertRestartCount (== 1)] + , andP + [assertEventType WorkerRestarted, assertRestartCount (== 2)] + ] + [] + Nothing + ] + , testGroup + "with temporary strategy" + [ testCase "does not restart on completion" $ testCapatazStream + [] + ( \capataz -> do + _workerId <- SUT.forkWorker + SUT.defWorkerOptions { SUT.workerRestartStrategy = SUT.Temporary } + (return ()) + capataz + return () + ) + [assertEventType WorkerStarted, assertEventType WorkerCompleted] + [assertEventType CapatazTerminated] + (Just $ not . assertEventType WorkerRestarted) + , testCase "does not restart on termination" $ testCapatazStream + [] + ( \capataz -> do + workerId <- SUT.forkWorker + SUT.defWorkerOptions { SUT.workerRestartStrategy = SUT.Temporary } + (forever $ threadDelay 1000100) + capataz + SUT.terminateWorker "termination test (1)" workerId capataz + threadDelay 100 + ) + [assertEventType WorkerStarted, assertEventType WorkerTerminated] + [assertEventType CapatazTerminated] + (Just $ not . assertEventType WorkerRestarted) + , testCase "does not restart on failure" $ testCapatazStream + [] + ( \capataz -> do + _workerId <- SUT.forkWorker + SUT.defWorkerOptions { SUT.workerRestartStrategy = SUT.Temporary } + (panic "worker failed!") + capataz + threadDelay 100 + ) + [assertEventType WorkerStarted, assertEventType WorkerFailed] + [assertEventType CapatazTerminated] + (Just $ not . assertEventType WorkerRestarted) + ] + ] + , testGroup + "multiple supervised IO sub-routines" + [ testCase "terminates all supervised sub-routines on teardown" + $ testCapatazStream + [] + ( \capataz -> do + _workerA <- SUT.forkWorker + SUT.defWorkerOptions { SUT.workerName = "A" + , SUT.workerRestartStrategy = SUT.Permanent + } + (forever $ threadDelay 1000100) + capataz + + + _workerB <- SUT.forkWorker + SUT.defWorkerOptions { SUT.workerName = "B" + , SUT.workerRestartStrategy = SUT.Permanent + } + (forever $ threadDelay 1000100) + capataz + + return () + ) + [ andP [assertEventType WorkerStarted, assertWorkerName "A"] + , andP [assertEventType WorkerStarted, assertWorkerName "B"] + ] + [ andP [assertEventType WorkerTerminated, assertWorkerName "A"] + , andP [assertEventType WorkerTerminated, assertWorkerName "B"] + , assertEventType CapatazTerminated + ] + Nothing + , testGroup + "with one for one capataz restart strategy" + [ testCase "restarts failing sub-routine only" + $ testCapatazStreamWithOptions + [] + ( \supOptions -> + supOptions { SUT.capatazRestartStrategy = SUT.OneForOne } + ) + ( \capataz -> do + _workerA <- SUT.forkWorker + SUT.defWorkerOptions + { SUT.workerName = "A" + , SUT.workerRestartStrategy = SUT.Temporary + } + (forever $ threadDelay 1000100) + capataz + + ioB <- mkFailingSubRoutine 1 + + _workerB <- SUT.forkWorker + SUT.defWorkerOptions + { SUT.workerName = "B" + , SUT.workerRestartStrategy = SUT.Permanent + } + (forever $ ioB >> threadDelay 1000100) + capataz + + return () + ) + [andP [assertEventType WorkerRestarted, assertWorkerName "B"]] + [] + ( Just $ not . andP + [assertEventType WorkerRestarted, assertWorkerName "A"] + ) + ] + , testGroup + "with all for one capataz restart strategy with newest first order" + [ testCase "does terminate all other workers that did not fail" + $ testCapatazStreamWithOptions + [] + ( \supOptions -> supOptions + { SUT.capatazRestartStrategy = SUT.AllForOne + , SUT.capatazWorkerTerminationOrder = SUT.OldestFirst + } + ) + ( \capataz -> do + -- This lockVar guarantees that workerB executes before workerA + lockVar <- newEmptyMVar + + ioA <- mkFailingSubRoutine 1 + + _workerA <- SUT.forkWorker + SUT.defWorkerOptions + { SUT.workerName = "A" + , SUT.workerRestartStrategy = SUT.Permanent + } + (forever $ readMVar lockVar >> ioA) + capataz + + _workerB <- SUT.forkWorker + SUT.defWorkerOptions + { SUT.workerName = "B" + , SUT.workerRestartStrategy = SUT.Permanent + } + (putMVar lockVar () >> forever (threadDelay 10)) + capataz + + return () + ) + [ andP [assertEventType WorkerStarted, assertWorkerName "A"] + , andP [assertEventType WorkerStarted, assertWorkerName "B"] + , andP [assertEventType WorkerFailed, assertWorkerName "A"] + , andP [assertEventType WorkerRestarted, assertWorkerName "A"] + , andP [assertEventType WorkerTerminated, assertWorkerName "B"] + , andP [assertEventType WorkerRestarted, assertWorkerName "B"] + ] + [] + Nothing + , testCase "does not restart sub-routines that are temporary" + $ testCapatazStreamWithOptions + [] + ( \supOptions -> supOptions + { SUT.capatazRestartStrategy = SUT.AllForOne + , SUT.capatazWorkerTerminationOrder = SUT.OldestFirst + } + ) + ( \capataz -> do + lockVar <- newEmptyMVar + + ioA <- mkFailingSubRoutine 1 + + _workerA <- SUT.forkWorker + SUT.defWorkerOptions + { SUT.workerName = "A" + , SUT.workerRestartStrategy = SUT.Permanent + } + (forever $ readMVar lockVar >> ioA) + capataz + + _workerB <- SUT.forkWorker + SUT.defWorkerOptions + { SUT.workerName = "B" + , SUT.workerRestartStrategy = SUT.Temporary + } + (putMVar lockVar () >> forever (threadDelay 10)) + capataz + + return () + ) + [ andP [assertEventType WorkerStarted, assertWorkerName "A"] + , andP [assertEventType WorkerStarted, assertWorkerName "B"] + , andP [assertEventType WorkerFailed, assertWorkerName "A"] + , andP [assertEventType WorkerRestarted, assertWorkerName "A"] + , andP [assertEventType WorkerTerminated, assertWorkerName "B"] + ] + [] + ( Just $ not . andP + [assertEventType WorkerRestarted, assertWorkerName "B"] + ) + , testCase "restarts sub-routines that are not temporary" + $ testCapatazStreamWithOptions + [] + ( \supOptions -> supOptions + { SUT.capatazRestartStrategy = SUT.AllForOne + , SUT.capatazWorkerTerminationOrder = SUT.NewestFirst + } + ) + ( \capataz -> do + ioA <- mkFailingSubRoutine 1 + + lockVar <- newEmptyMVar + + _workerA <- SUT.forkWorker + SUT.defWorkerOptions + { SUT.workerName = "A" + , SUT.workerRestartStrategy = SUT.Permanent + } + (forever $ readMVar lockVar >> ioA) + capataz + + _workerB <- SUT.forkWorker + SUT.defWorkerOptions + { SUT.workerName = "B" + , SUT.workerRestartStrategy = SUT.Transient + } + (putMVar lockVar () >> forever (threadDelay 10)) + capataz + + return () + ) + [ andP [assertEventType WorkerRestarted, assertWorkerName "B"] + , andP [assertEventType WorkerRestarted, assertWorkerName "A"] + ] + [] + Nothing + ] + , testGroup + "with all for one capataz restart strategy with oldest first order" + [ testCase "does not restart sub-routines that are temporary" + $ testCapatazStreamWithOptions + [] + ( \supOptions -> supOptions + { SUT.capatazRestartStrategy = SUT.AllForOne + , SUT.capatazWorkerTerminationOrder = SUT.OldestFirst + } + ) + ( \capataz -> do + ioA <- mkFailingSubRoutine 1 + + -- This lockVar guarantees that workerB executes before workerA + lockVar <- newEmptyMVar + + _workerA <- SUT.forkWorker + SUT.defWorkerOptions + { SUT.workerName = "A" + , SUT.workerRestartStrategy = SUT.Permanent + } + (forever $ readMVar lockVar >> ioA) + capataz + + _workerB <- SUT.forkWorker + SUT.defWorkerOptions + { SUT.workerName = "B" + , SUT.workerRestartStrategy = SUT.Temporary + } + (putMVar lockVar () >> forever (threadDelay 10)) + capataz + + return () + ) + [andP [assertEventType WorkerRestarted, assertWorkerName "A"]] + [] + ( Just $ not . andP + [assertEventType WorkerRestarted, assertWorkerName "B"] + ) + , testCase "restarts sub-routines that are not temporary" + $ testCapatazStreamWithOptions + [] + ( \supOptions -> supOptions + { SUT.capatazRestartStrategy = SUT.AllForOne + , SUT.capatazWorkerTerminationOrder = SUT.OldestFirst + } + ) + ( \capataz -> do + ioA <- mkFailingSubRoutine 1 + + -- This lockVar guarantees that workerB executes before workerA + lockVar <- newEmptyMVar + + _workerA <- SUT.forkWorker + SUT.defWorkerOptions + { SUT.workerName = "A" + , SUT.workerRestartStrategy = SUT.Permanent + } + (forever $ readMVar lockVar >> ioA) + capataz + + _workerB <- SUT.forkWorker + SUT.defWorkerOptions + { SUT.workerName = "B" + , SUT.workerRestartStrategy = SUT.Transient + } + (putMVar lockVar () >> forever (threadDelay 10)) + capataz + + return () + ) + [ andP [assertEventType WorkerRestarted, assertWorkerName "A"] + , andP [assertEventType WorkerRestarted, assertWorkerName "B"] + ] + [] + Nothing + ] + ] + ] diff --git a/test/testsuite/Main.hs b/test/testsuite/Main.hs new file mode 100644 --- /dev/null +++ b/test/testsuite/Main.hs @@ -0,0 +1,18 @@ +{-# LANGUAGE DuplicateRecordFields #-} +{-# LANGUAGE NamedFieldPuns #-} +{-# LANGUAGE NoImplicitPrelude #-} +{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE ScopedTypeVariables #-} +module Main where + +import Protolude + +import Control.Concurrent.CapatazTest +import Test.Tasty (defaultMainWithIngredients, testGroup) +import Test.Tasty.Ingredients.Rerun (rerunningTests) +import Test.Tasty.Runners (consoleTestReporter, listingTests) + +main :: IO () +main = defaultMainWithIngredients + [rerunningTests [listingTests, consoleTestReporter]] + (testGroup "capataz" tests)