parallel-io (empty) → 0.2
raw patch · 8 files changed
+518/−0 lines, 8 filesdep +HUnitdep +basedep +containerssetup-changed
Dependencies added: HUnit, base, containers, extensible-exceptions, random, test-framework, test-framework-hunit, time
Files
- Control/Concurrent/ParallelIO.hs +8/−0
- Control/Concurrent/ParallelIO/Benchmark.hs +27/−0
- Control/Concurrent/ParallelIO/Global.hs +57/−0
- Control/Concurrent/ParallelIO/Local.hs +263/−0
- Control/Concurrent/ParallelIO/Tests.hs +78/−0
- LICENSE +22/−0
- Setup.lhs +4/−0
- parallel-io.cabal +59/−0
+ Control/Concurrent/ParallelIO.hs view
@@ -0,0 +1,8 @@+module Control.Concurrent.ParallelIO (+ module Control.Concurrent.ParallelIO.Global+ ) where++-- By default, just export the user-friendly Global interface.+-- Those who want more power can import Local explicitly.+import Control.Concurrent.ParallelIO.Global+
+ Control/Concurrent/ParallelIO/Benchmark.hs view
@@ -0,0 +1,27 @@+module Main where++import Data.IORef+import Data.Time.Clock++import Control.Concurrent.ParallelIO.Global+++n :: Int+n = 1000000++main :: IO ()+main = do+ r <- newIORef (0 :: Int)+ let incRef = atomicModifyIORef r (\a -> (a, a))+ time $ parallel_ $ replicate n $ incRef+ v <- readIORef r+ stopGlobalPool+ print v++time :: IO a -> IO a+time action = do+ start <- getCurrentTime+ result <- action+ stop <- getCurrentTime+ print $ stop `diffUTCTime` start+ return result
+ Control/Concurrent/ParallelIO/Global.hs view
@@ -0,0 +1,57 @@+-- | Parallelism combinators with an implicit global thread-pool.+--+-- The most basic example of usage is:+--+-- > main = parallel_ [putStrLn "Echo", putStrLn " in parallel"] >> stopGlobalPool+--+-- Make sure that you compile with @-threaded@ and supply @+RTS -N2 -RTS@+-- to the generated Haskell executable, or you won't get any parallelism.+--+-- The "Control.Concurrent.ParallelIO.Local" module provides a more general+-- interface which allows explicit passing of pools and control of their size.+-- 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,+ + parallel_, parallel, parallelInterleaved+ ) where++import GHC.Conc++import System.IO.Unsafe++import qualified Control.Concurrent.ParallelIO.Local as L+++{-# NOINLINE globalPool #-}+globalPool :: L.Pool+globalPool = unsafePerformIO $ L.startPool numCapabilities++-- | In order to reliably make use of the global parallelism combinators,+-- you must invoke this function after all calls to those combinators have+-- finished. A good choice might be at the end of 'main'.+--+-- See also 'L.stopPool'.+stopGlobalPool :: IO ()+stopGlobalPool = L.stopPool globalPool++-- | Execute the given actions in parallel on the global thread pool.+--+-- See also 'L.parallel_'.+parallel_ :: [IO a] -> IO ()+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.+--+-- See also 'L.parallel'.+parallel :: [IO a] -> IO [a]+parallel = L.parallel globalPool++-- | Execute the given actions in parallel on the global thread pool,+-- returning the results in the approximate order of completion.+--+-- See also 'L.parallelInterleaved'.+parallelInterleaved :: [IO a] -> IO [a]+parallelInterleaved = L.parallelInterleaved globalPool
+ Control/Concurrent/ParallelIO/Local.hs view
@@ -0,0 +1,263 @@+{-# LANGUAGE ScopedTypeVariables #-}+-- | Parallelism combinators with explicit thread-pool creation and+-- passing.+--+-- The most basic example of usage is:+--+-- > main = withPool 2 $ \pool -> parallel_ pool [putStrLn "Echo", putStrLn " in parallel"]+--+-- Make sure that you compile with @-threaded@ and supply @+RTS -N2 -RTS@+-- to the generated Haskell executable, or you won't get any parallelism.+--+-- The "Control.Concurrent.ParallelIO.Global" module is implemented+-- on top of this one by maintaining a shared global thread pool+-- with one thread per capability.+module Control.Concurrent.ParallelIO.Local (+ WorkItem, WorkQueue, Pool,+ withPool, startPool, stopPool,+ enqueueOnPool, spawnPoolWorkerFor,+ + parallel_, parallel, parallelInterleaved+ ) where++import qualified Control.Concurrent.ParallelIO.ConcurrentSet as CS++import Control.Concurrent+import Control.Exception.Extensible as E+import Control.Monad++import System.IO+++-- | Type of work items you can put onto the queue. The 'Bool'+-- returned from the 'IO' action specifies whether the invoking+-- thread should terminate itself immediately.+type WorkItem = IO Bool++-- | A 'WorkQueue' is used to communicate 'WorkItem's to the workers.+type WorkQueue = CS.ConcurrentSet WorkItem++-- | The type of thread pools used by 'ParallelIO'.+-- The best way to construct one of these is using 'withPool'.+data Pool = Pool {+ pool_threadcount :: Int,+ pool_spawnedby :: ThreadId,+ pool_queue :: WorkQueue+ }++-- | A slightly unsafe way to construct a pool. Make a pool from the maximum+-- number of threads you wish it to execute (including the main thread+-- in the count).+-- +-- If you use this variant then ensure that you insert a call to 'stopPool'+-- somewhere in your program after all users of that pool have finished.+--+-- A better alternative is to see if you can use the 'withPool' variant.+startPool :: Int -> IO Pool+startPool threadcount = do+ threadId <- myThreadId+ queue <- CS.new+ let pool = Pool {+ pool_threadcount = threadcount,+ pool_spawnedby = threadId,+ pool_queue = queue+ }+ + replicateM_ (threadcount - 1) (spawnPoolWorkerFor pool)+ return pool++-- | Clean up a thread pool. If you don't call this then no one holds the queue,+-- the queue gets GC'd, the threads find themselves blocked indefinitely, and you get+-- exceptions.+-- +-- This cleanly shuts down the threads so the queue isn't important and you don't get+-- exceptions.+--+-- 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++-- | 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+++-- | Internal method for scheduling work on a pool.+enqueueOnPool :: Pool -> WorkItem -> IO ()+enqueueOnPool pool = CS.insert (pool_queue pool)++-- | Internal method for adding extra unblocked threads to a pool if one is going to be+-- temporarily blocked.+spawnPoolWorkerFor :: Pool -> IO ()+spawnPoolWorkerFor pool = do+ _ <- forkIO $ workerLoop `E.catch` \(e :: E.SomeException) -> do+ hPutStrLn stderr $ "Exception on thread: " ++ show e+ throwTo (pool_spawnedby pool) $ ErrorCall $ "Control.Concurrent.ParallelIO: parallel thread died.\n" ++ show e+ return ()+ where+ workerLoop :: IO ()+ workerLoop = do+ kill <- join $ CS.delete (pool_queue pool)+ unless kill workerLoop+++-- | Run the list of computations in parallel.+--+-- 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 'parallel_'.+-- This should minimize contention and hence pre-emption, while also preventing+-- starvation.+--+-- 2. On return all actions have been performed.+--+-- 3. The function returns in a timely manner as soon as all actions have+-- been performed.+--+-- 4. The above properties are true even if 'parallel_' is used by an+-- action which is itself being executed by 'parallel_'.+parallel_ :: Pool -> [IO a] -> IO ()+parallel_ _ [] = return ()+parallel_ pool xs | pool_threadcount pool <= 1 = sequence_ xs+parallel_ _ [x] = x >> return ()+parallel_ pool (x1:xs) = do+ count <- newMVar $ length xs+ pause <- newEmptyMVar+ forM_ xs $ \x ->+ enqueueOnPool pool $ do+ _ <- x+ modifyMVar count $ \i -> do+ let i' = i - 1+ kill = i' == 0+ when kill $ putMVar pause ()+ return (i', kill)+ _ <- 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++-- | Run the list of computations in parallel, returning the results in the+-- same order as the corresponding actions.+--+-- 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 'parallel_'.+-- This should minimize contention and hence pre-emption, while also preventing+-- starvation.+--+-- 2. On return all actions have been performed.+--+-- 3. The function returns in a timely manner as soon as all actions have+-- been performed.+--+-- 4. The above properties are true even if 'parallel' is used by an+-- action which is itself being executed by 'parallel'.+parallel :: Pool -> [IO a] -> IO [a]+parallel _ [] = return []+parallel pool xs | pool_threadcount pool <= 1 = sequence xs+parallel _ [x] = fmap return x+parallel pool (x1:xs) = do+ count <- newMVar $ length xs+ resultvars <- forM xs $ \x -> do+ resultvar <- newEmptyMVar+ enqueueOnPool pool $ do+ x >>= putMVar resultvar+ modifyMVar count $ \i -> let i' = i - 1 in return (i', i' == 0)+ return resultvar+ result1 <- 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++-- | Run the list of computations in parallel, returning the results in the+-- approximate order of completion.+--+-- 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 'parallel_'.+-- This should minimize contention and hence pre-emption, while also preventing+-- starvation.+--+-- 2. On return all actions have been performed.+--+-- 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+-- action which is itself being executed by 'parallelInterleaved'.+parallelInterleaved :: Pool -> [IO a] -> IO [a]+parallelInterleaved _ [] = return []+parallelInterleaved pool xs | pool_threadcount pool <= 1 = sequence xs+parallelInterleaved _ [x] = fmap return x+parallelInterleaved pool (x1:xs) = do+ let thecount = length xs+ count <- newMVar $ thecount+ resultschan <- newChan+ forM_ xs $ \x -> do+ enqueueOnPool pool $ do+ x >>= writeChan resultschan+ modifyMVar count $ \i -> let i' = i - 1 in return (i', i' == 0)+ result1 <- 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++-- An alternative implementation of parallel_ might:+--+-- 1. Avoid spawning an additional thread+--+-- 2. Remove the need for the pause mvar+--+-- 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:+--+-- 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.+--+-- 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.+--+-- For posterity, the implementation was:+--+-- @+-- 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)+-- @+--+-- NB: in this scheme, kill is only True when the program is exiting.
+ Control/Concurrent/ParallelIO/Tests.hs view
@@ -0,0 +1,78 @@+{-# OPTIONS_GHC -fno-warn-missing-signatures #-}+module Main where++import Data.IORef+import Data.List++import Test.Framework+import Test.Framework.Providers.HUnit+import Test.HUnit ((@?))++import GHC.Conc++import Control.Monad++import qualified Control.Concurrent.ParallelIO.Global as Global+import Control.Concurrent.ParallelIO.Local+++main :: IO ()+main = do+ defaultMain tests+ Global.stopGlobalPool++tests :: [Test]+tests = [ testCase "parallel_ executes correct number of actions" $ repeatTest parallel__execution_count_correct+ , testCase "parallel_ doesn't spawn too many threads" $ repeatTest parallel__doesnt_spawn_too_many_threads+ , testCase "parallel executes correct actions" $ repeatTest parallel_executes_correct_actions+ , testCase "parallel doesn't spawn too many threads" $ repeatTest parallel_doesnt_spawn_too_many_threads+ , testCase "parallelInterleaved executes correct actions" $ repeatTest parallelInterleaved_executes_correct_actions+ , testCase "parallelInterleaved doesn't spawn too many threads" $ repeatTest parallelInterleaved_doesnt_spawn_too_many_threads+ ]++parallel__execution_count_correct n = do+ ref <- newIORef 0+ Global.parallel_ (replicate n (atomicModifyIORef_ ref (+ 1)))+ fmap (==n) $ readIORef ref++parallel_executes_correct_actions n = fmap (expected ==) actual+ where actual = Global.parallel (map (return . (+1)) [0..n])+ expected = [(1 :: Int)..n + 1]++parallelInterleaved_executes_correct_actions n = fmap ((expected ==) . sort) actual+ where actual = Global.parallelInterleaved (map (return . (+1)) [0..n])+ expected = [(1 :: Int)..n + 1]++parallel__doesnt_spawn_too_many_threads = doesnt_spawn_too_many_threads parallel_+parallel_doesnt_spawn_too_many_threads = doesnt_spawn_too_many_threads parallel+parallelInterleaved_doesnt_spawn_too_many_threads = doesnt_spawn_too_many_threads parallelInterleaved++doesnt_spawn_too_many_threads the_parallel n = do+ threadcountref <- newIORef 0+ maxref <- newIORef 0+ -- NB: we use a local pool rather than the global one because otherwise we get interference effects+ -- when we run the testsuite in parallel+ withPool numCapabilities $ \pool -> do+ _ <- the_parallel pool $ replicate n $ do+ tc' <- atomicModifyIORef_ threadcountref (+ 1)+ _ <- atomicModifyIORef_ maxref (`max` tc')+ -- This delay and 'yield' combination was experimentally determined. The test+ -- can and does still nondeterministically fail with a non-zero probability+ -- dependening on runtime scheduling behaviour. It seems that the first instance+ -- of this test to run in the process is especially vulnerable.+ yield+ threadDelay 20000+ yield+ atomicModifyIORef_ threadcountref (\tc -> tc - 1)+ seenmax <- readIORef maxref+ let expected_max_concurrent_threads = numCapabilities `min` n+ if expected_max_concurrent_threads == seenmax+ then return True+ else putStrLn ("Expected at most " ++ show expected_max_concurrent_threads ++ ", got " ++ show seenmax) >> return False+++atomicModifyIORef_ :: IORef a -> (a -> a) -> IO a+atomicModifyIORef_ ref f = atomicModifyIORef ref (\x -> let x' = f x in x' `seq` (x', x'))++repeatTest :: (Int -> IO Bool) -> IO ()+repeatTest testcase = forM_ [0..100] $ \n -> testcase n @? "n=" ++ show n
+ LICENSE view
@@ -0,0 +1,22 @@+Copyright (c) 2008, Maximilian Bolingbroke+All rights reserved.++Redistribution and use in source and binary forms, with or without modification, are permitted+provided that the following conditions are met:++ * Redistributions of source code must retain the above copyright notice, this list of+ conditions and the following disclaimer.+ * Redistributions in binary form must reproduce the above copyright notice, this list of+ conditions and the following disclaimer in the documentation and/or other materials+ provided with the distribution.+ * Neither the name of Maximilian Bolingbroke nor the names of other contributors may be used to+ endorse or promote products derived from this software without specific prior written permission.++THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND ANY EXPRESS OR+IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND+FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR+CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL+DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,+DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER+IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT+OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ Setup.lhs view
@@ -0,0 +1,4 @@+#! /usr/bin/env runhaskell++> import Distribution.Simple+> main = defaultMain
+ parallel-io.cabal view
@@ -0,0 +1,59 @@+Name: parallel-io+Version: 0.2+Cabal-Version: >= 1.2+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+ thread pool is guaranteed to contain a fixed number of unblocked threads, minimizing+ contention. Furthermore, the parallel combinators can be used re-entrently - your parallel+ actions can spawn more parallel actions - without violating this property.+ .+ 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.+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>+Maintainer: Max Bolingbroke <batterseapower@hotmail.com>+Build-Type: Simple+++Flag Benchmark+ Description: Build the benchmarking tool+ Default: False++Flag Tests+ Description: Build the test runner+ Default: False++Library+ Exposed-Modules:+ Control.Concurrent.ParallelIO+ Control.Concurrent.ParallelIO.Global+ Control.Concurrent.ParallelIO.Local+ + Build-Depends: base >= 3 && < 5, extensible-exceptions > 0.1.0.1, containers >= 0.3 && < 0.4, random >= 1.0 && < 1.1++Executable benchmark+ Main-Is: Control/Concurrent/ParallelIO/Benchmark.hs+ + if !flag(benchmark)+ Buildable: False+ else+ Build-Depends: base >= 3 && < 5, extensible-exceptions > 0.1.0.1, containers >= 0.3 && < 0.4, random >= 1.0 && < 1.1,+ time >= 1+ + Ghc-Options: -threaded++Executable tests+ Main-Is: Control/Concurrent/ParallelIO/Tests.hs+ + if !flag(tests)+ Buildable: False+ else+ Build-Depends: base >= 3 && < 5, extensible-exceptions > 0.1.0.1, containers >= 0.3 && < 0.4, random >= 1.0 && < 1.1,+ test-framework >= 0.1.1, test-framework-hunit >= 0.1.1, HUnit >= 1.2 && < 2+ + Ghc-Options: -threaded