monad-par 0.3.4.2 → 0.3.4.3
raw patch · 7 files changed
+162/−141 lines, 7 filesdep +chaselev-dequedep ~monad-par-extrasPVP ok
version bump matches the API change (PVP)
Dependencies added: chaselev-deque
Dependency ranges changed: monad-par-extras
API changes (from Hackage documentation)
Files
- Control/Monad/Par.hs +1/−1
- Control/Monad/Par/Scheds/Direct.hs +112/−105
- Control/Monad/Par/Scheds/DirectInternal.hs +23/−21
- Control/Monad/Par/Scheds/Trace.hs +4/−5
- Control/Monad/Par/Scheds/TraceInternal.hs +6/−7
- monad-par.cabal +11/−2
- tests/ParTests.hs +5/−0
Control/Monad/Par.hs view
@@ -165,5 +165,5 @@ import Control.Monad.Par.Class hiding ( spawn, spawn_, spawnP, put, put_ , get, newFull, new, fork, newFull_ )-import Control.Monad.Par.Scheds.Direct+import Control.Monad.Par.Scheds.Trace import Control.Monad.Par.Combinator
Control/Monad/Par/Scheds/Direct.hs view
@@ -2,8 +2,7 @@ ExistentialQuantification, CPP, ScopedTypeVariables, TypeSynonymInstances, MultiParamTypeClasses, GeneralizedNewtypeDeriving, PackageImports,- ParallelListComp- #-}+ ParallelListComp #-} {- OPTIONS_GHC -Wall -fno-warn-name-shadowing -fno-warn-unused-do-bind -}@@ -17,7 +16,7 @@ -- trace data structure). module Control.Monad.Par.Scheds.Direct (- Sched(..), + Sched(..), Par, -- abstract: Constructor not exported. IVar(..), IVarContents(..), -- sched,@@ -50,10 +49,17 @@ import qualified Data.Map as M import qualified Data.Set as S import Data.Maybe (catMaybes)+import Data.Word (Word64)+ import Data.Concurrent.Deque.Class (WSDeque)+#ifdef USE_CHASELEV+#warning "Note: using Chase-Lev lockfree workstealing deques..."+import Data.Concurrent.Deque.ChaseLev.DequeInstance+import Data.Concurrent.Deque.ChaseLev as R+#else import Data.Concurrent.Deque.Reference.DequeInstance import Data.Concurrent.Deque.Reference as R-import Data.Word (Word64)+#endif import qualified Control.Exception as E @@ -69,7 +75,6 @@ -- Configuration Toggles -------------------------------------------------------------------------------- --- #define DEBUG -- [2012.08.30] This shows a 10X improvement on nested parfib: -- #define NESTED_SCHEDS #define PARPUTS@@ -86,7 +91,8 @@ -- conditionals and trust dead-code-elimination. -------------------------------------------------------------------- -#ifdef DEBUG+#ifdef DEBUG_DIRECT+#warning "DEBUG: Activating debugging for Direct.hs" import Debug.Trace (trace) import System.Environment (getEnvironment) theEnv = unsafePerformIO $ getEnvironment@@ -140,7 +146,7 @@ data IVarContents a = Full a | Empty | Blocked [a -> IO ()] -unsafeParIO :: IO a -> Par a +unsafeParIO :: IO a -> Par a unsafeParIO iom = Par (lift$ lift iom) io :: IO a -> Par a@@ -171,13 +177,13 @@ -- becomes an active worker. wp <- readIORef globalWorkerPool return (M.lookup tid wp)-registerWorker tid sched = - atomicModifyIORef globalWorkerPool $ +registerWorker tid sched =+ atomicModifyIORef globalWorkerPool $ \ mp -> (M.insert tid sched mp, ())-unregisterWorker tid = - atomicModifyIORef globalWorkerPool $ +unregisterWorker tid =+ atomicModifyIORef globalWorkerPool $ \ mp -> (M.delete tid mp, ())-#else +#else amINested _ = return Nothing registerWorker _ _ = return () unregisterWorker _tid = return ()@@ -189,9 +195,9 @@ {-# INLINE popWork #-} popWork :: Sched -> IO (Maybe (Par ()))-popWork Sched{ workpool, no } = do - mb <- R.tryPopL workpool - when dbg $ case mb of +popWork Sched{ workpool, no } = do+ mb <- R.tryPopL workpool+ when dbg $ case mb of Nothing -> return () Just _ -> do sn <- makeStableName mb printf " [%d] -> POP work unit %d\n" no (hashStableName sn)@@ -244,12 +250,12 @@ runNewSessionAndWait :: String -> Sched -> Par b -> IO b runNewSessionAndWait name sched userComp = do tid <- myThreadId -- TODO: remove when done debugging- sid <- modifyHotVar (sessionCounter sched) (\ x -> (x+1,x)) + sid <- modifyHotVar (sessionCounter sched) (\ x -> (x+1,x)) _ <- modifyHotVar (activeSessions sched) (\ set -> (S.insert sid set, ()))- + -- Here we have an extra IORef... ugly. ref <- newIORef (error$ "Empty session-result ref ("++name++") should never be touched (sid "++ show sid++", "++show tid ++")")- newFlag <- newHotVar False + newFlag <- newHotVar False -- Push the new session: _ <- modifyHotVar (sessions sched) (\ ls -> ((Session sid newFlag) : ls, ())) @@ -269,7 +275,7 @@ kont n = trivialCont$ "("++name++", sid "++show sid++", round "++show n++")" loop :: Word64 -> ROnly () loop n = do flg <- liftIO$ readIORef newFlag- unless flg $ do + unless flg $ do when dbg $ liftIO$ do tid4 <- myThreadId printf " [%d %s] BOUNCE %d... going into reschedule until finished.\n" (no sched) (show tid4) n@@ -294,7 +300,7 @@ then tl else error$ "Tried to pop the session stack and found we ("++show sid ++") were not on the top! (instead "++show sid2++")"- + -- By returning here we ARE implicitly reengaging the scheduler, since we -- are already inside the rescheduleR loop on this thread -- (before runParIO was called in a nested fashion).@@ -303,7 +309,7 @@ {-# NOINLINE runParIO #-} runParIO userComp = do- tid <- myThreadId + tid <- myThreadId #if __GLASGOW_HASKELL__ >= 701 /* 20110301 */ -- -- We create a thread on each CPU with forkOn. The CPU on which@@ -324,8 +330,8 @@ let main_cpu = 0 #endif maybSched <- amINested tid- tidorig <- myThreadId -- TODO: remove when done debugging - case maybSched of + tidorig <- myThreadId -- TODO: remove when done debugging+ case maybSched of Just (sched) -> do -- Here the current thread is ALREADY a worker. All we need to -- do is plug the users new computation in.@@ -336,19 +342,19 @@ ------------------------------------------------------------ -- Non-nested case, make a new set of worker threads:- ------------------------------------------------------------ + ------------------------------------------------------------ Nothing -> do allscheds <- makeScheds main_cpu [Session _ topSessFlag] <- readHotVar$ sessions$ head allscheds- + mfin <- newEmptyMVar doneFlags <- forM (zip [0..] allscheds) $ \(cpu,sched) -> do- workerDone <- newEmptyMVar + workerDone <- newEmptyMVar ---------------------------------------- let wname = ("(worker "++show cpu++" of originator "++show tidorig++")") -- forkOn cpu $ do- _ <- forkWithExceptions (forkOn cpu) wname $ do - ------------------------------------------------------------STRT WORKER THREAD + _ <- forkWithExceptions (forkOn cpu) wname $ do+ ------------------------------------------------------------STRT WORKER THREAD tid2 <- myThreadId registerWorker tid2 sched if (cpu /= main_cpu)@@ -362,15 +368,15 @@ writeIORef topSessFlag True when dbg$ do printf " *** Out of entire runContT user computation on main thread %s.\n" (show tid2) -- sanityCheck allscheds- putMVar mfin x + putMVar mfin x unregisterWorker tid ------------------------------------------------------------END WORKER THREAD return (if cpu == main_cpu then Nothing else Just workerDone) - when _WAIT_FOR_WORKERS $ do + when _WAIT_FOR_WORKERS $ do when dbg$ printf " *** [%s] Originator thread: waiting for workers to complete." (show tidorig)- forM_ (catMaybes doneFlags) $ \ mv -> do + forM_ (catMaybes doneFlags) $ \ mv -> do n <- readMVar mv -- n <- A.wait mv when dbg$ printf " * [%s] Worker %s completed\n" (show tidorig) (show n)@@ -381,9 +387,9 @@ -- GHC is expensive like a forkOS thread. ---------------------------------------- -- DEBUGGING ---#ifdef DEBUG+#ifdef DEBUG_DIRECT busyTakeMVar (" The global wait "++ show tidorig) mfin -- Final value.--- dbgTakeMVar "global waiting thread" mfin -- Final value. +-- dbgTakeMVar "global waiting thread" mfin -- Final value. #else takeMVar mfin -- Final value. #endif@@ -391,11 +397,11 @@ -- Create the default scheduler(s) state: makeScheds :: Int -> IO [Sched]-makeScheds main = do +makeScheds main = do when dbg$ do tid <- myThreadId printf "[initialization] Creating %d worker threads, currently on %s\n" numCapabilities (show tid) workpools <- replicateM numCapabilities $ R.newQ- rngs <- replicateM numCapabilities $ Random.create >>= newHotVar + rngs <- replicateM numCapabilities $ Random.create >>= newHotVar idle <- newHotVar [] -- The STACKs are per-worker.. but the root finished flag is shared between all anonymous system workers: sessionFinished <- newHotVar False@@ -436,19 +442,19 @@ -- | read the value in a @IVar@. The 'get' can only return when the -- value has been written by a prior or parallel @put@ to the same -- @IVar@.-get (IVar vr) = do - callCC $ \kont -> +get (IVar vr) = do+ callCC $ \kont -> do e <- io$ readIORef vr case e of Full a -> return a _ -> do sch <- RD.ask-# ifdef DEBUG+# ifdef DEBUG_DIRECT sn <- io$ makeStableName vr -- Should probably do the MutVar inside...- let resched = trace (" ["++ show (no sch) ++ "] - Rescheduling on unavailable ivar "++show (hashStableName sn)++"!") + let resched = trace (" ["++ show (no sch) ++ "] - Rescheduling on unavailable ivar "++show (hashStableName sn)++"!") #else- let resched = + let resched = # endif longjmpSched -- Invariant: kont must not be lost. -- Because we continue on the same processor the Sched stays the same:@@ -464,9 +470,9 @@ -- be accessed by importing Control.Monad.Par.Unsafe. {-# INLINE unsafePeek #-} unsafePeek :: IVar a -> Par (Maybe a)-unsafePeek (IVar v) = do +unsafePeek (IVar v) = do e <- io$ readIORef v- case e of + case e of Full a -> return (Just a) _ -> return Nothing @@ -476,20 +482,20 @@ -- In this scheduler, puts immediately execute woken work in the current thread. put_ (IVar vr) !content = do sched <- RD.ask- ks <- io$ do + ks <- io$ do ks <- atomicModifyIORef vr $ \e -> case e of Empty -> (Full content, []) Full _ -> error "multiple put" Blocked ks -> (Full content, ks)-#ifdef DEBUG- when (dbglvl >= 3) $ do +#ifdef DEBUG_DIRECT+ when (dbglvl >= 3) $ do sn <- makeStableName vr- printf " [%d] Put value %s into IVar %d. Waking up %d continuations.\n" + printf " [%d] Put value %s into IVar %d. Waking up %d continuations.\n" (no sched) (show content) (hashStableName sn) (length ks) return ()-#endif +#endif return ks- wakeUp sched ks content + wakeUp sched ks content -- | NOTE unsafeTryPut is NOT exposed directly through this module. (So -- this module remains SAFE in the Safe Haskell sense.) It can only@@ -497,15 +503,15 @@ {-# INLINE unsafeTryPut #-} unsafeTryPut (IVar vr) !content = do -- Head strict rather than fully strict.- sched <- RD.ask - (ks,res) <- io$ do + sched <- RD.ask+ (ks,res) <- io$ do pr <- atomicModifyIORef vr $ \e -> case e of Empty -> (Full content, ([], content)) Full x -> (Full x, ([], x)) Blocked ks -> (Full content, (ks, content))-#ifdef DEBUG+#ifdef DEBUG_DIRECT sn <- makeStableName vr- printf " [%d] unsafeTryPut: value %s in IVar %d. Waking up %d continuations.\n" + printf " [%d] unsafeTryPut: value %s in IVar %d. Waking up %d continuations.\n" (no sched) (show content) (hashStableName sn) (length (fst pr)) #endif return pr@@ -533,7 +539,7 @@ -- _ -> spawn_$ do spawn_$ io$ kont arg -- io$ parchain rest -- error$"FINISHME - wake "++show (length ks)++" conts"- else + else -- This version sacrifices a parallelism opportunity and -- imposes additional serialization. --@@ -541,7 +547,7 @@ -- This "optimization" should not be on the table. -- mapM_ ($arg) ks do io$ kont arg- loop rest + loop rest return () pMap kont [] = io$ kont arg@@ -552,37 +558,37 @@ -- parchain [kont] = kont arg -- parchain (kont:rest) = do spawn$ io$ kont arg -- parchain rest- + ------------------------------------------------------------ {-# INLINE fork #-} fork :: Par () -> Par () fork task = -- Forking the "parent" means offering up the continuation of the -- fork rather than the task argument for stealing:- case _FORKPARENT of - True -> do - sched <- RD.ask + case _FORKPARENT of+ True -> do+ sched <- RD.ask callCC$ \parent -> do let wrapped = parent () io$ pushWork sched wrapped -- Then execute the child task and return to the scheduler when it is complete:- task + task -- If we get to this point we have finished the child task: longjmpSched -- We reschedule to pop the cont we pushed. -- TODO... OPTIMIZATION: we could also try the pop directly, and if it succeeds return normally.... io$ printf " !!! ERROR: Should never reach this point #1\n" - when dbg$ do - sched2 <- RD.ask + when dbg$ do+ sched2 <- RD.ask io$ printf " - called parent continuation... was on worker [%d] now on worker [%d]\n" (no sched) (no sched2) return () - False -> do + False -> do sch <- RD.ask when dbg$ io$ printf " [%d] forking task...\n" (no sch) io$ pushWork sch task- + -- This routine "longjmp"s to the scheduler, throwing out its own continuation. longjmpSched :: Par a -- longjmpSched = Par $ C.ContT rescheduleR@@ -592,7 +598,7 @@ -- then it finally invokes its continuation. rescheduleR :: Word64 -> (a -> ROnly ()) -> ROnly () rescheduleR cnt kont = do- mysched <- RD.ask + mysched <- RD.ask when dbg$ liftIO$ do tid <- myThreadId sess <- readSessions mysched null <- R.nullQ (workpool mysched)@@ -602,24 +608,24 @@ case mtask of Nothing -> do (Session _ finRef):_ <- liftIO$ readIORef $ sessions mysched- fin <- liftIO$ readIORef finRef + fin <- liftIO$ readIORef finRef if fin then do when (dbglvl >= 1) $ liftIO $ do tid <- myThreadId sess <- readSessions mysched- printf " [%d %s] - DROP out of reschedule loop, sessionFinished=%s, all sessions %s\n" + printf " [%d %s] - DROP out of reschedule loop, sessionFinished=%s, all sessions %s\n" (no mysched) (show tid) (show fin) (show sess) empt <- R.nullQ$ workpool mysched when (not empt) $ do- printf " [%d %s] - WARNING - leaving rescheduleR while local workpoll is nonempty\n" - (no mysched) (show tid) - + printf " [%d %s] - WARNING - leaving rescheduleR while local workpoll is nonempty\n"+ (no mysched) (show tid)+ kont (error "Direct.hs: The result value from rescheduleR should not be used.") else do -- when (dbglvl >= 1) $ liftIO $ do- -- tid <- myThreadId + -- tid <- myThreadId -- sess <- readSessions mysched- -- printf " [%d %s] - Apparently NOT finished with head session... trying to steal, all sessions %s\n" + -- printf " [%d %s] - Apparently NOT finished with head session... trying to steal, all sessions %s\n" -- (no mysched) (show tid) (show sess) liftIO$ steal mysched #ifdef WAKEIDLE@@ -631,16 +637,16 @@ -- When popping work from our own queue the Sched (Reader value) stays the same: when dbg $ do sn <- liftIO$ makeStableName task liftIO$ printf " [%d] popped work %d from own queue\n" (no mysched) (hashStableName sn)- let C.ContT fn = unPar task + let C.ContT fn = unPar task -- Run the stolen task with a continuation that returns to the scheduler if the task exits normally:- fn (\ _ -> do + fn (\ _ -> do sch <- RD.ask when dbg$ liftIO$ printf " + task finished successfully on cpu %d, calling reschedule continuation..\n" (no sch) rescheduleR 0 kont) -- | Attempt to steal work or, failing that, give up and go idle.--- +-- -- The current policy is to do a burst of of N tries without -- yielding or pausing inbetween. steal :: Sched -> IO ()@@ -657,7 +663,7 @@ ---------------------------------------- -- IDLING behavior:- go 0 _ | _IDLING_ON = + go 0 _ | _IDLING_ON = do m <- newEmptyMVar r <- modifyHotVar idle $ \is -> (m:is, is) if length r == numCapabilities - 1@@ -689,14 +695,14 @@ when (dbglvl>=2)$ printf " [%d] | trying steal from %d\n" my_no (no schd) -- let dq = workpool schd :: WSDeque (Par ())- let dq = workpool schd + let dq = workpool schd r <- R.tryPopR dq case r of Just task -> do when dbg$ do sn <- makeStableName task printf " [%d] | stole work (unit %d) from cpu %d\n" my_no (hashStableName sn) (no schd)- runReaderWith mysched $ + runReaderWith mysched $ C.runContT (unPar task) (\_ -> do when dbg$ do sn <- liftIO$ makeStableName task@@ -711,8 +717,8 @@ errK = error "Error cont: this closure shouldn't be used" trivialCont :: String -> a -> ROnly ()-trivialCont str _ = do -#ifdef DEBUG+trivialCont str _ = do+#ifdef DEBUG_DIRECT -- trace (str ++" trivialCont evaluated!") liftIO$ printf " !! trivialCont evaluated, msg: %s\n" str #endif@@ -728,14 +734,14 @@ {-# INLINE spawn1_ #-} -- Spawn a one argument function instead of a thunk. This is good for debugging if the value supports "Show".-spawn1_ f x = -#ifdef DEBUG+spawn1_ f x =+#ifdef DEBUG_DIRECT do sn <- io$ makeStableName f sch <- RD.ask; when dbg$ io$ printf " [%d] spawning fn %d with arg %s\n" (no sch) (hashStableName sn) (show x) #endif spawn_ (f x) --- The following is usually inefficient! +-- The following is usually inefficient! newFull_ a = do v <- new put_ v a return v@@ -750,7 +756,7 @@ spawnP a = spawn (return a) -- In Debug mode we require that IVar contents be Show-able:-#ifdef DEBUG+#ifdef DEBUG_DIRECT put :: (Show a, NFData a) => IVar a -> a -> Par () spawn :: (Show a, NFData a) => Par a -> Par (IVar a) spawn_ :: Show a => Par a -> Par (IVar a)@@ -758,8 +764,8 @@ spawnP :: (Show a, NFData a) => a -> Par (IVar a) put_ :: Show a => IVar a -> a -> Par () get :: Show a => IVar a -> Par a-runPar :: Show a => Par a -> a -runParIO :: Show a => Par a -> IO a +runPar :: Show a => Par a -> a+runParIO :: Show a => Par a -> IO a newFull :: (Show a, NFData a) => a -> Par (IVar a) newFull_ :: Show a => a -> Par (IVar a) unsafeTryPut :: Show b => IVar b -> b -> Par b@@ -771,8 +777,8 @@ put_ :: IVar a -> a -> Par () put :: NFData a => IVar a -> a -> Par () get :: IVar a -> Par a-runPar :: Par a -> a -runParIO :: Par a -> IO a +runPar :: Par a -> a+runParIO :: Par a -> IO a newFull :: NFData a => a -> Par (IVar a) newFull_ :: a -> Par (IVar a) unsafeTryPut :: IVar b -> b -> Par b@@ -822,7 +828,7 @@ sanityCheck allscheds = do forM_ allscheds $ \ Sched{no, workpool} -> do b <- R.nullQ workpool- when (not b) $ do + when (not b) $ do () <- printf "WARNING: After main thread exited non-empty queue remains for worker %d\n" no return () printf "Sanity check complete.\n"@@ -830,20 +836,21 @@ -- | This tries to localize the blocked-indefinitely exception: dbgTakeMVar :: String -> MVar a -> IO a-dbgTakeMVar msg mv = +dbgTakeMVar msg mv = -- catch (takeMVar mv) ((\_ -> doDebugStuff) :: BlockedIndefinitelyOnMVar -> IO a) E.catch (takeMVar mv) ((\_ -> doDebugStuff) :: IOError -> IO a)- where + where doDebugStuff = do printf "This takeMVar blocked indefinitely!: %s\n" msg error "failed" -- | For debugging purposes. This can help us figure out (but an ugly -- process of elimination) which MVar reads are leading to a "Thread -- blocked indefinitely" exception.+{- busyTakeMVar :: String -> MVar a -> IO a busyTakeMVar msg mv = try (10 * 1000 * 1000)- where - try 0 = do + where+ try 0 = do when dbg $ do tid <- myThreadId -- After we've failed enough times, start complaining:@@ -851,20 +858,20 @@ try (100 * 1000) try n = do x <- tryTakeMVar mv- case x of + case x of Just y -> return y Nothing -> do yield; try (n-1)- +-} -- | Fork a thread but ALSO set up an error handler that suppresses -- MVar exceptions. forkIO_Suppress :: Int -> IO () -> IO ThreadId-forkIO_Suppress whre action = - forkOn whre $ - E.handle (\e -> +forkIO_Suppress whre action =+ forkOn whre $+ E.handle (\e -> case (e :: E.BlockedIndefinitelyOnMVar) of- _ -> do - putStrLn$"CAUGHT child thread exception: "++show e + _ -> do+ putStrLn$"CAUGHT child thread exception: "++show e return () ) action@@ -872,14 +879,14 @@ -- | Exceptions that walk up the fork tree of threads: forkWithExceptions :: (IO () -> IO ThreadId) -> String -> IO () -> IO ThreadId-forkWithExceptions forkit descr action = do +forkWithExceptions forkit descr action = do parent <- myThreadId forkit $ do tid <- myThreadId E.catch action- (\ e -> - case E.fromException e of - Just E.ThreadKilled -> printf -- hPrintf stderr + (\ e ->+ case E.fromException e of+ Just E.ThreadKilled -> printf -- hPrintf stderr "\nThreadKilled exception inside child thread, %s (not propagating!): %s\n" (show tid) (show descr) _ -> do printf -- hPrintf stderr "\nException inside child thread %s, %s: %s\n" (show descr) (show tid) (show e)@@ -893,5 +900,5 @@ ls <- readIORef (sessions sched) bools <- mapM (\ (Session _ r) -> readIORef r) ls return (zip (map (\ (Session sid _) -> sid) ls) bools)- - ++
Control/Monad/Par/Scheds/DirectInternal.hs view
@@ -1,6 +1,4 @@-{-# LANGUAGE PackageImports, CPP, - GeneralizedNewtypeDeriving - #-}+{-# LANGUAGE PackageImports, CPP, GeneralizedNewtypeDeriving #-} -- | Type definiton and some helpers. This is used mainly by -- Direct.hs but can also be used by other modules that want access to@@ -17,14 +15,18 @@ import Control.Concurrent hiding (yield) import GHC.Conc import Data.IORef-import Data.Concurrent.Deque.Class (WSDeque)--- import Data.Concurrent.Deque.Reference.DequeInstance--- import Data.Concurrent.Deque.Reference as R+import qualified Data.Set as S+import Data.Word (Word64) import Data.Concurrent.Deque.Class (WSDeque)++#ifdef USE_CHASELEV+#warning "Note: using Chase-Lev lockfree workstealing deques..."+import Data.Concurrent.Deque.ChaseLev.DequeInstance+import Data.Concurrent.Deque.ChaseLev as R+#else import Data.Concurrent.Deque.Reference.DequeInstance import Data.Concurrent.Deque.Reference as R-import qualified Data.Set as S-import Data.Word (Word64)+#endif -- Our monad stack looks like this: -- ---------@@ -34,7 +36,7 @@ -- --------- -- The ReaderT monad is there for retrieving the scheduler given the -- fact that the API calls do not get it as an argument.--- +-- -- Note that the result type for continuations is unit. Forked -- computations return nothing. --@@ -47,28 +49,28 @@ -- An ID along with a flag to signal completion: data Session = Session SessionID (HotVar Bool) -data Sched = Sched - { +data Sched = Sched+ { ---- Per worker ---- no :: {-# UNPACK #-} !Int, workpool :: WSDeque (Par ()), rng :: HotVar Random.GenIO, -- Random number gen for work stealing.- isMain :: Bool, -- Are we the main/master thread? + isMain :: Bool, -- Are we the main/master thread? -- The stack of nested sessions that THIS worker is participating in. -- When a session finishes, the worker can return to its Haskell -- calling context (it's "real" continuation). sessions :: HotVar [Session], -- (1) This is always non-empty, containing at least the root- -- session corresponding to the anonymous system workers. + -- session corresponding to the anonymous system workers. -- (2) The original invocation of runPar also counts as a session- -- and pushes a second + -- and pushes a second -- (3) Nested runPar invocations may push further sessions onto the stack.- + ---- Global data: ---- idle :: HotVar [MVar Bool], -- waiting idle workers scheds :: [Sched], -- A global list of schedulers.- + -- Any thread that enters runPar (original or nested) registers -- itself in this global list. When the list becomes null, -- worker threads may shut down or at least go idle.@@ -109,7 +111,7 @@ modifyHotVar_ v fn = atomicModifyIORef v (\a -> (fn a, ())) readHotVar = readIORef writeHotVar = writeIORef-instance Show (IORef a) where +instance Show (IORef a) where show ref = "<ioref>" -- hotVarTransaction = id@@ -118,7 +120,7 @@ writeHotVarRaw = writeHotVar -#elif HOTVAR == 2 +#elif HOTVAR == 2 #warning "Using MVars for hot atomic variables." -- This uses MVars that are always full with *something* type HotVar a = MVar a@@ -127,7 +129,7 @@ modifyHotVar_ v fn = modifyMVar_ v (return . fn) readHotVar = readMVar writeHotVar v x = do swapMVar v x; return ()-instance Show (MVar a) where +instance Show (MVar a) where show ref = "<mvar>" -- hotVarTransaction = id@@ -143,14 +145,14 @@ -- Simon Marlow said he saw better scaling with TVars (surprise to me): type HotVar a = TVar a newHotVar = newTVarIO-modifyHotVar tv fn = atomically (do x <- readTVar tv +modifyHotVar tv fn = atomically (do x <- readTVar tv let (x2,b) = fn x writeTVar tv x2 return b) modifyHotVar_ tv fn = atomically (do x <- readTVar tv; writeTVar tv (fn x)) readHotVar x = atomically $ readTVar x writeHotVar v x = atomically $ writeTVar v x-instance Show (TVar a) where +instance Show (TVar a) where show ref = "<tvar>" hotVarTransaction = atomically
Control/Monad/Par/Scheds/Trace.hs view
@@ -1,6 +1,5 @@ {-# LANGUAGE RankNTypes, NamedFieldPuns, BangPatterns,- ExistentialQuantification, MultiParamTypeClasses, CPP- #-}+ ExistentialQuantification, MultiParamTypeClasses, CPP #-} {- OPTIONS_GHC -Wall -fno-warn-name-shadowing -fwarn-unused-imports -} {- | This is the scheduler described in the paper "A Monad for@@ -13,7 +12,7 @@ module Control.Monad.Par.Scheds.Trace ( Par, runPar, runParIO, fork, IVar, new, newFull, newFull_, get, put, put_,- spawn, spawn_, spawnP + spawn, spawn_, spawnP ) where import qualified Control.Monad.Par.Class as PC@@ -46,8 +45,8 @@ spawn_ = spawn_ spawnP = spawnP -instance PC.ParIVar IVar Par where - fork = fork +instance PC.ParIVar IVar Par where+ fork = fork new = new put = put put_ = put_
Control/Monad/Par/Scheds/TraceInternal.hs view
@@ -1,6 +1,5 @@ {-# LANGUAGE RankNTypes, NamedFieldPuns, BangPatterns,- ExistentialQuantification, CPP- #-}+ ExistentialQuantification, CPP #-} {-# OPTIONS_GHC -Wall -fno-warn-name-shadowing -fno-warn-unused-do-bind #-} -- | This module exposes the internals of the @Par@ monad so that you@@ -41,7 +40,7 @@ -- | The main scheduler loop. sched :: Bool -> Sched -> Trace -> IO () sched _doSync queue t = loop t- where + where loop t = case t of New a f -> do r <- newIORef a@@ -77,7 +76,7 @@ -- threads work-queue because it can be stolen by other threads. -- else return () - Yield parent -> do + Yield parent -> do -- Go to the end of the worklist: let Sched { workpool } = queue -- TODO: Perhaps consider Data.Seq here.@@ -180,9 +179,9 @@ -- From outside the Par computation we can peek. But this is nondeterministic. pollIVar :: IVar a -> IO (Maybe a)-pollIVar (IVar ref) = +pollIVar (IVar ref) = do contents <- readIORef ref- case contents of + case contents of Full x -> return (Just x) _ -> return (Nothing) @@ -244,7 +243,7 @@ -- | An asynchronous version in which the main thread of control in a -- Par computation can return while forked computations still run in--- the background. +-- the background. runParAsync :: Par a -> a runParAsync = unsafePerformIO . runPar_internal False
monad-par.cabal view
@@ -1,5 +1,5 @@ Name: monad-par-Version: 0.3.4.2+Version: 0.3.4.3 Synopsis: A library for parallel programming based on a monad @@ -20,6 +20,7 @@ -- 0.3.4 : switch to direct scheduler as default (only 1-level nesting allowed) -- 0.3.4.1 : fix build with GHC 7.0, and fix test -- 0.3.4.2 : Bugfix, 0.3.4.1 was released with debugging switches flipped.+-- 0.3.4.3 : Bugfix, Trace scheduler is now the default Description: The 'Par' monad offers a simple API for parallel programming. The@@ -71,6 +72,10 @@ tests/hs_cassandra_microbench.hs tests/hs_cassandra_microbench2.hs +Flag chaselev+ Description: Use Chase-Lev Deques for higher-perf work-stealing.+ Default: False+ Library Source-repository head type: git@@ -98,13 +103,17 @@ , abstract-par , abstract-deque >= 0.1.4 -- Extras such as parMap, RNG, State- , monad-par-extras == 0.3.*+ , monad-par-extras >= 0.3 , deepseq >= 1.1 , array >= 0.3 , mwc-random >= 0.11 , containers , parallel >= 3.1 , mtl >= 2.0.1.0++ if flag(chaselev)+ cpp-options: -DUSE_CHASELEV+ build-depends: chaselev-deque ghc-options: -O2 Other-modules:
tests/ParTests.hs view
@@ -98,7 +98,12 @@ -- both :: Par a -> Par a -> Par a -- both a b = Par $ \c -> Fork (runCont a c) (runCont b c) ++-- | A reduction test. case_test_pmrr1 :: Assertion+-- Saw a failure here using Direct:+-- http://tester-lin.soic.indiana.edu:8080/job/HackageReleased_monad-par/GHC_VERS=7.0.4,label=tank.cs.indiana.edu/40/console+-- Exception inside child thread "(worker 0 of originator ThreadId 5)", ThreadId 10: thread blocked indefinitely in an MVar operation case_test_pmrr1 = par 5050 $ parMapReduceRangeThresh 1 (InclusiveRange 1 100) (return) (return `bincomp` (+)) 0