monad-par 0.3 → 0.3.4
raw patch · 9 files changed
+1030/−823 lines, 9 filesdep ~base
Dependency ranges changed: base
Files
- Control/Monad/Par.hs +28/−23
- Control/Monad/Par/IO.hs +44/−0
- Control/Monad/Par/Scheds/Direct.hs +534/−272
- Control/Monad/Par/Scheds/DirectInternal.hs +160/−0
- Control/Monad/Par/Scheds/Trace.hs +1/−1
- Control/Monad/Par/Scheds/TraceInternal.hs +154/−457
- monad-par.cabal +43/−37
- tests/ParTests.hs +51/−21
- tests/TestHelpers.hs +15/−12
Control/Monad/Par.hs view
@@ -1,17 +1,14 @@ -{-| (NOTE: This module reexports a default Par scheduler. A generic- interface can be found in "Control.Monad.Par.Class" and other- schedulers, sometimes with different capabilities, can be found in- "Control.Monad.Par.Scheds".)+{-| - The @monad-par@ package provides a family of @Par@ monads, for speeding up pure- computations using parallel processors. They cannot be used for- speeding up computations that use IO (for that, see- @Control.Concurrent@). The result of a given @Par@ computation is- always the same - ie. it is deterministic, but the computation may- be performed more quickly if there are processors available to- share the work.+ The @monad-par@ package provides a family of @Par@ monads, for+ speeding up pure computations using parallel processors. (for a similar+ programming model for use with @IO@, see "Control.Monad.Par.IO".) + The result of a given @Par@ computation is always the same - i.e. it+ is deterministic, but the computation may be performed more quickly+ if there are processors available to share the work.+ For example, the following program fragment computes the values of @(f x)@ and @(g x)@ in parallel, and returns a pair of their results: @@ -69,25 +66,35 @@ parallel work are only created by @fork@ and a few other combinators. - The implementation is based on a work-stealing scheduler that- divides the work as evenly as possible between the available- processors at runtime.+ The default implementation is based on a work-stealing scheduler+ that divides the work as evenly as possible between the available+ processors at runtime. Other schedulers are available that are+ based on different policies and have different performance+ characteristics. To use one of these other schedulers, just import+ its module instead of "Control.Monad.Par": + * "Control.Monad.Par.Scheds.Trace"++ * "Control.Monad.Par.Scheds.Sparks"+ For more information on the programming model, please see these sources: - * The wiki/tutorial (<http://www.haskell.org/haskellwiki/Par_Monad:_A_Parallelism_Tutorial>)+ * The wiki\/tutorial (<http://www.haskell.org/haskellwiki/Par_Monad:_A_Parallelism_Tutorial>)+ * The original paper (<http://www.cs.indiana.edu/~rrnewton/papers/haskell2011_monad-par.pdf>)+ * Tutorial slides (<http://community.haskell.org/~simonmar/slides/CUFP.pdf>)- * Other slides: <http://www.cs.ox.ac.uk/ralf.hinze/WG2.8/28/slides/simon.pdf>, - <http://www.cs.indiana.edu/~rrnewton/talks/2011_HaskellSymposium_ParMonad.pdf> + * Other slides: (<http://www.cs.ox.ac.uk/ralf.hinze/WG2.8/28/slides/simon.pdf>,+ <http://www.cs.indiana.edu/~rrnewton/talks/2011_HaskellSymposium_ParMonad.pdf>)+ -} module Control.Monad.Par ( -- * The Par Monad Par, - runPar, + runPar, runParIO, fork, -- | forks a computation to happen in parallel. The forked@@ -156,9 +163,7 @@ ) where --- (0.3) Export 'Par' operators via the generic interface.-import Control.Monad.Par.Class-import Control.Monad.Par.Scheds.Trace hiding (spawn_, spawn, spawnP, put, get, new, newFull, fork, put_, newFull_)--- import Control.Monad.Par.Scheds.Direct -+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.Combinator
+ Control/Monad/Par/IO.hs view
@@ -0,0 +1,44 @@+{-# LANGUAGE GeneralizedNewtypeDeriving, PackageImports #-}+{- |+ This module is an alternative version of "Control.Monad.Par" in+ which the `Par` type provides `IO` operations, by means of `liftIO`.+ The price paid is that only `runParIO` is available, not the pure `runPar`.++ This module uses the same default scheduler as "Control.Monad.Par",+ and tasks scheduled by the two can share the same pool of worker+ threads. + -}++module Control.Monad.Par.IO+ ( ParIO, P.IVar, runParIO+ -- And instances! + )+ where++-- import qualified Control.Monad.Par as P+-- import qualified Control.Monad.Par.Scheds.Trace as P+-- import qualified Control.Monad.Par.Scheds.TraceInternal as TI++import qualified Control.Monad.Par.Scheds.DirectInternal as PI+import qualified Control.Monad.Par.Scheds.Direct as P+import Control.Monad.Par.Class+import Control.Applicative+import "mtl" Control.Monad.Trans (lift, liftIO, MonadIO)++-- | A wrapper around an underlying Par type which allows IO.+newtype ParIO a = ParIO { unPar :: PI.Par a }+ deriving (Functor, Applicative, Monad,+ ParFuture P.IVar, ParIVar P.IVar)++-- | A run method which allows actual IO to occur on top of the Par+-- monad. Of course this means that all the normal problems of+-- parallel IO computations are present, including nondeterminsm.+--+-- A simple example program:+--+-- > runParIO (liftIO$ putStrLn "hi" :: ParIO ())+runParIO :: ParIO a -> IO a+runParIO = P.runParIO . unPar++instance MonadIO ParIO where+ liftIO io = ParIO (PI.Par (lift$ lift io))
Control/Monad/Par/Scheds/Direct.hs view
@@ -1,9 +1,11 @@ {-# LANGUAGE RankNTypes, NamedFieldPuns, BangPatterns, ExistentialQuantification, CPP, ScopedTypeVariables, TypeSynonymInstances, MultiParamTypeClasses,- GeneralizedNewtypeDeriving, PackageImports+ GeneralizedNewtypeDeriving, PackageImports,+ ParallelListComp #-} + {- OPTIONS_GHC -Wall -fno-warn-name-shadowing -fno-warn-unused-do-bind -} -- {- LANGUAGE Trustworthy -}@@ -15,10 +17,11 @@ -- trace data structure). module Control.Monad.Par.Scheds.Direct (- Sched(..), Par,+ Sched(..), + Par, -- abstract: Constructor not exported. IVar(..), IVarContents(..), -- sched,- runPar, + runPar, runParIO, new, get, put_, fork, newFull, newFull_, put, spawn, spawn_, spawnP,@@ -29,35 +32,31 @@ import Control.Applicative import Control.Concurrent hiding (yield)-import Debug.Trace-import Data.IORef-import Text.Printf-import GHC.Conc-import "mtl" Control.Monad.Cont as C+import Data.IORef (IORef,newIORef,readIORef,writeIORef,atomicModifyIORef)+import Text.Printf (printf, hPrintf)+import GHC.Conc (numCapabilities,yield)+import "mtl" Control.Monad.Cont as C import qualified "mtl" Control.Monad.Reader as RD--- import qualified Data.Array as A--- import qualified Data.Vector as A-import qualified Data.Sequence as Seq-import System.Random.MWC as Random-import System.IO.Unsafe (unsafePerformIO)-import System.Mem.StableName-import qualified Control.Monad.Par.Class as PC-import qualified Control.Monad.Par.Unsafe as UN+import qualified System.Random.MWC as Random+import System.IO (stderr)+import System.IO.Unsafe (unsafePerformIO)+import System.Mem.StableName (makeStableName, hashStableName)+import qualified Control.Monad.Par.Class as PC+import qualified Control.Monad.Par.Unsafe as UN+import Control.Monad.Par.Scheds.DirectInternal+ (Par(..), Sched(..), HotVar, SessionID, Session(Session),+ newHotVar, readHotVar, modifyHotVar, modifyHotVar_, writeHotVarRaw) import Control.DeepSeq---- import Data.Concurrent.Deque.Class as DQ-#ifdef REACTOR_DEQUE--- These performed ABYSMALLY:-import Data.Concurrent.Deque.ChaseLev-import Data.Concurrent.Deque.ChaseLev.DequeInstance-import qualified Data.Concurrent.Deque.ReactorDeque as R-import Data.Array.IO-#else+import qualified Data.Map as M+import qualified Data.Set as S+import Data.Maybe (catMaybes) import Data.Concurrent.Deque.Class (WSDeque) import Data.Concurrent.Deque.Reference.DequeInstance import Data.Concurrent.Deque.Reference as R-#endif+import Data.Word (Word64) +import qualified Control.Exception as E+ import Prelude hiding (null) import qualified Prelude @@ -65,143 +64,120 @@ -- Configuration Toggles -------------------------------------------------------------------------------- --- define DEBUG+-- #define DEBUG+-- [2012.08.30] This shows a 10X improvement on nested parfib:+-- #define NESTED_SCHEDS+#define PARPUTS+-- #define FORKPARENT+-- #define IDLING_ON+ -- Next, IF idling is on, should we do wakeups?:+-- #define WAKEIDLE++-- #define WAIT_FOR_WORKERS++-------------------------------------------------------------------+-- Ifdefs for the above preprocessor defines. Try to MINIMIZE code+-- that lives in this dangerous region, and instead do normal+-- conditionals and trust dead-code-elimination.+--------------------------------------------------------------------+ #ifdef DEBUG+import Debug.Trace (trace)+import System.Environment (getEnvironment)+theEnv = unsafePerformIO $ getEnvironment dbg = True+dbglvl = 1 #else dbg = False+dbglvl = 0 #endif+dbg :: Bool+dbglvl :: Int -#define FORKPARENT-#define WAKEIDLE+_PARPUTS :: Bool+#ifdef PARPUTS+_PARPUTS = True+#else+_PARPUTS = False+#endif ------------------------------------------------------------------------------------ Core type definitions---------------------------------------------------------------------------------+_FORKPARENT :: Bool+#ifdef FORKPARENT+_FORKPARENT = True+#else+#warning "FORKPARENT POLICY NOT USED; THIS IS GENERALLY WORSE"+_FORKPARENT = False+#endif --- Our monad stack looks like this:--- ------------ ContT--- ReaderT--- IO--- ------------ 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.----newtype Par a = Par { unPar :: C.ContT () ROnly a }- deriving (Monad, MonadCont, RD.MonadReader Sched)-type ROnly = RD.ReaderT Sched IO+_IDLING_ON :: Bool+#ifdef IDLING_ON+_IDLING_ON = True+#else+_IDLING_ON = False+#endif -data Sched = Sched - { - ---- Per worker ----- no :: {-# UNPACK #-} !Int,-#ifdef REACTOR_DEQUE- workpool :: R.Deque IOArray (Par ()),+_WAIT_FOR_WORKERS :: Bool+#ifdef WAIT_FOR_WORKERS+_WAIT_FOR_WORKERS = True #else- workpool :: WSDeque (Par ()),+_WAIT_FOR_WORKERS = False #endif- rng :: HotVar GenIO, -- Random number gen for work stealing.- isMain :: Bool, -- Are we the main/master thread? - ---- Global data: ----- killflag :: HotVar Bool,- idle :: HotVar [MVar Bool],- scheds :: [Sched] -- A global list of schedulers.- } -newtype IVar a = IVar (IORef (IVarContents a)) -data IVarContents a = Full a | Empty | Blocked [a -> IO ()]--unsafeParIO :: IO a -> Par a -unsafeParIO io = Par (lift$ lift io)-io = unsafeParIO -- shorthand used below- ----------------------------------------------------------------------------------- Helpers #1: Atomic Variables+-- Core type definitions ----------------------------------------------------------------------------------- TEMP: Experimental -#ifndef HOTVAR-#define HOTVAR 1-#endif-newHotVar :: a -> IO (HotVar a)-modifyHotVar :: HotVar a -> (a -> (a,b)) -> IO b-modifyHotVar_ :: HotVar a -> (a -> a) -> IO ()-writeHotVar :: HotVar a -> a -> IO ()-readHotVar :: HotVar a -> IO a--- readHotVarRaw :: HotVar a -> m a--- writeHotVarRaw :: HotVar a -> m a--{-# INLINE newHotVar #-}-{-# INLINE modifyHotVar #-}-{-# INLINE modifyHotVar_ #-}-{-# INLINE readHotVar #-}-{-# INLINE writeHotVar #-}---#if HOTVAR == 1-type HotVar a = IORef a-newHotVar = newIORef-modifyHotVar = atomicModifyIORef-modifyHotVar_ v fn = atomicModifyIORef v (\a -> (fn a, ()))-readHotVar = readIORef-writeHotVar = writeIORef-instance Show (IORef a) where - show ref = "<ioref>"---- hotVarTransaction = id-hotVarTransaction = error "Transactions not currently possible for IO refs"-readHotVarRaw = readHotVar-writeHotVarRaw = writeHotVar+type ROnly = RD.ReaderT Sched IO +newtype IVar a = IVar (IORef (IVarContents a)) -#elif HOTVAR == 2 -#warning "Using MVars for hot atomic variables."--- This uses MVars that are always full with *something*-type HotVar a = MVar a-newHotVar x = do v <- newMVar; putMVar v x; return v-modifyHotVar v fn = modifyMVar v (return . fn)-modifyHotVar_ v fn = modifyMVar_ v (return . fn)-readHotVar = readMVar-writeHotVar v x = do swapMVar v x; return ()-instance Show (MVar a) where - show ref = "<mvar>"+data IVarContents a = Full a | Empty | Blocked [a -> IO ()] --- hotVarTransaction = id--- We could in theory do this by taking the mvar to grab the lock.--- But we'd need some temporary storage....-hotVarTransaction = error "Transactions not currently possible for MVars"-readHotVarRaw = readHotVar-writeHotVarRaw = writeHotVar+unsafeParIO :: IO a -> Par a +unsafeParIO iom = Par (lift$ lift iom) +io :: IO a -> Par a+io = unsafeParIO -- shorthand used below -#elif HOTVAR == 3-#warning "Using TVars for hot atomic variables."--- 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 - 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 - show ref = "<tvar>"+--------------------------------------------------------------------------------+-- Global State+-------------------------------------------------------------------------------- -hotVarTransaction = atomically-readHotVarRaw = readTVar-writeHotVarRaw = writeTVar+-- This keeps track of ALL worker threads across all unreated+-- `runPar` instantiations. This is used to detect nested invocations+-- of `runPar` and avoid reinitialization.+-- globalWorkerPool :: IORef (Data.IntMap ())+globalWorkerPool :: IORef (M.Map ThreadId Sched)+globalWorkerPool = unsafePerformIO $ newIORef M.empty+-- TODO! Make this semi-local! (not shared between "top-level" runPars) +{-# INLINE amINested #-}+{-# INLINE registerWorker #-}+{-# INLINE unregisterWorker #-}+amINested :: ThreadId -> IO (Maybe Sched)+registerWorker :: ThreadId -> Sched -> IO ()+unregisterWorker :: ThreadId -> IO ()+#ifdef NESTED_SCHEDS+-- | If the current threadID is ALREADY a worker, return the corresponding Sched structure.+amINested tid = do+ -- There is no race here. Each thread inserts itself before it+ -- becomes an active worker.+ wp <- readIORef globalWorkerPool+ return (M.lookup tid wp)+registerWorker tid sched = + atomicModifyIORef globalWorkerPool $ + \ mp -> (M.insert tid sched mp, ())+unregisterWorker tid = + atomicModifyIORef globalWorkerPool $ + \ mp -> (M.delete tid mp, ())+#else +amINested _ = return Nothing+registerWorker _ _ = return ()+unregisterWorker _tid = return () #endif - ----------------------------------------------------------------------------- -- Helpers #2: Pushing and popping work. -----------------------------------------------------------------------------@@ -210,28 +186,26 @@ popWork :: Sched -> IO (Maybe (Par ())) popWork Sched{ workpool, no } = do mb <- R.tryPopL workpool - if dbg - then case mb of - Nothing -> return Nothing- Just x -> do sn <- makeStableName mb- printf " [%d] -> POP work unit %d\n" no (hashStableName sn)- return mb- else return mb+ when dbg $ case mb of + Nothing -> return ()+ Just _ -> do sn <- makeStableName mb+ printf " [%d] -> POP work unit %d\n" no (hashStableName sn)+ return mb {-# INLINE pushWork #-} pushWork :: Sched -> Par () -> IO () pushWork Sched { workpool, idle, no, isMain } task = do--- modifyHotVar_ workpool (`pushL` task) R.pushL workpool task when dbg $ do sn <- makeStableName task printf " [%d] -> PUSH work unit %d\n" no (hashStableName sn)-#ifdef WAKEIDLE+#if defined(IDLING_ON) && defined(WAKEIDLE) --when isMain$ -- Experimenting with reducing contention by doing this only from a single thread. -- TODO: We need to have a proper binary wakeup-tree. tryWakeIdle idle #endif-+ return () +tryWakeIdle :: HotVar [MVar Bool] -> IO () tryWakeIdle idle = do -- NOTE: I worry about having the idle var hammmered by all threads on their spawn-path: -- If any worker is idle, wake one up and give it work to do.@@ -239,11 +213,11 @@ when (not (Prelude.null idles)) $ do when dbg$ printf "Waking %d idle thread(s).\n" (length idles) r <- modifyHotVar idle (\is -> case is of- [] -> ([], return ())- (i:is) -> (is, putMVar i False))+ [] -> ([], return ())+ (i:ils) -> (ils, putMVar i False)) r -- wake an idle worker up by putting an MVar. -rand :: HotVar GenIO -> IO Int+rand :: HotVar Random.GenIO -> IO Int rand ref = Random.uniformR (0, numCapabilities-1) =<< readHotVar ref --------------------------------------------------------------------------------@@ -253,18 +227,88 @@ instance NFData (IVar a) where rnf _ = () -runPar userComp = unsafePerformIO $ do- +{-# NOINLINE runPar #-}+runPar = unsafePerformIO . runParIO+++-- | This procedure creates a new worker on the current thread (with a+-- new session ID) and plugs it into the work-stealing environment.+-- This new worker extracts itself from the work stealing pool when+-- `userComp` has completed, thus freeing the current thread (this+-- procedure) to return normally.+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)) + _ <- 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 + -- Push the new session:+ _ <- modifyHotVar (sessions sched) (\ ls -> ((Session sid newFlag) : ls, ()))++ let userComp' = do when dbg$ io$ do+ tid2 <- myThreadId+ printf " [%d %s] Starting Par computation on %s.\n" (no sched) (show tid2) name+ ans <- userComp+ -- This add-on to userComp will run only after userComp has completed successfully,+ -- but that does NOT guarantee that userComp-forked computations have terminated:+ io$ do when (dbglvl>=1) $ do+ tid3 <- myThreadId+ printf " [%d %s] Continuation for %s called, finishing it up (%d)...\n" (no sched) (show tid3) name sid+ writeIORef ref ans+ writeHotVarRaw newFlag True+ modifyHotVar (activeSessions sched) (\ set -> (S.delete sid set, ()))+ kont :: Word64 -> a -> ROnly ()+ kont n = trivialCont$ "("++name++", sid "++show sid++", round "++show n++")"+ loop :: Word64 -> ROnly ()+ loop n = do flg <- liftIO$ readIORef newFlag+ 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+ rescheduleR 0 $ trivialCont$ "("++name++", sid "++show sid++")"+ loop (n+1)++ -- THIS IS RETURNING TOO EARLY!!:+ runReaderWith sched (C.runContT (unPar userComp') (kont 0)) -- Does this ASSUME child stealing?+ runReaderWith sched (loop 1)++ -- TODO: Ideally we would wait for ALL outstanding (stolen) work on this "team" to complete.++ when (dbglvl>=1)$ do+ active <- readHotVar (activeSessions sched)+ sess@True <- readHotVar newFlag -- ASSERT!+ printf " [%d %s] RETURN from %s (sessFin %s) runContT (%d) active set %s\n"+ (no sched) (show tid) name (show sess) sid (show active)++ -- Here we pop off the frame we added to the session stack:+ modifyHotVar_ (sessions sched) $ \ (Session sid2 _ : tl) ->+ if sid == sid2+ 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).+ readIORef ref+++{-# NOINLINE runParIO #-}+runParIO userComp = do+ tid <- myThreadId #if __GLASGOW_HASKELL__ >= 701 /* 20110301 */ --- -- We create a thread on each CPU with forkOnIO. The CPU on which+ -- We create a thread on each CPU with forkOn. The CPU on which -- the current thread is running will host the main thread; the -- other CPUs will host worker threads. -- -- Note: GHC 7.1.20110301 is required for this to work, because that -- is when threadCapability was added. --- (main_cpu, _) <- threadCapability =<< myThreadId+ (main_cpu, _) <- threadCapability tid #else -- -- Lacking threadCapability, we always pick CPU #0 to run the main@@ -274,60 +318,102 @@ -- let main_cpu = 0 #endif- allscheds <- makeScheds main_cpu-- m <- newEmptyMVar- forM_ (zip [0..] allscheds) $ \(cpu,sched) ->- forkOnIO cpu $- if (cpu /= main_cpu)- then do when dbg$ printf " [%d] Entering scheduling loop.\n" cpu- runReaderWith sched $ rescheduleR errK- when dbg$ printf " [%d] Exited scheduling loop. FINISHED.\n" cpu- else do- let userComp' = do when dbg$ io$ printf " [%d] Starting Par computation on main thread.\n" main_cpu- res <- userComp- finalSched <- RD.ask - when dbg$ io$ printf " [%d] Out of Par computation on main thread. Writing MVar...\n" (no finalSched)+ maybSched <- amINested tid+ 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. - -- Sanity check our work queues:- when dbg $ io$ sanityCheck allscheds- io$ putMVar m res- - RD.runReaderT (C.runContT (unPar userComp') trivialCont) sched- when dbg$ do putStrLn " *** Out of entire runContT user computation on main thread."- sanityCheck allscheds- -- Not currently requiring that other scheduler threads have exited before we - -- (the main thread) exit. But we do signal here that they should terminate:- writeIORef (killflag sched) True+ sid0 <- readHotVar (sessionCounter sched)+ when (dbglvl>=1)$ printf " [%d %s] runPar called from existing worker thread, new session (%d)....\n" (no sched) (show tid) (sid0 + 1)+ runNewSessionAndWait "nested runPar" sched userComp - when dbg$ do putStrLn " *** Reading final MVar on main thread."- takeMVar m -- Final value.+ ------------------------------------------------------------+ -- 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 + ----------------------------------------+ let wname = ("(worker "++show cpu++" of originator "++show tidorig++")")+-- forkOn cpu $ do+ _ <- forkWithExceptions (forkOn cpu) wname $ do + ------------------------------------------------------------STRT WORKER THREAD + tid2 <- myThreadId+ registerWorker tid2 sched+ if (cpu /= main_cpu)+ then do when dbg$ printf " [%d %s] Anonymous worker entering scheduling loop.\n" cpu (show tid2)+ runReaderWith sched $ rescheduleR 0 (trivialCont (wname++show tid2))+ when dbg$ printf " [%d] Anonymous worker exited scheduling loop. FINISHED.\n" cpu+ putMVar workerDone cpu+ return ()+ else do x <- runNewSessionAndWait "top-lvl main worker" sched userComp+ -- When the main worker finishes we can tell the anonymous "system" workers:+ 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 + unregisterWorker tid+ ------------------------------------------------------------END WORKER THREAD+ return (if cpu == main_cpu then Nothing else Just workerDone) --- Make sure there is no work left in any deque after exiting.-sanityCheck :: [Sched] -> IO ()-sanityCheck allscheds = do- forM_ allscheds $ \ Sched{no, workpool} -> do- b <- R.nullQ workpool- when (not b) $ do - printf "WARNING: After main thread exited non-empty queue remains for worker %d\n" no- putStrLn "Sanity check complete."+ when _WAIT_FOR_WORKERS $ do + when dbg$ printf " *** [%s] Originator thread: waiting for workers to complete." (show tidorig)+ forM_ (catMaybes doneFlags) $ \ mv -> do + n <- readMVar mv+ -- n <- A.wait mv+ when dbg$ printf " * [%s] Worker %s completed\n" (show tidorig) (show n) + when dbg$ do printf " *** [%s] Reading final MVar on originator thread.\n" (show tidorig)+ -- We don't directly use the thread we come in on. Rather, that thread waits+ -- waits. One reason for this is that the main/progenitor thread in+ -- GHC is expensive like a forkOS thread.+ ----------------------------------------+ -- DEBUGGING -- +-- takeMVar mfin -- Final value.+-- dbgTakeMVar "global waiting thread" mfin -- Final value.+ busyTakeMVar (" The global wait "++ show tidorig) mfin -- Final value. + ---------------------------------------- -- Create the default scheduler(s) state:-makeScheds main = do+makeScheds :: Int -> IO [Sched]+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 - idle <- newHotVar [] - killflag <- newHotVar False- let allscheds = [ Sched { no=x, idle, killflag, isMain= (x==main),- workpool=wp, scheds=allscheds, rng=rng- }- | (x,wp,rng) <- zip3 [0..] workpools rngs]+ idle <- newHotVar []+ -- The STACKs are per-worker.. but the root finished flag is shared between all anonymous system workers:+ sessionFinished <- newHotVar False+ sessionStacks <- mapM newHotVar (replicate numCapabilities [Session baseSessionID sessionFinished])+ activeSessions <- newHotVar S.empty+ sessionCounter <- newHotVar (baseSessionID + 1)+ let allscheds = [ Sched { no=x, idle, isMain= (x==main),+ workpool=wp, scheds=allscheds, rng=rng,+ sessions = stck,+ activeSessions=activeSessions,+ sessionCounter=sessionCounter+ }+ -- | (x,wp,rng,stck) <- zip4 [0..] workpools rngs sessionStacks+ | x <- [0 .. numCapabilities-1]+ | wp <- workpools+ | rng <- rngs+ | stck <- sessionStacks+ ] return allscheds +-- The ID of top-level runPar sessions.+baseSessionID :: SessionID+baseSessionID = 1000 + -------------------------------------------------------------------------------- -- IVar operations --------------------------------------------------------------------------------@@ -342,146 +428,217 @@ -- | 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 iv@(IVar v) = do - callCC $ \cont -> +get (IVar vr) = do + callCC $ \kont -> do- e <- io$ readIORef v+ e <- io$ readIORef vr case e of Full a -> return a _ -> do sch <- RD.ask # ifdef DEBUG- sn <- io$ makeStableName iv+ sn <- io$ makeStableName vr -- Should probably do the MutVar inside... let resched = trace (" ["++ show (no sch) ++ "] - Rescheduling on unavailable ivar "++show (hashStableName sn)++"!") #else let resched = # endif- reschedule+ longjmpSched -- Invariant: kont must not be lost. -- Because we continue on the same processor the Sched stays the same:- -- TODO: Try NOT using monads as first class values here. Check for performance effect:- r <- io$ atomicModifyIORef v $ \e -> case e of- Empty -> (Blocked [pushWork sch . cont], resched)- Full a -> (Full a, return a)- Blocked ks -> (Blocked (pushWork sch . cont:ks), resched)+ -- TODO: Try NOT using monadic values as first class. Check for performance effect:+ r <- io$ atomicModifyIORef vr $ \x -> case x of+ Empty -> (Blocked [pushWork sch . kont], resched)+ Full a -> (Full a, return a) -- kont is implicit here.+ Blocked ks -> (Blocked (pushWork sch . kont:ks), resched) r -- | NOTE unsafePeek is NOT exposed directly through this module. (So -- this module remains SAFE in the Safe Haskell sense.) It can only -- be accessed by importing Control.Monad.Par.Unsafe. {-# INLINE unsafePeek #-}-unsafePeek iv@(IVar v) = do +unsafePeek :: IVar a -> Par (Maybe a)+unsafePeek (IVar v) = do e <- io$ readIORef v case e of Full a -> return (Just a) _ -> return Nothing +------------------------------------------------------------ {-# INLINE put_ #-} -- | @put_@ is a version of @put@ that is head-strict rather than fully-strict.-put_ iv@(IVar v) !content = do- sched <- RD.ask - io$ do - ks <- atomicModifyIORef v $ \e -> case e of+-- In this scheduler, puts immediately execute woken work in the current thread.+put_ (IVar vr) !content = do+ sched <- RD.ask+ ks <- io$ do + ks <- atomicModifyIORef vr $ \e -> case e of Empty -> (Full content, []) Full _ -> error "multiple put" Blocked ks -> (Full content, ks)- #ifdef DEBUG- sn <- makeStableName iv- printf " [%d] Put value %s into IVar %d. Waking up %d continuations.\n" - (no sched) (show content) (hashStableName sn) (length ks)-#endif- mapM_ ($content) ks- return ()-+ when (dbglvl >= 3) $ do + sn <- makeStableName vr+ printf " [%d] Put value %s into IVar %d. Waking up %d continuations.\n" + (no sched) (show content) (hashStableName sn) (length ks)+ return ()+#endif + return ks+ 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 -- be accessed by importing Control.Monad.Par.Unsafe. {-# INLINE unsafeTryPut #-}-unsafeTryPut iv@(IVar v) !content = do+unsafeTryPut (IVar vr) !content = do -- Head strict rather than fully strict. sched <- RD.ask - io$ do - (ks,res) <- atomicModifyIORef v $ \e -> case e of+ (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- sn <- makeStableName iv+ sn <- makeStableName vr printf " [%d] unsafeTryPut: value %s in IVar %d. Waking up %d continuations.\n" - (no sched) (show content) (hashStableName sn) (length ks)+ (no sched) (show content) (hashStableName sn) (length (fst pr)) #endif- mapM_ ($content) ks- return res+ return pr+ wakeUp sched ks content+ return res +-- | When an IVar is filled in, continuations wake up.+{-# INLINE wakeUp #-}+wakeUp :: Sched -> [a -> IO ()]-> a -> Par ()+wakeUp _sched ks arg = loop ks+ where+ loop [] = return ()+ loop (kont:rest) = do+ -- FIXME -- without strict firewalls keeping ivars from moving+ -- between runPar sessions, if we allow nested scheduler use+ -- we could potentially wake up work belonging to a different+ -- runPar and thus bring it into our worker and delay our own+ -- continuation until its completion.+ if _PARPUTS then+ -- We do NOT force the putting thread to postpone its continuation.+ do spawn_$ pMap kont rest+ return ()+ -- case rest of+ -- [] -> spawn_$ io$ kont arg+ -- _ -> spawn_$ do spawn_$ io$ kont arg+ -- io$ parchain rest+ -- error$"FINISHME - wake "++show (length ks)++" conts"+ else + -- This version sacrifices a parallelism opportunity and+ -- imposes additional serialization.+ --+ -- [2012.08.31] WARNING -- this serialzation CAN cause deadlock.+ -- This "optimization" should not be on the table.+ -- mapM_ ($arg) ks+ do io$ kont arg+ loop rest + return () --- TODO: Continuation (parent) stealing version.+ pMap kont [] = io$ kont arg+ pMap kont (more:rest) =+ do spawn_$ io$ kont arg+ pMap more rest++ -- parchain [kont] = kont arg+ -- parchain (kont:rest) = do spawn$ io$ kont arg+ -- parchain rest+ ++------------------------------------------------------------ {-# INLINE fork #-} fork :: Par () -> Par ()-#ifdef FORKPARENT-#warning "FORK PARENT POLICY USED"-fork task = do - sched <- RD.ask - callCC$ \parent -> do- let wrapped = parent ()- -- Is it possible to slip in a new Sched here?- -- let wrapped = lift$ RD.runReaderT (parent ()) undefined- io$ pushWork sched wrapped- -- Then execute the child task and return to the scheduler when it is complete:- task - -- If we get to this point we have finished the child task:- reschedule -- We reschedule to pop the cont we pushed.- io$ putStrLn " !!! ERROR: Should not reach this point #1" +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 + 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 + -- 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 - io$ printf " called parent continuation... was on cpu %d now on cpu %d\n" (no sched) (no sched2)+ 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 () -#else-fork task = do- sch <- RD.ask- io$ when dbg$ printf " [%d] forking task...\n" (no sch)- io$ pushWork sch task-#endif+ 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.-reschedule :: Par a -reschedule = Par $ C.ContT rescheduleR+longjmpSched :: Par a+-- longjmpSched = Par $ C.ContT rescheduleR+longjmpSched = Par $ C.ContT (\ _k -> rescheduleR 0 (trivialCont "longjmpSched")) --- Reschedule ignores its continuation.--- It runs the scheduler loop indefinitely, until it observers killflag==True-rescheduleR :: ignoredCont -> ROnly ()-rescheduleR k = do+-- Reschedule the scheduler loop until it observes sessionFinished==True, and+-- then it finally invokes its continuation.+rescheduleR :: Word64 -> (a -> ROnly ()) -> ROnly ()+rescheduleR cnt kont = do mysched <- RD.ask - when dbg$ liftIO$ printf " [%d] - Reschedule...\n" (no mysched)+ when dbg$ liftIO$ do tid <- myThreadId+ sess <- readSessions mysched+ null <- R.nullQ (workpool mysched)+ printf " [%d %s] - Reschedule #%d... sessions %s, pool empty %s\n"+ (no mysched) (show tid) cnt (show sess) (show null) mtask <- liftIO$ popWork mysched case mtask of- Nothing -> do k <- liftIO$ readIORef (killflag mysched) - unless k $ do + Nothing -> do+ (Session _ finRef):_ <- liftIO$ readIORef $ sessions mysched+ 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" + (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) + + kont (error "Direct.hs: The result value from rescheduleR should not be used.")+ else do+ -- when (dbglvl >= 1) $ liftIO $ do+ -- tid <- myThreadId + -- sess <- readSessions mysched+ -- 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 -- io$ tryWakeIdle (idle mysched) #endif- rescheduleR errK+ liftIO yield+ rescheduleR (cnt+1) kont Just task -> do -- 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 -- 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 errK)--{-# INLINE runReaderWith #-}-runReaderWith state m = RD.runReaderT m state+ 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 () steal mysched@Sched{ idle, scheds, rng, no=my_no } = do- when dbg$ printf " [%d] + stealing\n" my_no+ when (dbglvl>=2)$ do tid <- myThreadId+ printf " [%d %s] + stealing\n" my_no (show tid) i <- getnext (-1 :: Int) go maxtries i where@@ -492,13 +649,13 @@ ---------------------------------------- -- IDLING behavior:- go 0 _ = + go 0 _ | _IDLING_ON = do m <- newEmptyMVar r <- modifyHotVar idle $ \is -> (m:is, is) if length r == numCapabilities - 1 then do when dbg$ printf " [%d] | initiating shutdown\n" my_no- mapM_ (\m -> putMVar m True) r+ mapM_ (\vr -> putMVar vr True) r else do done <- takeMVar m if done@@ -509,14 +666,19 @@ when dbg$ printf " [%d] | woken up\n" my_no i <- getnext (-1::Int) go maxtries i++ -- We need to return from this loop to check sessionFinished and exit the scheduler if necessary.+ go 0 _i | _IDLING_ON == False = yield+ ---------------------------------------- go tries i | i == my_no = do i' <- getnext i go (tries-1) i' | otherwise = do+ -- We ONLY go through the global sched array to access victims: let schd = scheds!!i- when dbg$ printf " [%d] | trying steal from %d\n" my_no (no schd)+ when (dbglvl>=2)$ printf " [%d] | trying steal from %d\n" my_no (no schd) -- let dq = workpool schd :: WSDeque (Par ()) let dq = workpool schd @@ -536,10 +698,15 @@ Nothing -> do i' <- getnext i go (tries-1) i' -errK = error "this closure shouldn't be used"-trivialCont _ = +-- | The continuation which should not be called.+errK :: t+errK = error "Error cont: this closure shouldn't be used"++trivialCont :: String -> a -> ROnly ()+trivialCont str _ = do #ifdef DEBUG- trace "trivialCont evaluated!"+-- trace (str ++" trivialCont evaluated!")+ liftIO$ printf " !! trivialCont evaluated, msg: %s\n" str #endif return () @@ -580,23 +747,29 @@ spawn :: (Show a, NFData a) => Par a -> Par (IVar a) spawn_ :: Show a => Par a -> Par (IVar a) spawn1_ :: (Show a, Show b) => (a -> Par b) -> a -> Par (IVar b)+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 newFull :: (Show a, NFData a) => a -> Par (IVar a) newFull_ :: Show a => a -> Par (IVar a)+unsafeTryPut :: Show b => IVar b -> b -> Par b #else spawn :: NFData a => Par a -> Par (IVar a) spawn_ :: Par a -> Par (IVar a) spawn1_ :: (a -> Par b) -> a -> Par (IVar b)+spawnP :: NFData a => a -> Par (IVar a) 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 newFull :: NFData a => a -> Par (IVar a) newFull_ :: a -> Par (IVar a)-+unsafeTryPut :: IVar b -> b -> Par b +-- We can't make proper instances with the extra Show constraints: instance PC.ParFuture IVar Par where get = get spawn = spawn@@ -624,3 +797,92 @@ pure = return -- </boilerplate> --------------------------------------------------------------------------------+++{-# INLINE runReaderWith #-}+-- | Arguments flipped for convenience.+runReaderWith :: r -> RD.ReaderT r m a -> m a+runReaderWith state m = RD.runReaderT m state+++--------------------------------------------------------------------------------+-- DEBUGGING TOOLs+--------------------------------------------------------------------------------++-- Make sure there is no work left in any deque after exiting.+sanityCheck :: [Sched] -> IO ()+sanityCheck allscheds = do+ forM_ allscheds $ \ Sched{no, workpool} -> do+ b <- R.nullQ workpool+ 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"+++-- | This tries to localize the blocked-indefinitely exception:+dbgTakeMVar :: String -> MVar a -> IO a+dbgTakeMVar msg mv = +-- catch (takeMVar mv) ((\_ -> doDebugStuff) :: BlockedIndefinitelyOnMVar -> IO a)+ E.catch (takeMVar mv) ((\_ -> doDebugStuff) :: IOError -> IO a)+ 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 + tid <- myThreadId+ -- After we've failed enough times, start complaining:+ printf "%s not getting anywhere, msg: %s\n" (show tid) msg + try (100 * 1000)+ try n = do+ x <- tryTakeMVar mv+ 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 -> + case (e :: E.BlockedIndefinitelyOnMVar) of+ _ -> do + putStrLn$"CAUGHT child thread exception: "++show e + return ()+ )+ action+++-- | Exceptions that walk up the fork tree of threads:+forkWithExceptions :: (IO () -> IO ThreadId) -> String -> IO () -> IO ThreadId+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 + "\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)+ E.throwTo parent (e :: E.SomeException)+ )+++-- Do all the memory reads to snapshot the current session stack:+readSessions :: Sched -> IO [(SessionID, Bool)]+readSessions sched = do+ 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
@@ -0,0 +1,160 @@+{-# 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+-- the internals of the scheduler (i.e. the private `Par` type constructor).++module Control.Monad.Par.Scheds.DirectInternal where++import Control.Applicative+import "mtl" Control.Monad.Cont as C+import qualified "mtl" Control.Monad.Reader as RD++import qualified System.Random.MWC as Random++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 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)++-- Our monad stack looks like this:+-- ---------+-- ContT+-- ReaderT+-- IO+-- ---------+-- 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.+--+newtype Par a = Par { unPar :: C.ContT () ROnly a }+ deriving (Monad, MonadCont, RD.MonadReader Sched)+type ROnly = RD.ReaderT Sched IO++type SessionID = Word64++-- An ID along with a flag to signal completion:+data Session = Session SessionID (HotVar Bool)++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? ++ -- 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. + -- (2) The original invocation of runPar also counts as a session+ -- 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.+ activeSessions :: HotVar (S.Set SessionID),++ -- A counter to support unique session IDs:+ sessionCounter :: HotVar SessionID+ }+++--------------------------------------------------------------------------------+-- Helpers #1: Atomic Variables+--------------------------------------------------------------------------------+-- TEMP: Experimental++#ifndef HOTVAR+#define HOTVAR 1+#endif+newHotVar :: a -> IO (HotVar a)+modifyHotVar :: HotVar a -> (a -> (a,b)) -> IO b+modifyHotVar_ :: HotVar a -> (a -> a) -> IO ()+writeHotVar :: HotVar a -> a -> IO ()+readHotVar :: HotVar a -> IO a+-- readHotVarRaw :: HotVar a -> m a+-- writeHotVarRaw :: HotVar a -> m a++{-# INLINE newHotVar #-}+{-# INLINE modifyHotVar #-}+{-# INLINE modifyHotVar_ #-}+{-# INLINE readHotVar #-}+{-# INLINE writeHotVar #-}+++#if HOTVAR == 1+type HotVar a = IORef a+newHotVar = newIORef+modifyHotVar = atomicModifyIORef+modifyHotVar_ v fn = atomicModifyIORef v (\a -> (fn a, ()))+readHotVar = readIORef+writeHotVar = writeIORef+instance Show (IORef a) where + show ref = "<ioref>"++-- hotVarTransaction = id+hotVarTransaction = error "Transactions not currently possible for IO refs"+readHotVarRaw = readHotVar+writeHotVarRaw = writeHotVar+++#elif HOTVAR == 2 +#warning "Using MVars for hot atomic variables."+-- This uses MVars that are always full with *something*+type HotVar a = MVar a+newHotVar x = do v <- newMVar; putMVar v x; return v+modifyHotVar v fn = modifyMVar v (return . fn)+modifyHotVar_ v fn = modifyMVar_ v (return . fn)+readHotVar = readMVar+writeHotVar v x = do swapMVar v x; return ()+instance Show (MVar a) where + show ref = "<mvar>"++-- hotVarTransaction = id+-- We could in theory do this by taking the mvar to grab the lock.+-- But we'd need some temporary storage....+hotVarTransaction = error "Transactions not currently possible for MVars"+readHotVarRaw = readHotVar+writeHotVarRaw = writeHotVar+++#elif HOTVAR == 3+#warning "Using TVars for hot atomic variables."+-- 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 + 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 + show ref = "<tvar>"++hotVarTransaction = atomically+readHotVarRaw = readTVar+writeHotVarRaw = writeTVar++#endif
Control/Monad/Par/Scheds/Trace.hs view
@@ -11,7 +11,7 @@ -} module Control.Monad.Par.Scheds.Trace (- Par, runPar, fork,+ Par, runPar, runParIO, fork, IVar, new, newFull, newFull_, get, put, put_, spawn, spawn_, spawnP ) where
Control/Monad/Par/Scheds/TraceInternal.hs view
@@ -1,7 +1,7 @@ {-# LANGUAGE RankNTypes, NamedFieldPuns, BangPatterns,- ExistentialQuantification, CPP, ParallelListComp+ ExistentialQuantification, CPP #-}-{- OPTIONS_GHC -Wall -fno-warn-name-shadowing -fno-warn-unused-do-bind -}+{-# OPTIONS_GHC -Wall -fno-warn-name-shadowing -fno-warn-unused-do-bind #-} -- | This module exposes the internals of the @Par@ monad so that you -- can build your own scheduler or other extensions. Do not use this@@ -12,28 +12,24 @@ Trace(..), Sched(..), Par(..), IVar(..), IVarContents(..), sched,- runPar, runParAsync, runParAsyncHelper,+ runPar, runParIO, runParAsync,+ -- runParAsyncHelper, new, newFull, newFull_, get, put_, put, pollIVar, yield, ) where -import Control.Monad as M hiding (sequence, join)-import Prelude hiding (mapM, sequence)+import Control.Monad as M hiding (mapM, sequence, join)+import Prelude hiding (mapM, sequence, head,tail) import Data.IORef import System.IO.Unsafe import Control.Concurrent hiding (yield) import GHC.Conc hiding (yield) import Control.DeepSeq import Control.Applicative-import Data.Array-import Data.List (partition, find)---import Text.Printf-+-- import Text.Printf -- ------------------------------------------------------------------------------ MAIN SCHEDULING AND RUNNING--- --------------------------------------------------------------------------- data Trace = forall a . Get (IVar a) (a -> Trace) | forall a . Put (IVar a) a Trace@@ -42,63 +38,9 @@ | Done | Yield Trace -data Sched = Sched- { no :: {-# UNPACK #-} !ThreadNumber,- -- ^ The threadnumber of this worker- workpool :: IORef WorkPool,- -- ^ The workpool for this worker- status :: IORef AllStatus,- -- ^ The Schedulers' status- scheds :: Array ThreadNumber Sched,- -- ^ The list of all workers by thread- tId :: IORef ThreadId- -- ^ The ThreadId of this worker- }--type ThreadNumber = Int-type UId = Int-type CountRef = IORef Int-type WorkLimit = (UId, CountRef)--- ^ The UId and the count of tasks left or Nothing if there's no limit--- When the UId is -1, it means that the worker will remain alive until --- purposely killed (by globalThreadShutdown).------ The reason for a work limit is to make sure that nested threads properly exit.--- Imagine a scenario where thread A, a worker thread, encounters a runPar. It --- recursively enters worker status, but it needs ot leave worker status at some --- point to finish the task that caused it to call runPar. Suppose now that it --- encounters another call to runPar. If it has the ability to finish and return, --- we must make sure it returns first for the nested runPar or else it will return --- to the wrong place! The work limit helps achieve this.------ TODO: Perhaps the work limit need not restrict what a thread can work on, but --- instead it simply provides the singular point that a thread is allowed to return --- from. The only concern is some potential for bad blocking - is that a legit --- concern?--isWLUId :: WorkLimit -> (UId -> Bool) -> Bool---isWLUId Nothing _ = False-isWLUId (uid, _) op = op uid--shouldEndWorkSet :: WorkLimit -> IO Bool-shouldEndWorkSet (u,_) | u == -1 = return False-shouldEndWorkSet (_, cr) = do- c <- readIORef cr- return (c == 0)--idleAtWL :: WorkLimit -> MVar Bool -> Idle---idleAtWL Nothing m = Idle Nothing m-idleAtWL (uid, _) m = Idle uid m- -- | The main scheduler loop.--- This takes the synchrony flag, our Sched, the particular work queue we're--- currently working on, the uid of the work queue (for pushing work), our --- work limit, and the already-popped, first trace in the work queue.------ INVARIANT: This should only be called by threads who ARE currently marked--- as working.-sched :: Bool -> WorkLimit -> Sched -> (IORef [Trace]) -> UId -> Trace -> IO ()-sched _doSync wl q@Sched{status, workpool} queueref uid t = loop t+sched :: Bool -> Sched -> Trace -> IO ()+sched _doSync queue t = loop t where loop t = case t of New a f -> do@@ -110,74 +52,50 @@ Full a -> loop (c a) _other -> do r <- atomicModifyIORef v $ \e -> case e of- Empty -> (Blocked [c], go)- Full a -> (Full a, loop (c a))- Blocked cs -> (Blocked (c:cs), go)+ Empty -> (Blocked [c], reschedule queue)+ Full a -> (Full a, loop (c a))+ Blocked cs -> (Blocked (c:cs), reschedule queue) r Put (IVar v) a t -> do cs <- atomicModifyIORef v $ \e -> case e of Empty -> (Full a, []) Full _ -> error "multiple put" Blocked cs -> (Full a, cs)- mapM_ (pushWork status uid queueref . ($a)) cs+ mapM_ (pushWork queue. ($a)) cs loop t Fork child parent -> do- pushWork status uid queueref child+ pushWork queue child loop parent Done -> if _doSync- then go- -- We could fork an extra thread here to keep numCapabilities workers- -- even when the main thread returns to the runPar caller...- else do -- putStrLn " [par] Forking replacement thread..\n"- forkIO go; return ()- -- But even if we don't we are not orphaning any work in this- -- thread's work-queue because it can be stolen by other threads.- -- else return ()+ then reschedule queue+-- We could fork an extra thread here to keep numCapabilities workers+-- even when the main thread returns to the runPar caller...+ else do putStrLn " [par] Forking replacement thread..\n"+ forkIO (reschedule queue); return ()+-- But even if we don't we are not orphaning any work in this+-- threads work-queue because it can be stolen by other threads.+-- else return () Yield parent -> do -- Go to the end of the worklist:+ let Sched { workpool } = queue -- TODO: Perhaps consider Data.Seq here.- -- This would also be a chance to steal and work from opposite ends of the queue.- atomicModifyIORef queueref $ \ts -> (ts++[parent],())- go- go = do- mt <- atomicPopIORef queueref- case mt of- Just t -> loop t- Nothing -> do- -- SCARY: we better be working on the top queue in the pool!- cr <- wpRemoveWork uid workpool- workDone <- decWorkerCount uid cr status- -- If this uid is our workLimit id AND worker count == 0, then - -- we should just return () rather than calling reschedule q- unless (isWLUId wl (== uid) && workDone) $- reschedule wl q-+ -- This would also be a chance to steal and work from opposite ends of the queue.+ atomicModifyIORef workpool $ \ts -> (ts++[parent], ())+ reschedule queue --- | Process the next work queue on the work pool, or failing that, go into --- work stealing mode.------ INVARIANT: This should only be called by threads who are NOT currently --- marked as working (or if they are, the task they were working --- on executed a runPar).-reschedule :: WorkLimit -> Sched -> IO ()-reschedule wl q@Sched{ workpool, status } = do- wp <- readIORef workpool- case wp of- Work uid cr wqref _ | isWLUId wl (uid >=) -> do- incWorkerCount cr- nextTrace <- atomicPopIORef wqref- case nextTrace of- Just t -> sched True wl q wqref uid t- Nothing -> do- wpRemoveWork uid workpool- workDone <- decWorkerCount uid cr status- -- If this uid is our workLimit id AND worker count == 0, then - -- we should just return () rather than calling reschedule q- unless (isWLUId wl (== uid) && workDone) $- reschedule wl q- _ -> steal wl q+-- | Process the next item on the work queue or, failing that, go into+-- work-stealing mode.+reschedule :: Sched -> IO ()+reschedule queue@Sched{ workpool } = do+ e <- atomicModifyIORef workpool $ \ts ->+ case ts of+ [] -> ([], Nothing)+ (t:ts') -> (ts', Just t)+ case e of+ Nothing -> steal queue+ Just t -> sched True queue t -- RRN: Note -- NOT doing random work stealing breaks the traditional@@ -185,162 +103,57 @@ -- parallel) programs. -- | Attempt to steal work or, failing that, give up and go idle.-steal :: WorkLimit -> Sched -> IO ()-steal wl q@Sched{ status, scheds, no=my_no } = - -- printf "cpu %d stealing\n" my_no >> - go l+steal :: Sched -> IO ()+steal q@Sched{ idle, scheds, no=my_no } = do+ -- printf "cpu %d stealing\n" my_no+ go scheds where- (l,u) = bounds scheds- go n- | n > u = do- -- Prepare to go idle- m <- newEmptyMVar- atomicModifyIORef status $ addIdler (idleAtWL wl m)- -- Check to see if this workset is ready to close- s <- shouldEndWorkSet wl- if s- then do- -- Time to close this workset- --printf "cpu %d shutting down workset %d\n" my_no myPriLimit- endWorkSet status (fst wl)- return ()- else do- -- There's more work being done here, so I'll go idle- finished <- takeMVar m- unless finished $ go l- | n == my_no = go (n+1)- | otherwise = readIORef (workpool (scheds!n)) >>= tryToSteal- where- tryToSteal (Work uid cr wqref wp) | isWLUId wl (uid >=) = do- incWorkerCount cr- stolenTrace <- atomicPopIORef wqref- case stolenTrace of- Nothing -> decWorkerCount uid cr status >> tryToSteal wp- Just t -> do- sublst <- newIORef []- atomicModifyIORef (workpool q) $ \wp' -> (Work uid cr sublst wp', ())- sched True wl q sublst uid t- tryToSteal _ = go (n+1)----- ------------------------------------------------------------------------------ UTILITY FUNCTIONS--- ------------------------------------------------------------------------------- | Push work. Then, find an idle worker with uid less than the pushed work.--- If one is found, wake it up.-pushWork :: IORef AllStatus -> UId -> (IORef [Trace]) -> Trace -> IO ()-pushWork status uid wqref t = do- atomicModifyIORef wqref $ (\ts -> (t:ts, ()))- allstatus <- readIORef status- when (hasIdleWorker uid allstatus) $ do- r <- atomicModifyIORef status $ getIdleWorker uid- case r of- Just b -> putMVar b False- Nothing -> return ()---- | A utility function for decreasing the task count of a work set.--- If the count becomes 0, endWorkSet is called on the work set.-decWorkerCount :: UId -> CountRef -> IORef AllStatus -> IO Bool-decWorkerCount uid countref status = do- done <- atomicModifyIORef countref $ - (\n -> if n == 0 then error "Impossible value in decWorkerCount" else (n-1, n == 1))- when done $ (endWorkSet status uid >> globalWorkComplete uid)- return done---- | A utility function for increasing the task count of a work set.-incWorkerCount :: CountRef -> IO ()-incWorkerCount countref = do- atomicModifyIORef countref $ (\n -> (n+1, ()))---- | A utility for popping an element off of an IORef list.--- The return value is Just a where a is the head of the list--- or Nothing if the list is null.-atomicPopIORef :: IORef [a] -> IO (Maybe a)-atomicPopIORef ref = atomicModifyIORef ref $ \lst ->- case lst of- [] -> ([], Nothing)- (e:es) -> (es, Just e)---- ------------------------------------------------------------------------------ IDLING STATUS--- -----------------------------------------------------------------------------data Idle = Idle {-# UNPACK #-} !UId (MVar Bool)-data ExtIdle = ExtIdle {-# UNPACK #-} !UId (MVar ())-type AllStatus = ([Idle], [ExtIdle])---- | A new empty PQueue of Statuses-newStatus :: AllStatus-newStatus = ([], [])---- | Adds a new Idler to the AllStatus.-addIdler :: Idle -> AllStatus -> (AllStatus, ())-addIdler i@(Idle u _) (is, es) = ((insert is, es), ())- where insert [] = [i]- insert xs@(i'@(Idle u' _):xs') = if u <= u'- then i : xs- else i' : insert xs'---- | Adds a new External idler to the AllStatus.-addExtIdler :: ExtIdle -> AllStatus -> (AllStatus, ())-addExtIdler e (is, es) = ((is, e:es), ())---- | Returns an idle worker with uid less than or equal to the given one --- (if it exists) and removes it from the AllStatus-getIdleWorker :: UId -> AllStatus -> (AllStatus, Maybe (MVar Bool))-getIdleWorker u q = case q of- ([],_) -> (q, Nothing)- ((Idle u' m'):rst, es) -> if u' <= u then ((rst,es), Just m') else (q, Nothing)---- | Returns true if there is an idle worker with uid less than the given one-hasIdleWorker :: UId -> AllStatus -> Bool-hasIdleWorker uid q = case getIdleWorker uid q of- (_, Nothing) -> False- (_, Just _) -> True---- | Wakes up all idle workers at the given uid with the True signal-endWorkSet :: IORef AllStatus -> UId -> IO ()-endWorkSet status uid = do- (is, es) <- atomicModifyIORef status $ getAllAtID- mapM_ (\(ExtIdle _ mb) -> putMVar mb ()) es- mapM_ (\(Idle _ mb) -> putMVar mb True) is- where- getAllAtID (is, es) = ((is', es'), (elems1, elems2))- where- (elems1, is') = partition (\(Idle u _) -> u == uid) is- (elems2, es') = partition (\(ExtIdle u _) -> u == uid) es----- ------------------------------------------------------------------------------ WorkPool--- ------------------------------------------------------------------------------- | The WorkPool keeps a queue where each element has a UId, a list of --- traces, and the countRef of how many workers are working on Traces --- of this UId.------ It should be that by the natural pushing done in sched, this pool --- should always be in order. We take advantage of this by making --- guarantees but not actually checking at runtime whether they're true.-data WorkPool = Work {-# UNPACK #-} !UId CountRef (IORef [Trace]) WorkPool | NoWork---- | Pop the next work queue from the work pool. This should only be called --- if both the work pool contains a pool, and the queue in that pool is --- empty. Thus, it should only be called by the pool's owner.-wpRemoveWork :: UId -> IORef WorkPool -> IO CountRef-wpRemoveWork uid pRef = atomicModifyIORef pRef f- where f :: WorkPool -> (WorkPool, CountRef)- f (Work uid' cr' _ p') | uid == uid' = (p', cr')- f (Work uid' cr' wq' p') = - let (p'', cr'') = f p'- in (Work uid' cr' wq' p'', cr'')- f NoWork = error "Impossible state in wpRemoveWork"+ go [] = do m <- newEmptyMVar+ r <- atomicModifyIORef idle $ \is -> (m:is, is)+ if length r == numCapabilities - 1+ then do+ -- printf "cpu %d initiating shutdown\n" my_no+ mapM_ (\m -> putMVar m True) r+ else do+ done <- takeMVar m+ if done+ then do+ -- printf "cpu %d shutting down\n" my_no+ return ()+ else do+ -- printf "cpu %d woken up\n" my_no+ go scheds+ go (x:xs)+ | no x == my_no = go xs+ | otherwise = do+ r <- atomicModifyIORef (workpool x) $ \ ts ->+ case ts of+ [] -> ([], Nothing)+ (x:xs) -> (xs, Just x)+ case r of+ Just t -> do+ -- printf "cpu %d got work from cpu %d\n" my_no (no x)+ sched True q t+ Nothing -> go xs +-- | If any worker is idle, wake one up and give it work to do.+pushWork :: Sched -> Trace -> IO ()+pushWork Sched { workpool, idle } t = do+ atomicModifyIORef workpool $ \ts -> (t:ts, ())+ idles <- readIORef idle+ when (not (null idles)) $ do+ r <- atomicModifyIORef idle (\is -> case is of+ [] -> ([], return ())+ (i:is) -> (is, putMVar i False))+ r -- wake one up --- ------------------------------------------------------------------------------ PAR AND IVAR--- ---------------------------------------------------------------------------+data Sched = Sched+ { no :: {-# UNPACK #-} !Int,+ workpool :: IORef [Trace],+ idle :: IORef [MVar Bool],+ scheds :: [Sched] -- Global list of all per-thread workers.+ }+-- deriving Show newtype Par a = Par { runCont :: (a -> Trace) -> Trace@@ -360,14 +173,12 @@ newtype IVar a = IVar (IORef (IVarContents a)) -- data IVar a = IVar (IORef (IVarContents a)) -data IVarContents a = Full a | Empty | Blocked [a -> Trace]- -- Forcing evaluation of a IVar is fruitless. instance NFData (IVar a) where rnf _ = () --- From outside the Par computation we can peek. But this is--- nondeterministic; it should perhaps have "unsafe" in the name.++-- From outside the Par computation we can peek. But this is nondeterministic. pollIVar :: IVar a -> IO (Maybe a) pollIVar (IVar ref) = do contents <- readIORef ref@@ -376,220 +187,106 @@ _ -> return (Nothing) --- ------------------------------------------------------------------------------ GLOBAL THREAD IDENTIFICATION--- ------------------------------------------------------------------------------- Global thread identification is handled byt the globalThreadState object.--- The main way to interact with this object is to attempt to establish global --- Scheds, shut down the threads and clear the Scheds, or to mark a work set --- as complete.--data GlobalThreadState = GTS (Array ThreadNumber Sched) !UId !Int---- | This is the global thread state variable-globalThreadState :: IORef (Maybe GlobalThreadState)-globalThreadState = unsafePerformIO $ newIORef $ Nothing---- | This is called when a work set completes (see decWorkerCount).--- We do this so that we can know if it's okay to do a --- globalThreadShutdown.-globalWorkComplete :: UId -> IO ()-globalWorkComplete _ = - atomicModifyIORef globalThreadState f- where f Nothing = error "Impossible state in globalWorkComplete."- f (Just (GTS retA n c)) = (Just (GTS retA n (c+1)), ())---- | Attempts to set the global Scheds. If they are already extablished, --- this returns a Failure with a new UId (to interact with the global --- threads) and the current global Scheds. Otherwise, it establishes --- the given array as the global Scheds, and returns a Success containing --- the UId to use.-data GTSResult = Success UId | Failure UId (Array ThreadNumber Sched)-globalEstablishScheds :: Array ThreadNumber Sched -> IO GTSResult-globalEstablishScheds a = - atomicModifyIORef globalThreadState f- where f Nothing = (Just (GTS a 1 0), Success 0)- f (Just (GTS retA n c)) = (Just (GTS retA (n+1) c), Failure n retA)---- | Attempts to shutdown the global threads. If there are unfinished tasks, --- this shuts down nothing and returns False. Otherwise, this shuts down --- all threads, un-establishes the global Scheds, and returns True.--- If the Scheds are currently unestablished, this does nothing and returns --- False.------ TODO: This can sometimes leave threads hanging who are not doing any work --- but have not yet marked themselves as idle. Things won't exactly --- break, but there may be MVar errors that are thrown.-globalThreadShutdown :: IO Bool-globalThreadShutdown = do - ma <- atomicModifyIORef globalThreadState f- case ma of- Nothing -> return False- Just a -> do- let s = status $ a ! (fst $ bounds a)- (is, es) <- atomicModifyIORef s $ \x -> (newStatus, x)- mapM_ (\(ExtIdle _ m) -> putMVar m ()) es- mapM_ (\(Idle _ mb) -> putMVar mb True) is- return True- where f (Just (GTS a n c)) | n == c = (Nothing, Just a)- f gts = (gts, Nothing)----- ------------------------------------------------------------------------------ RUNPAR--- ------------------------------------------------------------------------------- [Notes on threadCapability]------ We create a thread on each CPU with forkOnIO. Ideally, the CPU on --- which the current thread is running will host the main thread; the --- other CPUs will host worker threads.------ This is possible using threadCapability, but this requires--- GHC 7.1.20110301, because that is when threadCapability was added.------ Lacking threadCapability, we always pick CPU #0 to run the main--- thread. If the current thread is not running on CPU #0, this--- will require some data to be shipped over the memory bus, and--- hence will be slightly slower than the version using threadCapability.------ If this is a nested runPar call, then we can do slightly better. We --- can look at the current workers' ThreadIds and see if we are one of --- them. If so, we do the work on that core. If not, we are once again --- forced to choose arbitrarily, so we send the work to CPU #0.---+data IVarContents a = Full a | Empty | Blocked [a -> Trace] {-# INLINE runPar_internal #-}-runPar_internal :: Bool -> Par a -> a-runPar_internal _doSync x = unsafePerformIO $ do- -- Set up the schedulers- myTId <- myThreadId- tIds <- replicateM numCapabilities $ newIORef myTId- workpools <- replicateM numCapabilities $ newIORef NoWork- statusRef <- newIORef newStatus- let states = listArray (0, numCapabilities-1)- [ Sched { no=n, workpool=wp, status=statusRef, scheds=states, tId=t }- | n <- [0..] | wp <- workpools | t <- tIds ]- res <- globalEstablishScheds states- case res of- Success uid -> do-#if __GLASGOW_HASKELL__ >= 701 /* 20110301 */- -- See [Notes on threadCapability] for more details- (main_cpu, _) <- threadCapability =<< myThreadId-#else- let main_cpu = 0-#endif- currentWorkers <- newIORef 1- let workLimit' = (-1, undefined)- let workLimit = (0, currentWorkers)- - m <- newEmptyMVar- rref <- newIORef Empty- atomicModifyIORef statusRef $ addExtIdler (ExtIdle uid m)- forM_ (elems states) $ \(state@Sched{no=cpu}) -> do- forkOnIO cpu $ do- myTId <- myThreadId- --printf "cpu %d setting threadId=%s\n" cpu (show myTId)- writeIORef (tId state) myTId- if (cpu /= main_cpu)- then reschedule workLimit' state- else do- sublst <- newIORef []- atomicModifyIORef (workpool state) $ \wp -> (Work uid currentWorkers sublst wp, ())- sched _doSync workLimit state sublst uid $ runCont (x >>= put_ (IVar rref)) (const Done)- takeMVar m- --printf "done\n"- r <- readIORef rref- - -- TODO: If we're doing this nested strategy, we should probably just keep the - -- threads alive indefinitely. After all, we can get some weird conditions - -- doing it this way. At the least, we should put this in steal where the - -- shutdown occurs.- b <- globalThreadShutdown--- putStrLn $ "Global thread shutdown: " ++ show b- case r of- Full a -> return a- _ -> error "no result"+runPar_internal :: Bool -> Par a -> IO a+runPar_internal _doSync x = do+ workpools <- replicateM numCapabilities $ newIORef []+ idle <- newIORef []+ let states = [ Sched { no=x, workpool=wp, idle, scheds=states }+ | (x,wp) <- zip [0..] workpools ] - Failure uid cScheds -> do #if __GLASGOW_HASKELL__ >= 701 /* 20110301 */- -- See [Notes on threadCapability] for more details- (main_cpu, _) <- threadCapability myTId- cTId <- readIORef $ tId $ cScheds ! main_cpu- let doWork = cTId == myTId+ --+ -- We create a thread on each CPU with forkOnIO. The CPU on which+ -- the current thread is running will host the main thread; the+ -- other CPUs will host worker threads.+ --+ -- Note: GHC 7.1.20110301 is required for this to work, because that+ -- is when threadCapability was added.+ --+ (main_cpu, _) <- threadCapability =<< myThreadId #else- cTIds <- mapM (\s -> (readIORef $ tId $ s) >>= (\t -> return (s,t))) (elems cScheds)- let (main_cpu, doWork) = case find ((== myTId) . snd) cTIds of- Nothing -> (0, False)- Just (s,_) -> (no s, True)+ --+ -- Lacking threadCapability, we always pick CPU #0 to run the main+ -- thread. If the current thread is not running on CPU #0, this+ -- will require some data to be shipped over the memory bus, and+ -- hence will be slightly slower than the version above.+ --+ let main_cpu = 0 #endif- - rref <- newIORef Empty- let task = runCont (x >>= put_ (IVar rref)) (const Done)- state = cScheds ! main_cpu- if doWork- then do- --printf "cpu %d using old threads, of which I am one\n" main_cpu- currentWorkers <- newIORef 1- sublst <- newIORef []- let workLimit = (uid, currentWorkers)- atomicModifyIORef (workpool state) $ \wp -> (Work uid currentWorkers sublst wp, ())- sched _doSync workLimit state sublst uid $ task- else do- --printf "cpu %d using old threads, of which I am not one\n" main_cpu- currentWorkers <- newIORef 0- sublst <- newIORef [task]- m <- newEmptyMVar- atomicModifyIORef (status state) $ addExtIdler (ExtIdle uid m)- atomicModifyIORef (workpool state) $ \wp -> (Work uid currentWorkers sublst wp, ())- takeMVar m- --printf "cpu %d finished in child\n" main_cpu- r <- readIORef rref--- globalThreadShutdown- case r of- Full a -> return a- _ -> error "no result" --- | The main way to run a Par computation+ m <- newEmptyMVar+ forM_ (zip [0..] states) $ \(cpu,state) ->+ forkOnIO cpu $+ if (cpu /= main_cpu)+ then reschedule state+ else do+ rref <- newIORef Empty+ sched _doSync state $ runCont (x >>= put_ (IVar rref)) (const Done)+ readIORef rref >>= putMVar m++ r <- takeMVar m+ case r of+ Full a -> return a+ _ -> error "no result"++ runPar :: Par a -> a-runPar = runPar_internal True+runPar = unsafePerformIO . runPar_internal True +-- | A version that avoids an internal `unsafePerformIO` for calling+-- contexts that are already in the `IO` monad.+runParIO :: Par a -> IO a+runParIO = runPar_internal True+ -- | An asynchronous version in which the main thread of control in a -- Par computation can return while forked computations still run in -- the background. runParAsync :: Par a -> a-runParAsync = runPar_internal False---- | An alternative version in which the consumer of the result has--- the option to "help" run the Par computation if results it is--- interested in are not ready yet.-runParAsyncHelper :: Par a -> (a, IO ())-runParAsyncHelper = undefined -- TODO: Finish Me.-+runParAsync = unsafePerformIO . runPar_internal False --- ------------------------------------------------------------------------------ PAR FUNCTIONS--- ---------------------------------------------------------------------------+-- ----------------------------------------------------------------------------- +-- | creates a new @IVar@ new :: Par (IVar a) new = Par $ New Empty +-- | creates a new @IVar@ that contains a value newFull :: NFData a => a -> Par (IVar a) newFull x = deepseq x (Par $ New (Full x)) +-- | creates a new @IVar@ that contains a value (head-strict only) newFull_ :: a -> Par (IVar a) newFull_ !x = Par $ New (Full x) +-- | 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 a -> Par a get v = Par $ \c -> Get v c +-- | like 'put', but only head-strict rather than fully-strict. put_ :: IVar a -> a -> Par () put_ v !a = Par $ \c -> Put v a (c ()) +-- | put a value into a @IVar@. Multiple 'put's to the same @IVar@+-- are not allowed, and result in a runtime error.+--+-- 'put' fully evaluates its argument, which therefore must be an+-- instance of 'NFData'. The idea is that this forces the work to+-- happen when we expect it, rather than being passed to the consumer+-- of the @IVar@ and performed later, which often results in less+-- parallelism than expected.+--+-- Sometimes partial strictness is more appropriate: see 'put_'.+-- put :: NFData a => IVar a -> a -> Par () put v a = deepseq a (Par $ \c -> Put v a (c ())) +-- | Allows other parallel computations to progress. (should not be+-- necessary in most cases). yield :: Par () yield = Par $ \c -> Yield (c ())
monad-par.cabal view
@@ -1,5 +1,5 @@ Name: monad-par-Version: 0.3+Version: 0.3.4 Synopsis: A library for parallel programming based on a monad @@ -16,46 +16,44 @@ -- 0.3 : Factored/reorganized modules and packages. -- *This* package is the original, core monad-par.---Description: This library offers an alternative parallel programming- API to that provided by the @parallel@ package.-- A 'Par' monad allows the simple description of- parallel computations, and can be used to add- parallelism to pure Haskell code. The basic API- is straightforward: the monad supports forking- and simple communication in terms of 'IVar's.-- The library comes with a work-stealing- implementation, but the internals are also- exposed so that you can build your own scheduler- if necessary.--- Examples of use can be found in the examples/ directory- of the source package.--- The modules below provide additionaly schedulers,- data structures, and other added capabilities- layered on top of the 'Par' monad.+-- 0.3.1 : fix for ghc 7.6.1, expose Par.IO+-- 0.3.4 : switch to direct scheduler as default (only 1-level nesting allowed) --- * Finish These--- * Module Descriptions+Description:+ The 'Par' monad offers a simple API for parallel programming. The+ library works for parallelising both pure and @IO@ computations,+ although only the pure version is deterministic.+ .+ For complete documentation see "Control.Monad.Par".+ .+ Some examples of use can be found in the @examples/@ directory of+ the source package.+ .+ Other related packages:+ .+ * @abstract-par@ provides the type classes that abstract over different+ implementations of the @Par@ monad.+ .+ * @monad-par-extras@ provides some extra combinators layered on top of+ the @Par@ monad.+ .+ Changes in 0.3.4 relative to 0.3:+ .+ * Fix bugs that cause "thread blocked indefinitely on MVar" crashes.+ .+ * Added "Control.Monad.Par.IO" Homepage: https://github.com/simonmar/monad-par License: BSD3 License-file: LICENSE-Author: Simon Marlow-Maintainer: Simon Marlow <marlowsd@gmail.com>+Author: Simon Marlow, Ryan Newton+Maintainer: Simon Marlow <marlowsd@gmail.com>, Ryan Newton <rrnewton@gmail.com> Copyright: (c) Simon Marlow 2011 Stability: Experimental Category: Control,Parallelism,Monads Build-type: Simple Cabal-version: >=1.8 - extra-source-files: tests/AListTest.hs tests/AllTests.hs@@ -70,9 +68,14 @@ tests/hs_cassandra_microbench2.hs Library+ Source-repository head+ type: git+ location: https://github.com/simonmar/monad-par+ Exposed-modules: -- The classic, simple monad-par interface: Control.Monad.Par+ , Control.Monad.Par.IO -- This is the default scheduler: , Control.Monad.Par.Scheds.Trace@@ -82,7 +85,8 @@ , Control.Monad.Par.Scheds.Direct -- This scheduler uses sparks rather than IO threads.- -- It only supports Futures, not full IVars:+ -- It only supports Futures, not full IVars. Fork+ -- becomes lighter weight. , Control.Monad.Par.Scheds.Sparks Build-depends: base >= 4 && < 5@@ -110,9 +114,11 @@ -- Internal logging framework: -- Control.Monad.Par.Logging, - -- Serial Elision is currently experimental:+ -- Serial Elision scheduling is currently experimental: -- Control.Monad.Par.Scheds.SerialElision + Control.Monad.Par.Scheds.DirectInternal+ ------------------------------------------------------------ -- Data Structures -- ------------------------------------------------------------@@ -132,14 +138,14 @@ ghc-options: -itests -rtsopts -threaded build-depends: base >= 4 && < 5 , abstract-par, monad-par-extras- , array >= 0.3+ , array >= 0.3 , deepseq >= 1.2 , time , QuickCheck, HUnit , test-framework-hunit, test-framework-quickcheck2 , test-framework, test-framework-th- -- , binary ---+ , abstract-deque >= 0.1.4+ , mwc-random >= 0.11+ , mtl >= 2.0.1.0+ , containers
tests/ParTests.hs view
@@ -4,44 +4,54 @@ import Control.Monad.Par.Combinator -import Control.Monad.Par.Scheds.Trace-import Control.Monad.Par.Scheds.TraceInternal (Par(..),Trace(Fork),runCont,runParAsync)+-- import Control.Monad.Par.Scheds.Trace+-- import Control.Monad.Par.Scheds.TraceInternal (Par(..),Trace(Fork),runCont,runParAsync) --- import Control.Monad.Par.Scheds.Direct+import Control.Monad.Par.Scheds.Direct -import Control.Concurrent.Chan-import Control.Exception -import System.IO.Unsafe-import Data.IORef-import Test.HUnit+-- import Control.Concurrent.Chan ()+import GHC.Conc (numCapabilities)+import Control.Exception (evaluate)+-- import System.IO.Unsafe+-- import Data.IORef+import Test.HUnit (Assertion, (@=?)) import Test.Framework.TH (testGroupGenerator)-import Test.Framework (defaultMain, testGroup)-import Test.Framework.Providers.HUnit+-- import Test.Framework (defaultMain, testGroup)+import qualified Test.Framework as TF+import Test.Framework.Providers.HUnit -- import Test.Framework.Providers.QuickCheck2 (testProperty)-import System.Timeout+import System.Timeout (timeout) -import TestHelpers+import TestHelpers (assertException, prnt, _prnt, _unsafeio, waste_time, collectOutput) -- ----------------------------------------------------------------------------- -- Testing -three = (3::Int)+three :: Int+three = 3++par :: (Eq a, Show a) => a -> Par a -> Assertion par res m = res @=? runPar m +case_justReturn :: Assertion case_justReturn = par three (return 3)++case_oneIVar :: Assertion case_oneIVar = par three (do r <- new; put r 3; get r) -- [2012.01.02] Apparently observing divergences here too:+case_forkNFill :: Assertion case_forkNFill = par three (do r <- new; fork (put r 3); get r) -- [2012.05.02] The nested Trace implementation sometimes fails to -- throw this exception, so we expect either the exception or a -- timeout. This is reasonable since we might expect a deadlock in a -- non-Trace scheduler. --ACF+case_getEmpty :: IO () case_getEmpty = do- _ <- timeout 100000 $ assertException "no result" $ + _ <- timeout 100000 $ assertException ["no result", "timeout"] $ runPar $ do r <- new; get r return () @@ -50,6 +60,7 @@ -- master branch with 16 threads: -- -- | Simple diamond test.+case_test_diamond :: Assertion case_test_diamond = 9 @=? (m :: Int) where m = runPar $ do@@ -64,7 +75,8 @@ -- -- NOTE: presently observing termination problems here. -- runPar is failing to exist after the exception?-disabled_case_multiput = assertException "multiple put" $ +disabled_case_multiput :: IO ()+disabled_case_multiput = assertException ["multiple put"] $ runPar $ do a <- new put a (3::Int)@@ -83,6 +95,7 @@ -- both :: Par a -> Par a -> Par a -- both a b = Par $ \c -> Fork (runCont a c) (runCont b c) +case_test_pmrr1 :: Assertion case_test_pmrr1 = par 5050 $ parMapReduceRangeThresh 1 (InclusiveRange 1 100) (return) (return `bincomp` (+)) 0@@ -91,29 +104,46 @@ ------------------------------------------------------------ --- Observe the real time ordering of events: --- A D B <pause> C E +-- | Observe the real time ordering of events:+--+-- Child-stealing: +-- A D B <pause> C E+-- +-- Parent-stealing:+-- A B D <pause> C E +--+-- Sequential:+-- A B <pause> C D E+-- +-- This is only for the TRACE scheduler right now.+case_async_test1 :: IO () case_async_test1 = do x <- res- case words x of - ["A","D","B","C",_,"E"] -> return ()- _ -> error$ "Bad output: "++ show (words x)+ case (numCapabilities, words x) of+ (1,["A","B","C",_,"D","E"]) -> return () + (n,["A","D","B","C",_,"E"]) | n > 1 -> return ()+ (n,["A","B","D","C",_,"E"]) | n > 1 -> return () + _ -> error$ "Bad temporal pattern: "++ show (words x) where res = collectOutput $ \ r -> do prnt r "A" evaluate$ runPar $- do + do iv <- new fork $ do _prnt r "B" x <- _unsafeio$ waste_time 0.5 _prnt r$ "C "++ show x -- _prnt r$ "C "++ show (_waste_time awhile)+ put iv () _prnt r "D"+ get iv prnt r$ "E"+ ------------------------------------------------------------ +tests :: [TF.Test] tests = [ $(testGroupGenerator) ]
tests/TestHelpers.hs view
@@ -9,14 +9,13 @@ import Data.IORef import Data.Time.Clock --- import Control.Monad.Par.Unsafe-import Control.Monad.Par.Scheds.Trace-import Control.Monad.Par.Scheds.TraceInternal (Par(..),Trace(Fork),runCont,runParAsync)+import Control.Monad.Par.Class ------------------------------------------------------------ -- Helpers -_unsafeio :: IO a -> Par a+-- _unsafeio :: IO a -> Par a+_unsafeio :: ParFuture iv p => IO a -> p a _unsafeio io = let x = unsafePerformIO io in x `seq` return x @@ -42,14 +41,16 @@ -- Obviously this takes a lot longer if it's interpreted: --awhile = 300000000+awhile :: Integer awhile = 3 * 1000 * 1000 -- awhile = 300000 +atomicModifyIORef_ :: IORef a -> (a -> a) -> IO () atomicModifyIORef_ rf fn = atomicModifyIORef rf (\x -> (fn x, ())) --- Haskell doesn't offer a way to create a Handle for in-memory output.--- So here we use IORefs instead+-- | Haskell doesn't offer a way to create a Handle for in-memory output.+-- So here we use IORefs instead... collectOutput :: (IORef [String] -> IO ()) -> IO String collectOutput fn = do c <- newIORef []@@ -60,7 +61,8 @@ prnt :: IORef [String] -> String -> IO () prnt ref str = atomicModifyIORef_ ref (str:) -_prnt :: IORef [String] -> String -> Par ()+-- _prnt :: IORef [String] -> String -> Par ()+_prnt :: ParFuture iv p => IORef [String] -> String -> p () _prnt ref = _unsafeio . prnt ref @@ -74,17 +76,18 @@ -- assertFailure $ "Expected exception: " ++ show ex -- where isWanted = guard . (== ex) --- Ensure that evaluating an expression returns an exception-assertException :: String -> a -> IO ()-assertException msg val = do+-- | Ensure that evaluating an expression returns an exception+-- containing one of the expected messages.+assertException :: [String] -> a -> IO ()+assertException msgs val = do x <- catch (do evaluate val; return Nothing) (\e -> do putStrLn$ "Good. Caught exception: " ++ show (e :: SomeException) return (Just$ show e)) case x of Nothing -> error "Failed to get an exception!" Just s -> - if isInfixOf msg s + if any (`isInfixOf` s) msgs then return () - else error$ "Got the wrong exception, expected to see the text: "++ show msg + else error$ "Got the wrong exception, expected to one of the strings: "++ show msgs ++ "\nInstead got this exception:\n " ++ show s