diff --git a/Control/Concurrent/ParallelIO/ConcurrentCollection.hs b/Control/Concurrent/ParallelIO/ConcurrentCollection.hs deleted file mode 100644 --- a/Control/Concurrent/ParallelIO/ConcurrentCollection.hs +++ /dev/null @@ -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 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 @@ -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) 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.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 . - Thanks to Neil Mitchell and Bulat Ziganshin for the code this package is based on. + The package is inspired by the thread . + 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