parallel-io 0.3.2.2 → 0.3.5
raw patch · 3 files changed
Files
- Control/Concurrent/ParallelIO/Global.hs +21/−1
- Control/Concurrent/ParallelIO/Local.hs +94/−3
- parallel-io.cabal +12/−7
Control/Concurrent/ParallelIO/Global.hs view
@@ -15,7 +15,9 @@ -- pool with one thread per capability. module Control.Concurrent.ParallelIO.Global ( -- * Executing actions- parallel_, parallelE_, parallel, parallelE, parallelInterleaved, parallelInterleavedE,+ parallel_, parallelE_, parallel, parallelE,+ parallelInterleaved, parallelInterleavedE,+ parallelFirst, parallelFirstE, -- * Global pool management globalPool, stopGlobalPool,@@ -124,3 +126,21 @@ -- See also 'L.parallelInterleavedE'. parallelInterleavedE :: [IO a] -> IO [Either SomeException a] parallelInterleavedE = L.parallelInterleavedE globalPool++-- | Run the list of computations in parallel, returning the result of the first+-- thread that completes with (Just x), if any.+--+-- Users of the global pool must call 'stopGlobalPool' from the main thread at the end of their program.+--+-- See also 'L.parallelFirst'.+parallelFirst :: [IO (Maybe a)] -> IO (Maybe a)+parallelFirst = L.parallelFirst globalPool++-- | Run the list of computations in parallel, returning the result of the first+-- thread that completes with (Just x), if any, and reporting any exception explicitly.+--+-- Users of the global pool must call 'stopGlobalPool' from the main thread at the end of their program.+--+-- See also 'L.parallelFirstE'.+parallelFirstE :: [IO (Maybe a)] -> IO (Maybe (Either SomeException a))+parallelFirstE = L.parallelFirstE globalPool
Control/Concurrent/ParallelIO/Local.hs view
@@ -16,7 +16,9 @@ -- with one thread per capability. module Control.Concurrent.ParallelIO.Local ( -- * Executing actions- parallel_, parallelE_, parallel, parallelE, parallelInterleaved, parallelInterleavedE,+ parallel_, parallelE_, parallel, parallelE,+ parallelInterleaved, parallelInterleavedE,+ parallelFirst, parallelFirstE, -- * Pool management Pool, withPool, startPool, stopPool,@@ -36,8 +38,14 @@ import System.IO +catchNonThreadKilled :: IO a -> (SomeException -> IO a) -> IO a+catchNonThreadKilled act handler = act `E.catch` \e -> case fromException e of Just ThreadKilled -> throwIO e; _ -> handler e++onNonThreadKilledException :: IO a -> IO b -> IO a+onNonThreadKilledException act handler = catchNonThreadKilled act (\e -> handler >> throwIO e)+ reflectExceptionsTo :: ThreadId -> IO () -> IO ()-reflectExceptionsTo tid act = act `E.catch` \e -> throwTo tid (e :: SomeException)+reflectExceptionsTo tid act = catchNonThreadKilled act (throwTo tid) -- | A thread pool, containing a maximum number of threads. The best way to@@ -280,7 +288,7 @@ -- 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.+-- while thread 1 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@@ -314,3 +322,86 @@ writeChan resultchan ei_e_res return () extraWorkerWhileBlocked pool (mapM (\_act -> readChan resultchan) acts)++-- | Run the list of computations in parallel, returning the result of the first+-- thread that completes with (Just x), if any+--+-- Has the following properties:+--+-- 1. Never creates more or less unblocked threads than are specified to+-- live in the pool. NB: this count includes the thread executing 'parallelInterleaved'.+-- This should minimize contention and hence pre-emption, while also preventing+-- starvation.+--+-- 2. On return all actions have either been performed or cancelled (with ThreadKilled exceptions).+--+-- 3. The above properties are true even if 'parallelFirst' is used by an+-- action which is itself being executed by one of the parallel combinators.+--+-- 4. If any of the IO actions throws an exception, the exception thrown by the first+-- throwing action in the input list will be thrown by 'parallelFirst'. Importantly, at the+-- time the exception is thrown there is no guarantee that the other parallel actions+-- have been completed or cancelled.+--+-- 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 1 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.+parallelFirst :: Pool -> [IO (Maybe a)] -> IO (Maybe a)+parallelFirst pool acts = mask $ \restore -> do+ main_tid <- myThreadId+ resultvar <- newEmptyMVar+ (tids, waits) <- liftM unzip $ forM acts $ \act -> do+ wait_var <- newEmptyMVar+ tid <- forkIO $ flip onNonThreadKilledException (tryPutMVar resultvar Nothing) $ -- If we throw an exception, unblock+ bracket_ (killPoolWorkerFor pool) (spawnPoolWorkerFor pool >> putMVar wait_var ()) $ -- the main thread so it can rethrow it+ reflectExceptionsTo main_tid $ do+ mb_res <- restore act+ case mb_res of+ Nothing -> return ()+ Just res -> tryPutMVar resultvar (Just res) >> return ()+ return (tid, wait_var)+ forkIO $ mapM_ takeMVar waits >> tryPutMVar resultvar Nothing >> return ()+ mb_res <- extraWorkerWhileBlocked pool (takeMVar resultvar)+ mapM_ killThread tids+ return mb_res++-- | As 'parallelFirst', but instead of throwing exceptions that are thrown by subcomputations,+-- they are returned in a data structure.+--+-- As a result, property 4 of 'parallelFirst' 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.+parallelFirstE :: Pool -> [IO (Maybe a)] -> IO (Maybe (Either SomeException a))+parallelFirstE pool acts = mask $ \restore -> do+ main_tid <- myThreadId+ resultvar <- newEmptyMVar+ (tids, waits) <- liftM unzip $ forM acts $ \act -> do+ wait_var <- newEmptyMVar+ tid <- forkIO $ bracket_ (killPoolWorkerFor pool) (spawnPoolWorkerFor pool >> putMVar wait_var ()) $ do+ ei_mb_res <- try (restore act)+ case ei_mb_res of+ -- NB: we aren't in danger of putting a "thread killed" exception into the MVar+ -- since we only kill the spawned threads *after* we have already taken from resultvar+ Left e -> tryPutMVar resultvar (Just (Left e)) >> return ()+ Right Nothing -> return ()+ Right (Just res) -> tryPutMVar resultvar (Just (Right res)) >> return ()+ return (tid, wait_var)+ forkIO $ mapM_ takeMVar waits >> tryPutMVar resultvar Nothing >> return ()+ mb_res <- extraWorkerWhileBlocked pool (takeMVar resultvar)+ mapM_ killThread tids+ return mb_res
parallel-io.cabal view
@@ -1,6 +1,6 @@ Name: parallel-io-Version: 0.3.2.2-Cabal-Version: >= 1.2+Version: 0.3.5+Cabal-Version: >= 1.10 Category: Concurrency Synopsis: Combinators for executing IO actions in parallel on a thread pool. Description: This package provides combinators for sequencing IO actions onto a thread pool. The@@ -35,6 +35,7 @@ Default: False Library+ Default-Language: Haskell98 Exposed-Modules: Control.Concurrent.ParallelIO Control.Concurrent.ParallelIO.Global@@ -42,44 +43,48 @@ Other-Modules: Control.Concurrent.ParallelIO.Compat - Build-Depends: base >= 4 && < 5, extensible-exceptions > 0.1.0.1, containers >= 0.2 && < 0.6, random >= 1.0 && < 1.1+ Build-Depends: base >= 4 && < 5, extensible-exceptions > 0.1.0.1, containers >= 0.2 && < 0.7, random >= 1.0 && < 1.3 Executable benchmark+ Default-Language: Haskell98 Main-Is: Control/Concurrent/ParallelIO/Benchmark.hs if !flag(benchmark) Buildable: False else- Build-Depends: base >= 4 && < 5, extensible-exceptions > 0.1.0.1, containers >= 0.2 && < 0.6, random >= 1.0 && < 1.1,+ Build-Depends: base >= 4 && < 5, extensible-exceptions > 0.1.0.1, containers >= 0.2 && < 0.7, random >= 1.0 && < 1.3, time >= 1 Ghc-Options: -threaded Executable tests+ Default-Language: Haskell98 Main-Is: Control/Concurrent/ParallelIO/Tests.hs if !flag(tests) Buildable: False else- Build-Depends: base >= 4 && < 5, extensible-exceptions > 0.1.0.1, containers >= 0.2 && < 0.6, random >= 1.0 && < 1.1,+ Build-Depends: base >= 4 && < 5, extensible-exceptions > 0.1.0.1, containers >= 0.2 && < 0.7, random >= 1.0 && < 1.3, test-framework >= 0.1.1, test-framework-hunit >= 0.1.1, HUnit >= 1.2 && < 2 Ghc-Options: -threaded -rtsopts Executable fuzz+ Default-Language: Haskell98 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.6, random >= 1.0 && < 1.1+ Build-Depends: base >= 4 && < 5, extensible-exceptions > 0.1.0.1, containers >= 0.2 && < 0.7, random >= 1.0 && < 1.3 Ghc-Options: -threaded -rtsopts Executable fuzz-seq+ Default-Language: Haskell98 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.6, random >= 1.0 && < 1.1+ Build-Depends: base >= 4 && < 5, extensible-exceptions > 0.1.0.1, containers >= 0.2 && < 0.7, random >= 1.0 && < 1.3