threads-supervisor 1.0.0.0 → 1.0.1.0

raw patch · 4 files changed

+190/−67 lines, 4 filesPVP: major bump suggested

API removals or changes: PVP suggests a major version bump

API changes (from Hackage documentation)

Files

• src/Control/Concurrent/Supervisor.hs +113/−40
• src/Control/Concurrent/Supervisor/Tutorial.hs +62/−17
• test/Tests.hs +8/−9
• threads-supervisor.cabal +7/−1

@@ -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))

@@ -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!

@@ -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)

@@ -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