monad-par 0.3.4.8 → 0.3.4.9
raw patch · 8 files changed
+176/−44 lines, 8 filesdep ~basedep ~mtl
Dependency ranges changed: base, mtl
Files
- Control/Monad/Par/IO.hs +2/−1
- Control/Monad/Par/Scheds/Direct.hs +32/−23
- Control/Monad/Par/Scheds/DirectInternal.hs +41/−5
- Control/Monad/Par/Scheds/Sparks.hs +13/−2
- Control/Monad/Par/Scheds/Trace.hs +3/−1
- Control/Monad/Par/Scheds/TraceInternal.hs +51/−5
- monad-par.cabal +1/−1
- tests/ParTests_shared.hs +33/−6
Control/Monad/Par/IO.hs view
@@ -19,10 +19,11 @@ import Control.Monad.Par.Class import Control.Applicative import Control.Monad.Trans (liftIO, MonadIO)+import Control.Monad.Fix (MonadFix) -- | A wrapper around an underlying Par type which allows IO. newtype ParIO a = ParIO (Par a)- deriving (Functor, Applicative, Monad, ParFuture IVar, ParIVar IVar)+ deriving (Functor, Applicative, Monad, ParFuture IVar, ParIVar IVar, MonadFix) -- | 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
Control/Monad/Par/Scheds/Direct.hs view
@@ -26,7 +26,7 @@ new, get, put_, fork, newFull, newFull_, put, spawn, spawn_, spawnP,- spawn1_+ spawn1_, fixPar, FixParException (..) -- runParAsync, runParAsyncHelper, -- yield, ) where@@ -34,30 +34,32 @@ import Control.Applicative import Control.Concurrent hiding (yield) import Data.IORef (IORef,newIORef,readIORef,writeIORef,atomicModifyIORef)-import Text.Printf (printf, hPrintf)+import Text.Printf (printf) import GHC.Conc (numCapabilities,yield) import "mtl" Control.Monad.Cont as C import qualified "mtl" Control.Monad.Reader as RD 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)+ newHotVar, readHotVar, modifyHotVar, modifyHotVar_,+ writeHotVarRaw, fixPar, FixParException (..)) #ifdef NEW_GENERIC import qualified Control.Par.Class as PN import qualified Control.Par.Class.Unsafe as PU #endif import Control.DeepSeq+#ifdef NESTED_SCHEDS import qualified Data.Map as M+#endif import qualified Data.Set as S import Data.Maybe (catMaybes) import Data.Word (Word64) -import Data.Concurrent.Deque.Class (WSDeque)+-- import Data.Concurrent.Deque.Class (WSDeque) #ifdef USE_CHASELEV #warning "Note: using Chase-Lev lockfree workstealing deques..." import Data.Concurrent.Deque.ChaseLev.DequeInstance@@ -166,8 +168,10 @@ -- `runPar` instantiations. This is used to detect nested invocations -- of `runPar` and avoid reinitialization. -- globalWorkerPool :: IORef (Data.IntMap ())+#ifdef NESTED_SCHEDS globalWorkerPool :: IORef (M.Map ThreadId Sched) globalWorkerPool = unsafePerformIO $ newIORef M.empty+#endif -- TODO! Make this semi-local! (not shared between "top-level" runPars) {-# INLINE amINested #-}@@ -211,7 +215,7 @@ {-# INLINE pushWork #-} pushWork :: Sched -> Par () -> IO ()-pushWork Sched { workpool, idle, no, isMain } task = do+pushWork Sched { workpool, idle, no } task = do R.pushL workpool task when dbg $ do sn <- makeStableName task printf " [%d] -> PUSH work unit %d\n" no (hashStableName sn)@@ -242,7 +246,7 @@ -------------------------------------------------------------------------------- instance NFData (IVar a) where- rnf _ = ()+ rnf !_ = () {-# NOINLINE runPar #-} runPar = unsafePerformIO . runParIO@@ -538,7 +542,7 @@ -- continuation until its completion. if _PARPUTS then -- We do NOT force the putting thread to postpone its continuation.- do spawn_$ pMap kont rest+ do _ <- spawn_$ pMap kont rest return () -- case rest of -- [] -> spawn_$ io$ kont arg@@ -558,7 +562,7 @@ pMap kont [] = io$ kont arg pMap kont (more:rest) =- do spawn_$ io$ kont arg+ do _ <- spawn_$ io$ kont arg pMap more rest -- parchain [kont] = kont arg@@ -581,7 +585,7 @@ -- 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.+ _ <- 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" @@ -674,10 +678,13 @@ r <- modifyHotVar idle $ \is -> (m:is, is) if length r == numCapabilities - 1 then do- when dbg$ printf " [%d] | initiating shutdown\n" my_no+ when dbg$ printf " [%d] | waking up all threads\n" my_no+ writeHotVarRaw idle [] mapM_ (\vr -> putMVar vr True) r else do- done <- takeMVar m+ (Session _ finRef):_ <- readIORef $ sessions mysched+ fin <- readIORef finRef+ done <- if fin then pure True else takeMVar m if done then do when dbg$ printf " [%d] | shutting down\n" my_no@@ -719,14 +726,16 @@ go (tries-1) i' -- | The continuation which should not be called.-errK :: t-errK = error "Error cont: this closure shouldn't be used"+_errK :: t+_errK = error "Error cont: this closure shouldn't be used" trivialCont :: String -> a -> ROnly ()-trivialCont str _ = do #ifdef DEBUG_DIRECT+trivialCont str _ = do -- trace (str ++" trivialCont evaluated!") liftIO$ printf " !! trivialCont evaluated, msg: %s\n" str+#else+trivialCont _str _ = do #endif return () @@ -849,8 +858,8 @@ -------------------------------------------------------------------------------- -- Make sure there is no work left in any deque after exiting.-sanityCheck :: [Sched] -> IO ()-sanityCheck allscheds = do+_sanityCheck :: [Sched] -> IO ()+_sanityCheck allscheds = do forM_ allscheds $ \ Sched{no, workpool} -> do b <- R.nullQ workpool when (not b) $ do@@ -860,8 +869,8 @@ -- | This tries to localize the blocked-indefinitely exception:-dbgTakeMVar :: String -> MVar a -> IO a-dbgTakeMVar msg mv =+_dbgTakeMVar :: String -> MVar a -> IO a+_dbgTakeMVar msg mv = -- catch (takeMVar mv) ((\_ -> doDebugStuff) :: BlockedIndefinitelyOnMVar -> IO a) E.catch (takeMVar mv) (\(_::IOError) -> doDebugStuff) where@@ -890,8 +899,8 @@ -- | Fork a thread but ALSO set up an error handler that suppresses -- MVar exceptions.-forkIO_Suppress :: Int -> IO () -> IO ThreadId-forkIO_Suppress whre action =+_forkIO_Suppress :: Int -> IO () -> IO ThreadId+_forkIO_Suppress whre action = forkOn whre $ E.handle (\e -> case (e :: E.BlockedIndefinitelyOnMVar) of@@ -911,9 +920,9 @@ E.catch action (\ e -> case E.fromException e of- Just E.ThreadKilled -> printf -- hPrintf stderr+ Just E.ThreadKilled -> printf "\nThreadKilled exception inside child thread, %s (not propagating!): %s\n" (show tid) (show descr)- _ -> do printf -- hPrintf stderr+ _ -> do printf "\nException inside child thread %s, %s: %s\n" (show descr) (show tid) (show e) E.throwTo parent (e :: E.SomeException) )
Control/Monad/Par/Scheds/DirectInternal.hs view
@@ -9,6 +9,7 @@ import Control.Applicative import "mtl" Control.Monad.Cont as C import qualified "mtl" Control.Monad.Reader as RD+import Control.Monad.IO.Class (liftIO) import qualified System.Random.MWC as Random @@ -18,16 +19,22 @@ import qualified Data.Set as S import Data.Word (Word64) import Data.Concurrent.Deque.Class (WSDeque)+import Control.Monad.Fix (MonadFix (mfix))+#if MIN_VERSION_base(4,4,0)+import GHC.IO.Unsafe (unsafeDupableInterleaveIO)+#else+import GHC.IO.Unsafe (unsafeInterleaveIO)+#endif #ifdef USE_CHASELEV #warning "Note: using Chase-Lev lockfree workstealing deques..." import Data.Concurrent.Deque.ChaseLev.DequeInstance import Data.Concurrent.Deque.ChaseLev as R-#else-import Data.Concurrent.Deque.Reference.DequeInstance-import Data.Concurrent.Deque.Reference as R #endif +import Control.Exception (Exception, throwIO, BlockedIndefinitelyOnMVar (..),+ catch)+ -- Our monad stack looks like this: -- --------- -- ContT@@ -44,6 +51,33 @@ deriving (Functor, Applicative, Monad, MonadCont, RD.MonadReader Sched) type ROnly = RD.ReaderT Sched IO +instance MonadFix Par where+ mfix = fixPar++-- | Take the monadic fixpoint of a 'Par' computation. This is+-- the definition of 'mfix' for 'Par'. Throws 'FixParException'+-- if the result is demanded strictly within the computation.+fixPar :: (a -> Par a) -> Par a+-- We do this IO-style, rather than ST-style, in order to get a+-- consistent exception type. Using the ST-style mfix, a strict+-- argument could lead us to *either* a <<loop>> exception *or*+-- (if the wrong sort of computation gets re-run) a "multiple-put"+-- error.+fixPar f = Par $ ContT $ \ar -> RD.ReaderT $ \sched -> do+ mv <- newEmptyMVar+ ans <- unsafeDupableInterleaveIO (readMVar mv `catch`+ \ ~BlockedIndefinitelyOnMVar -> throwIO FixParException)+ flip RD.runReaderT sched $+ runContT (unPar (f ans)) $ \a -> liftIO (putMVar mv a) >> ar a++#if !MIN_VERSION_base(4,4,0)+unsafeDupableInterleaveIO :: IO a -> IO a+unsafeDupableInterleaveIO = unsafeInterleaveIO+#endif++data FixParException = FixParException deriving Show+instance Exception FixParException+ type SessionID = Word64 -- An ID along with a flag to signal completion:@@ -112,10 +146,12 @@ readHotVar = readIORef writeHotVar = writeIORef instance Show (IORef a) where- show ref = "<ioref>"+ show _ref = "<ioref>" +writeHotVarRaw :: HotVar a -> a -> IO () -- hotVarTransaction = id hotVarTransaction = error "Transactions not currently possible for IO refs"+readHotVarRaw :: HotVar a -> IO a readHotVarRaw = readHotVar writeHotVarRaw = writeHotVar @@ -130,7 +166,7 @@ readHotVar = readMVar writeHotVar v x = do swapMVar v x; return () instance Show (MVar a) where- show ref = "<mvar>"+ show _ref = "<mvar>" -- hotVarTransaction = id -- We could in theory do this by taking the mvar to grab the lock.
Control/Monad/Par/Scheds/Sparks.hs view
@@ -8,7 +8,7 @@ ( Par(..), Future(..), runPar, - get, spawn, spawn_, spawnP+ get, spawn, spawn_, spawnP, fixPar ) where @@ -17,12 +17,13 @@ import Control.DeepSeq import Control.Parallel import qualified Control.Monad.Par.Class as PC+import Control.Monad.Fix (MonadFix (mfix)) -- import Control.Parallel.Strategies (rpar)-import System.IO.Unsafe (unsafePerformIO) #ifdef NEW_GENERIC import qualified Control.Par.Class as PN import qualified Control.Par.Class.Unsafe as PU+import System.IO.Unsafe (unsafePerformIO) #endif @@ -70,6 +71,16 @@ instance Applicative Par where (<*>) = ap pure = Done++instance MonadFix Par where+ mfix = fixPar++-- | Take the monadic fixpoint of a 'Par' computation. This is+-- the definition of 'mfix' for 'Par'.+fixPar :: (a -> Par a) -> Par a+fixPar f =+ let fr = f (case fr of Done x -> x)+ in fr #ifdef NEW_GENERIC doio :: IO a -> Par a
Control/Monad/Par/Scheds/Trace.hs view
@@ -14,7 +14,7 @@ module Control.Monad.Par.Scheds.Trace ( Par, runPar, runParIO, fork, IVar, new, newFull, newFull_, get, put, put_,- spawn, spawn_, spawnP+ spawn, spawn_, spawnP, fixPar, FixParException (..) ) where import qualified Control.Monad.Par.Class as PC@@ -39,7 +39,9 @@ -- -- Standard instances: -- <boilerplate>+spawn :: NFData a => Par a -> Par (IVar a) spawn p = do r <- new; fork (p >>= put r); return r+spawn_ :: Par a -> Par (IVar a) spawn_ p = do r <- new; fork (p >>= put_ r); return r -- </boilerplate>>
Control/Monad/Par/Scheds/TraceInternal.hs view
@@ -14,7 +14,7 @@ runPar, runParIO, runParAsync, -- runParAsyncHelper, new, newFull, newFull_, get, put_, put,- pollIVar, yield,+ pollIVar, yield, fixPar, FixParException (..) ) where @@ -22,9 +22,17 @@ import Prelude hiding (mapM, sequence, head,tail) import Data.IORef import System.IO.Unsafe+#if MIN_VERSION_base(4,4,0)+import GHC.IO.Unsafe (unsafeDupableInterleaveIO)+#else+import GHC.IO.Unsafe (unsafeInterleaveIO)+#endif import Control.Concurrent hiding (yield) import GHC.Conc (numCapabilities) import Control.DeepSeq+import Control.Monad.Fix (MonadFix (mfix))+import Control.Exception (Exception, throwIO, BlockedIndefinitelyOnMVar (..),+ catch) -- import Text.Printf #if !MIN_VERSION_base(4,8,0)@@ -97,6 +105,9 @@ r <- io loop (c r) +data FixParException = FixParException deriving Show+instance Exception FixParException+ -- | Process the next item on the work queue or, failing that, go into -- work-stealing mode. reschedule :: Sched -> IO ()@@ -182,6 +193,31 @@ (<*>) = ap pure a = Par ($ a) +instance MonadFix Par where+ mfix = fixPar++-- | Take the monadic fixpoint of a 'Par' computation. This is+-- the definition of 'mfix' for 'Par'. Throws 'FixParException'+-- if the result is demanded strictly within the computation.+fixPar :: (a -> Par a) -> Par a+-- We do this IO-style, rather than ST-style, in order to get a+-- consistent exception type. Using the ST-style mfix, a strict+-- argument could lead us to *either* a <<loop>> exception *or*+-- (if the wrong sort of computation gets re-run) a "multiple-put"+-- error.+fixPar f = Par $ \ c ->+ LiftIO (do+ mv <- newEmptyMVar+ ans <- unsafeDupableInterleaveIO (readMVar mv+ `catch` \ ~BlockedIndefinitelyOnMVar -> throwIO FixParException)+ case f ans of+ Par q -> pure $ q $ \a -> LiftIO (putMVar mv a) (\ ~() -> c a)) id++#if !MIN_VERSION_base(4,4,0)+unsafeDupableInterleaveIO :: IO a -> IO a+unsafeDupableInterleaveIO = unsafeInterleaveIO+#endif+ newtype IVar a = IVar (IORef (IVarContents a)) -- data IVar a = IVar (IORef (IVarContents a)) @@ -189,9 +225,8 @@ instance Eq (IVar a) where (IVar r1) == (IVar r2) = r1 == r2 --- Forcing evaluation of a IVar is fruitless. instance NFData (IVar a) where- rnf _ = ()+ rnf !_ = () -- From outside the Par computation we can peek. But this is nondeterministic.@@ -281,7 +316,14 @@ -- | Creates a new @IVar@ that contains a value newFull :: NFData a => a -> Par (IVar a)-newFull x = deepseq x (Par $ New (Full x))+-- What are we doing here? We're manually raising the arity+-- of newFull from 2 to 3, which seems like it's probably what+-- we want most of the time. Notably, fmapping over the result+-- gives really awful-looking Core if we don't do this.+-- Regardless, I think we logically want to force the+-- value when it's installed in the IVar rather than+-- when we create the action to install it in the IVar.+newFull x = Par $ \c -> x `deepseq` New (Full x) c -- | Creates a new @IVar@ that contains a value (head-strict only) newFull_ :: a -> Par (IVar a)@@ -309,7 +351,11 @@ -- 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 ()))+-- Manually raise the arity, which seems likely to be what+-- we want most of the time. We really want to force the+-- value when it's installed in the IVar, not when we+-- create the Par action to install it in the IVar.+put v a = Par $ \c -> a `deepseq` Put v a (c ()) -- | Allows other parallel computations to progress. (should not be -- necessary in most cases).
monad-par.cabal view
@@ -1,5 +1,5 @@ Name: monad-par-Version: 0.3.4.8+Version: 0.3.4.9 Synopsis: A library for parallel programming based on a monad
@@ -25,6 +25,30 @@ par :: (Eq a, Show a) => a -> Par a -> Assertion par res m = res @=? runPar m +-- From https://github.com/simonmar/monad-par/pull/49+case_parallelFilter :: Assertion+case_parallelFilter = run 200 where+ run 0 = pure ()+ run i = do+ par result (parfilter p xs)+ run (i-1)++ p x = x `mod` 2 == 0+ xs = [0..10] :: [Int]+ result = filter p xs++ parfilter _ [] = pure []+ parfilter f [x] = pure (if f x then [x] else [])+ parfilter f xs = do+ let (as, bs) = halve xs+ v1 <- spawn $ parfilter f as+ v2 <- spawn $ parfilter f bs+ left <- get v1+ right <- get v2+ pure (left ++ right)++ halve xs = splitAt (length xs `div` 2) xs+ -- | Make sure there's no problem with bringing the worker threads up and down many -- times. 10K runPar's takes about 6.3 seconds. case_lotsaRunPar :: Assertion@@ -85,12 +109,15 @@ case_test_diamond = 9 @=? (m :: Int) where m = runPar $ do- [a,b,c,d] <- sequence [new,new,new,new]- fork $ do x <- get a; put b (x+1)- fork $ do x <- get a; put c (x+2)- fork $ do x <- get b; y <- get c; put d (x+y)- fork $ do put a 3- get d+ abcd <- sequence [new,new,new,new]+ case abcd of+ [a,b,c,d] -> do+ fork $ do x <- get a; put b (x+1)+ fork $ do x <- get a; put c (x+2)+ fork $ do x <- get b; y <- get c; put d (x+y)+ fork $ do put a 3+ get d+ _ -> error "Oops" -- | Violate IVar single-assignment: --