threads-supervisor (empty) → 1.0.0.0

raw patch · 7 files changed

+576/−0 lines, 7 filesdep +QuickCheckdep +basedep +bytestringsetup-changed

Dependencies added: QuickCheck, base, bytestring, stm, tasty, tasty-hunit, tasty-quickcheck, threads-supervisor, time, transformers, unordered-containers

Files

• LICENSE +20/−0
• Setup.hs +2/−0
• src/Control/Concurrent/Supervisor.hs +167/−0
• src/Control/Concurrent/Supervisor/Tutorial.hs +115/−0
• test/Main.hs +30/−0
• test/Tests.hs +190/−0
• threads-supervisor.cabal +52/−0

@@ -0,0 +1,20 @@+Copyright (c) 2014 Alfredo Di Napoli++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.

@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain

@@ -0,0 +1,167 @@++{-+  Humble module inspired to Erlang supervisors,+  with minimal dependencies.+-}++{-# LANGUAGE GADTs #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE BangPatterns #-}+{-# LANGUAGE RecordWildCards #-}+{-# LANGUAGE ScopedTypeVariables #-}++module Control.Concurrent.Supervisor +  ( Supervisor+  , DeadLetter+  , Child(..)+  , RestartAction+  , SupervisionEvent(..)+  , RestartStrategy(..)+  , newSupervisor+  , shutdownSupervisor+  , eventStream+  , activeChildren+  , supervise+  , forkSupervised+  ) where++import qualified Data.HashMap.Strict as Map+import           Control.Concurrent+import           Control.Concurrent.STM+import           Data.IORef+import           Control.Exception+import           Control.Monad+import           Data.Time++--------------------------------------------------------------------------------+data Uninitialised+data Initialised++--------------------------------------------------------------------------------+data Supervisor_ a where+     NewSupervisor :: {+        _ns_myTid    :: !(Maybe ThreadId)+      , _ns_children :: !(IORef (Map.HashMap ThreadId Child))+      , _ns_mailbox :: TQueue DeadLetter+      , _ns_eventStream :: TBQueue SupervisionEvent+      } -> Supervisor_ Uninitialised+     Supervisor :: {+        _sp_myTid    :: !(Maybe ThreadId)+      , _sp_children :: !(IORef (Map.HashMap ThreadId Child))+      , _sp_mailbox :: TQueue DeadLetter+      , _sp_eventStream :: TBQueue SupervisionEvent+      } -> Supervisor_ Initialised++type SupervisorSpec = Supervisor_ Uninitialised+type Supervisor = Supervisor_ Initialised++--------------------------------------------------------------------------------+data DeadLetter = DeadLetter ThreadId SomeException++--------------------------------------------------------------------------------+data Child = Child !RestartStrategy RestartAction++--------------------------------------------------------------------------------+type RestartAction = ThreadId -> IO ThreadId++--------------------------------------------------------------------------------+data SupervisionEvent =+     ChildBorn !ThreadId !UTCTime+   | ChildDied !ThreadId !SomeException !UTCTime+   | ChildRestarted !ThreadId !ThreadId !RestartStrategy !UTCTime+   | ChildFinished !ThreadId !UTCTime+   deriving Show++--------------------------------------------------------------------------------+-- | Erlang inspired strategies.+data RestartStrategy =+     OneForOne+     deriving Show++--------------------------------------------------------------------------------+-- | Creates a new supervisor+newSupervisor :: IO SupervisorSpec+newSupervisor = do+  tkn <- newTQueueIO+  evt <- newTBQueueIO 1000+  ref <- newIORef Map.empty+  return $ NewSupervisor Nothing ref tkn evt++--------------------------------------------------------------------------------+eventStream :: Supervisor -> TBQueue SupervisionEvent+eventStream (Supervisor _ _ _ e) = e++--------------------------------------------------------------------------------+activeChildren :: Supervisor -> IO Int+activeChildren (Supervisor _ chRef _ _) = do+  readIORef chRef >>= return . length . Map.keys++--------------------------------------------------------------------------------+-- | Shutdown the given supervisor. This will cause the supervised children to+-- be killed as well.+shutdownSupervisor :: Supervisor -> IO ()+shutdownSupervisor (Supervisor sId chRef _ _) = do+  case sId of +    Nothing -> return ()+    Just tid -> do+      readIORef chRef >>= \chMap -> forM_ (Map.keys chMap) killThread+      killThread tid++--------------------------------------------------------------------------------+-- | Fork a thread in a supervised mode.+forkSupervised :: Supervisor+               -- ^ The 'Supervisor'+               -> RestartStrategy+               -- ^ The 'RestartStrategy' to use+               -> IO ()+               -- ^ The computation to run+               -> IO ThreadId+forkSupervised sup@Supervisor{..} str act =+  bracket (supervised sup act) return $ \newChild -> do+    let ch = Child str (const (supervised sup act))+    atomicModifyIORef' _sp_children $ \chMap -> (Map.insert newChild ch chMap, ())+    now <- getCurrentTime+    writeIfNotFull _sp_eventStream (ChildBorn newChild now)+    return newChild++--------------------------------------------------------------------------------+writeIfNotFull :: TBQueue SupervisionEvent -> SupervisionEvent -> IO ()+writeIfNotFull q evt = atomically $ do+  isFull <- isFullTBQueue q+  unless isFull $ writeTBQueue q evt++--------------------------------------------------------------------------------+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)+  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

@@ -0,0 +1,115 @@+{-# LANGUAGE RecordWildCards #-}+{-# LANGUAGE ScopedTypeVariables #-}++{-| Use @threads-supervisor@ if you want the "poor-man's Erlang supervisors".+    @threads-supervisor@ is an IO-based library  with minimal dependencies+    which does only one thing: It provides you a 'Supervisor' entity  you can use+    to monitor your forked computations. If one of the managed threads dies,+    you can decide if and how to restart it. This gives you:++    * Protection against silent exceptions which might terminate your workers.+    * A simple but powerful way of structure your program into a supervision tree,+    where the leaves are the worker threads, and the nodes can be other+    supervisors being monitored.+    * A disaster recovery mechanism.++  You can install the @threads-supervisor@ library by running:+  > $ cabal install threads-supervisor+-}++module Control.Concurrent.Supervisor.Tutorial+ (  -- * Introduction+    -- $introduction++    -- * Different type of jobs+    -- $jobs++    -- * Creating a SupervisorSpec+    -- $createSpec++    -- * Creating a Supervisor+    -- $createSupervisor++    -- * Supervising and choosing a 'RestartStrategy'+    -- $supervising++    -- * Wrapping up+    -- $conclusions++ ) where++-- $introduction+-- Who worked with Haskell's concurrency primitives would be surely familiar+-- with the `forkIO` function, which allow us to fork an IO computation in a separate+-- green thread. `forkIO` is great, but is also very low level, and has a+-- couple of subtleties, as you can read from this passage of the documentation:+--+--    The newly created thread has an exception handler that discards the exceptions +--    `BlockedIndefinitelyOnMVar`,`BlockedIndefinitelyOnSTM`, and `ThreadKilled`, +--    and passes all other exceptions to the uncaught exception handler.+--+-- To mitigate this, we have a couple of libraries available,  for example+-- <http://hackage.haskell.org/package/async> and <http://hackage.haskell.org/package/slave-thread>.+-- +-- But what about if I do not want to take explicit action, but instead specifying upfront+-- how to react to disaster, and leave the library work out the details?+-- This is what this library aims to do.++-- $jobs+-- In this example, let's create four different threads:+--+-- > job1 :: IO ()+-- > job1 = do+-- >   threadDelay 5000000+-- >   fail "Dead"+-- This job will die after five seconds.+--+-- > job2 :: ThreadId -> IO ()+-- > job2 tid = do+-- >   threadDelay 3000000+-- >   killThread tid+--+-- This other job instead, we have waited three seconds, and then kill a target thread,+-- generating an asynchronous exception.+--+-- > job3 :: IO ()+-- > job3 = do+-- >   threadDelay 5000000+-- >   error "Oh boy, I'm good as dead"+--+-- This guy is very similar to the first one, except for the fact `error` is used instead of `fail`.+--+-- > job4 :: IO ()+-- > job4 = threadDelay 7000000+-- @job4@ is what we wish for all our passing cross computation: smooth sailing.+--+-- These jobs represent a significant pool of our everyday computations in the IO monad+++-- $createSpec+-- Here I discuss the creation of a 'SupervisorSpec'++-- $createSupervisor+-- Here I discuss the creation of a 'Supervisor' from a 'SupervisionSpec'++-- $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!+--+-- > 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)+-- >   where+-- >    go eS = do+-- >      newE <- atomically $ readTBQueue eS+-- >      print newE+-- >      go eS

@@ -0,0 +1,30 @@+{-# LANGUAGE OverloadedStrings #-}+module Main where++import           Test.Tasty+import           Test.Tasty.QuickCheck+import           Test.QuickCheck.Monadic+import           Tests+++--------------------------------------------------------------------------------+main :: IO ()+main = defaultMain allTests+++--------------------------------------------------------------------------------+withQuickCheckDepth :: TestName -> Int -> [TestTree] -> TestTree+withQuickCheckDepth tn depth tests =+  localOption (QuickCheckTests depth) (testGroup tn tests)+++--------------------------------------------------------------------------------+allTests :: TestTree+allTests = testGroup "All Tests" [+    withQuickCheckDepth "Control.Concurrent.Supervisor" 10 [+        testProperty "1 supervised thread, no exceptions" (monadicIO test1SupThreadNoEx)+      , testProperty "1 supervised thread, premature exception" (monadicIO test1SupThreadPrematureDemise)+      , testProperty "killing spree" (monadicIO testKillingSpree)+      , testProperty "cleanup" (monadicIO testSupCleanup)+    ]+  ]

@@ -0,0 +1,190 @@+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE RecordWildCards #-}+{-# LANGUAGE BangPatterns #-}+{-# LANGUAGE ScopedTypeVariables #-}+module Tests where++import           Test.Tasty.HUnit as HUnit+import           Test.Tasty.QuickCheck+import           Test.QuickCheck.Monadic as QM+import qualified Data.List as List+import           Control.Monad+import           Control.Monad.Trans.Class+import           Control.Applicative+import           Control.Concurrent+import           Control.Concurrent.STM+import           Control.Exception+import           Control.Concurrent.Supervisor+import           Data.Time++--------------------------------------------------------------------------------+type IOProperty = PropertyM IO++-- How much a thread will live.+newtype TTL = TTL Int deriving Show++-- | Generate a random thread live time between 0.5 sec and 2 secs.+randomLiveTime :: Gen Int+randomLiveTime = choose (500000, 2000000)++instance Arbitrary TTL where+  arbitrary = TTL <$> randomLiveTime++data ThreadAction =+    Live+  | DieAfter TTL --natural death+  | ThrowAfter TTL+  deriving Show++instance Arbitrary ThreadAction where+  arbitrary = do+    act <- elements [const Live, DieAfter, ThrowAfter]+    ttl <- arbitrary+    return $ act ttl++-- We cannot easily deal with async exceptions+-- being thrown at us.+data ExecutionPlan = ExecutionPlan {+    toSpawn :: Int+  , actions :: [ThreadAction]+  } deriving Show++instance Arbitrary ExecutionPlan where+  arbitrary = do+    ts <- choose (1,20)+    acts <- vectorOf ts arbitrary+    return $ ExecutionPlan ts acts++--------------------------------------------------------------------------------+howManyRestarted :: ExecutionPlan -> Int+howManyRestarted (ExecutionPlan _ acts) = length . filter pred_ $ acts+  where+    pred_ (ThrowAfter _) = True+    pred_ _ = False++--------------------------------------------------------------------------------+howManyLiving :: ExecutionPlan -> Int+howManyLiving (ExecutionPlan _ acts) = length . filter pred_ $ acts+  where+    pred_ Live = True+    pred_ _ = False++--------------------------------------------------------------------------------+assertActiveThreads :: Supervisor -> (Int -> Bool) -> IOProperty ()+assertActiveThreads sup p = do+  ac <- lift (activeChildren sup)+  QM.assert (p ac)++--------------------------------------------------------------------------------+qToList :: TBQueue SupervisionEvent -> IO [SupervisionEvent]+qToList q = do+  nextEl <- atomically (tryReadTBQueue q)+  case nextEl of+    (Just el) -> (el :) <$> qToList q+    Nothing -> return []++--------------------------------------------------------------------------------+assertContainsNMsg :: (SupervisionEvent -> Bool) +                   -> Int+                   -> [SupervisionEvent] +                   -> IOProperty ()+assertContainsNMsg _ 0 _ = QM.assert True+assertContainsNMsg _ x [] = lift $+  HUnit.assertBool ("assertContainsNMsg: list exhausted and " ++ show x ++ " left.") False+assertContainsNMsg matcher !n (x:xs) = case matcher x of+  True  -> assertContainsNMsg matcher (n - 1) xs+  False -> assertContainsNMsg matcher n xs++--------------------------------------------------------------------------------+assertContainsNRestartMsg :: Int -> [SupervisionEvent] -> IOProperty ()+assertContainsNRestartMsg = assertContainsNMsg matches+  where+    matches (ChildRestarted{}) = True+    matches _ = False++--------------------------------------------------------------------------------+assertContainsNFinishedMsg :: Int -> [SupervisionEvent] -> IOProperty ()+assertContainsNFinishedMsg = assertContainsNMsg matches+  where+    matches (ChildFinished{}) = True+    matches _ = False++--------------------------------------------------------------------------------+assertContainsRestartMsg :: [SupervisionEvent] -> ThreadId -> IOProperty ()+assertContainsRestartMsg [] _ = QM.assert False+assertContainsRestartMsg (x:xs) tid = case x of+  ((ChildRestarted old _ _ _)) -> +    if old == tid then QM.assert True else assertContainsRestartMsg xs tid+  _ -> assertContainsRestartMsg xs tid++--------------------------------------------------------------------------------+-- Control.Concurrent.Supervisor tests+test1SupThreadNoEx :: IOProperty ()+test1SupThreadNoEx = forAllM randomLiveTime $ \ttl -> do+  supSpec <- lift newSupervisor+  sup <- lift $ supervise supSpec+  _ <- lift (forkSupervised sup OneForOne (forever $ threadDelay ttl))+  assertActiveThreads sup (== 1)+  lift $ shutdownSupervisor sup++--------------------------------------------------------------------------------+test1SupThreadPrematureDemise :: IOProperty ()+test1SupThreadPrematureDemise = forAllM randomLiveTime $ \ttl -> do+  supSpec <- lift newSupervisor+  sup <- lift $ supervise supSpec+  tid <- lift (forkSupervised sup OneForOne (forever $ threadDelay ttl))+  lift $ do+    throwTo tid (AssertionFailed "You must die")+    threadDelay ttl --give time to restart the thread+  assertActiveThreads sup (== 1)+  q <- lift $ qToList (eventStream sup)+  assertContainsNRestartMsg 1 q+  lift $ shutdownSupervisor sup++--------------------------------------------------------------------------------+fromAction :: Supervisor -> ThreadAction -> IO ThreadId+fromAction s Live = forkSupervised s OneForOne (forever $ threadDelay 100000000)+fromAction s (DieAfter (TTL ttl)) = forkSupervised s OneForOne (threadDelay ttl)+fromAction s (ThrowAfter (TTL ttl)) = forkSupervised s OneForOne (do+  threadDelay ttl +  throwIO $ AssertionFailed "die")++--------------------------------------------------------------------------------+maxWait :: [ThreadAction] -> Int+maxWait ta = go ta []+  where+    go [] [] = 0+    go [] acc = List.maximum acc+    go (Live:xs) acc = go xs acc+    go ((DieAfter (TTL t)):xs) acc = go xs (t : acc)+    go ((ThrowAfter (TTL t)):xs) acc = go xs (t : acc)++--------------------------------------------------------------------------------+-- In this test, we generate random IO actions for the threads to be+-- executed, then we calculate how many of them needs to be alive after all+-- the side effects strikes.+testKillingSpree :: IOProperty ()+testKillingSpree = forAllM arbitrary $ \ep@(ExecutionPlan _ acts) -> do+  supSpec <- lift newSupervisor+  sup <- lift $ supervise supSpec+  _ <- forM acts $ lift . fromAction sup+  lift (threadDelay $ maxWait acts * 2)+  q <- lift $ qToList (eventStream sup)+  assertActiveThreads sup (>= howManyLiving ep)+  assertContainsNRestartMsg (howManyRestarted ep) q+  lift $ shutdownSupervisor sup++--------------------------------------------------------------------------------+-- In this test, we test that the supervisor does not leak memory by removing+-- children who finished+testSupCleanup :: IOProperty ()+testSupCleanup = forAllM (vectorOf 100 arbitrary) $ \ttls -> do+  let acts = map DieAfter ttls+  supSpec <- lift newSupervisor+  sup <- lift $ supervise supSpec+  _ <- forM acts $ lift . fromAction sup+  lift (threadDelay $ maxWait acts * 2)+  q <- lift $ qToList (eventStream sup)+  assertActiveThreads sup (== 0)+  assertContainsNFinishedMsg (length acts) q+  lift $ shutdownSupervisor sup

@@ -0,0 +1,52 @@+name:                threads-supervisor+version:             1.0.0.0+synopsis:            Simple, IO-based library for Erlang-style thread supervision+description:         Simple, IO-based library for Erlang-style thread supervision+license:             MIT+license-file:        LICENSE+author:              Alfredo Di Napoli+maintainer:          alfredo.dinapoli@gmail.com+copyright:           Alfredo Di Napoli+category:            Concurrency+build-type:          Simple+cabal-version:       >=1.10++source-repository head+  type: git+  location: https://github.com/adinapoli/threads-supervisor++library+  exposed-modules:+    Control.Concurrent.Supervisor+    Control.Concurrent.Supervisor.Tutorial+  build-depends:+    base                 >= 4.6 && < 5,+    unordered-containers >= 0.2.0.0 && < 0.3.0.0,+    stm                  >= 2.4,+    time                 >= 1.2+  hs-source-dirs:+    src+  default-language:    Haskell2010++test-suite threads-supervisor-tests+  type:+    exitcode-stdio-1.0+  main-is:+    Main.hs+  other-modules:+    Tests+  hs-source-dirs:+    test+  default-language:+    Haskell2010+  build-depends:+      threads-supervisor -any+    , base+    , bytestring+    , QuickCheck+    , tasty >= 0.9.0.1+    , tasty-quickcheck+    , tasty-hunit+    , time+    , stm+    , transformers