diff --git a/Control/Monad/Par.hs b/Control/Monad/Par.hs --- a/Control/Monad/Par.hs +++ b/Control/Monad/Par.hs @@ -61,7 +61,7 @@ available processors. Unlike @Control.Parallel@, in @Control.Monad.Par@ parallelism is - not combined with laziness, so sharing and granulairty are + not combined with laziness, so sharing and granularity are completely under the control of the programmer. New units of parallel work are only created by @fork@ and a few other combinators. @@ -97,7 +97,7 @@ runPar, runParIO, fork, - -- | forks a computation to happen in parallel. The forked + -- | Forks a computation to happen in parallel. The forked -- computation may exchange values with other computations using -- @IVar@s. @@ -105,21 +105,21 @@ IVar, new, - -- | creates a new @IVar@ + -- creates a new @IVar@ newFull, - -- | creates a new @IVar@ that contains a value + -- creates a new @IVar@ that contains a value newFull_, - -- | creates a new @IVar@ that contains a value (head-strict only) + -- creates a new @IVar@ that contains a value (head-strict only) get, - -- | read the value in an @IVar@. 'get' can only return when the + -- read the value in an @IVar@. 'get' can only return when the -- value has been written by a prior or parallel @put@ to the same -- @IVar@. put, - -- | put a value into a @IVar@. Multiple 'put's to the same @IVar@ + -- put a value into an @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 @@ -132,11 +132,11 @@ -- put_, - -- | like 'put', but only head-strict rather than fully-strict. + -- like 'put', but only head-strict rather than fully-strict. -- * Operations spawn, - -- | Like 'fork', but returns a @IVar@ that can be used to query the + -- | Like 'fork', but returns an @IVar@ that can be used to query the -- result of the forked computataion. Therefore @spawn@ provides /futures/ or /promises/. -- -- > spawn p = do 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 @@ -205,8 +205,8 @@ mb <- R.tryPopL workpool when dbg $ case mb of Nothing -> return () - Just _ -> do sn <- makeStableName mb - printf " [%d] -> POP work unit %d\n" no (hashStableName sn) + Just _ -> do sn <- makeStableName mb + printf " [%d] -> POP work unit %d\n" no (hashStableName sn) return mb {-# INLINE pushWork #-} @@ -214,7 +214,7 @@ pushWork Sched { workpool, idle, no, isMain } task = do R.pushL workpool task when dbg $ do sn <- makeStableName task - printf " [%d] -> PUSH work unit %d\n" no (hashStableName sn) + printf " [%d] -> PUSH work unit %d\n" no (hashStableName sn) #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. @@ -415,11 +415,11 @@ activeSessions <- newHotVar S.empty sessionCounter <- newHotVar (baseSessionID + 1) let allscheds = [ Sched { no=x, idle, isMain= (x==main), - workpool=wp, scheds=allscheds, rng=rng, + 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 @@ -439,13 +439,13 @@ -------------------------------------------------------------------------------- {-# INLINE new #-} --- | creates a new @IVar@ +-- | Creates a new @IVar@ new :: Par (IVar a) new = io$ do r <- newIORef Empty return (IVar r) {-# INLINE get #-} --- | read the value in a @IVar@. The 'get' can only return when the +-- | Read the value in an @IVar@. The 'get' operation can only return when the -- value has been written by a prior or parallel @put@ to the same -- @IVar@. get (IVar vr) = do @@ -453,8 +453,8 @@ do e <- io$ readIORef vr case e of - Full a -> return a - _ -> do + Full a -> return a + _ -> do sch <- RD.ask # ifdef DEBUG_DIRECT sn <- io$ makeStableName vr -- Should probably do the MutVar inside... @@ -462,14 +462,14 @@ #else let resched = # endif - longjmpSched -- Invariant: kont must not be lost. + longjmpSched -- Invariant: kont must not be lost. -- Because we continue on the same processor the Sched stays the same: -- 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 + 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 @@ -512,13 +512,13 @@ sched <- RD.ask (ks,res) <- io$ do pr <- atomicModifyIORef vr $ \e -> case e of - Empty -> (Full content, ([], content)) - Full x -> (Full x, ([], x)) - Blocked ks -> (Full content, (ks, content)) + Empty -> (Full content, ([], content)) + Full x -> (Full x, ([], x)) + Blocked ks -> (Full content, (ks, content)) #ifdef DEBUG_DIRECT sn <- makeStableName vr printf " [%d] unsafeTryPut: value %s in IVar %d. Waking up %d continuations.\n" - (no sched) (show content) (hashStableName sn) (length (fst pr)) + (no sched) (show content) (hashStableName sn) (length (fst pr)) #endif return pr wakeUp sched ks content @@ -615,7 +615,7 @@ Nothing -> do (Session _ finRef):_ <- liftIO$ readIORef $ sessions mysched fin <- liftIO$ readIORef finRef - if fin + if fin then do when (dbglvl >= 1) $ liftIO $ do tid <- myThreadId sess <- readSessions mysched @@ -633,22 +633,22 @@ -- 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 + liftIO$ steal mysched #ifdef WAKEIDLE -- io$ tryWakeIdle (idle mysched) #endif liftIO yield - rescheduleR (cnt+1) kont + 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) + 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 sch <- RD.ask when dbg$ liftIO$ printf " + task finished successfully on cpu %d, calling reschedule continuation..\n" (no sch) - rescheduleR 0 kont) + rescheduleR 0 kont) -- | Attempt to steal work or, failing that, give up and go idle. @@ -684,7 +684,7 @@ return () else do when dbg$ printf " [%d] | woken up\n" my_no - i <- getnext (-1::Int) + i <- getnext (-1::Int) go maxtries i -- We need to return from this loop to check sessionFinished and exit the scheduler if necessary. @@ -693,7 +693,7 @@ ---------------------------------------- go tries i | i == my_no = do i' <- getnext i - go (tries-1) i' + go (tries-1) i' | otherwise = do -- We ONLY go through the global sched array to access victims: @@ -707,16 +707,16 @@ case r of Just task -> do when dbg$ do sn <- makeStableName task - printf " [%d] | stole work (unit %d) from cpu %d\n" my_no (hashStableName sn) (no schd) - runReaderWith mysched $ - C.runContT (unPar task) - (\_ -> do - when dbg$ do sn <- liftIO$ makeStableName task - liftIO$ printf " [%d] | DONE running stolen work (unit %d) from %d\n" my_no (hashStableName sn) (no schd) - return ()) + printf " [%d] | stole work (unit %d) from cpu %d\n" my_no (hashStableName sn) (no schd) + runReaderWith mysched $ + C.runContT (unPar task) + (\_ -> do + when dbg$ do sn <- liftIO$ makeStableName task + liftIO$ printf " [%d] | DONE running stolen work (unit %d) from %d\n" my_no (hashStableName sn) (no schd) + return ()) Nothing -> do i' <- getnext i - go (tries-1) i' + go (tries-1) i' -- | The continuation which should not be called. errK :: t @@ -728,7 +728,7 @@ -- trace (str ++" trivialCont evaluated!") liftIO$ printf " !! trivialCont evaluated, msg: %s\n" str #endif - return () + return () ---------------------------------------------------------------------------------------------------- @@ -749,8 +749,8 @@ -- The following is usually inefficient! newFull_ a = do v <- new - put_ v a - return v + put_ v a + return v newFull a = deepseq a (newFull_ a) @@ -813,7 +813,7 @@ #ifdef NEW_GENERIC instance PU.ParMonad Par where - fork = fork + fork = fork internalLiftIO io = Par (lift $ lift io) instance PU.ParThreadSafe Par where @@ -826,14 +826,14 @@ spawn = spawn spawn_ = spawn_ spawnP = spawnP - + instance PN.ParIVar Par where new = new put_ = put_ newFull = newFull newFull_ = newFull_ #endif - + -- -------------------------------------------------------------------------------- @@ -898,7 +898,7 @@ _ -> do putStrLn$"CAUGHT child thread exception: "++show e return () - ) + ) action @@ -909,14 +909,14 @@ forkit $ do tid <- myThreadId E.catch action - (\ e -> + (\ 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 + _ -> 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: @@ -925,5 +925,3 @@ 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/Sparks.hs b/Control/Monad/Par/Scheds/Sparks.hs --- a/Control/Monad/Par/Scheds/Sparks.hs +++ b/Control/Monad/Par/Scheds/Sparks.hs @@ -55,7 +55,7 @@ -- instance Monad Par where - return x = Done x + return = pure Done x >>= k = k x instance PC.ParFuture Future Par where @@ -69,7 +69,7 @@ instance Applicative Par where (<*>) = ap - pure = return + pure = Done #ifdef NEW_GENERIC doio :: IO a -> Par a 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 @@ -25,9 +25,12 @@ import Control.Concurrent hiding (yield) import GHC.Conc (numCapabilities) import Control.DeepSeq -import Control.Applicative -- import Text.Printf +#if !MIN_VERSION_base(4,8,0) +import Control.Applicative +#endif + #if __GLASGOW_HASKELL__ <= 700 import GHC.Conc (forkOnIO) forkOn = forkOnIO @@ -74,22 +77,22 @@ loop parent Done -> if _doSync - then reschedule queue + 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 () +-- 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. + -- This would also be a chance to steal and work from opposite ends of the queue. atomicModifyIORef workpool $ \ts -> (ts++[parent], ()) - reschedule queue + reschedule queue LiftIO io c -> do r <- io loop (c r) @@ -172,12 +175,12 @@ fmap f m = Par $ \c -> runCont m (c . f) instance Monad Par where - return a = Par ($ a) + return = pure m >>= k = Par $ \c -> runCont m $ \a -> runCont (k a) c instance Applicative Par where (<*>) = ap - pure = return + pure a = Par ($ a) newtype IVar a = IVar (IORef (IVarContents a)) -- data IVar a = IVar (IORef (IVarContents a)) @@ -247,11 +250,20 @@ _ -> error "no result" +-- | Run a parallel, deterministic computation and return its result. +-- +-- Note: you must NOT return an IVar in the output of the parallel +-- computation. This is unfortunately not enforced, as it is with +-- `runST` or with newer libraries that export a Par monad, such as +-- `lvish`. runPar :: Par a -> a runPar = unsafePerformIO . runPar_internal True -- | A version that avoids an internal `unsafePerformIO` for calling -- contexts that are already in the `IO` monad. +-- +-- Returning any value containing IVar is still disallowed, as it +-- can compromise type safety. runParIO :: Par a -> IO a runParIO = runPar_internal True @@ -263,29 +275,29 @@ -- ----------------------------------------------------------------------------- --- | creates a new @IVar@ +-- | Creates a new @IVar@ new :: Par (IVar a) new = Par $ New Empty --- | creates a new @IVar@ that contains a value +-- | 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) +-- | 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 +-- | Read the value in an @IVar@. The 'get' operation 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. +-- | 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@ +-- | Put a value into an @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 diff --git a/monad-par.cabal b/monad-par.cabal --- a/monad-par.cabal +++ b/monad-par.cabal @@ -1,5 +1,5 @@ Name: monad-par -Version: 0.3.4.7 +Version: 0.3.4.8 Synopsis: A library for parallel programming based on a monad