parallel-io 0.3.0.2 → 0.3.1
raw patch · 5 files changed
+176/−101 lines, 5 filesdep ~containersnew-component:exe:fuzz-seq
Dependency ranges changed: containers
Files
- Control/Concurrent/ParallelIO/Compat.hs +16/−0
- Control/Concurrent/ParallelIO/ConcurrentCollection.hs +36/−18
- Control/Concurrent/ParallelIO/Global.hs +29/−1
- Control/Concurrent/ParallelIO/Local.hs +81/−76
- parallel-io.cabal +14/−6
+ Control/Concurrent/ParallelIO/Compat.hs view
@@ -0,0 +1,16 @@+{-# LANGUAGE CPP #-}+module Control.Concurrent.ParallelIO.Compat (+ mask, mask_+ ) where++#if MIN_VERSION_base(4,3,0)+import Control.Exception ( mask, 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++mask_ :: IO a -> IO a+mask_ io = mask $ \_ -> io+#endif
Control/Concurrent/ParallelIO/ConcurrentCollection.hs view
@@ -2,6 +2,8 @@ ConcurrentSet, Chan, ConcurrentCollection(..) ) where +import Control.Concurrent.ParallelIO.Compat+ import Control.Concurrent.MVar import Control.Concurrent.Chan import Control.Monad@@ -32,44 +34,60 @@ -- 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, Either (MVar ()) (IM.IntMap a)))+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, Left mvar)) newStdGen newEmptyMVar >>= newMVar+ 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, ei_mvar_ys) = do+ where go (gen, contents) = do let (i, gen') = random gen- case ei_mvar_ys of- Left wait_mvar -> do+ case contents of+ EmptyWithWaiters 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))+ --+ -- 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- ei_wait_x <- modifyMVar set_mvar go- case ei_wait_x of- Left wait_mvar -> 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!- takeMVar wait_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- Right x -> return x+ NonEmpty x -> return x - go (gen, Left wait_mvar) = return ((gen, Left wait_mvar), Left wait_mvar)- go (gen, Right xs) = do+ 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 Left newEmptyMVar- else return (Right xs')- return ((gen, new_value), Right chosen)+ then fmap EmptyWithWaiters newEmptyMVar+ else return (NonEmpty xs')+ return ((gen, new_value), NonEmpty chosen) instance ConcurrentCollection Chan where
Control/Concurrent/ParallelIO/Global.hs view
@@ -15,7 +15,7 @@ -- pool with one thread per capability. module Control.Concurrent.ParallelIO.Global ( -- * Executing actions- parallel_, parallel, parallelInterleaved,+ parallel_, parallelE_, parallel, parallelE, parallelInterleaved, parallelInterleavedE, -- * Global pool management globalPool, stopGlobalPool,@@ -27,6 +27,8 @@ import GHC.Conc +import Control.Exception+ import System.IO.Unsafe import qualified Control.Concurrent.ParallelIO.Local as L@@ -79,6 +81,14 @@ parallel_ :: [IO a] -> IO () parallel_ = L.parallel_ globalPool +-- | Execute the given actions in parallel on the global thread pool, reporting exceptions explicitly.+--+-- Users of the global pool must call 'stopGlobalPool' from the main thread at the end of their program.+--+-- See also 'L.parallelE_'.+parallelE_ :: [IO a] -> IO [Maybe SomeException]+parallelE_ = L.parallelE_ globalPool+ -- | Execute the given actions in parallel on the global thread pool, -- returning the results in the same order as the corresponding actions. --@@ -89,6 +99,15 @@ parallel = L.parallel globalPool -- | Execute the given actions in parallel on the global thread pool,+-- returning the results in the same order as the corresponding actions and reporting exceptions explicitly.+--+-- Users of the global pool must call 'stopGlobalPool' from the main thread at the end of their program.+--+-- See also 'L.parallelE'.+parallelE :: [IO a] -> IO [Either SomeException a]+parallelE = L.parallelE 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.@@ -96,3 +115,12 @@ -- See also 'L.parallelInterleaved'. parallelInterleaved :: [IO a] -> IO [a] parallelInterleaved = L.parallelInterleaved globalPool++-- | Execute the given actions in parallel on the global thread pool,+-- returning the results in the approximate order of completion and reporting exceptions explicitly.+--+-- Users of the global pool must call 'stopGlobalPool' from the main thread at the end of their program.+--+-- See also 'L.parallelInterleavedE'.+parallelInterleavedE :: [IO a] -> IO [Either SomeException a]+parallelInterleavedE = L.parallelInterleavedE globalPool
Control/Concurrent/ParallelIO/Local.hs view
@@ -1,4 +1,4 @@-{-# LANGUAGE CPP, ScopedTypeVariables #-}+{-# LANGUAGE ScopedTypeVariables #-} -- | Parallelism combinators with explicit thread-pool creation and -- passing. --@@ -16,7 +16,7 @@ -- with one thread per capability. module Control.Concurrent.ParallelIO.Local ( -- * Executing actions- parallel_, parallel, parallelInterleaved,+ parallel_, parallelE_, parallel, parallelE, parallelInterleaved, parallelInterleavedE, -- * Pool management Pool, withPool, startPool, stopPool,@@ -26,23 +26,18 @@ spawnPoolWorkerFor, killPoolWorkerFor ) where +import Control.Concurrent.ParallelIO.Compat import qualified Control.Concurrent.ParallelIO.ConcurrentCollection as CC import Control.Concurrent-import Control.Exception.Extensible as E+import Control.Exception import Control.Monad -import System.IO-+import Data.IORef -#if MIN_VERSION_base(4,3,0)-import Control.Exception ( mask )-#else-import Control.Exception ( blocked, block, unblock )+import System.IO -mask :: ((IO a -> IO a) -> IO b) -> IO b-mask io = blocked >>= \b -> if b then io id else block $ io unblock-#endif+import Prelude hiding (catch) -- TODO: I should deal nicely with exceptions raised by the actions on other threads.@@ -57,14 +52,14 @@ -- | 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---- FIXME: I saw deadlocks very quickly with the fuzzer using ConcurrentSet.--- Is ConcurrentSet incorrect, or was it exposing a bug here?---type WorkQueue = CC.ConcurrentSet WorkItem+--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'.@@ -111,7 +106,7 @@ -- | 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. withPool :: Int -> (Pool -> IO a) -> IO a-withPool threadcount = E.bracket (startPool threadcount) stopPool+withPool threadcount = bracket (startPool threadcount) stopPool -- | Internal method for scheduling work on a pool.@@ -140,14 +135,14 @@ -- -- > 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)+extraWorkerWhileBlocked pool wait = bracket (spawnPoolWorkerFor pool) (\() -> killPoolWorkerFor pool) (\() -> wait) -- | 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.+-- 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 `E.catch` \(e :: E.SomeException) -> 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@@ -159,7 +154,10 @@ --hPutStrLn stderr $ "[waiting] " ++ show tid work_item <- CC.delete (pool_queue pool) --hPutStrLn stderr $ "[working] " ++ show tid- kill <- work_item+ + -- 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 -- | Internal method for removing threads from a pool after one of the threads on the pool@@ -186,33 +184,16 @@ -- 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, undefined behaviour will result.--- If you want safety, wrap your actions in 'Control.Exception.try'.+-- 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_'. parallel_ :: Pool -> [IO a] -> IO ()-parallel_ _ [] = 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.---parallel_ pool xs | pool_threadcount pool <= 1 = sequence_ xs-parallel_ _ [x] = x >> return ()-parallel_ pool (x1:xs) = mask $ \restore -> do- count <- newMVar $ length xs- pause <- newEmptyMVar- forM_ xs $ \x ->- enqueueOnPool pool $ do- _ <- restore x- modifyMVar count $ \i -> do- let i' = i - 1- kill = i' == 0- when kill $ {- putStrLn "Natural death" >> -} putMVar pause ()- return (i', kill)- _ <- 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- takeMVar pause+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.+parallelE_ :: Pool -> [IO a] -> IO [Maybe SomeException]+parallelE_ pool = fmap (map (either Just (\_ -> Nothing))) . parallelE pool+ -- | Run the list of computations in parallel, returning the results in the -- same order as the corresponding actions. --@@ -231,26 +212,40 @@ -- 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, undefined behaviour will result.--- If you want safety, wrap your actions in 'Control.Exception.try'.+-- 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'. 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) = mask $ \restore -> do- count <- newMVar $ length xs+parallel pool xs = do+ ei_e_ress <- parallelE pool xs+ mapM (either throw return) ei_e_ress++-- | As 'parallel', but instead of throwing exceptions that are thrown by subcomputations,+-- they are returned in a data structure.+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 resultvar <- newEmptyMVar enqueueOnPool pool $ do- restore x >>= putMVar resultvar- modifyMVar count $ \i -> let i' = i - 1 in return (i', i' == 0)+ ei_e_res <- try (restore x)+ -- 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 resultvar- result1 <- restore x1+ 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 (result1:) $ mapM takeMVar resultvars+ fmap (ei_e_res1:) $ mapM takeMVar resultvars -- | Run the list of computations in parallel, returning the results in the -- approximate order of completion.@@ -270,32 +265,42 @@ -- 3. 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, undefined behaviour will result.--- If you want safety, wrap your actions in 'Control.Exception.try'.+-- 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 _ [] = return []--- It is important that we do not include this special case (see parallel_ for why)+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-parallelInterleaved _ [x] = fmap return x-parallelInterleaved pool (x1:xs) = mask $ \restore -> do+parallelInterleavedE_lazy _ [x] = fmap return (try x)+parallelInterleavedE_lazy pool (x1:xs) = mask $ \restore -> do let thecount = length xs- count <- newMVar $ thecount+ count <- newIORef (length xs) resultschan <- newChan forM_ xs $ \x -> do enqueueOnPool pool $ do- restore x >>= writeChan resultschan- modifyMVar count $ \i -> let i' = i - 1 in return (i', i' == 0)- result1 <- restore x1+ 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- results <- fmap ((result1:) . take thecount) $ getChanContents resultschan- return $ seqList results--seqList :: [a] -> [a]-seqList [] = []-seqList (x:xs) = x `seq` xs' `seq` (x:xs')- where xs' = seqList xs+ -- 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: --
parallel-io.cabal view
@@ -1,5 +1,5 @@ Name: parallel-io-Version: 0.3.0.2+Version: 0.3.1 Cabal-Version: >= 1.2 Category: Concurrency Synopsis: Combinators for executing IO actions in parallel on a thread pool.@@ -39,7 +39,8 @@ Control.Concurrent.ParallelIO Control.Concurrent.ParallelIO.Global Control.Concurrent.ParallelIO.Local- Other-Modules:+ 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@@ -50,7 +51,7 @@ if !flag(benchmark) Buildable: False else- Build-Depends: base >= 4 && < 5, extensible-exceptions > 0.1.0.1, containers >= 0.2 && < 0.4, random >= 1.0 && < 1.1,+ Build-Depends: base >= 4 && < 5, extensible-exceptions > 0.1.0.1, containers >= 0.2 && < 0.5, random >= 1.0 && < 1.1, time >= 1 Ghc-Options: -threaded@@ -61,7 +62,7 @@ if !flag(tests) Buildable: False else- Build-Depends: base >= 4 && < 5, extensible-exceptions > 0.1.0.1, containers >= 0.2 && < 0.4, random >= 1.0 && < 1.1,+ 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@@ -72,7 +73,14 @@ if !flag(fuzz) Buildable: False else- Build-Depends: base >= 4 && < 5, extensible-exceptions > 0.1.0.1, containers >= 0.2 && < 0.4, random >= 1.0 && < 1.1,- test-framework >= 0.1.1, test-framework-hunit >= 0.1.1, HUnit >= 1.2 && < 2+ Build-Depends: base >= 4 && < 5, extensible-exceptions > 0.1.0.1, containers >= 0.2 && < 0.5, random >= 1.0 && < 1.1 Ghc-Options: -threaded++Executable fuzz-seq+ Main-Is: Control/Concurrent/ParallelIO/Fuzz.hs++ if !flag(fuzz)+ Buildable: False+ else+ Build-Depends: base >= 4 && < 5, extensible-exceptions > 0.1.0.1, containers >= 0.2 && < 0.5, random >= 1.0 && < 1.1