diff --git a/Control/Concurrent/ParallelIO/ConcurrentSet.hs b/Control/Concurrent/ParallelIO/ConcurrentSet.hs new file mode 100644 --- /dev/null +++ b/Control/Concurrent/ParallelIO/ConcurrentSet.hs @@ -0,0 +1,67 @@ +-- | A set that elements can be added to and remove from concurrently. +-- +-- The main difference between this and a queue is that 'ConcurrentSet' does not +-- make any guarantees about the order in which things will come out -- in fact, +-- it will go out of its way to make sure that they are unordered! +-- +-- The reason that I use this primitive rather than 'Chan' is that: +-- 1) At Standard Chartered we saw intermitted deadlocks when using 'Chan', +-- but Neil tells me that he stopped seeing them when they moved to a 'ConcurrentSet' +-- like thing. We never found the reason for the deadlocks though... +-- 2) It's better to dequeue parallel tasks in pseudo random order for many +-- common applications, because (e.g. in Shake) lots of tasks that require the same +-- machine resources (i.e. CPU or RAM) tend to be next to each other in the list. +-- Thus, reducing access locality means that we tend to choose tasks that require +-- different resources. +module Control.Concurrent.ParallelIO.ConcurrentSet ( + ConcurrentSet, new, insert, delete + ) where + +import Control.Concurrent.MVar +import Control.Monad + +import qualified Data.IntMap as IM + +import System.Random + + +data ConcurrentSet a = CS (MVar (StdGen, Either (MVar ()) (IM.IntMap a))) + +new :: IO (ConcurrentSet a) +new = fmap CS $ liftM2 (\gen mvar -> (gen, Left mvar)) newStdGen newEmptyMVar >>= newMVar + +insert :: ConcurrentSet a -> a -> IO () +insert (CS set_mvar) x = modifyMVar_ set_mvar go + where go (gen, ei_mvar_ys) = do + let (i, gen') = random gen + case ei_mvar_ys of + Left wait_mvar -> do + -- Wake up all waiters (if any): any one of them may want this item + putMVar wait_mvar () + return (gen', Right (IM.singleton i x)) + Right ys -> return (gen', Right (IM.insert i x ys)) + +delete :: ConcurrentSet a -> IO a +delete (CS set_mvar) = loop + where + loop = do + ei_wait_x <- modifyMVar set_mvar go + case ei_wait_x of + Left wait_mvar -> do + -- NB: it's very important that we don't do this while we are holding the set_mvar! + takeMVar wait_mvar + -- Someone put data in the MVar, but we might have to wait again if someone snaffles + -- it before we got there. + -- + -- TODO: make this fairer -- there is definite starvation potential here, though it + -- doesn't matter for the application I have in mind (Shake) + loop + Right x -> return x + + go (gen, Left wait_mvar) = return ((gen, Left wait_mvar), Left wait_mvar) + go (gen, Right xs) = do + let (chosen, xs') = IM.deleteFindMin xs + new_value <- if IM.null xs' + then fmap Left newEmptyMVar + else return (Right xs') + return ((gen, new_value), Right chosen) 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 @@ -12,7 +12,8 @@ -- 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 ( - stopGlobalPool, + globalPool, stopGlobalPool, + extraWorkerWhileBlocked, spawnPoolWorker, killPoolWorker, parallel_, parallel, parallelInterleaved ) where @@ -35,6 +36,27 @@ -- See also 'L.stopPool'. stopGlobalPool :: IO () stopGlobalPool = L.stopPool globalPool + +-- | 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. +-- +-- See also 'L.extraWorkerWhileBlocked'. +extraWorkerWhileBlocked :: IO () -> IO () +extraWorkerWhileBlocked = L.extraWorkerWhileBlocked globalPool + +-- | Internal method for adding extra unblocked threads to a pool if one is going to be +-- temporarily blocked. +-- +-- See also 'L.spawnPoolWorkerFor'. +spawnPoolWorker :: IO () +spawnPoolWorker = L.spawnPoolWorkerFor globalPool + +-- | Internal method for removing threads from a pool after we become unblocked. +-- +-- See also 'L.killPoolWorkerFor'. +killPoolWorker :: IO () +killPoolWorker = L.killPoolWorkerFor globalPool -- | Execute the given actions in parallel on the global thread pool. -- 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 @@ -15,7 +15,8 @@ module Control.Concurrent.ParallelIO.Local ( WorkItem, WorkQueue, Pool, withPool, startPool, stopPool, - enqueueOnPool, spawnPoolWorkerFor, + enqueueOnPool, + extraWorkerWhileBlocked, spawnPoolWorkerFor, killPoolWorkerFor, parallel_, parallel, parallelInterleaved ) where @@ -76,7 +77,7 @@ -- Only call this /after/ all users of the pool have completed, or your program may -- block indefinitely. stopPool :: Pool -> IO () -stopPool pool = replicateM_ (pool_threadcount pool - 1) $ enqueueOnPool pool $ return True +stopPool pool = replicateM_ (pool_threadcount pool - 1) $ killPoolWorkerFor pool -- | A safe wrapper around 'startPool' and 'stopPool'. Executes an 'IO' action using a newly-created -- pool with the specified number of threads and cleans it up at the end. @@ -88,6 +89,17 @@ 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 +-- 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. +extraWorkerWhileBlocked :: Pool -> IO () -> IO () +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. spawnPoolWorkerFor :: Pool -> IO () @@ -101,6 +113,10 @@ workerLoop = do kill <- join $ CS.delete (pool_queue pool) unless kill workerLoop + +-- | Internal method for removing threads from a pool after we become unblocked. +killPoolWorkerFor :: Pool -> IO () +killPoolWorkerFor pool = enqueueOnPool pool $ return True -- | Run the list of computations in parallel. diff --git a/parallel-io.cabal b/parallel-io.cabal --- a/parallel-io.cabal +++ b/parallel-io.cabal @@ -1,5 +1,5 @@ Name: parallel-io -Version: 0.2 +Version: 0.2.1 Cabal-Version: >= 1.2 Category: Concurrency Synopsis: Combinators for executing IO actions in parallel on a thread pool. @@ -13,9 +13,9 @@ License: BSD3 License-File: LICENSE Homepage: http://batterseapower.github.com/parallel-io -Author: Neil Mitchell , - Bulat Ziganshin , - Max Bolingbroke +Author: Max Bolingbroke , + Neil Mitchell , + Bulat Ziganshin Maintainer: Max Bolingbroke Build-Type: Simple @@ -33,6 +33,8 @@ Control.Concurrent.ParallelIO Control.Concurrent.ParallelIO.Global Control.Concurrent.ParallelIO.Local + Other-Modules: + Control.Concurrent.ParallelIO.ConcurrentSet Build-Depends: base >= 3 && < 5, extensible-exceptions > 0.1.0.1, containers >= 0.3 && < 0.4, random >= 1.0 && < 1.1