diff --git a/LICENSE b/LICENSE new file mode 100644 --- /dev/null +++ b/LICENSE @@ -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. 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/src/Control/Concurrent/Supervisor.hs b/src/Control/Concurrent/Supervisor.hs new file mode 100644 --- /dev/null +++ b/src/Control/Concurrent/Supervisor.hs @@ -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 diff --git a/src/Control/Concurrent/Supervisor/Tutorial.hs b/src/Control/Concurrent/Supervisor/Tutorial.hs new file mode 100644 --- /dev/null +++ b/src/Control/Concurrent/Supervisor/Tutorial.hs @@ -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 +-- and . +-- +-- 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 diff --git a/test/Main.hs b/test/Main.hs new file mode 100644 --- /dev/null +++ b/test/Main.hs @@ -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) + ] + ] diff --git a/test/Tests.hs b/test/Tests.hs new file mode 100644 --- /dev/null +++ b/test/Tests.hs @@ -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 diff --git a/threads-supervisor.cabal b/threads-supervisor.cabal new file mode 100644 --- /dev/null +++ b/threads-supervisor.cabal @@ -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