diff --git a/BroadcastChan/Extra.hs b/BroadcastChan/Extra.hs --- a/BroadcastChan/Extra.hs +++ b/BroadcastChan/Extra.hs @@ -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 diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -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 diff --git a/README.md b/README.md --- a/README.md +++ b/README.md @@ -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) diff --git a/broadcast-chan.cabal b/broadcast-chan.cabal --- a/broadcast-chan.cabal +++ b/broadcast-chan.cabal @@ -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