supervisors (empty) → 0.1.0.0

raw patch · 8 files changed

+266/−0 lines, 8 filesdep +asyncdep +basedep +containerssetup-changed

Dependencies added: async, base, containers, hspec, stm, supervisors, unliftio

Files

• .gitignore +4/−0
• CHANGELOG.md +5/−0
• LICENSE +20/−0
• README.md +40/−0
• Setup.hs +2/−0
• src/Supervisors.hs +121/−0
• supervisors.cabal +68/−0
• tests/Main.hs +6/−0

@@ -0,0 +1,4 @@+dist+dist-newstyle+*.ghc.*+cabal.project.local

@@ -0,0 +1,5 @@+# Revision history for supervisors++## 0.1.0.0 -- YYYY-mm-dd++* First version. Released on an unsuspecting world.

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

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

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

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

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

@@ -0,0 +1,6 @@+module Main (main) where++import Test.Hspec++main :: IO ()+main = hspec $ pure ()