diff --git a/Control/Monad/Par.hs b/Control/Monad/Par.hs --- a/Control/Monad/Par.hs +++ b/Control/Monad/Par.hs @@ -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 diff --git a/Control/Monad/Par/Scheds/Direct.hs b/Control/Monad/Par/Scheds/Direct.hs --- a/Control/Monad/Par/Scheds/Direct.hs +++ b/Control/Monad/Par/Scheds/Direct.hs @@ -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) - - + + diff --git a/Control/Monad/Par/Scheds/DirectInternal.hs b/Control/Monad/Par/Scheds/DirectInternal.hs --- a/Control/Monad/Par/Scheds/DirectInternal.hs +++ b/Control/Monad/Par/Scheds/DirectInternal.hs @@ -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 = "" -- 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 = "" -- 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 = "" hotVarTransaction = atomically diff --git a/Control/Monad/Par/Scheds/Trace.hs b/Control/Monad/Par/Scheds/Trace.hs --- a/Control/Monad/Par/Scheds/Trace.hs +++ b/Control/Monad/Par/Scheds/Trace.hs @@ -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_ diff --git a/Control/Monad/Par/Scheds/TraceInternal.hs b/Control/Monad/Par/Scheds/TraceInternal.hs --- a/Control/Monad/Par/Scheds/TraceInternal.hs +++ b/Control/Monad/Par/Scheds/TraceInternal.hs @@ -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 diff --git a/monad-par.cabal b/monad-par.cabal --- a/monad-par.cabal +++ b/monad-par.cabal @@ -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: diff --git a/tests/ParTests.hs b/tests/ParTests.hs --- a/tests/ParTests.hs +++ b/tests/ParTests.hs @@ -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