parallel-io 0.2 → 0.2.1

raw patch · 4 files changed

+114/−7 lines, 4 filesdep ~base

Dependency ranges changed: base

Files

• Control/Concurrent/ParallelIO/ConcurrentSet.hs +67/−0
• Control/Concurrent/ParallelIO/Global.hs +23/−1
• Control/Concurrent/ParallelIO/Local.hs +18/−2
• parallel-io.cabal +6/−4

@@ -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)

@@ -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. --

@@ -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.

@@ -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 <ndmitchell@gmail.com>,-                    Bulat Ziganshin <bulat.ziganshin@gmail.com>,-                    Max Bolingbroke <batterseapower@hotmail.com>+Author:             Max Bolingbroke <batterseapower@hotmail.com>,+                    Neil Mitchell <ndmitchell@gmail.com>,+                    Bulat Ziganshin <bulat.ziganshin@gmail.com> Maintainer:         Max Bolingbroke <batterseapower@hotmail.com> 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