broadcast-chan 0.2.0.2 → 0.2.1

raw patch · 4 files changed

+125/−20 lines, 4 filesdep +transformersPVP: major bump suggested

API removals or changes: PVP suggests a major version bump

Dependencies added: transformers

API changes (from Hackage documentation)

Files

• BroadcastChan/Extra.hs +114/−13
• CHANGELOG.md +6/−0
• README.md +1/−0
• broadcast-chan.cabal +4/−7

@@ -25,9 +25,12 @@     ( Action(..)     , BracketOnError(..)     , Handler(..)+    , ThreadBracket(..)     , mapHandler     , runParallel+    , runParallelWith     , runParallel_+    , runParallelWith_     ) where  #if !MIN_VERSION_base(4,8,0)@@ -37,15 +40,19 @@ import Control.Concurrent.MVar import Control.Concurrent.QSem import Control.Concurrent.QSemN-import Control.Exception (Exception(..), SomeException(..))+import Control.Exception (Exception(..), SomeException(..), bracketOnError) import qualified Control.Exception as Exc import Control.Monad ((>=>), replicateM, void)+import Control.Monad.Trans.Cont (ContT(..)) import Control.Monad.IO.Unlift (MonadIO(..)) import Data.Typeable (Typeable) import System.Mem.Weak (Weak, deRefWeak)  import BroadcastChan.Internal +evalContT :: Monad m => ContT r m r -> m r+evalContT m = runContT m return+ -- DANGER! Breaks the invariant that you can't write to closed channels! -- Only meant to be used in 'parallelCore'! unsafeWriteBChan :: MonadIO m => BroadcastChan In a -> a -> m ()@@ -98,6 +105,33 @@     --   finish and threads to terminate.     } +-- | Datatype for specifying additional setup/cleanup around forking threads.+-- Used by 'runParallelWith' and 'runParallelWith_' to fix resource management+-- in @broadcast-chan-conduit@.+--+-- If the allocation action can fail/abort with an exception it __MUST__ take+-- care not to leak resources in these cases. In other words, IFF 'setupFork'+-- succeeds then this library will ensure the corresponding cleanup runs.+--+-- @since 0.2.1+data ThreadBracket+    = ThreadBracket+    { setupFork :: IO ()+    -- ^ Setup action to run before spawning a new thread.+    , cleanupFork :: IO ()+    -- ^ Normal cleanup action upon thread termination.+    , cleanupForkError :: IO ()+    -- ^ Exceptional cleanup action in case thread terminates due to an+    -- exception.+    }++noopBracket :: ThreadBracket+noopBracket = ThreadBracket+    { setupFork = return ()+    , cleanupFork = return ()+    , cleanupForkError = return ()+    }+ -- | Convenience function for changing the monad the exception handler runs in. mapHandler :: (m Action -> n Action) -> Handler m a -> Handler n a mapHandler _ (Simple act) = Simple act@@ -111,9 +145,10 @@     => Handler IO a     -> Int     -> IO ()+    -> ThreadBracket     -> (a -> IO ())     -> m (IO [Weak ThreadId], [Weak ThreadId] -> IO (), a -> IO (), m ())-parallelCore hndl threads onDrop f = liftIO $ do+parallelCore hndl threads onDrop threadBracket f = liftIO $ do     originTid <- myThreadId     inChanIn <- newBroadcastChan     inChanOut <- newBChanListener inChanIn@@ -144,21 +179,28 @@                     f a `Exc.catch` handler a                     processInput -        allocate :: IO [Weak ThreadId]-        allocate = liftIO $ do-            tids <- replicateM threads . forkFinally processInput $ \exit -> do+        unsafeAllocThread :: IO (Weak ThreadId)+        unsafeAllocThread = do+            setupFork+            tid <- forkFinally processInput $ \exit -> do                 signalQSemN shutdownSem 1                 case exit of                     Left exc-                      | Just Shutdown <- fromException exc -> return ()+                      | Just Shutdown <- fromException exc -> cleanupForkError                       | otherwise ->                           Exc.throwTo originTid exc `Exc.catch` shutdownHandler-                    Right () -> return ()+                    Right () -> cleanupFork -            mapM mkWeakThreadId tids+            mkWeakThreadId tid           where             shutdownHandler Shutdown = return () +        allocThread :: ContT r IO (Weak ThreadId)+        allocThread = ContT $ bracketOnError unsafeAllocThread killWeakThread++        allocateThreads :: IO [Weak ThreadId]+        allocateThreads = evalContT $ replicateM threads allocThread+         cleanup :: [Weak ThreadId] -> IO ()         cleanup threadIds = liftIO . Exc.uninterruptibleMask_ $ do             mapM_ killWeakThread threadIds@@ -169,8 +211,10 @@             closeBChan inChanIn             liftIO $ waitQSemN endSem threads -    return (allocate, cleanup, bufferValue, wait)+    return (allocateThreads, cleanup, bufferValue, wait)   where+    ThreadBracket{setupFork,cleanupFork,cleanupForkError} = threadBracket+     killWeakThread :: Weak ThreadId -> IO ()     killWeakThread wTid = do         tid <- deRefWeak wTid@@ -216,7 +260,37 @@     -> ((a -> m ()) -> (a -> m (Maybe b)) -> n r)     -- ^ \"Stream\" processing function     -> n (BracketOnError n r)-runParallel yielder hndl threads work pipe = do+runParallel = runParallelWith noopBracket++-- | Like 'runParallel', but accepts a setup and cleanup action that will be+-- run before spawning a new thread and upon thread exit respectively.+--+-- The main use case is to properly manage the resource reference counts of+-- 'Control.Monad.Trans.Resource.ResourceT'.+--+-- If the setup throws an 'IO' exception or otherwise aborts, it __MUST__+-- ensure any allocated resource are freed. If it completes without an+-- exception, the cleanup is guaranteed to run (assuming proper use of+-- bracketing with the returned 'BracketOnError').+--+-- @since 0.2.1+runParallelWith+    :: forall a b m n r+     . (MonadIO m, MonadIO n)+    => ThreadBracket+    -- ^ Bracketing action used to manage resources across thread spawns+    -> Either (b -> n r) (r -> b -> n r)+    -- ^ Output yielder+    -> Handler IO a+    -- ^ Parallel processing exception handler+    -> Int+    -- ^ Number of threads to use+    -> (a -> IO b)+    -- ^ Function to run in parallel+    -> ((a -> m ()) -> (a -> m (Maybe b)) -> n r)+    -- ^ \"Stream\" processing function+    -> n (BracketOnError n r)+runParallelWith threadBracket yielder hndl threads work pipe = do     outChanIn <- newBroadcastChan     outChanOut <- newBChanListener outChanIn @@ -227,7 +301,7 @@         notifyDrop = void $ writeBChan outChanIn Nothing      (allocate, cleanup, bufferValue, wait) <--        parallelCore hndl threads notifyDrop process+        parallelCore hndl threads notifyDrop threadBracket process      let queueAndYield :: a -> m (Maybe b)         queueAndYield x = do@@ -280,13 +354,40 @@     -> ((a -> m ()) -> n r)     -- ^ \"Stream\" processing function     -> n (BracketOnError n r)-runParallel_ hndl threads workFun processElems = do+runParallel_ = runParallelWith_ noopBracket++-- | Like 'runParallel_', but accepts a setup and cleanup action that will be+-- run before spawning a new thread and upon thread exit respectively.+--+-- The main use case is to properly manage the resource reference counts of+-- 'Control.Monad.Trans.Resource.ResourceT'.+--+-- If the setup throws an 'IO' exception or otherwise aborts, it __MUST__+-- ensure any allocated resource are freed. If it completes without an+-- exception, the cleanup is guaranteed to run (assuming proper use of+-- bracketing with the returned 'BracketOnError').+--+-- @since 0.2.1+runParallelWith_+    :: (MonadIO m, MonadIO n)+    => ThreadBracket+    -- ^ Bracketing action used to manage resources across thread spawns+    -> Handler IO a+    -- ^ Parallel processing exception handler+    -> Int+    -- ^ Number of threads to use+    -> (a -> IO ())+    -- ^ Function to run in parallel+    -> ((a -> m ()) -> n r)+    -- ^ \"Stream\" processing function+    -> n (BracketOnError n r)+runParallelWith_ threadBracket hndl threads workFun processElems = do     sem <- liftIO $ newQSem threads      let process x = signalQSem sem >> workFun x      (allocate, cleanup, bufferValue, wait) <--        parallelCore hndl threads (return ()) process+        parallelCore hndl threads (return ()) threadBracket process      let action = do             result <- processElems $ \v -> liftIO $ do

@@ -1,3 +1,9 @@+0.2.1 [2019.11.17]+------------------+* Adds `ThreadBracket`, `runParallelWith`, and `runParallelWith_` to+  `BroadcastChan.Extra` to support thread related resource management. This is+  required to fix `broadcast-chan-conduit`'s use of `MonadResource`.+ 0.2.0.2 [2019.03.30] -------------------- * GHC 8.6/MonadFail compatibility fix

@@ -2,6 +2,7 @@ ================================================================ [![BSD3](https://img.shields.io/badge/License-BSD-blue.svg)](https://en.wikipedia.org/wiki/BSD_License) [![Hackage](https://img.shields.io/hackage/v/broadcast-chan.svg)](https://hackage.haskell.org/package/broadcast-chan)+[![hackage-ci](https://matrix.hackage.haskell.org/api/v2/packages/broadcast-chan/badge)](https://matrix.hackage.haskell.org/#/package/broadcast-chan) [![Stackage](https://www.stackage.org/package/broadcast-chan/badge/lts?label=Stackage)](https://www.stackage.org/package/broadcast-chan) [![Build Status](https://travis-ci.org/merijn/broadcast-chan.svg)](https://travis-ci.org/merijn/broadcast-chan) 

@@ -1,5 +1,5 @@ Name:               broadcast-chan-Version:            0.2.0.2+Version:            0.2.1  Homepage:           https://github.com/merijn/broadcast-chan Bug-Reports:        https://github.com/merijn/broadcast-chan/issues@@ -15,7 +15,7 @@ Cabal-Version:      >= 1.10 Build-Type:         Simple Tested-With:        GHC == 7.8.4, GHC == 7.10.3, GHC == 8.0.2, GHC == 8.2.2,-                    GHC == 8.4.4, GHC == 8.6.4+                    GHC == 8.4.4, GHC == 8.6.5, GHC == 8.8.1  Extra-Source-Files: README.md                   , CHANGELOG.md@@ -101,7 +101,8 @@                         Trustworthy                         TupleSections -  Build-Depends:        base >= 4.7 && < 4.13+  Build-Depends:        base >= 4.7 && < 4.14+               ,        transformers >= 0.2 && < 0.6                ,        unliftio-core >= 0.1.1 && < 0.2  Benchmark sync@@ -170,7 +171,3 @@ Source-Repository head   Type:     git   Location: ssh://github.com:merijn/broadcast-chan.git--Source-Repository head-  Type:     mercurial-  Location: https://bitbucket.org/merijnv/broadcast-chan