diff --git a/Control/Concurrent/ParallelIO.hs b/Control/Concurrent/ParallelIO.hs --- a/Control/Concurrent/ParallelIO.hs +++ b/Control/Concurrent/ParallelIO.hs @@ -1,3 +1,11 @@ +-- | Combinators for executing IO actions in parallel on a thread pool. +-- +-- This module just reexports "Control.Concurrent.ParallelIO.Global": this contains versions of +-- the combinators that make use of a single global thread pool with as many threads as there are +-- capabilities. +-- +-- For finer-grained control, you can use "Control.Concurrent.ParallelIO.Local" instead, which +-- gives you control over the creation of the pool. module Control.Concurrent.ParallelIO ( module Control.Concurrent.ParallelIO.Global ) where diff --git a/Control/Concurrent/ParallelIO/Global.hs b/Control/Concurrent/ParallelIO/Global.hs --- a/Control/Concurrent/ParallelIO/Global.hs +++ b/Control/Concurrent/ParallelIO/Global.hs @@ -7,15 +7,22 @@ -- Make sure that you compile with @-threaded@ and supply @+RTS -N2 -RTS@ -- to the generated Haskell executable, or you won't get any parallelism. -- +-- If you plan to allow your worker items to block, then you should read the documentation for 'extraWorkerWhileBlocked'. +-- -- The "Control.Concurrent.ParallelIO.Local" module provides a more general -- interface which allows explicit passing of pools and control of their size. -- This module is implemented on top of that one by maintaining a shared global thread -- pool with one thread per capability. module Control.Concurrent.ParallelIO.Global ( + -- * Executing actions + parallel_, parallel, parallelInterleaved, + + -- * Global pool management globalPool, stopGlobalPool, - extraWorkerWhileBlocked, spawnPoolWorker, killPoolWorker, + extraWorkerWhileBlocked, - parallel_, parallel, parallelInterleaved + -- * Advanced global pool management + spawnPoolWorker, killPoolWorker ) where import GHC.Conc @@ -24,7 +31,9 @@ import qualified Control.Concurrent.ParallelIO.Local as L - +-- | The global thread pool. Contains as many threads as there are capabilities. +-- +-- Users of the global pool must call 'stopGlobalPool' from the main thread at the end of their program. {-# NOINLINE globalPool #-} globalPool :: L.Pool globalPool = unsafePerformIO $ L.startPool numCapabilities @@ -36,6 +45,7 @@ -- See also 'L.stopPool'. stopGlobalPool :: IO () stopGlobalPool = L.stopPool globalPool + -- TODO: could I lift the requirement to call this function with a touchPool function after the parallel combinators? -- | Wrap any IO action used from your worker threads that may block with this method: -- it temporarily spawns another worker thread to make up for the loss of the old blocked @@ -45,14 +55,17 @@ extraWorkerWhileBlocked :: IO a -> IO a extraWorkerWhileBlocked = L.extraWorkerWhileBlocked globalPool --- | Internal method for adding extra unblocked threads to a pool if one is going to be --- temporarily blocked. +-- | Internal method for adding extra unblocked threads to a pool if one of the current +-- worker threads is going to be temporarily blocked. Unrestricted use of this is unsafe, +-- so we reccomend that you use the 'extraWorkerWhileBlocked' function instead if possible. -- -- See also 'L.spawnPoolWorkerFor'. spawnPoolWorker :: IO () spawnPoolWorker = L.spawnPoolWorkerFor globalPool --- | Internal method for removing threads from a pool after we become unblocked. +-- | Internal method for removing threads from a pool after one of the threads on the pool +-- becomes newly unblocked. Unrestricted use of this is unsafe, so we reccomend that you use +-- the 'extraWorkerWhileBlocked' function instead if possible. -- -- See also 'L.killPoolWorkerFor'. killPoolWorker :: IO () @@ -60,6 +73,8 @@ -- | Execute the given actions in parallel on the global thread pool. -- +-- Users of the global pool must call 'stopGlobalPool' from the main thread at the end of their program. +-- -- See also 'L.parallel_'. parallel_ :: [IO a] -> IO () parallel_ = L.parallel_ globalPool @@ -67,12 +82,16 @@ -- | Execute the given actions in parallel on the global thread pool, -- returning the results in the same order as the corresponding actions. -- +-- Users of the global pool must call 'stopGlobalPool' from the main thread at the end of their program. +-- -- See also 'L.parallel'. parallel :: [IO a] -> IO [a] parallel = L.parallel globalPool -- | Execute the given actions in parallel on the global thread pool, -- returning the results in the approximate order of completion. +-- +-- Users of the global pool must call 'stopGlobalPool' from the main thread at the end of their program. -- -- See also 'L.parallelInterleaved'. parallelInterleaved :: [IO a] -> IO [a] diff --git a/Control/Concurrent/ParallelIO/Local.hs b/Control/Concurrent/ParallelIO/Local.hs --- a/Control/Concurrent/ParallelIO/Local.hs +++ b/Control/Concurrent/ParallelIO/Local.hs @@ -1,4 +1,4 @@ -{-# LANGUAGE ScopedTypeVariables #-} +{-# LANGUAGE CPP, ScopedTypeVariables #-} -- | Parallelism combinators with explicit thread-pool creation and -- passing. -- @@ -9,16 +9,21 @@ -- Make sure that you compile with @-threaded@ and supply @+RTS -N2 -RTS@ -- to the generated Haskell executable, or you won't get any parallelism. -- +-- If you plan to allow your worker items to block, then you should read the documentation for 'extraWorkerWhileBlocked'. +-- -- The "Control.Concurrent.ParallelIO.Global" module is implemented -- on top of this one by maintaining a shared global thread pool -- with one thread per capability. module Control.Concurrent.ParallelIO.Local ( - WorkItem, WorkQueue, Pool, - withPool, startPool, stopPool, - enqueueOnPool, - extraWorkerWhileBlocked, spawnPoolWorkerFor, killPoolWorkerFor, + -- * Executing actions + parallel_, parallel, parallelInterleaved, + + -- * Pool management + Pool, withPool, startPool, stopPool, + extraWorkerWhileBlocked, - parallel_, parallel, parallelInterleaved + -- * Advanced pool management + spawnPoolWorkerFor, killPoolWorkerFor ) where import qualified Control.Concurrent.ParallelIO.ConcurrentSet as CS @@ -30,7 +35,26 @@ import System.IO --- | Type of work items you can put onto the queue. The 'Bool' +#if MIN_VERSION_base(4,3,0) +import Control.Exception ( mask ) +#else +import Control.Exception ( blocked, block, unblock ) + +mask :: ((IO a -> IO a) -> IO b) -> IO b +mask io = blocked >>= \b -> if b then io id else block $ io unblock +#endif + + +-- TODO: I should deal nicely with exceptions raised by the actions on other threads. +-- Probably I should provide variants of the functions that report exceptions in lieu +-- of values. +-- +-- When I introduce this, I want to preserve the current behaviour that causes the +-- application to die promptly if we are using the unsafe variants of the combinators, +-- and one of the nested actions dies. + + +-- | Type of work items that are put onto the queue internally. The 'Bool' -- returned from the 'IO' action specifies whether the invoking -- thread should terminate itself immediately. type WorkItem = IO Bool @@ -38,8 +62,8 @@ -- | A 'WorkQueue' is used to communicate 'WorkItem's to the workers. type WorkQueue = CS.ConcurrentSet WorkItem --- | The type of thread pools used by 'ParallelIO'. --- The best way to construct one of these is using 'withPool'. +-- | A thread pool, containing a maximum number of threads. The best way to +-- construct one of these is using 'withPool'. data Pool = Pool { pool_threadcount :: Int, pool_spawnedby :: ThreadId, @@ -69,9 +93,8 @@ replicateM_ (threadcount - 1) (spawnPoolWorkerFor pool) return pool --- | Clean up a thread pool. If you don't call this then no one holds the queue, --- the queue gets GC'd, the threads find themselves blocked indefinitely, and you get --- exceptions. +-- | Clean up a thread pool. If you don't call this from the main thread then no one holds the queue, +-- the queue gets GC'd, the threads find themselves blocked indefinitely, and you get exceptions. -- -- This cleanly shuts down the threads so the queue isn't important and you don't get -- exceptions. @@ -91,22 +114,36 @@ enqueueOnPool :: Pool -> WorkItem -> IO () enqueueOnPool pool = CS.insert (pool_queue pool) --- | Wrap any IO action used from your worker threads that may block with this method: --- it temporarily spawns another worker thread to make up for the loss of the old blocked +-- | You should wrap any IO action used from your worker threads that may block with this method. +-- It temporarily spawns another worker thread to make up for the loss of the old blocked -- worker. -- -- This is particularly important if the unblocking is dependent on worker threads actually doing -- work. If you have this situation, and you don't use this method to wrap blocking actions, then -- you may get a deadlock if all your worker threads get blocked on work that they assume will be -- done by other worker threads. +-- +-- An example where something goes wrong if you don't use this to wrap blocking actions is the following example: +-- +-- > newEmptyMVar >>= \mvar -> parallel_ pool [readMVar mvar, putMVar mvar ()] +-- +-- If we only have one thread, we will sometimes get a schedule where the 'readMVar' action is run +-- before the 'putMVar'. Unless we wrap the read with 'extraWorkerWhileBlocked', if the pool has a +-- single thread our program to deadlock, because the worker will become blocked and no other thread +-- will be available to execute the 'putMVar'. +-- +-- The correct code is: +-- +-- > newEmptyMVar >>= \mvar -> parallel_ pool [extraWorkerWhileBlocked pool (readMVar mvar), putMVar mvar ()] extraWorkerWhileBlocked :: Pool -> IO a -> IO a extraWorkerWhileBlocked pool wait = E.bracket (spawnPoolWorkerFor pool) (\() -> killPoolWorkerFor pool) (\() -> wait) --- | Internal method for adding extra unblocked threads to a pool if one is going to be --- temporarily blocked. +-- | Internal method for adding extra unblocked threads to a pool if one of the current +-- worker threads is going to be temporarily blocked. Unrestricted use of this is unsafe, +-- so we reccomend that you use the 'extraWorkerWhileBlocked' function instead if possible. spawnPoolWorkerFor :: Pool -> IO () spawnPoolWorkerFor pool = do - _ <- forkIO $ workerLoop `E.catch` \(e :: E.SomeException) -> do + _ <- mask $ \restore -> forkIO $ restore workerLoop `E.catch` \(e :: E.SomeException) -> do hPutStrLn stderr $ "Exception on thread: " ++ show e throwTo (pool_spawnedby pool) $ ErrorCall $ "Control.Concurrent.ParallelIO: parallel thread died.\n" ++ show e return () @@ -116,7 +153,9 @@ kill <- join $ CS.delete (pool_queue pool) unless kill workerLoop --- | Internal method for removing threads from a pool after we become unblocked. +-- | Internal method for removing threads from a pool after one of the threads on the pool +-- becomes newly unblocked. Unrestricted use of this is unsafe, so we reccomend that you use +-- the 'extraWorkerWhileBlocked' function instead if possible. killPoolWorkerFor :: Pool -> IO () killPoolWorkerFor pool = enqueueOnPool pool $ return True @@ -136,7 +175,10 @@ -- been performed. -- -- 4. The above properties are true even if 'parallel_' is used by an --- action which is itself being executed by 'parallel_'. +-- action which is itself being executed by one of the parallel combinators. +-- +-- If any of the IO actions throws an exception, undefined behaviour will result. +-- If you want safety, wrap your actions in 'Control.Exception.try'. parallel_ :: Pool -> [IO a] -> IO () parallel_ _ [] = return () -- It is very important that we *don't* include this special case! @@ -145,18 +187,18 @@ -- to allow processing to continue even before it has finished executing. --parallel_ pool xs | pool_threadcount pool <= 1 = sequence_ xs parallel_ _ [x] = x >> return () -parallel_ pool (x1:xs) = do +parallel_ pool (x1:xs) = mask $ \restore -> do count <- newMVar $ length xs pause <- newEmptyMVar forM_ xs $ \x -> enqueueOnPool pool $ do - _ <- x + _ <- restore x modifyMVar count $ \i -> do let i' = i - 1 kill = i' == 0 when kill $ putMVar pause () return (i', kill) - _ <- x1 + _ <- restore x1 -- NB: it is safe to spawn a worker because at least one will die - the -- length of xs must be strictly greater than 0. spawnPoolWorkerFor pool @@ -168,7 +210,7 @@ -- Has the following properties: -- -- 1. Never creates more or less unblocked threads than are specified to --- live in the pool. NB: this count includes the thread executing 'parallel_'. +-- live in the pool. NB: this count includes the thread executing 'parallel'. -- This should minimize contention and hence pre-emption, while also preventing -- starvation. -- @@ -178,21 +220,24 @@ -- been performed. -- -- 4. The above properties are true even if 'parallel' is used by an --- action which is itself being executed by 'parallel'. +-- action which is itself being executed by one of the parallel combinators. +-- +-- If any of the IO actions throws an exception, undefined behaviour will result. +-- If you want safety, wrap your actions in 'Control.Exception.try'. parallel :: Pool -> [IO a] -> IO [a] parallel _ [] = return [] -- It is important that we do not include this special case (see parallel_ for why) --parallel pool xs | pool_threadcount pool <= 1 = sequence xs parallel _ [x] = fmap return x -parallel pool (x1:xs) = do +parallel pool (x1:xs) = mask $ \restore -> do count <- newMVar $ length xs resultvars <- forM xs $ \x -> do resultvar <- newEmptyMVar enqueueOnPool pool $ do - x >>= putMVar resultvar + restore x >>= putMVar resultvar modifyMVar count $ \i -> let i' = i - 1 in return (i', i' == 0) return resultvar - result1 <- x1 + result1 <- restore x1 -- NB: it is safe to spawn a worker because at least one will die - the -- length of xs must be strictly greater than 0. spawnPoolWorkerFor pool @@ -204,7 +249,7 @@ -- Has the following properties: -- -- 1. Never creates more or less unblocked threads than are specified to --- live in the pool. NB: this count includes the thread executing 'parallel_'. +-- live in the pool. NB: this count includes the thread executing 'parallelInterleaved'. -- This should minimize contention and hence pre-emption, while also preventing -- starvation. -- @@ -214,20 +259,24 @@ -- is likely to be very similar to the order of completion. -- -- 3. The above properties are true even if 'parallelInterleaved' is used by an --- action which is itself being executed by 'parallelInterleaved'. +-- action which is itself being executed by one of the parallel combinators. +-- +-- If any of the IO actions throws an exception, undefined behaviour will result. +-- If you want safety, wrap your actions in 'Control.Exception.try'. parallelInterleaved :: Pool -> [IO a] -> IO [a] parallelInterleaved _ [] = return [] -parallelInterleaved pool xs | pool_threadcount pool <= 1 = sequence xs +-- It is important that we do not include this special case (see parallel_ for why) +--parallelInterleaved pool xs | pool_threadcount pool <= 1 = sequence xs parallelInterleaved _ [x] = fmap return x -parallelInterleaved pool (x1:xs) = do +parallelInterleaved pool (x1:xs) = mask $ \restore -> do let thecount = length xs count <- newMVar $ thecount resultschan <- newChan forM_ xs $ \x -> do enqueueOnPool pool $ do - x >>= writeChan resultschan + restore x >>= writeChan resultschan modifyMVar count $ \i -> let i' = i - 1 in return (i', i' == 0) - result1 <- x1 + result1 <- restore x1 -- NB: it is safe to spawn a worker because at least one will die - the -- length of xs must be strictly greater than 0. spawnPoolWorkerFor pool diff --git a/parallel-io.cabal b/parallel-io.cabal --- a/parallel-io.cabal +++ b/parallel-io.cabal @@ -1,12 +1,14 @@ Name: parallel-io -Version: 0.2.2 +Version: 0.3.0 Cabal-Version: >= 1.2 Category: Concurrency Synopsis: Combinators for executing IO actions in parallel on a thread pool. Description: This package provides combinators for sequencing IO actions onto a thread pool. The - thread pool is guaranteed to contain a fixed number of unblocked threads, minimizing - contention. Furthermore, the parallel combinators can be used re-entrently - your parallel - actions can spawn more parallel actions - without violating this property. + thread pool is guaranteed to contain no more unblocked threads than a user-specified upper limit, thus + minimizing contention. + . + Furthermore, the parallel combinators can be used reentrantly - your parallel + actions can spawn more parallel actions - without violating this property of the thread pool. . The package is heavily inspired by the thread . Thanks to Neil Mitchell and Bulat Ziganshin for the code this package is based on.