diff --git a/Control/Concurrent/ParallelIO.hs b/Control/Concurrent/ParallelIO.hs new file mode 100644 --- /dev/null +++ b/Control/Concurrent/ParallelIO.hs @@ -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 + diff --git a/Control/Concurrent/ParallelIO/Benchmark.hs b/Control/Concurrent/ParallelIO/Benchmark.hs new file mode 100644 --- /dev/null +++ b/Control/Concurrent/ParallelIO/Benchmark.hs @@ -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 diff --git a/Control/Concurrent/ParallelIO/Global.hs b/Control/Concurrent/ParallelIO/Global.hs new file mode 100644 --- /dev/null +++ b/Control/Concurrent/ParallelIO/Global.hs @@ -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 diff --git a/Control/Concurrent/ParallelIO/Local.hs b/Control/Concurrent/ParallelIO/Local.hs new file mode 100644 --- /dev/null +++ b/Control/Concurrent/ParallelIO/Local.hs @@ -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. diff --git a/Control/Concurrent/ParallelIO/Tests.hs b/Control/Concurrent/ParallelIO/Tests.hs new file mode 100644 --- /dev/null +++ b/Control/Concurrent/ParallelIO/Tests.hs @@ -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 diff --git a/LICENSE b/LICENSE new file mode 100644 --- /dev/null +++ b/LICENSE @@ -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. diff --git a/Setup.lhs b/Setup.lhs new file mode 100644 --- /dev/null +++ b/Setup.lhs @@ -0,0 +1,4 @@ +#! /usr/bin/env runhaskell + +> import Distribution.Simple +> main = defaultMain diff --git a/parallel-io.cabal b/parallel-io.cabal new file mode 100644 --- /dev/null +++ b/parallel-io.cabal @@ -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 . + 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 , + Bulat Ziganshin , + Max Bolingbroke +Maintainer: Max Bolingbroke +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