parallel-io 0.3.0.1 → 0.3.0.2
raw patch · 5 files changed
+172/−78 lines, 5 filesdep ~containersnew-component:exe:fuzz
Dependency ranges changed: containers
Files
- Control/Concurrent/ParallelIO/ConcurrentCollection.hs +78/−0
- Control/Concurrent/ParallelIO/ConcurrentSet.hs +0/−67
- Control/Concurrent/ParallelIO/Fuzz.hs +59/−0
- Control/Concurrent/ParallelIO/Local.hs +18/−9
- parallel-io.cabal +17/−2
+ Control/Concurrent/ParallelIO/ConcurrentCollection.hs view
@@ -0,0 +1,78 @@+module Control.Concurrent.ParallelIO.ConcurrentCollection (+ ConcurrentSet, Chan, ConcurrentCollection(..)+ ) where++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, Either (MVar ()) (IM.IntMap a)))++instance ConcurrentCollection ConcurrentSet where+ new = fmap CS $ liftM2 (\gen mvar -> (gen, Left mvar)) newStdGen newEmptyMVar >>= newMVar++ insert (CS set_mvar) x = modifyMVar_ set_mvar go+ where go (gen, ei_mvar_ys) = do+ let (i, gen') = random gen+ case ei_mvar_ys of+ Left 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))++ delete (CS set_mvar) = loop+ where+ loop = do+ ei_wait_x <- modifyMVar set_mvar go+ case ei_wait_x of+ Left wait_mvar -> do+ -- NB: it's very important that we don't do this while we are holding the set_mvar!+ takeMVar 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+ + go (gen, Left wait_mvar) = return ((gen, Left wait_mvar), Left wait_mvar)+ go (gen, Right 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)+++instance ConcurrentCollection Chan where+ new = newChan+ insert = writeChan+ delete = readChan
− Control/Concurrent/ParallelIO/ConcurrentSet.hs
@@ -1,67 +0,0 @@--- | 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.-module Control.Concurrent.ParallelIO.ConcurrentSet (- ConcurrentSet, new, insert, delete- ) where--import Control.Concurrent.MVar-import Control.Monad--import qualified Data.IntMap as IM--import System.Random---data ConcurrentSet a = CS (MVar (StdGen, Either (MVar ()) (IM.IntMap a)))--new :: IO (ConcurrentSet a)-new = fmap CS $ liftM2 (\gen mvar -> (gen, Left mvar)) newStdGen newEmptyMVar >>= newMVar--insert :: ConcurrentSet a -> a -> IO ()-insert (CS set_mvar) x = modifyMVar_ set_mvar go- where go (gen, ei_mvar_ys) = do- let (i, gen') = random gen- case ei_mvar_ys of- Left 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))--delete :: ConcurrentSet a -> IO a-delete (CS set_mvar) = loop- where- loop = do- ei_wait_x <- modifyMVar set_mvar go- case ei_wait_x of- Left wait_mvar -> do- -- NB: it's very important that we don't do this while we are holding the set_mvar!- takeMVar 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- - go (gen, Left wait_mvar) = return ((gen, Left wait_mvar), Left wait_mvar)- go (gen, Right 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)
+ Control/Concurrent/ParallelIO/Fuzz.hs view
@@ -0,0 +1,59 @@+module Main where++import Data.IORef+import qualified Numeric++import System.Random++import Control.Concurrent+import Control.Concurrent.ParallelIO.Local++import Control.Monad+++-- | Range for number of threads to spawn+sPAWN_RANGE = (0, 100)++-- | Delay range in microseconds+dELAY_RANGE = (0, 1000000)++-- | Out of 100 parallel actions, how many should recursively spawn?+sPAWN_PERCENTAGE :: Int+sPAWN_PERCENTAGE = 2++-- | Number of threads to have processing work items+mAX_WORKERS = 3+++showFloat :: RealFloat a => a -> String+showFloat x = Numeric.showFFloat (Just 2) x ""+++expected :: Fractional b => (Int, Int) -> b+expected (top, bottom) = fromIntegral (top + bottom) / 2++main :: IO ()+main = do+ -- Birth rate is the rate at which new work items enter the queue+ putStrLn $ "Expected birth rate: " ++ showFloat ((expected sPAWN_RANGE * (fromIntegral sPAWN_PERCENTAGE / 100) * fromIntegral mAX_WORKERS) / (expected dELAY_RANGE / 1000000) :: Double) ++ " items/second"+ -- Service rate is the rate at which work items are removed from the pool+ putStrLn $ "Expected service rate: " ++ showFloat (fromIntegral mAX_WORKERS / (expected dELAY_RANGE / 1000000) :: Double) ++ " items/second"+ -- We are balanced on average if birth rate == service rate, i.e. expected sPAWN_RANGE * (fromIntegral sPAWN_PERCENTAGE / 100) == 1+ putStrLn $ "Balance factor (should be 1): " ++ showFloat (expected sPAWN_RANGE * (fromIntegral sPAWN_PERCENTAGE / 100) :: Double)+ withPool mAX_WORKERS $ \pool -> forever (fuzz pool)++fuzz pool = do+ n <- randomRIO sPAWN_RANGE+ tid <- myThreadId+ putStrLn $ show tid ++ ":\t" ++ show n+ parallel_ pool $ flip map [1..n] $ \i -> do+ should_spawn <- fmap (<= sPAWN_PERCENTAGE) $ randomRIO (1, 100)+ nested_tid <- myThreadId+ + putStrLn $ show nested_tid ++ ":\trunning " ++ show i ++ if should_spawn then " (recursing)" else ""+ + randomRIO dELAY_RANGE >>= threadDelay+ + putStrLn $ show nested_tid ++ ":\twoke up"+ + when should_spawn $ fuzz pool
Control/Concurrent/ParallelIO/Local.hs view
@@ -26,7 +26,7 @@ spawnPoolWorkerFor, killPoolWorkerFor ) where -import qualified Control.Concurrent.ParallelIO.ConcurrentSet as CS+import qualified Control.Concurrent.ParallelIO.ConcurrentCollection as CC import Control.Concurrent import Control.Exception.Extensible as E@@ -60,8 +60,12 @@ type WorkItem = IO Bool -- | A 'WorkQueue' is used to communicate 'WorkItem's to the workers.-type WorkQueue = CS.ConcurrentSet WorkItem+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+ -- | A thread pool, containing a maximum number of threads. The best way to -- construct one of these is using 'withPool'. data Pool = Pool {@@ -83,7 +87,7 @@ | threadcount < 1 = error $ "startPool: thread count must be strictly positive (was " ++ show threadcount ++ ")" | otherwise = do threadId <- myThreadId- queue <- CS.new+ queue <- CC.new let pool = Pool { pool_threadcount = threadcount, pool_spawnedby = threadId,@@ -112,7 +116,7 @@ -- | Internal method for scheduling work on a pool. enqueueOnPool :: Pool -> WorkItem -> IO ()-enqueueOnPool pool = CS.insert (pool_queue pool)+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@@ -142,22 +146,27 @@ -- 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. spawnPoolWorkerFor :: Pool -> IO ()-spawnPoolWorkerFor pool = do+spawnPoolWorkerFor pool = {- putStrLn "spawnPoolWorkerFor" >> -} do _ <- mask $ \restore -> forkIO $ restore workerLoop `E.catch` \(e :: E.SomeException) -> do- hPutStrLn stderr $ "Exception on thread: " ++ show e+ 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- kill <- join $ CS.delete (pool_queue pool)+ --tid <- myThreadId+ --hPutStrLn stderr $ "[waiting] " ++ show tid+ work_item <- CC.delete (pool_queue pool)+ --hPutStrLn stderr $ "[working] " ++ show tid+ kill <- work_item unless kill workerLoop -- | 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 = enqueueOnPool pool $ return True+killPoolWorkerFor pool = {- putStrLn "killPoolWorkerFor" >> -} enqueueOnPool pool (return True) -- | Run the list of computations in parallel.@@ -196,7 +205,7 @@ modifyMVar count $ \i -> do let i' = i - 1 kill = i' == 0- when kill $ putMVar pause ()+ 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
parallel-io.cabal view
@@ -1,5 +1,5 @@ Name: parallel-io-Version: 0.3.0.1+Version: 0.3.0.2 Cabal-Version: >= 1.2 Category: Concurrency Synopsis: Combinators for executing IO actions in parallel on a thread pool.@@ -26,6 +26,10 @@ Description: Build the benchmarking tool Default: False +Flag Fuzz+ Description: Build the fuzzing tool for discovering deadlocks+ Default: False+ Flag Tests Description: Build the test runner Default: False@@ -36,7 +40,7 @@ Control.Concurrent.ParallelIO.Global Control.Concurrent.ParallelIO.Local Other-Modules:- Control.Concurrent.ParallelIO.ConcurrentSet+ 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 @@ -60,4 +64,15 @@ 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 + Ghc-Options: -threaded++Executable fuzz+ 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.4, random >= 1.0 && < 1.1,+ test-framework >= 0.1.1, test-framework-hunit >= 0.1.1, HUnit >= 1.2 && < 2+ Ghc-Options: -threaded