diff --git a/Control/Monad/Par.hs b/Control/Monad/Par.hs --- a/Control/Monad/Par.hs +++ b/Control/Monad/Par.hs @@ -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 () + * The wiki\/tutorial () + * The original paper () + * Tutorial slides () - * Other slides: , - + * Other slides: (, + ) + -} 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 diff --git a/Control/Monad/Par/IO.hs b/Control/Monad/Par/IO.hs new file mode 100644 --- /dev/null +++ b/Control/Monad/Par/IO.hs @@ -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)) 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 @@ -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 = "" - --- 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 = "" +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 = "" +-------------------------------------------------------------------------------- +-- 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 -- -------------------------------------------------------------------------------- + + +{-# 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) + + diff --git a/Control/Monad/Par/Scheds/DirectInternal.hs b/Control/Monad/Par/Scheds/DirectInternal.hs new file mode 100644 --- /dev/null +++ b/Control/Monad/Par/Scheds/DirectInternal.hs @@ -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 = "" + +-- 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 = "" + +-- 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 = "" + +hotVarTransaction = atomically +readHotVarRaw = readTVar +writeHotVarRaw = writeTVar + +#endif 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 @@ -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 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,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 ()) 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 +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 +Author: Simon Marlow, Ryan Newton +Maintainer: Simon Marlow , Ryan Newton 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 diff --git a/tests/ParTests.hs b/tests/ParTests.hs --- a/tests/ParTests.hs +++ b/tests/ParTests.hs @@ -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 C E +-- | Observe the real time ordering of events: +-- +-- Child-stealing: +-- A D B C E +-- +-- Parent-stealing: +-- A B D C E +-- +-- Sequential: +-- A B 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) ] diff --git a/tests/TestHelpers.hs b/tests/TestHelpers.hs --- a/tests/TestHelpers.hs +++ b/tests/TestHelpers.hs @@ -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