diff --git a/src/Control/Concurrent/Supervisor.hs b/src/Control/Concurrent/Supervisor.hs --- a/src/Control/Concurrent/Supervisor.hs +++ b/src/Control/Concurrent/Supervisor.hs @@ -9,26 +9,41 @@ {-# LANGUAGE BangPatterns #-} {-# LANGUAGE RecordWildCards #-} {-# LANGUAGE ScopedTypeVariables #-} +{-# LANGUAGE DeriveDataTypeable #-} module Control.Concurrent.Supervisor - ( Supervisor + ( SupervisorSpec + , Supervisor , DeadLetter - , Child(..) , RestartAction , SupervisionEvent(..) , RestartStrategy(..) + -- * Creating a new supervisor spec + -- $new + , newSupervisorSpec + -- * Creating a new supervisor + -- $sup , newSupervisor + -- * Stopping a supervisor + -- $shutdown , shutdownSupervisor + -- * Accessing Supervisor event log + -- $log , eventStream , activeChildren - , supervise + -- * Supervise a forked thread + -- $fork , forkSupervised + -- * Monitor another supervisor + -- $monitor + , monitor ) where import qualified Data.HashMap.Strict as Map import Control.Concurrent import Control.Concurrent.STM import Data.IORef +import Data.Typeable import Control.Exception import Control.Monad import Data.Time @@ -42,13 +57,13 @@ NewSupervisor :: { _ns_myTid :: !(Maybe ThreadId) , _ns_children :: !(IORef (Map.HashMap ThreadId Child)) - , _ns_mailbox :: TQueue DeadLetter + , _ns_mailbox :: TChan DeadLetter , _ns_eventStream :: TBQueue SupervisionEvent } -> Supervisor_ Uninitialised Supervisor :: { _sp_myTid :: !(Maybe ThreadId) , _sp_children :: !(IORef (Map.HashMap ThreadId Child)) - , _sp_mailbox :: TQueue DeadLetter + , _sp_mailbox :: TChan DeadLetter , _sp_eventStream :: TBQueue SupervisionEvent } -> Supervisor_ Initialised @@ -59,7 +74,8 @@ data DeadLetter = DeadLetter ThreadId SomeException -------------------------------------------------------------------------------- -data Child = Child !RestartStrategy RestartAction +data Child = Worker !RestartStrategy RestartAction + | Supvsr !RestartStrategy !(Supervisor_ Initialised) -------------------------------------------------------------------------------- type RestartAction = ThreadId -> IO ThreadId @@ -73,40 +89,79 @@ deriving Show -------------------------------------------------------------------------------- --- | Erlang inspired strategies. +-- | Erlang inspired strategies. At the moment only the 'OneForOne' is +-- implemented. data RestartStrategy = OneForOne deriving Show +-- $new +-- In order to create a new supervisor, you need a `SupervisorSpec`, +-- which can be acquired by a call to `newSupervisor`: + -------------------------------------------------------------------------------- --- | Creates a new supervisor -newSupervisor :: IO SupervisorSpec -newSupervisor = do - tkn <- newTQueueIO +-- | Creates a new 'SupervisorSpec'. The reason it doesn't return a +-- 'Supervisor' is to force you to call 'supervise' explicitly, in order to start the +-- supervisor thread. +newSupervisorSpec :: IO SupervisorSpec +newSupervisorSpec = do + tkn <- newTChanIO evt <- newTBQueueIO 1000 ref <- newIORef Map.empty return $ NewSupervisor Nothing ref tkn evt +-- $supervise + -------------------------------------------------------------------------------- +newSupervisor :: SupervisorSpec -> IO Supervisor +newSupervisor spec = forkIO (handleEvents spec) >>= \tid -> do + mbx <- atomically $ dupTChan (_ns_mailbox spec) + return $ Supervisor { + _sp_myTid = Just tid + , _sp_mailbox = mbx + , _sp_children = _ns_children spec + , _sp_eventStream = _ns_eventStream spec + } + +-- $log + +-------------------------------------------------------------------------------- +-- | Gives you access to the event this supervisor is generating, allowing you +-- to react. It's using a bounded queue to explicitly avoid memory leaks in case +-- you do not want to drain the queue to listen to incoming events. eventStream :: Supervisor -> TBQueue SupervisionEvent eventStream (Supervisor _ _ _ e) = e -------------------------------------------------------------------------------- +-- | Returns the number of active threads at a given moment in time. activeChildren :: Supervisor -> IO Int activeChildren (Supervisor _ chRef _ _) = do readIORef chRef >>= return . length . Map.keys +-- $shutdown + -------------------------------------------------------------------------------- -- | Shutdown the given supervisor. This will cause the supervised children to --- be killed as well. +-- be killed as well. To do so, we explore the children tree, killing workers as we go, +-- and recursively calling `shutdownSupervisor` in case we hit a monitored `Supervisor`. shutdownSupervisor :: Supervisor -> IO () shutdownSupervisor (Supervisor sId chRef _ _) = do - case sId of + case sId of Nothing -> return () Just tid -> do - readIORef chRef >>= \chMap -> forM_ (Map.keys chMap) killThread + chMap <- readIORef chRef + processChildren (Map.toList chMap) killThread tid + where + processChildren [] = return () + processChildren (x:xs) = do + case x of + (tid, Worker _ _) -> killThread tid + (_, Supvsr _ s) -> shutdownSupervisor s + processChildren xs +-- $fork + -------------------------------------------------------------------------------- -- | Fork a thread in a supervised mode. forkSupervised :: Supervisor @@ -118,7 +173,7 @@ -> IO ThreadId forkSupervised sup@Supervisor{..} str act = bracket (supervised sup act) return $ \newChild -> do - let ch = Child str (const (supervised sup act)) + let ch = Worker str (const (supervised sup act)) atomicModifyIORef' _sp_children $ \chMap -> (Map.insert newChild ch chMap, ()) now <- getCurrentTime writeIfNotFull _sp_eventStream (ChildBorn newChild now) @@ -134,34 +189,52 @@ supervised :: Supervisor -> IO () -> IO ThreadId supervised Supervisor{..} act = forkFinally act $ \res -> case res of Left ex -> bracket myThreadId return $ \myId -> atomically $ - writeTQueue _sp_mailbox (DeadLetter myId ex) + writeTChan _sp_mailbox (DeadLetter myId ex) Right _ -> bracket myThreadId return $ \myId -> do now <- getCurrentTime atomicModifyIORef' _sp_children $ \chMap -> (Map.delete myId chMap, ()) writeIfNotFull _sp_eventStream (ChildFinished myId now) -------------------------------------------------------------------------------- -supervise :: SupervisorSpec -> IO Supervisor -supervise NewSupervisor{..} = forkIO go >>= \tid -> return $ Supervisor { - _sp_myTid = Just tid - , _sp_mailbox = _ns_mailbox - , _sp_children = _ns_children - , _sp_eventStream = _ns_eventStream - } - where - go = do - (DeadLetter newDeath ex) <- atomically $ readTQueue _ns_mailbox - now <- getCurrentTime - writeIfNotFull _ns_eventStream (ChildDied newDeath ex now) - case asyncExceptionFromException ex of - Just ThreadKilled -> go - _ -> do - chMap <- readIORef _ns_children - case Map.lookup newDeath chMap of - Nothing -> go - Just ch@(Child str act) -> case str of - OneForOne -> do - newThreadId <- act newDeath - writeIORef _ns_children (Map.insert newThreadId ch $! Map.delete newDeath chMap) - writeIfNotFull _ns_eventStream (ChildRestarted newDeath newThreadId str now) - go +handleEvents :: SupervisorSpec -> IO () +handleEvents sp@(NewSupervisor myId myChildren myMailbox myStream) = do + (DeadLetter newDeath ex) <- atomically $ readTChan myMailbox + now <- getCurrentTime + writeIfNotFull myStream (ChildDied newDeath ex now) + case asyncExceptionFromException ex of + Just ThreadKilled -> handleEvents sp + _ -> do + chMap <- readIORef myChildren + case Map.lookup newDeath chMap of + Nothing -> handleEvents sp + Just ch@(Worker str act) -> case str of + OneForOne -> do + newThreadId <- act newDeath + writeIORef myChildren (Map.insert newThreadId ch $! Map.delete newDeath chMap) + writeIfNotFull myStream (ChildRestarted newDeath newThreadId str now) + handleEvents sp + Just ch@(Supvsr str (Supervisor _ mbx cld es)) -> case str of + OneForOne -> do + let node = Supervisor myId myChildren myMailbox myStream + newThreadId <- supervised node (handleEvents $ NewSupervisor Nothing mbx cld es) + writeIORef myChildren (Map.insert newThreadId ch $! Map.delete newDeath chMap) + writeIfNotFull myStream (ChildRestarted newDeath newThreadId str now) + handleEvents sp + +-- $monitor + +newtype MonitorRequest = MonitoredSupervision ThreadId deriving (Show, Typeable) + +instance Exception MonitorRequest + +-------------------------------------------------------------------------------- +-- | Monitor another supervisor. To achieve these, we simulate a new 'DeadLetter', +-- so that the first supervisor will effectively restart the monitored one. +-- Thanks to the fact that for the supervisor the restart means we just copy over +-- its internal state, it should be perfectly fine to do so. +monitor :: Supervisor -> Supervisor -> IO () +monitor (Supervisor _ _ mbox _) (Supervisor mbId _ _ _) = do + case mbId of + Nothing -> return () + Just tid -> atomically $ + writeTChan mbox (DeadLetter tid (toException $ MonitoredSupervision tid)) diff --git a/src/Control/Concurrent/Supervisor/Tutorial.hs b/src/Control/Concurrent/Supervisor/Tutorial.hs --- a/src/Control/Concurrent/Supervisor/Tutorial.hs +++ b/src/Control/Concurrent/Supervisor/Tutorial.hs @@ -1,5 +1,4 @@ -{-# LANGUAGE RecordWildCards #-} -{-# LANGUAGE ScopedTypeVariables #-} +{-# OPTIONS_GHC -fno-warn-unused-imports #-} {-| Use @threads-supervisor@ if you want the "poor-man's Erlang supervisors". @threads-supervisor@ is an IO-based library with minimal dependencies @@ -85,31 +84,77 @@ -- -- These jobs represent a significant pool of our everyday computations in the IO monad - -- $createSpec --- Here I discuss the creation of a 'SupervisorSpec' +-- A 'SupervisorSpec' simply holds the state of our supervision, and can be safely shared +-- between supervisors. Under the hood, both the `SupervisorSpec` and the `Supervisor` +-- share the same structure; in fact, they are just type synonyms: +-- +-- > type SupervisorSpec = Supervisor_ Uninitialised +-- > type Supervisor = Supervisor_ Initialised +-- The important difference though, is that the `SupervisorSpec` does not imply the creation +-- of an asynchronous thread, which the latter does. To keep separated the initialisation +-- of the data structure from the logic of supervising, we use GADTs and type synonyms to +-- force you create a spec first. +-- Creating a spec it just a matter of calling `newSupervisorSpec`. -- $createSupervisor --- Here I discuss the creation of a 'Supervisor' from a 'SupervisionSpec' +-- Creating a 'Supervisor' from a 'SupervisionSpec', is as simple as calling `newSupervisor`. +-- immediately after doing so, a new thread will be started, monitoring any subsequent IO actions +-- submitted to it. -- $supervising --- Here I discuss how you can supervise other threads. - --- $conclusions --- I hope that you are now convinced that this library can be of some use to you! +-- Let's wrap everything together into a full blown example: -- -- > main :: IO () -- > main = bracketOnError (do --- > supSpec <- newSupervisor --- > sup <- supervise supSpec --- > j1 <- forkSupervised sup OneForOne job1 --- > _ <- forkSupervised sup OneForOne (job2 j1) --- > _ <- forkSupervised sup OneForOne job3 --- > _ <- forkSupervised sup OneForOne job4 --- > _ <- forkIO (go (eventStream sup)) --- > return sup) shutdownSupervisor (\_ -> threadDelay 10000000000) +-- > supSpec <- newSupervisorSpec +-- > +-- > sup1 <- newSupervisor supSpec +-- > sup2 <- newSupervisor supSpec +-- > +-- > sup1 `monitor` sup2 +-- > +-- > _ <- forkSupervised sup2 OneForOne job3 +-- > +-- > j1 <- forkSupervised sup1 OneForOne job1 +-- > _ <- forkSupervised sup1 OneForOne (job2 j1) +-- > _ <- forkSupervised sup1 OneForOne job4 +-- > _ <- forkIO (go (eventStream sup1)) +-- > return sup1) shutdownSupervisor (\_ -> threadDelay 10000000000) -- > where -- > go eS = do -- > newE <- atomically $ readTBQueue eS -- > print newE -- > go eS +-- +-- What we have done here, was to spawn our supervisor out from a spec, +-- any using our swiss knife `forkSupervised` to spawn for supervised +-- IO computations. As you can see, if we partially apply `forkSupervised`, +-- its type resemble `forkIO` one; this is by design, as we want to keep +-- this API as IO-friendly as possible +-- in the very same example, we also create another supervisor +-- (from the same spec, but you can create a separate one as well) +-- and we ask the first supervisor to monitor the second one. +-- +-- If you run this program, hopefully you should see on stdout +-- something like this: +-- +-- > ChildBorn ThreadId 62 2015-02-13 11:51:15.293882 UTC +-- > ChildBorn ThreadId 63 2015-02-13 11:51:15.293897 UTC +-- > ChildBorn ThreadId 64 2015-02-13 11:51:15.293904 UTC +-- > ChildDied ThreadId 61 (MonitoredSupervision ThreadId 61) 2015-02-13 11:51:15.293941 UTC +-- > ChildBorn ThreadId 65 2015-02-13 11:51:15.294014 UTC +-- > ChildFinished ThreadId 64 2015-02-13 11:51:18.294797 UTC +-- > ChildDied ThreadId 63 thread killed 2015-02-13 11:51:18.294909 UTC +-- > ChildDied ThreadId 62 Oh boy, I'm good as dead 2015-02-13 11:51:20.294861 UTC +-- > ChildRestarted ThreadId 62 ThreadId 68 OneForOne 2015-02-13 11:51:20.294861 UTC +-- > ChildFinished ThreadId 65 2015-02-13 11:51:22.296089 UTC +-- > ChildDied ThreadId 68 Oh boy, I'm good as dead 2015-02-13 11:51:25.296189 UTC +-- > ChildRestarted ThreadId 68 ThreadId 69 OneForOne 2015-02-13 11:51:25.296189 UTC +-- > ChildDied ThreadId 69 Oh boy, I'm good as dead 2015-02-13 11:51:30.297464 UTC +-- > ChildRestarted ThreadId 69 ThreadId 70 OneForOne 2015-02-13 11:51:30.297464 UTC +-- > ChildDied ThreadId 70 Oh boy, I'm good as dead 2015-02-13 11:51:35.298123 UTC +-- > ChildRestarted ThreadId 70 ThreadId 71 OneForOne 2015-02-13 11:51:35.298123 UTC + +-- $conclusions +-- I hope that you are now convinced that this library can be of some use to you! diff --git a/test/Tests.hs b/test/Tests.hs --- a/test/Tests.hs +++ b/test/Tests.hs @@ -15,7 +15,6 @@ import Control.Concurrent.STM import Control.Exception import Control.Concurrent.Supervisor -import Data.Time -------------------------------------------------------------------------------- type IOProperty = PropertyM IO @@ -121,8 +120,8 @@ -- Control.Concurrent.Supervisor tests test1SupThreadNoEx :: IOProperty () test1SupThreadNoEx = forAllM randomLiveTime $ \ttl -> do - supSpec <- lift newSupervisor - sup <- lift $ supervise supSpec + supSpec <- lift newSupervisorSpec + sup <- lift $ newSupervisor supSpec _ <- lift (forkSupervised sup OneForOne (forever $ threadDelay ttl)) assertActiveThreads sup (== 1) lift $ shutdownSupervisor sup @@ -130,8 +129,8 @@ -------------------------------------------------------------------------------- test1SupThreadPrematureDemise :: IOProperty () test1SupThreadPrematureDemise = forAllM randomLiveTime $ \ttl -> do - supSpec <- lift newSupervisor - sup <- lift $ supervise supSpec + supSpec <- lift newSupervisorSpec + sup <- lift $ newSupervisor supSpec tid <- lift (forkSupervised sup OneForOne (forever $ threadDelay ttl)) lift $ do throwTo tid (AssertionFailed "You must die") @@ -165,8 +164,8 @@ -- the side effects strikes. testKillingSpree :: IOProperty () testKillingSpree = forAllM arbitrary $ \ep@(ExecutionPlan _ acts) -> do - supSpec <- lift newSupervisor - sup <- lift $ supervise supSpec + supSpec <- lift newSupervisorSpec + sup <- lift $ newSupervisor supSpec _ <- forM acts $ lift . fromAction sup lift (threadDelay $ maxWait acts * 2) q <- lift $ qToList (eventStream sup) @@ -180,8 +179,8 @@ testSupCleanup :: IOProperty () testSupCleanup = forAllM (vectorOf 100 arbitrary) $ \ttls -> do let acts = map DieAfter ttls - supSpec <- lift newSupervisor - sup <- lift $ supervise supSpec + supSpec <- lift newSupervisorSpec + sup <- lift $ newSupervisor supSpec _ <- forM acts $ lift . fromAction sup lift (threadDelay $ maxWait acts * 2) q <- lift $ qToList (eventStream sup) diff --git a/threads-supervisor.cabal b/threads-supervisor.cabal --- a/threads-supervisor.cabal +++ b/threads-supervisor.cabal @@ -1,5 +1,5 @@ name: threads-supervisor -version: 1.0.0.0 +version: 1.0.1.0 synopsis: Simple, IO-based library for Erlang-style thread supervision description: Simple, IO-based library for Erlang-style thread supervision license: MIT @@ -27,6 +27,9 @@ hs-source-dirs: src default-language: Haskell2010 + ghc-options: + -Wall + -funbox-strict-fields test-suite threads-supervisor-tests type: @@ -39,6 +42,9 @@ test default-language: Haskell2010 + ghc-options: + -threaded + "-with-rtsopts=-N" build-depends: threads-supervisor -any , base