parallel-io 0.3.1 → 0.3.2
raw patch · 3 files changed
+135/−264 lines, 3 files
Files
- Control/Concurrent/ParallelIO/ConcurrentCollection.hs +0/−96
- Control/Concurrent/ParallelIO/Local.hs +130/−162
- parallel-io.cabal +5/−6
− Control/Concurrent/ParallelIO/ConcurrentCollection.hs
@@ -1,96 +0,0 @@-module Control.Concurrent.ParallelIO.ConcurrentCollection (- ConcurrentSet, Chan, ConcurrentCollection(..)- ) where--import Control.Concurrent.ParallelIO.Compat--import Control.Concurrent.MVar-import Control.Concurrent.Chan-import Control.Monad--import qualified Data.IntMap as IM--import System.Random---class ConcurrentCollection p where- new :: IO (p a)- insert :: p a -> a -> IO ()- delete :: p a -> IO a----- | 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.-data ConcurrentSet a = CS (MVar (StdGen, Contents (IM.IntMap a)))--data Contents a = EmptyWithWaiters (MVar ())- | NonEmpty a--instance ConcurrentCollection ConcurrentSet where- new = fmap CS $ liftM2 (\gen mvar -> (gen, EmptyWithWaiters mvar)) newStdGen newEmptyMVar >>= newMVar-- -- We don't mask asynchronous exceptions here because it's OK if we signal the wait_mvar- -- but the set still doesn't contain anything: the readers (i.e. in "delete") will just- -- discover that and start waiting again, just as if another thread had deleted before- -- they got a chance to read from a newly non-empty set- insert (CS set_mvar) x = modifyMVar_ set_mvar go- where go (gen, contents) = do- let (i, gen') = random gen- case contents of- EmptyWithWaiters wait_mvar -> do- -- Wake up all waiters (if any): any one of them may want this item- --- -- NB: we don't use putMvar here (even though it would be safe) because- -- this way I get an obvious exception if I've done something daft.- True <- tryPutMVar wait_mvar ()- return (gen', NonEmpty (IM.singleton i x))- NonEmpty ys -> return (gen', NonEmpty (IM.insert i x ys))-- delete (CS set_mvar) = loop- where- loop = do- contents <- modifyMVar set_mvar peek_inside- case contents of- EmptyWithWaiters wait_mvar -> do- -- NB: it's very important that we don't do this while we are holding the set_mvar!- --- -- We are careful to readMVar here rather than takeMVar, because *there may be more- -- than one waiter*. This does lead to a bit of a scrummage, because every single- -- waiter will get woken up and go for newly-added data simultaneously, but the alternative- -- is disconcertingly subtle.- () <- readMVar 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- NonEmpty x -> return x- - peek_inside (gen, EmptyWithWaiters wait_mvar) = return ((gen, EmptyWithWaiters wait_mvar), EmptyWithWaiters wait_mvar)- peek_inside (gen, NonEmpty xs) = do- let (chosen, xs') = IM.deleteFindMin xs- new_value <- if IM.null xs'- then fmap EmptyWithWaiters newEmptyMVar- else return (NonEmpty xs')- return ((gen, new_value), NonEmpty chosen)---instance ConcurrentCollection Chan where- new = newChan- insert = writeChan- delete = readChan
Control/Concurrent/ParallelIO/Local.hs view
@@ -27,46 +27,25 @@ ) where import Control.Concurrent.ParallelIO.Compat-import qualified Control.Concurrent.ParallelIO.ConcurrentCollection as CC import Control.Concurrent import Control.Exception import Control.Monad -import Data.IORef- import System.IO import Prelude hiding (catch) --- 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.+reflectExceptionsTo :: ThreadId -> IO () -> IO ()+reflectExceptionsTo tid act = act `catch` \e -> throwTo tid (e :: SomeException) --- | 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.------ INVARIANT: all 'WorkItem's do not throw synchronous exceptions. It is acceptable--- for them to throw asynchronous exceptions and to be interruptible.-type WorkItem = IO Bool---- | A 'WorkQueue' is used to communicate 'WorkItem's to the workers.---type WorkQueue = CC.Chan WorkItem-type WorkQueue = CC.ConcurrentSet WorkItem- -- | 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,- pool_queue :: WorkQueue+ pool_sem :: QSem } -- | A slightly unsafe way to construct a pool. Make a pool from the maximum@@ -80,17 +59,7 @@ startPool :: Int -> IO Pool startPool threadcount | threadcount < 1 = error $ "startPool: thread count must be strictly positive (was " ++ show threadcount ++ ")"- | otherwise = do- threadId <- myThreadId- queue <- CC.new- let pool = Pool {- pool_threadcount = threadcount,- pool_spawnedby = threadId,- pool_queue = queue- }- - replicateM_ (threadcount - 1) (spawnPoolWorkerFor pool)- return pool+ | otherwise = fmap (Pool threadcount) $ newQSem (threadcount - 1) -- | 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.@@ -109,10 +78,6 @@ withPool threadcount = bracket (startPool threadcount) stopPool --- | Internal method for scheduling work on a pool.-enqueueOnPool :: Pool -> WorkItem -> IO ()-enqueueOnPool pool = CC.insert (pool_queue pool)- -- | 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.@@ -135,36 +100,19 @@ -- -- > newEmptyMVar >>= \mvar -> parallel_ pool [extraWorkerWhileBlocked pool (readMVar mvar), putMVar mvar ()] extraWorkerWhileBlocked :: Pool -> IO a -> IO a-extraWorkerWhileBlocked pool wait = bracket (spawnPoolWorkerFor pool) (\() -> killPoolWorkerFor pool) (\() -> wait)+extraWorkerWhileBlocked pool = bracket_ (spawnPoolWorkerFor pool) (killPoolWorkerFor pool) -- | 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 recommend that you use the 'extraWorkerWhileBlocked' function instead if possible. spawnPoolWorkerFor :: Pool -> IO ()-spawnPoolWorkerFor pool = {- putStrLn "spawnPoolWorkerFor" >> -} do- _ <- mask $ \restore -> forkIO $ restore workerLoop `catch` \(e :: SomeException) -> do- tid <- myThreadId- hPutStrLn stderr $ "Exception on " ++ show tid ++ ": " ++ show e- throwTo (pool_spawnedby pool) $ ErrorCall $ "Control.Concurrent.ParallelIO: parallel thread died.\n" ++ show e- return ()- where- workerLoop :: IO ()- workerLoop = do- --tid <- myThreadId- --hPutStrLn stderr $ "[waiting] " ++ show tid- work_item <- CC.delete (pool_queue pool)- --hPutStrLn stderr $ "[working] " ++ show tid- - -- If we get an asynchronous exception on a worker thread, don't make any attempt to handle it: just die.- -- The one concession we make is that we are careful not to lose work items from the queue.- kill <- work_item `onException` CC.insert (pool_queue pool) work_item- unless kill workerLoop+spawnPoolWorkerFor pool = signalQSem (pool_sem pool) -- | 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 = {- putStrLn "killPoolWorkerFor" >> -} enqueueOnPool pool (return True)+killPoolWorkerFor pool = waitQSem (pool_sem pool) -- | Run the list of computations in parallel.@@ -184,13 +132,41 @@ -- 4. The above properties are true even if 'parallel_' is used by an -- action which is itself being executed by one of the parallel combinators. ----- If any of the IO actions throws an exception, the exception thrown by the first--- failing action in the input list will be thrown by 'parallel_'.+-- 5. If any of the IO actions throws an exception this does not prevent any of the+-- other actions from being performed.+--+-- 6. If any of the IO actions throws an exception, the exception thrown by the first+-- failing action in the input list will be thrown by 'parallel_'. Importantly, at the+-- time the exception is thrown there is no guarantee that the other parallel actions+-- have completed.+--+-- The motivation for this choice is that waiting for the all threads to either return+-- or throw before throwing the first exception will almost always cause GHC to show the+-- "Blocked indefinitely in MVar operation" exception rather than the exception you care about.+--+-- The reason for this behaviour can be seen by considering this machine state:+--+-- 1. The main thread has used the parallel combinators to spawn two threads, thread 1 and thread 2.+-- It is blocked on both of them waiting for them to return either a result or an exception via an MVar.+--+-- 2. Thread 1 and thread 2 share another (empty) MVar, the "wait handle". Thread 2 is waiting on the handle,+-- while thread 2 will eventually put into the handle.+-- +-- Consider what happens when thread 1 is buggy and throws an exception before putting into the handle. Now+-- thread 2 is blocked indefinitely, and so the main thread is also blocked indefinetly waiting for the result+-- of thread 2. GHC has no choice but to throw the uninformative exception. However, what we really wanted to+-- see was the original exception thrown in thread 1!+--+-- By having the main thread abandon its wait for the results of the spawned threads as soon as the first exception+-- comes in, we give this exception a chance to actually be displayed. parallel_ :: Pool -> [IO a] -> IO () parallel_ pool xs = parallel pool xs >> return () -- | As 'parallel_', but instead of throwing exceptions that are thrown by subcomputations, -- they are returned in a data structure.+--+-- As a result, property 6 of 'parallel_' is not preserved, and therefore if your IO actions can depend on each other+-- and may throw exceptions your program may die with "blocked indefinitely" exceptions rather than informative messages. parallelE_ :: Pool -> [IO a] -> IO [Maybe SomeException] parallelE_ pool = fmap (map (either Just (\_ -> Nothing))) . parallelE pool @@ -212,40 +188,62 @@ -- 4. The above properties are true even if 'parallel' is used by an -- action which is itself being executed by one of the parallel combinators. ----- If any of the IO actions throws an exception, the exception thrown by the first--- failing action in the input list will be thrown by 'parallel'.+-- 5. If any of the IO actions throws an exception this does not prevent any of the+-- other actions from being performed.+--+-- 6. If any of the IO actions throws an exception, the exception thrown by the first+-- failing action in the input list will be thrown by 'parallel'. Importantly, at the+-- time the exception is thrown there is no guarantee that the other parallel actions+-- have completed.+--+-- The motivation for this choice is that waiting for the all threads to either return+-- or throw before throwing the first exception will almost always cause GHC to show the+-- "Blocked indefinitely in MVar operation" exception rather than the exception you care about.+--+-- The reason for this behaviour can be seen by considering this machine state:+--+-- 1. The main thread has used the parallel combinators to spawn two threads, thread 1 and thread 2.+-- It is blocked on both of them waiting for them to return either a result or an exception via an MVar.+--+-- 2. Thread 1 and thread 2 share another (empty) MVar, the "wait handle". Thread 2 is waiting on the handle,+-- while thread 2 will eventually put into the handle.+-- +-- Consider what happens when thread 1 is buggy and throws an exception before putting into the handle. Now+-- thread 2 is blocked indefinitely, and so the main thread is also blocked indefinetly waiting for the result+-- of thread 2. GHC has no choice but to throw the uninformative exception. However, what we really wanted to+-- see was the original exception thrown in thread 1!+--+-- By having the main thread abandon its wait for the results of the spawned threads as soon as the first exception+-- comes in, we give this exception a chance to actually be displayed. parallel :: Pool -> [IO a] -> IO [a]-parallel pool xs = do- ei_e_ress <- parallelE pool xs- mapM (either throw return) ei_e_ress+parallel pool acts = mask $ \restore -> do+ main_tid <- myThreadId+ resultvars <- forM acts $ \act -> do+ resultvar <- newEmptyMVar+ _tid <- forkIO $ bracket_ (killPoolWorkerFor pool) (spawnPoolWorkerFor pool) $ reflectExceptionsTo main_tid $ do+ res <- restore act+ -- Use tryPutMVar instead of putMVar so we get an exception if my brain has failed+ True <- tryPutMVar resultvar res+ return ()+ return resultvar+ extraWorkerWhileBlocked pool (mapM takeMVar resultvars) -- | As 'parallel', but instead of throwing exceptions that are thrown by subcomputations, -- they are returned in a data structure.+--+-- As a result, property 6 of 'parallel' is not preserved, and therefore if your IO actions can depend on each other+-- and may throw exceptions your program may die with "blocked indefinitely" exceptions rather than informative messages. parallelE :: Pool -> [IO a] -> IO [Either SomeException a]-parallelE _ [] = return []--- It is very important that we *don't* include this special case!--- The reason is that even if there is only one worker thread in the pool, one of--- the items we process might depend on the ability to use extraWorkerWhileBlocked--- to allow processing to continue even before it has finished executing.---parallelE pool xs | pool_threadcount pool <= 1 = sequence xs-parallelE _ [x] = fmap return (try x)-parallelE pool (x1:xs) = mask $ \restore -> do- count <- newIORef $ length xs- resultvars <- forM xs $ \x -> do+parallelE pool acts = mask $ \restore -> do+ resultvars <- forM acts $ \act -> do resultvar <- newEmptyMVar- enqueueOnPool pool $ do- ei_e_res <- try (restore x)+ _tid <- forkIO $ bracket_ (killPoolWorkerFor pool) (spawnPoolWorkerFor pool) $ do+ ei_e_res <- try (restore act) -- Use tryPutMVar instead of putMVar so we get an exception if my brain has failed- -- This also has the bonus that tryPutMVar is non-blocking, so we cannot get any- -- asynchronous exceptions from it (it is not "interruptable") True <- tryPutMVar resultvar ei_e_res- atomicModifyIORef count $ \i -> let i' = i - 1 in (i', i' == 0)+ return () return resultvar- ei_e_res1 <- try (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- fmap (ei_e_res1:) $ mapM takeMVar resultvars+ extraWorkerWhileBlocked pool (mapM takeMVar resultvars) -- | Run the list of computations in parallel, returning the results in the -- approximate order of completion.@@ -262,88 +260,58 @@ -- 3. The result of running actions appear in the list in undefined order, but which -- is likely to be very similar to the order of completion. ----- 3. The above properties are true even if 'parallelInterleaved' is used by an+-- 4. The above properties are true even if 'parallelInterleaved' is used by an -- action which is itself being executed by one of the parallel combinators. ----- If any of the IO actions throws an exception, the exception thrown by the first--- completing action in the input list will be thrown by 'parallelInterleaved'.-parallelInterleaved :: Pool -> [IO a] -> IO [a]-parallelInterleaved pool xs = do- ei_e_ress <- parallelInterleavedE_lazy pool xs- mapM (either throw return) ei_e_ress----- | As 'parallelInterleaved', but instead of throwing exceptions that are thrown by subcomputations,--- they are returned in a data structure.-parallelInterleavedE, parallelInterleavedE_lazy :: Pool -> [IO a] -> IO [Either SomeException a]-parallelInterleavedE pool xs = do- ei_e_ress <- parallelInterleavedE_lazy pool xs- mapM return ei_e_ress -- Force the output list: we should not return until all actions are done--parallelInterleavedE_lazy _ [] = return []--- It is important that we do not include this special case (see parallel for why)---parallelInterleaved pool xs | pool_threadcount pool <= 1 = sequence xs-parallelInterleavedE_lazy _ [x] = fmap return (try x)-parallelInterleavedE_lazy pool (x1:xs) = mask $ \restore -> do- let thecount = length xs- count <- newIORef (length xs)- resultschan <- newChan- forM_ xs $ \x -> do- enqueueOnPool pool $ do- ei_e_res <- try (restore x)- -- Although writeChan is interruptible, it unblocks promptly- writeChan resultschan ei_e_res- atomicModifyIORef count $ \i -> let i' = i - 1 in (i', i' == 0)- ei_e_res1 <- try (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- -- Yield results as they are output to the channel- ei_e_ress_infinite <- getChanContents resultschan- return (ei_e_res1:take thecount ei_e_ress_infinite)---- An alternative implementation of parallel_ might:------ 1. Avoid spawning an additional thread+-- 5. If any of the IO actions throws an exception this does not prevent any of the+-- other actions from being performed. ----- 2. Remove the need for the pause mvar+-- 6. If any of the IO actions throws an exception, the exception thrown by the first+-- failing action in the input list will be thrown by 'parallelInterleaved'. Importantly, at the+-- time the exception is thrown there is no guarantee that the other parallel actions+-- have completed. ----- By having the thread invoking parallel_ also pull stuff from the--- work pool, and poll the count variable after every item to see--- if everything has been processed (which would cause it to stop--- processing work pool items). However:+-- The motivation for this choice is that waiting for the all threads to either return+-- or throw before throwing the first exception will almost always cause GHC to show the+-- "Blocked indefinitely in MVar operation" exception rather than the exception you care about. ----- 1. This is less timely, because the main thread might get stuck--- processing a big work item not related to the current parallel_--- invocation, and wouldn't poll (and return) until that was done.+-- The reason for this behaviour can be seen by considering this machine state: ----- 2. It actually performs a bit less well too - or at least it did on--- my benchmark with lots of cheap actions, where polling would--- be relatively frequent. Went from 8.8s to 9.1s.+-- 1. The main thread has used the parallel combinators to spawn two threads, thread 1 and thread 2.+-- It is blocked on both of them waiting for them to return either a result or an exception via an MVar. ----- For posterity, the implementation was:+-- 2. Thread 1 and thread 2 share another (empty) MVar, the "wait handle". Thread 2 is waiting on the handle,+-- while thread 2 will eventually put into the handle.+-- +-- Consider what happens when thread 1 is buggy and throws an exception before putting into the handle. Now+-- thread 2 is blocked indefinitely, and so the main thread is also blocked indefinetly waiting for the result+-- of thread 2. GHC has no choice but to throw the uninformative exception. However, what we really wanted to+-- see was the original exception thrown in thread 1! ----- @--- parallel_ :: [IO a] -> IO ()--- parallel_ xs | numCapabilities <= 1 = sequence_ xs--- parallel_ [] = return ()--- parallel_ [x] = x >> return ()--- parallel_ (x1:xs) = do--- count <- newMVar $ length xs--- forM_ xs $ \x ->--- enqueueOnPool globalPool $ do--- x--- modifyMVar_ count $ \i -> return (i - 1)--- return False--- x1--- done <- fmap (== 0) $ readMVar count--- unless done $ myWorkerLoop globalPool count--- --- myWorkerLoop :: Pool -> MVar Int -> IO ()--- myWorkerLoop pool count = do--- kill <- join $ readChan (pool_queue pool)--- done <- fmap (== 0) $ readMVar count--- unless (kill || done) (myWorkerLoop pool count)--- @+-- By having the main thread abandon its wait for the results of the spawned threads as soon as the first exception+-- comes in, we give this exception a chance to actually be displayed.+parallelInterleaved :: Pool -> [IO a] -> IO [a]+parallelInterleaved pool acts = mask $ \restore -> do+ main_tid <- myThreadId+ resultchan <- newChan+ forM_ acts $ \act -> do+ _tid <- forkIO $ bracket_ (killPoolWorkerFor pool) (spawnPoolWorkerFor pool) $ reflectExceptionsTo main_tid $ do+ res <- restore act+ writeChan resultchan res+ return ()+ extraWorkerWhileBlocked pool (mapM (\_act -> readChan resultchan) acts)++-- | As 'parallelInterleaved', but instead of throwing exceptions that are thrown by subcomputations,+-- they are returned in a data structure. ----- NB: in this scheme, kill is only True when the program is exiting.+-- As a result, property 6 of 'parallelInterleaved' is not preserved, and therefore if your IO actions can depend on each other+-- and may throw exceptions your program may die with "blocked indefinitely" exceptions rather than informative messages.+parallelInterleavedE :: Pool -> [IO a] -> IO [Either SomeException a]+parallelInterleavedE pool acts = mask $ \restore -> do+ resultchan <- newChan+ forM_ acts $ \act -> do+ _tid <- forkIO $ bracket_ (killPoolWorkerFor pool) (spawnPoolWorkerFor pool) $ do+ ei_e_res <- try (restore act)+ writeChan resultchan ei_e_res+ return ()+ extraWorkerWhileBlocked pool (mapM (\_act -> readChan resultchan) acts)
parallel-io.cabal view
@@ -1,5 +1,5 @@ Name: parallel-io-Version: 0.3.1+Version: 0.3.2 Cabal-Version: >= 1.2 Category: Concurrency Synopsis: Combinators for executing IO actions in parallel on a thread pool.@@ -10,8 +10,8 @@ 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.+ The package is inspired by the thread <http://thread.gmane.org/gmane.comp.lang.haskell.cafe/56499/focus=56521>.+ Thanks to Neil Mitchell and Bulat Ziganshin for some of the code this package is based on. License: BSD3 License-File: LICENSE Homepage: http://batterseapower.github.com/parallel-io@@ -41,7 +41,6 @@ Control.Concurrent.ParallelIO.Local Other-Modules: Control.Concurrent.ParallelIO.Compat- Control.Concurrent.ParallelIO.ConcurrentCollection Build-Depends: base >= 4 && < 5, extensible-exceptions > 0.1.0.1, containers >= 0.2 && < 0.5, random >= 1.0 && < 1.1 @@ -65,7 +64,7 @@ Build-Depends: base >= 4 && < 5, extensible-exceptions > 0.1.0.1, containers >= 0.2 && < 0.5, random >= 1.0 && < 1.1, test-framework >= 0.1.1, test-framework-hunit >= 0.1.1, HUnit >= 1.2 && < 2 - Ghc-Options: -threaded+ Ghc-Options: -threaded -rtsopts Executable fuzz Main-Is: Control/Concurrent/ParallelIO/Fuzz.hs@@ -75,7 +74,7 @@ else Build-Depends: base >= 4 && < 5, extensible-exceptions > 0.1.0.1, containers >= 0.2 && < 0.5, random >= 1.0 && < 1.1 - Ghc-Options: -threaded+ Ghc-Options: -threaded -rtsopts Executable fuzz-seq Main-Is: Control/Concurrent/ParallelIO/Fuzz.hs