diff --git a/.gitignore b/.gitignore new file mode 100644 --- /dev/null +++ b/.gitignore @@ -0,0 +1,4 @@ +dist +dist-newstyle +*.ghc.* +cabal.project.local diff --git a/CHANGELOG.md b/CHANGELOG.md new file mode 100644 --- /dev/null +++ b/CHANGELOG.md @@ -0,0 +1,5 @@ +# Revision history for supervisors + +## 0.1.0.0 -- YYYY-mm-dd + +* First version. Released on an unsuspecting world. diff --git a/LICENSE b/LICENSE new file mode 100644 --- /dev/null +++ b/LICENSE @@ -0,0 +1,20 @@ +Copyright (c) 2018 Ian Denhardt + +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,40 @@ +[![hackage][hackage-img]][hackage] + +# Haskell Supervisors + +The `supervisors` package provides a useful abstraction for managing the +groups of Haskell threads, which may not have a strictly hierarchical +structure to their lifetimes. + +One way to think of it is that `supervisors` is to [async][async] as +[resourcet][resourcet] is to [bracket][bracket]. + +Most of the time you can manage these things in a hierarchical manner: +for bracket, acquire a resource, do stuff with it, and release it. For +async, spawn some tasks, wait for some or all of them, maybe kill the +remaining ones, and return. The memory used by all of these threads is +not reclaimed until the entire subtree finishes. + +But sometimes, your concurrency patterns don't fit neatly into a tree; +that is what this package is for. + +This package was originally written for use in the rpc layer of the +[capnp][capnp] package, where the various threads handling rpc calls +can have essentially arbitrary lifetimes, but we often want to make +sure they are all shut down when a connection is closed. + +Concretely, the library provides a `Supervisor` construct, which can be +used to safely spawn threads while guaranteeing that: + +* When the supervisor is killed, all of the threads it supervises will be + killed. +* Child threads can terminate in any order, and memory usage will always + be proportional to the number of *live* supervised threads. + +[async]: https://hackage.haskell.org/package/async +[bracket]: http://hackage.haskell.org/package/base-4.12.0.0/docs/Control-Exception-Base.html#v:bracket +[resourcet]: https://hackage.haskell.org/package/resourcet +[capnp]: https://hackage.haskell.org/package/capnp + +[hackage-img]: https://img.shields.io/hackage/v/supervisors.svg +[hackage]: https://hackage.haskell.org/package/supervisors 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/Supervisors.hs b/src/Supervisors.hs new file mode 100644 --- /dev/null +++ b/src/Supervisors.hs @@ -0,0 +1,121 @@ +{-# LANGUAGE BangPatterns #-} +{-# LANGUAGE LambdaCase #-} +{-| +Module: Supervisors +Description: Montior a pool of threads. + +This module exposes a 'Supervisor' construct, which can be used to safely +spawn threads while guaranteeing that: + +* When the supervisor is killed, all of the threads it supervises will be + killed. +* Child threads can terminate in any order, and memory usage will always + be proportional to the number of *live* supervised threads. +-} +module Supervisors + ( Supervisor + , withSupervisor + , supervise + , superviseSTM + ) where + +import UnliftIO.STM + +import Control.Concurrent + (ThreadId, forkIO, myThreadId, threadDelay, throwTo) +import Control.Concurrent.Async (withAsync) +import Control.Concurrent.STM (throwSTM) +import Control.Monad (forever, void) +import Data.Foldable (traverse_) +import UnliftIO.Exception + (Exception, SomeException, bracket, bracket_, toException, withException) + +import qualified Data.Set as S + +-- | A handle for a supervisor, which montiors a pool of threads. +data Supervisor = Supervisor + { stateVar :: TVar (Either SomeException (S.Set ThreadId)) + , runQ :: TQueue (IO ()) + } + +-- | Start a new supervisor, and return it. +newSupervisor :: IO Supervisor +newSupervisor = do + stateVar <- newTVarIO $ Right S.empty + runQ <- newTQueueIO + let sup = Supervisor + { stateVar = stateVar + , runQ = runQ + } + pure sup + +-- | Run the logic associated with the supervisor. This never returns until +-- the supervisor receives an (asynchronous) exception. When it does return, +-- all of the supervised threads will be killed. +runSupervisor :: Supervisor -> IO () +runSupervisor sup@Supervisor{runQ=q} = + forever (atomically (readTQueue q) >>= supervise sup) + `withException` + \e -> throwKids sup (e :: SomeException) + +-- | Run an IO action with access to a supervisor. Threads spawned using the +-- supervisor will be killed when the action returns. +withSupervisor :: (Supervisor -> IO ()) -> IO () +withSupervisor f = do + sup <- newSupervisor + withAsync (runSupervisor sup) $ const (f sup) + +-- | Throw an exception to all of a supervisor's children, using 'throwTo'. +throwKids :: Exception e => Supervisor -> e -> IO () +throwKids Supervisor{stateVar=stateVar} exn = + bracket + (atomically $ readTVar stateVar >>= \case + Left _ -> + pure S.empty + Right kids -> do + writeTVar stateVar $ Left (toException exn) + pure kids) + (traverse_ (`throwTo` exn)) + (\_ -> pure ()) + +-- | Launch the IO action in a thread, monitored by the 'Supervisor'. If the +-- supervisor receives an exception, the exception will also be raised in the +-- child thread. +supervise :: Supervisor -> IO () -> IO () +supervise Supervisor{stateVar=stateVar} task = + void $ forkIO $ bracket_ addMe removeMe task + where + -- | Add our ThreadId to the supervisor. + addMe = do + me <- myThreadId + atomically $ do + supState <- readTVar stateVar + case supState of + Left e -> + throwSTM e + Right kids -> do + let !newKids = S.insert me kids + writeTVar stateVar $ Right newKids + -- | Remove our ThreadId from the supervisor, so we don't leak it. + removeMe = do + me <- myThreadId + atomically $ modifyTVar' stateVar $ \case + state@(Left _) -> + -- The supervisor is already stopped; we don't need to + -- do anything. + state + Right kids -> + -- We need to remove ourselves from the list of children; + -- if we don't, we'll leak our ThreadId until the supervisor + -- exits. + -- + -- The use of $! here is very important, because even though + -- modifyTVar' is strict, it only does whnf, so it would leave + -- the state only evaluated as far as @Right (S.delete me kids)@; + -- in that case we would still leak @me@. + Right $! S.delete me kids + +-- | Like 'supervise', but can be used from inside 'STM'. The thread will be +-- spawned if and only if the transaction commits. +superviseSTM :: Supervisor -> IO () -> STM () +superviseSTM Supervisor{runQ=q} = writeTQueue q diff --git a/supervisors.cabal b/supervisors.cabal new file mode 100644 --- /dev/null +++ b/supervisors.cabal @@ -0,0 +1,68 @@ +cabal-version: 2.2 +name: supervisors +version: 0.1.0.0 +stability: Experimental +synopsis: Monitor groups of threads with non-hierarchical lifetimes. +description: + The @supervisors@ package provides a useful abstraction for managing the + groups of Haskell threads, which may not have a strictly hierarchical + structure to their lifetimes. + . + Concretely, the library provides a `Supervisor` construct, which can be + used to safely spawn threads while guaranteeing that: + . + * When the supervisor is killed, all of the threads it supervises will be + killed. + * Child threads can terminate in any order, and memory usage will always + be proportional to the number of *live* supervised threads. + . + One way to think of it is that @supervisors@ is to @async@ as + @resourcet@ is to @bracket@. + . + Note that this package is EXPERIMENTAL; it needs more careful testing before + I can earnestly recommend relying on it. + . + See the README and module documentation for more information. +homepage: https://github.com/zenhack/haskell-supervisors +bug-reports: https://github.com/zenhack/haskell-supervisors/issues +license: MIT +license-file: LICENSE +author: Ian Denhardt +maintainer: ian@zenhack.net +copyright: 2018 Ian Denhardt +category: Concurrency +build-type: Simple +extra-source-files: + CHANGELOG.md + , README.md + , .gitignore + +common shared-opts + build-depends: + base ^>=4.12 + +library + import: shared-opts + exposed-modules: Supervisors + hs-source-dirs: src/ + build-depends: + stm ^>=2.5 + , containers ^>=0.6 + , unliftio ^>=0.2.8 + , async ^>=2.2.1 + default-language: Haskell2010 + +test-suite tests + import: shared-opts + type: exitcode-stdio-1.0 + main-is: Main.hs + hs-source-dirs: tests/ + build-depends: + supervisors + , hspec ^>=2.6.0 + default-language: Haskell2010 + +source-repository head + type: git + branch: master + location: https://github.com/zenhack/haskell-supervisors.git diff --git a/tests/Main.hs b/tests/Main.hs new file mode 100644 --- /dev/null +++ b/tests/Main.hs @@ -0,0 +1,6 @@ +module Main (main) where + +import Test.Hspec + +main :: IO () +main = hspec $ pure ()