packages feed

parallel-io 0.2.2 → 0.3.0

raw patch · 4 files changed

+121/−43 lines, 4 files

Files

Control/Concurrent/ParallelIO.hs view
@@ -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
Control/Concurrent/ParallelIO/Global.hs view
@@ -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]
Control/Concurrent/ParallelIO/Local.hs view
@@ -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
parallel-io.cabal view
@@ -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 <http://thread.gmane.org/gmane.comp.lang.haskell.cafe/56499/focus=56521>.                     Thanks to Neil Mitchell and Bulat Ziganshin for the code this package is based on.