meta-par (empty) → 0.3
raw patch · 14 files changed
+1299/−0 lines, 14 filesdep +abstract-dequedep +abstract-pardep +basesetup-changed
Dependencies added: abstract-deque, abstract-par, base, bytestring, containers, deepseq, mtl, mwc-random, transformers, vector
Files
- Control/Monad/Par/Meta.hs +602/−0
- Control/Monad/Par/Meta/HotVar/IORef.hs +44/−0
- Control/Monad/Par/Meta/Resources/Backoff.hs +61/−0
- Control/Monad/Par/Meta/Resources/Debugging.hs +155/−0
- Control/Monad/Par/Meta/Resources/SMP.hs +127/−0
- Control/Monad/Par/Meta/Resources/SingleThreaded.hs +53/−0
- Control/Monad/Par/Meta/SMP.hs +43/−0
- Control/Monad/Par/Meta/Serial.hs +37/−0
- LICENSE +30/−0
- Setup.hs +3/−0
- System/Posix/Affinity.hs +25/−0
- cbits/pin.c +30/−0
- include/pin.h +2/−0
- meta-par.cabal +87/−0
+ Control/Monad/Par/Meta.hs view
@@ -0,0 +1,602 @@+{-# LANGUAGE BangPatterns #-}+{-# LANGUAGE CPP #-}+{-# LANGUAGE DeriveDataTypeable #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE NamedFieldPuns #-}+{-# LANGUAGE PackageImports #-}+{-# LANGUAGE TupleSections #-}+{-# OPTIONS_GHC -Wall -fno-warn-name-shadowing #-}++-- The Meta scheduler which can be parameterized over various+-- "Resources", including:+-- * Serial execution+-- * Shared memory SMP+-- * GPU accelerators+-- * Remote Machine "accelerators" (i.e. distributed)+++module Control.Monad.Par.Meta +( +-- * Core Meta-Par types+ Par+, IVar+-- * Operations+, PC.ParFuture(..)+, PC.ParIVar(..)+-- * Entrypoints+, runMetaPar+, runMetaParIO+-- * Implementation API+, Sched(..)+, GlobalState+-- ** Execution Resources+, Resource(..)+, Startup(..)+, WorkSearch(..)+-- ** Utilities+, forkWithExceptions+, spawnWorkerOnCPU+) where++import Control.Applicative+import Control.Concurrent ( MVar+ , newEmptyMVar+ , putMVar+ , readMVar+ , takeMVar+ , tryPutMVar+ , tryTakeMVar+ )+import Control.DeepSeq+import Control.Monad+import "mtl" Control.Monad.Cont (ContT(..), MonadCont, callCC, runContT)+import "mtl" Control.Monad.Reader (ReaderT, MonadReader, runReaderT, ask)+import Control.Monad.IO.Class+import Control.Exception (catch, throwTo, SomeException)++import Data.Concurrent.Deque.Class (WSDeque)+import Data.Concurrent.Deque.Reference.DequeInstance ()+import Data.Concurrent.Deque.Reference as R+import qualified Data.ByteString.Char8 as BS+import Data.Monoid+import Data.Set (Set)+import qualified Data.Set as Set+import Data.Typeable (Typeable)+import Data.IORef (IORef, writeIORef, newIORef)+import Data.Vector (Vector)+import qualified Data.Vector as V++import System.IO.Unsafe (unsafePerformIO)+import System.IO (stderr)+import System.Random.MWC++import Text.Printf+import qualified Debug.Trace as DT++#ifdef AFFINITY+import System.Posix.Affinity (setAffinityOS)+#endif++import Control.Monad.Par.Meta.Resources.Debugging (dbgTaggedMsg)+import Control.Monad.Par.Meta.HotVar.IORef+import qualified Control.Monad.Par.Class as PC++#if __GLASGOW_HASKELL__ >= 702+import GHC.Conc (forkOn, ThreadId, myThreadId, threadCapability)+import Control.Concurrent (getNumCapabilities)+threadCapability' tid = Just <$> threadCapability tid+#else+import GHC.Conc (forkOnIO, ThreadId, myThreadId, numCapabilities)+forkOn :: Int -> IO () -> IO ThreadId+forkOn = forkOnIO+getNumCapabilities :: IO Int+getNumCapabilities = return numCapabilities++-- This is a best-effort recreation of threadCapability for older GHC+-- versions that lack it. If the calling thread is a meta-par worker,+-- it must have been spawned with forkOn, so we return an answer that+-- indicates it is pinned.+-- +-- It does a search through the ThreadIds stored in the global+-- scheduler structure, and if it finds a match for the current+-- thread, it returns the 'no' field associated with that+-- ThreadId. Otherwise it returns Nothing.+threadCapability' tid = do+ vec <- readHotVar globalScheds+ let f Nothing = return $ mempty+ f (Just Sched{ no, tids }) = do+ set <- readHotVar tids+ case Set.member tid set of+ False -> return $ mempty+ True -> return $ First (Just no)+ cap <- getFirst . mconcat <$> mapM f (V.toList vec)+ return ((,True) <$> cap)+#endif+threadCapability' :: ThreadId -> IO (Maybe (Int, Bool))+++#if __GLASGOW_HASKELL__ < 700+void = fmap (const ())+#endif+++dbg :: Bool+#ifdef DEBUG+dbg = True+#else+dbg = False+#endif+++--------------------------------------------------------------------------------+-- Types++-- | The Meta-Par monad with its full suite of instances. Note that+-- the 'MonadIO' instance, while essential for building new+-- 'Resource's, is unsafe in client code when combined with+-- 'runMetaPar'. This type should therefore be exposed to client code+-- as a @newtype@ that omits the 'MonadIO' instance.+newtype Par a = Par { unPar :: ContT () ROnly a }+ deriving (Monad, MonadCont, MonadReader Sched, + MonadIO, Applicative, Functor, Typeable)++type ROnly = ReaderT Sched IO++-- | An 'IVar' is a /write-once/, /read-many/ structure for+-- communication between 'Par' threads.+newtype IVar a = IVar (HotVar (IVarContents a))++instance NFData (IVar a) where+ rnf _ = ()++data IVarContents a = Full a | Empty | Blocked [a -> IO ()]++-- | A 'GlobalState' structure tracks the state of all Meta-Par+-- workers in a program in a 'Data.Vector' indexed by capability+-- number.+type GlobalState = Vector (Maybe Sched)++-- | The 'Startup' component of a 'Resource' is a callback that+-- implements initialization behavior. For example, the SMP 'Startup'+-- calls 'spawnWorkerOnCPU' a number of times. The arguments to+-- 'Startup' are the combined 'Resource' of the current scheduler and+-- a thread-safe reference to the 'GlobalState'.+newtype Startup = St { runSt ::+ WorkSearch + -> HotVar GlobalState + -> IO () + }++instance Show Startup where+ show _ = "<Startup>"++instance Monoid Startup where+ mempty = St $ \_ _ -> return ()+ (St st1) `mappend` (St st2) = St st'+ where st' ws schedMap = st1 ws schedMap >> st2 ws schedMap + +-- | The 'WorkSearch' component of a 'Resource' is a callback that+-- responds to requests for work from Meta-Par workers. The arguments+-- to 'WorkSearch' are the 'Sched' for the current thread and a+-- thread-safe reference to the 'GlobalState'.+newtype WorkSearch = WS { runWS ::+ Sched+ -> HotVar GlobalState+ -> IO (Maybe (Par ()))+ }++instance Show WorkSearch where+ show _ = "<WorkSearch>"++instance Monoid WorkSearch where+ mempty = WS $ \_ _ -> return Nothing+ (WS ws1) `mappend` (WS ws2) = WS ws'+ where ws' sched schedMap = do+ mwork <- ws1 sched schedMap+ case mwork of+ Nothing -> ws2 sched schedMap+ _ -> return mwork ++-- | A 'Resource' provides an abstraction of heterogeneous execution+-- resources, and may be combined using 'Data.Monoid'+-- operations. Composition of resources is left-biased; for example,+-- if @resource1@ always returns work from its 'WorkSearch', the+-- composed resource @resource1 `mappend` resource2@ will never+-- request work from @resource2@.+data Resource = Resource {+ startup :: Startup+ , workSearch :: WorkSearch+ } deriving (Show)++instance Monoid Resource where+ mempty = Resource mempty mempty+ Resource st1 ws1 `mappend` Resource st2 ws2 =+ Resource (st1 `mappend` st2) (ws1 `mappend` ws2)++data Sched = Sched + { + ---- Per capability ----+ -- | Capability number+ no :: {-# UNPACK #-} !Int,+ -- | The 'ThreadId's of all worker threads on this capability+ tids :: HotVar (Set ThreadId),+ -- | The local 'WSDeque' for this worker. The worker may push+ -- and pop from the left of its own 'workpool', but workers on+ -- other threads may only steal from the right.+ workpool :: WSDeque (Par ()),+ -- | A 'GenIO' for random work stealing.+ rng :: HotVar GenIO,+ -- | A counter of how many extra workers are working on this+ -- capability. This situation arises during nested calls to+ -- 'runMetaPar', and the worker loop kills workers as necessary+ -- to keep this value at @1@.+ mortals :: HotVar Int, ++ -- | Tracks the number of consecutive times this worker has+ -- invoked a 'WorkSearch' and received 'Nothing'. This is used+ -- to implement backoff in+ -- 'Control.Monad.Par.Meta.Resources.Backoff'.+ consecutiveFailures :: IORef Int,++ -- | A per-thread source of unique identifiers for+ -- 'IVar's. Multiply this value by 'getNumCapabilities' and add+ -- 'no' for uniqueness.+ ivarUID :: HotVar Int,++ ---- Meta addition ----+ -- | The 'WorkSearch' of this worker's associated 'Resource'.+ schedWs :: WorkSearch+ }++instance Show Sched where+ show Sched{ no } = printf "Sched{ no=%d }" no ++--------------------------------------------------------------------------------+-- Helpers++-- | Produces a variant of 'forkOn' that allows exceptions from child+-- threads to propagate up to the parent thread.+forkWithExceptions :: (IO () -> IO ThreadId) -- ^ The basic 'forkOn' implementation+ -> String -- ^ A name for the child thread in error messages+ -> (IO () -> IO ThreadId)+forkWithExceptions forkit descr action = do + parent <- myThreadId+ forkit $ + Control.Exception.catch action+ (\ e -> do+ BS.hPutStrLn stderr $ BS.pack $ "Exception inside child thread "++show descr++": "++show e+ throwTo parent (e::SomeException)+ )++{-# INLINE ensurePinned #-}+-- Ensure that we execute an action within a pinned thread:+ensurePinned :: IO a -> IO a+ensurePinned action = do + tid <- myThreadId+ mp <- threadCapability' tid + case mp of+ Just (_, True) -> action+ Just (cap, _ ) -> do+ mv <- newEmptyMVar + void $ forkOn cap (action >>= putMVar mv)+ takeMVar mv+ Nothing -> do+ -- Older GHC case: we only consider a thread pinned if it's one+ -- of the threads manaaged by the global sched state. If it's+ -- not, we have no choice but to assume that it's not pinned,+ -- and spawn a new thread on CPU 0, which is an arbitrary+ -- choice.+ mv <- newEmptyMVar + void $ forkOn 0 (action >>= putMVar mv)+ takeMVar mv++--------------------------------------------------------------------------------+-- Work queue helpers++{-# INLINE popWork #-}+popWork :: Sched -> IO (Maybe (Par ()))+popWork Sched{ workpool, no } = do+ when dbg $ do+#if __GLASGOW_HASKELL__ >= 702+ (cap, _) <- threadCapability =<< myThreadId+ dbgTaggedMsg 4 $ BS.pack $ "[meta: cap "++show cap++ "] trying to pop local work on Sched "++ show no+#else+ dbgTaggedMsg 4 $ BS.pack $ "[meta] trying to pop local work on Sched "++ show no+#endif+ R.tryPopL workpool++{-# INLINE pushWork #-}+pushWork :: Sched -> Par () -> IO ()+pushWork Sched{ workpool } work = R.pushL workpool work++{-# INLINE pushWorkEnsuringWorker #-}+pushWorkEnsuringWorker :: Sched -> Par () -> IO (Maybe ())+pushWorkEnsuringWorker _ work = do+ no <- takeMVar workerPresentBarrier+ sched@Sched { tids } <- getSchedForCap no+ set <- readHotVar tids+ case Set.null set of+ False -> do+ when dbg $ printf "[meta] pushing ensured work onto cap %d\n" no+ Just <$> pushWork sched work+ True -> error $ printf "[meta] worker barrier filled by non-worker %d\n" no+ ++{-+{-# INLINE pushWorkEnsuringWorker #-}+pushWorkEnsuringWorker :: Sched -> Par () -> IO (Maybe ())+pushWorkEnsuringWorker Sched { no } work = do+ let attempt n = do+ sched@Sched { no, tids } <- getSchedForCap n+ set <- readHotVar tids+ case Set.null set of+ False -> do+ when dbg $ printf "[meta] pushing ensured work onto cap %d\n" no+ Just <$> pushWork sched work+ True -> return Nothing+ loop [] = return Nothing+ loop (n:ns) = do+ msucc <- attempt n+ case msucc of+ Just () -> return $ Just ()+ Nothing -> loop ns+ msucc <- attempt no+ schedNos <- IntMap.keys <$> readHotVar globalScheds+ case msucc of+ Just () -> return $ Just ()+ Nothing -> loop schedNos+-}++--------------------------------------------------------------------------------+-- Global structures and helpers for proper nesting behavior++{-# NOINLINE globalScheds #-}+globalScheds :: HotVar GlobalState+globalScheds = unsafePerformIO $ do+ n <- getNumCapabilities+ newHotVar $ V.replicate n Nothing++{-# NOINLINE workerPresentBarrier #-}+-- | Starts empty. Each new worker spawned tries to put its CPU+-- number. 'pushWorkEnsuringWorker' waits on this to ensure it pushes+-- the initial computation on a CPU with a worker.+workerPresentBarrier :: MVar Int+workerPresentBarrier = unsafePerformIO newEmptyMVar++{-# NOINLINE startBarrier #-}+startBarrier :: MVar ()+startBarrier = unsafePerformIO newEmptyMVar++-- | Warning: partial!+getSchedForCap :: Int -> IO Sched+getSchedForCap cap = do+ scheds <- readHotVar globalScheds+ case scheds V.! cap of+ Just sched -> return sched+ Nothing -> error $ + printf "tried to get a Sched for capability %d before initializing" cap+++makeOrGetSched :: WorkSearch -> Int -> IO Sched+makeOrGetSched ws cap = do+ sched <- Sched cap <$> newHotVar (Set.empty) -- tids+ <*> R.newQ -- workpool+ <*> (newHotVar =<< create) -- rng+ <*> newHotVar 0 -- mortals+ <*> newIORef 0 -- consecutiveFailures+ <*> newHotVar 0 -- ivarUID+ <*> pure ws -- workSearch+ modifyHotVar globalScheds $ \scheds ->+ case scheds V.! cap of+ Just sched -> (scheds, sched)+ Nothing -> if dbg+ then DT.trace (printf "[%d] created scheduler" cap)+ (scheds V.// [(cap, Just sched)], sched)+ else (scheds V.// [(cap, Just sched)], sched)++--------------------------------------------------------------------------------+-- Worker routines++forkOn' :: Int -> IO () -> IO ThreadId+#ifdef AFFINITY+forkOn' cap k = forkOn cap $ setAffinityOS cap >> k+#else+forkOn' = forkOn+#endif++-- | Spawn a Meta-Par worker that will stay on a given capability.+-- +-- Note: this does not check whether workers already exist on the+-- capability, and should be called appropriately. In particular, it+-- is the caller's responsibility to manage things like the 'mortal'+-- count of the given capability.+spawnWorkerOnCPU :: WorkSearch -- ^ The 'WorkSearch' called by the new worker+ -> Int -- ^ Capability+ -> IO ThreadId+spawnWorkerOnCPU ws cap = + forkWithExceptions (forkOn' cap) "spawned Par worker" $ do+ me <- myThreadId+ sched@Sched{ tids } <- makeOrGetSched ws cap+ modifyHotVar_ tids (Set.insert me)+ when dbg$ dbgTaggedMsg 2 $ BS.pack $+ printf "[meta: cap %d] spawning new worker" cap+ -- at least this worker is ready, so try filling the MVar+ _ <- tryPutMVar workerPresentBarrier cap+ -- wait on the barrier to start+ readMVar startBarrier+ when dbg$ dbgTaggedMsg 2 $ BS.pack $ + printf "[meta: cap %d] new working entering loop" cap+ runReaderT (workerLoop 0 errK) sched++errK :: a+errK = error "this closure shouldn't be used"++reschedule :: Par a+reschedule = Par $ ContT (workerLoop 0)++workerLoop :: Int -> ignoredCont -> ROnly ()+workerLoop failCount _k = do+ mysched@Sched{ no, mortals, schedWs=ws, consecutiveFailures } <- ask+ mwork <- liftIO $ popWork mysched+ case mwork of+ Just work -> do+ when dbg $ liftIO $ printf "[meta %d] popped work from own queue\n" no+ runContT (unPar work) $ const (workerLoop 0 _k)+ Nothing -> do+ -- check if we need to die+ die <- liftIO $ modifyHotVar mortals $ \ms ->+ case ms of+ 0 -> (0, False) + n | n > 0 -> (n-1, True)+ n -> error $+ printf "unexpected mortals count %d on cap %d" n no+ unless die $ do+ -- Before steal we make sure the consecutiveFailures field is up+ -- to date. This would seem to be a very ugly method of+ -- passing an extra argument to the steal action, and if we+ -- could tolerate it, it should perhaps become an additional argument:+ liftIO$ writeIORef consecutiveFailures failCount+ mwork <- liftIO (runWS ws mysched globalScheds)+ case mwork of+ Just work -> runContT (unPar work) $ const (workerLoop 0 _k)+ Nothing -> do + when dbg $ liftIO $ dbgTaggedMsg 4 $ BS.pack $ "[meta: cap "++show no++"] failed to find work; looping" + workerLoop (failCount + 1) _k ++{-# INLINE fork #-}+fork :: Par () -> Par ()+fork child = do+ sched <- ask+ callCC $ \parent -> do+ let wrapped = parent ()+ liftIO $ pushWork sched wrapped+ child >> reschedule++--------------------------------------------------------------------------------+-- IVar actions++{-# INLINE new #-}+new :: Par (IVar a)+new = liftIO $ IVar <$> newHotVar Empty++{-# INLINE get #-}+get :: IVar a -> Par a+get (IVar hv) = callCC $ \cont -> do+ contents <- liftIO $ readHotVar hv+ case contents of+ Full a -> return a+ _ -> do+ sch <- ask+ join . liftIO $ modifyHotVar hv $ \contents ->+ case contents of+ Empty -> (Blocked [pushWork sch . cont] , reschedule)+ Blocked ks -> (Blocked (pushWork sch . cont : ks), reschedule)+ Full a -> (Full a , return a)++{-# INLINE put_ #-}+put_ :: IVar a -> a -> Par ()+put_ (IVar hv) !content = do+ liftIO $ do+ ks <- modifyHotVar hv $ \contents ->+ case contents of+ Empty -> (Full content, [])+ Blocked ks -> (Full content, ks)+ Full _ -> error "multiple put"+ mapM_ ($content) ks++{-# INLINE put #-}+put :: NFData a => IVar a -> a -> Par ()+put iv a = deepseq a (put_ iv a)++{-# INLINE spawn #-}+spawn :: NFData a => Par a -> Par (IVar a)+spawn p = do r <- new; fork (p >>= put r); return r++{-# INLINE spawn_ #-}+spawn_ :: Par a -> Par (IVar a)+spawn_ p = do r <- new; fork (p >>= put_ r); return r+++--------------------------------------------------------------------------------+-- Entrypoint++-- | Run a 'Par' computation in the 'IO' monad, allowing+-- non-deterministic Meta-Par variants to be safely executed.+runMetaParIO :: Resource -> Par a -> IO a+runMetaParIO Resource{ startup=st, workSearch=ws } work = ensurePinned $ + do+ -- gather information+ tid <- myThreadId+ mp <- threadCapability' tid+ let cap = case mp of+ Just (n, _) -> n+ -- Older GHC case: default to CPU 0 if the calling+ -- thread is not already managed by meta-par+ Nothing -> 0++ sched@Sched{ tids, mortals } <- makeOrGetSched ws cap++ -- make the MVar for this answer, and wrap the incoming work, and+ -- push it on the current scheduler+ ansMVar <- newEmptyMVar+ let wrappedComp = do + ans <- work+ liftIO $ do+ dbgTaggedMsg 2 $ BS.pack "[meta] runMetaParIO computation finished, putting final MVar..."+ putMVar ansMVar ans+ -- if we're nested, we need to shut down the extra thread on+ -- our capability. If non-nested, we're done with the whole+ -- thing, and should really shut down.+ modifyHotVar_ mortals (1+)+ dbgTaggedMsg 2 $ BS.pack "[meta] runMetaParIO: done putting mvar and incrementing mortals."++ -- determine whether this is a nested call+ isNested <- Set.member tid <$> readHotVar tids+ if isNested then+ -- if it is, we need to spawn a replacement worker while we wait on ansMVar+ void $ spawnWorkerOnCPU ws cap+ -- if it's not, we need to run the init action + else runSt st ws globalScheds++ -- push the work, and then wait for the answer+ msucc <- pushWorkEnsuringWorker sched wrappedComp+ when (msucc == Nothing)+ $ error "[meta] could not find a scheduler with an active worker!"+ dbgTaggedMsg 2 $ BS.pack+ "[meta] runMetaParIO: Work pushed onto queue, now waiting on final MVar..."+ -- trigger the barrier so that workers start+ _ <- tryPutMVar startBarrier ()+ -- make sure the worker barrier is clear for subsequent runPars+ _ <- tryTakeMVar workerPresentBarrier+ ans <- takeMVar ansMVar++ -- TODO: Invariant checking -- make sure there is no work left:+ -- sanityCheck++ return ans++{-# INLINE runMetaPar #-}+-- | Run a 'Par' computation, and return its result as a pure+-- value. If the choice of 'Resource' introduces non-determinism, use+-- 'runMetaParIO' instead, as non-deterministic computations are not+-- referentially-transparent.+runMetaPar :: Resource -> Par a -> a+runMetaPar rsrc work = unsafePerformIO $ runMetaParIO rsrc work++--------------------------------------------------------------------------------+-- Boilerplate++spawnP :: NFData a => a -> Par (IVar a)+spawnP = spawn . return++instance PC.ParFuture IVar Par where+ get = get+ spawn = spawn+ spawn_ = spawn_+ spawnP = spawnP++instance PC.ParIVar IVar Par where+ fork = fork+ new = new+ put_ = put_
+ Control/Monad/Par/Meta/HotVar/IORef.hs view
@@ -0,0 +1,44 @@+{-# LANGUAGE TypeSynonymInstances #-}+{-# OPTIONS_GHC -Wall -fno-warn-orphans #-}++-- | A very simple layer of abstraction for frequently modified shared+-- variables. This additional indirection is only present for+-- benchmarking `IORef` operations vs. `MVar` or `TVar` implementations.+module Control.Monad.Par.Meta.HotVar.IORef ( HotVar+ , modifyHotVar+ , modifyHotVar_+ , newHotVar+ , readHotVar+ , readHotVarRaw+ , writeHotVar+ , writeHotVarRaw+ ) where++import Data.IORef++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++type HotVar a = IORef a+newHotVar = newIORef+modifyHotVar = atomicModifyIORef+modifyHotVar_ v fn = atomicModifyIORef v (\a -> (fn a, ()))+readHotVar = readIORef+writeHotVar = writeIORef+instance Show (HotVar a) where + show _ = "<ioref>"++readHotVarRaw :: HotVar a -> IO a+writeHotVarRaw :: HotVar a -> a -> IO ()+readHotVarRaw = readHotVar+writeHotVarRaw = writeHotVar++{- INLINE newHotVar -}+{- INLINE modifyHotVar -}+{- INLINE modifyHotVar_ -}+{- INLINE readHotVar -}+{- INLINE writeHotVar -}+
+ Control/Monad/Par/Meta/Resources/Backoff.hs view
@@ -0,0 +1,61 @@+{-# LANGUAGE OverloadedStrings, NamedFieldPuns #-}++-- | This module implements exponential backoff so as to prevent+-- spamming of steal actions. This is always a good idea, and+-- especially so in the distributed case where steal attempts send+-- actual messages.+--+-- Normally backoff functionality is baked into the scheduler loop.+-- One nice aspect of the Meta scheduler design is that backoff can+-- become "just another resource". Most schedulers (compositions)+-- should include this at tho bottom of their stack.++module Control.Monad.Par.Meta.Resources.Backoff ( defaultStartup+ , mkWorkSearch+ , mkResource+ ) where ++import Data.IORef (readIORef)+import Data.Word (Word64)+import Control.Monad.Par.Meta+import Control.Monad.Par.Meta.Resources.Debugging (dbgTaggedMsg)+import Control.Concurrent (threadDelay)+import qualified Data.ByteString.Char8 as BS++mkResource :: Word64 -> Word64 -> Resource+mkResource shortest longest =+ Resource defaultStartup (mkWorkSearch shortest longest)++defaultStartup :: Startup+defaultStartup = St (\ _ _ -> return () )++-- | To construct a WorkSearch we need to know the minimum and+-- maximum amount of time (nanoseconds) to sleep. The exponential+-- backoff policy is always the same: it starts at 1ns and doubles.+-- +-- The thing that changes over time is whether sleeping actually+-- *occurs*. For example, `mkWorkSearch 1000 100000` will not sleep+-- for the first ten invocations (until 1024), and then will sleep an+-- amount that doubles each time until it surpasses the maximum, at+-- which point each sleep will be for the maximum: 100ms.+mkWorkSearch :: Word64 -> Word64 -> WorkSearch+-- Sleeping ZERO time means not sleeping at all:+mkWorkSearch _ 0 = WS$ \ _ _ -> return Nothing+mkWorkSearch shortest longest = WS ws + where + ws Sched{consecutiveFailures} _ = do+ failCount <- readIORef consecutiveFailures+ let nanos = if failCount >= 64+ then longest + else 2 ^ failCount+ if nanos >= shortest then do + let capped = min longest nanos+ dbgTaggedMsg 3 $ "Backoff: Sleeping, nanoseconds = " `BS.append` BS.pack (show capped)+ threadDelay (fromIntegral capped)+ else do + dbgTaggedMsg 4 $ "Backoff: NOT yet sleeping, nanoseconds = " `BS.append` BS.pack (show nanos)+ -- QUESTION - do we want to yield here? Probably not.+ -- Maybe for some ranges of nanosecond intervals we could yield instead...+ -- yield+ return Nothing+
+ Control/Monad/Par/Meta/Resources/Debugging.hs view
@@ -0,0 +1,155 @@+{-# LANGUAGE MagicHash, UnboxedTuples, CPP #-}+{-# LANGUAGE OverloadedStrings #-}+{-# OPTIONS_GHC -Wall #-}++module Control.Monad.Par.Meta.Resources.Debugging ( dbg+ , dbgTaggedMsg+ , dbgDelay+ , dbgCharMsg+ , meaningless_alloc+ , taggedmsg_global_mode+ , verbosity+ ) where++----------------------------------------+-- For tracing events:+-- import Foreign+import Foreign.C (CString)+import GHC.Exts (traceEvent#, Ptr(Ptr))+-- import GHC.IO hiding (liftIO)+import GHC.IO (IO(IO))+----------------------------------------++import qualified Data.ByteString.Char8 as BS+import Data.IORef (IORef, newIORef, readIORef)+import Data.Monoid (mappend, Monoid)+import Control.Monad (when)+import Control.Concurrent (myThreadId, threadDelay)+import System.IO (hFlush, stderr)+import System.IO.Unsafe (unsafePerformIO)+import System.Environment (getEnvironment)+++-----------------------------------------------------------------------------------+-- VERBOSITY and DEBUGGING+-----------------------------------------------------------------------------------+-- This controls how much output will be printed, 0-5.+-- 5 is "debug" mode and affects other aspects of execution (see dbgDelay)++#define TMPDBG+#define EVENTLOG++-- RRN [2012.02.28] -- Eventually for performance reasons this+-- decision will probably be made statically. For now, in the heat of+-- debugging, it is nice to be able to change it dynamically.++-- Well, to change it truly DYNAMICALLY, it would have to be an IORef.+-- For now we just allow an environment variable to override the+-- setting at load time.++{-# NOINLINE verbosity #-}+verbosity :: Int+verbosity = unsafePerformIO$ do+-- putStrLn "GETTING ENV TO READ VERBOSITY..."+ env <- getEnvironment+-- putStrLn$ " ENV LENGTH " ++ show (length env)+ case lookup "VERBOSITY" env of+ -- if defined and not empty, default to 1+ Just "" -> return 1+ Just s -> do let n = read s+ when (n >= 2)$ putStrLn "(!) Responding to VERBOSITY environment variable!"+ return n+#ifdef DEBUG+ Nothing -> return 5+#else+ Nothing -> do -- putStrLn "DEFAULTING VERBOSITY TO 1" + return 1+#endif+++{-# NOINLINE binaryEventLog #-}+binaryEventLog :: Bool+binaryEventLog = unsafePerformIO$ do+ env <- getEnvironment+ case lookup "EVENTLOG" env of + Nothing -> return False+ Just "" -> return False + Just "0" -> return False + Just _ -> return True++-- When debugging is turned on we will do extra invariant checking:+dbg :: Bool+#ifdef DEBUG+dbg = True+#else+dbg = False+#endif++whenVerbosity :: Monad m => Int -> m () -> m ()+whenVerbosity n action = when (verbosity >= n) action++-- | dbgTaggedMsg is our routine for logging debugging output:+dbgTaggedMsg :: Int -> BS.ByteString -> IO ()+-- dbgTaggedMsg :: Int -> String -> IO ()+dbgTaggedMsg = if binaryEventLog then binaryLogMsg else textLogMsg++textLogMsg :: Int -> BS.ByteString -> IO ()+textLogMsg lvl s = + whenVerbosity lvl $ + do m <- readIORef taggedmsg_global_mode+ tid <- myThreadId+ BS.putStrLn$ " [distmeta" +++ m +++" "+++ sho tid +++"] "+++s++(+++) :: Monoid a => a -> a -> a+(+++) = mappend++sho :: Show a => a -> BS.ByteString+sho = BS.pack . show+++binaryLogMsg :: Int -> BS.ByteString -> IO ()+binaryLogMsg lvl s = do +-- meaningless_alloc -- This works as well as a print for preventing the inf loop [2012.03.01]!+ whenVerbosity lvl $ do + m <- readIORef taggedmsg_global_mode+ tid <- myThreadId+ let msg = " [distmeta"+++m+++" "+++ sho tid+++"] "+++s+ BS.useAsCString msg myTraceEvent+ return ()++myTraceEvent :: CString -> IO ()+myTraceEvent (Ptr msg) = IO $ \s -> case traceEvent# msg s of s' -> (# s', () #)+++-- | `dbgCharMsg` is for printing a small tag like '.' (with no line+-- termination) which produces a different kind of visual output.+-- dbgCharMsg :: Int -> String -> String -> IO ()+dbgCharMsg :: Int -> BS.ByteString -> BS.ByteString -> IO ()+dbgCharMsg lvl tag fullmsg = + if binaryEventLog + then dbgTaggedMsg lvl fullmsg -- It doesn't make sense to event-log a single character.+ else whenVerbosity lvl $ do BS.hPutStr stderr tag; hFlush stderr+++-- When debugging it is helpful to slow down certain fast paths to a human scale:+dbgDelay :: BS.ByteString -> IO ()+dbgDelay _ = + if dbg+ then threadDelay (200*1000)+ else return ()++meaningless_alloc :: IO ()+meaningless_alloc = + case length (fibls 5) of + 0 -> return (error "Never happen!")+ _ -> return ()+ where + fibls :: Int -> [Int]+ fibls n | n <= 1 = [1::Int] + fibls n = fibls (n-1) ++ fibls (n-2)+++{-# NOINLINE taggedmsg_global_mode #-}+-- Just for debugging, tracking global node as M (master) or S (slave):+taggedmsg_global_mode :: IORef BS.ByteString+taggedmsg_global_mode = unsafePerformIO$ newIORef "_M"
+ Control/Monad/Par/Meta/Resources/SMP.hs view
@@ -0,0 +1,127 @@+{-# LANGUAGE CPP #-}+{-# LANGUAGE NamedFieldPuns #-}++{-# OPTIONS_GHC -Wall -fno-warn-unused-do-bind -fno-warn-name-shadowing #-}++-- | This module implements a Meta-Par 'Resource' for SMP parallelism,+-- suitable as a base for combined execution resources (e.g.,+-- @Control.Monad.Par.Meta.AccSMP@).+module Control.Monad.Par.Meta.Resources.SMP (+ -- * Resource creation+ mkResource+ , mkResourceOn+ -- * Default implementation+ , defaultStartup+ , defaultWorkSearch+ -- * Customizable implementation+ , startupForCaps+ , wsForCaps+) where++import Control.Monad++import Data.Concurrent.Deque.Reference as R+import Data.List (nub)+import qualified Data.Vector as V+import qualified Data.ByteString.Char8 as BS++import System.Environment (getEnvironment)+import System.IO.Unsafe+import System.Random.MWC++import Text.Printf++import Control.Monad.Par.Meta++import Control.Monad.Par.Meta.HotVar.IORef+import Control.Monad.Par.Meta.Resources.Debugging (dbgTaggedMsg)++#if __GLASGOW_HASKELL__ >= 702+import Control.Concurrent (getNumCapabilities)+#else+import GHC.Conc (numCapabilities)+getNumCapabilities = return numCapabilities+#endif+++-- | Create an SMP resource for all capabilities. +mkResource :: Int -- ^ The number of steal attempts per 'WorkSearch' call.+ -> Resource+mkResource tries = Resource defaultStartup (defaultWorkSearch tries)++-- | Create an SMP resource for a configurable list of capabilities.+mkResourceOn :: [Int] -- ^ Capability list.+ -> Int -- ^ The number of steal attempts per 'WorkSearch' call.+ -> Resource+mkResourceOn caps tries = Resource (startupForCaps caps) (wsForCaps caps tries)++{-# NOINLINE getCaps #-}+getCaps :: [Int]+getCaps = unsafePerformIO $ do+ env <- getEnvironment+ case lookup "SMP_CAPS" env of+ Just cs -> do+ dbgTaggedMsg 2 $ BS.pack $ printf "[SMP] initialized with capability list %s\n" + (show ((read cs) :: [Int]))+ return $ read cs+ Nothing -> do + n <- getNumCapabilities+ dbgTaggedMsg 2 $ BS.pack $ printf "[SMP] initialized with capability list %s\n" + (show ([0..n-1] :: [Int])) + return [0..n-1]++-- | 'Startup' for spawning threads on all capabilities, or from a+-- 'read'-able list of capability numbers in the environment variable+-- @SMP_CAPS@.+defaultStartup :: Startup+defaultStartup = startupForCaps getCaps+ +-- | 'Startup' for spawning threads only on a particular set of+-- capabilities.+startupForCaps :: [Int] -> Startup+startupForCaps caps = St st+ where st ws _ = do+ dbgTaggedMsg 2 $ BS.pack $ printf "spawning worker threads for shared memory on caps:\n"+ dbgTaggedMsg 2 $ BS.pack $ printf "\t%s\n" (show caps)+ let caps' = nub caps+ forM_ caps' $ \n ->+ spawnWorkerOnCPU ws n >> return ()++{-# INLINE randModN #-}+randModN :: Int -> HotVar GenIO -> IO Int+randModN caps rngRef = uniformR (0, caps-1) =<< readHotVar rngRef++-- | 'WorkSearch' for all capabilities.+defaultWorkSearch :: Int -> WorkSearch+defaultWorkSearch = wsForCaps getCaps ++-- | Given a set of capabilities and a number of steals to attempt per+-- capability, return a 'WorkSearch'.+wsForCaps :: [Int] -> Int -> WorkSearch+wsForCaps caps triesPerCap = WS ws+ where + numCaps = length caps+ numTries = numCaps * triesPerCap+ capVec = V.fromList caps+ ws Sched { no, rng } schedsRef = do+ scheds <- readHotVar schedsRef+ let {-# INLINE getNext #-}+ getNext :: IO Int+ getNext = randModN numCaps rng+ -- | Main steal loop+ loop :: Int -> Int -> IO (Maybe (Par ()))+ loop 0 _ = return Nothing+ loop n i | capVec V.! i == no = loop (n-1) =<< getNext+ | otherwise =+ let target = capVec V.! i in+ case scheds V.! target of+ Nothing -> do + dbgTaggedMsg 2 $ BS.pack $ + printf "WARNING: no Sched for cap %d during steal\n" target+ loop (n-1) =<< getNext+ Just Sched { workpool = stealee } -> do+ mtask <- R.tryPopR stealee+ case mtask of+ Nothing -> loop (n-1) =<< getNext+ jtask -> return jtask + loop numTries =<< getNext
+ Control/Monad/Par/Meta/Resources/SingleThreaded.hs view
@@ -0,0 +1,53 @@+{-# LANGUAGE CPP #-}++{-# OPTIONS_GHC -Wall #-}++-- | A simple single-threaded 'Resource' that is a useful+-- accompaniment for testing non-CPU resources such as GPU or+-- distributed.+module Control.Monad.Par.Meta.Resources.SingleThreaded ( + -- * Resource creation+ mkResource+ -- * Implementation+ , defaultStartup+ , defaultWorkSearch+) where++import Control.Concurrent+import Control.Monad++import Text.Printf++import Control.Monad.Par.Meta++dbg :: Bool+#ifdef DEBUG+dbg = True+#else+dbg = True+#endif++-- | Create a single-threaded resource.+mkResource :: Resource+mkResource = Resource defaultStartup defaultWorkSearch++-- | Spawn a single Meta-Par worker.+defaultStartup :: Startup+defaultStartup = St st + where st ws _ = do+#if __GLASGOW_HASKELL__ >= 702+ (cap, _) <- threadCapability =<< myThreadId+#else+ -- Older GHCs: run on CPU 0 if we can't ask for caller's capability+ let cap = 0+#endif+ when dbg $ printf " [%d] spawning single worker\n" cap+ -- This startup is called from the "main" thread, we need+ -- to spawn a worker to do the actual work:+ spawnWorkerOnCPU ws cap >> return ()++-- | A single-threaded resource by itself is not aware of any other+-- sources of work, so its 'WorkSearch' always returns 'Nothing'.+defaultWorkSearch :: WorkSearch+defaultWorkSearch = WS ws + where ws _ _ = return Nothing
+ Control/Monad/Par/Meta/SMP.hs view
@@ -0,0 +1,43 @@+{-# LANGUAGE CPP #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-}++#if __GLASGOW_HASKELL__ >= 702+{-# Language Trustworthy #-}+#endif++{-# OPTIONS_GHC -Wall #-}++module Control.Monad.Par.Meta.SMP (+ -- * Meta-Par monad for SMP parallelism+ Par+ , Meta.IVar+ -- * Operations+ , ParFuture(..)+ , ParIVar(..)+ -- * Entrypoints+ , runPar+ , runParIO+) where++import Control.Applicative (Applicative)+import Control.Monad.Par.Class+import qualified Control.Monad.Par.Meta as Meta+import qualified Control.Monad.Par.Meta.Resources.SMP as SMP++-- | The Meta-Par monad specialized for SMP parallelism.+newtype Par a = Par { unPar :: Meta.Par a }+ deriving (Functor, Applicative, Monad,+ ParFuture Meta.IVar, ParIVar Meta.IVar)++-- | Number of times to attempt stealing from other SMP workers before+-- continuing in Meta-Par worker loop.+tries :: Int+tries = 20++resource :: Meta.Resource+resource = SMP.mkResource tries++runPar :: Par a -> a+runParIO :: Par a -> IO a+runPar = Meta.runMetaPar resource . unPar+runParIO = Meta.runMetaParIO resource . unPar
+ Control/Monad/Par/Meta/Serial.hs view
@@ -0,0 +1,37 @@+{-# LANGUAGE CPP #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-}++#if __GLASGOW_HASKELL__ >= 702+{-# Language Trustworthy #-}+#endif++{-# OPTIONS_GHC -Wall #-}++module Control.Monad.Par.Meta.Serial (+ -- * Meta-Par monad for single-threaded execution+ Par+ -- * Operations+ , ParFuture(..)+ , ParIVar(..)+ -- * Entrypoints+ , runPar+ , runParIO+) where++import Control.Applicative (Applicative)+import Control.Monad.Par.Class+import qualified Control.Monad.Par.Meta as Meta+import qualified Control.Monad.Par.Meta.Resources.SingleThreaded as Single++-- | The Meta-Par monad specialized for single-threaded execution.+newtype Par a = Par { unPar :: Meta.Par a }+ deriving (Functor, Applicative, Monad,+ ParFuture Meta.IVar, ParIVar Meta.IVar)++resource :: Meta.Resource+resource = Single.mkResource ++runPar :: Par a -> a+runParIO :: Par a -> IO a+runPar = Meta.runMetaPar resource . unPar+runParIO = Meta.runMetaParIO resource . unPar
+ LICENSE view
@@ -0,0 +1,30 @@+Copyright Simon Marlow 2011++All rights reserved.++Redistribution and use in source and binary forms, with or without+modification, are permitted provided that the following conditions are met:++ * Redistributions of source code must retain the above copyright+ notice, this list of conditions and the following disclaimer.++ * Redistributions in binary form must reproduce the above+ copyright notice, this list of conditions and the following+ disclaimer in the documentation and/or other materials provided+ with the distribution.++ * Neither the name of Simon Marlow nor the names of other+ contributors may be used to endorse or promote products derived+ from this software without specific prior written permission.++THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS+"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT+LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR+A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT+OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,+SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT+LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,+DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY+THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT+(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE+OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ Setup.hs view
@@ -0,0 +1,3 @@+#!/usr/bin/env runhaskell+import Distribution.Simple+main = defaultMain
+ System/Posix/Affinity.hs view
@@ -0,0 +1,25 @@+{-# LANGUAGE ForeignFunctionInterface #-}++module System.Posix.Affinity (setAffinityOS, setAffinityRange) where++import Control.Monad+import Foreign+import Foreign.C+import Foreign.Marshal.Array+import System.IO.Error++foreign import ccall "pin_pthread" pin_pthread :: CInt -> IO Errno+foreign import ccall "pin_process_multi" pin_process_multi :: Ptr CInt -> CInt -> IO Errno++setAffinityOS :: Int -> IO ()+setAffinityOS cpu = do+ err <- pin_pthread $ fromIntegral cpu+ when (err /= eOK) $ + ioError $ errnoToIOError "setAffinityOS" err Nothing Nothing++setAffinityRange :: [Int] -> IO ()+setAffinityRange cpus =+ withArrayLen cpus $ \len ptr -> do+ err <- pin_process_multi (castPtr ptr) (fromIntegral len)+ when (err /= eOK) $ + ioError $ errnoToIOError "setAffinityRangeOS" err Nothing Nothing
+ cbits/pin.c view
@@ -0,0 +1,30 @@+#define _GNU_SOURCE+#include <pthread.h>+#include <sched.h>+#include <stdlib.h>++int pin_pthread(int cpu) {+ cpu_set_t cs;+ pthread_t thread;++ thread = pthread_self();++ CPU_ZERO(&cs);+ CPU_SET(cpu, &cs);+ return pthread_setaffinity_np(thread, sizeof(cpu_set_t), &cs);+}++int pin_process_multi(int cpus[], int num) {+ cpu_set_t cs;+ int i;+ size_t size;++ size = CPU_ALLOC_SIZE(CPU_COUNT(&cs));++ CPU_ZERO(&cs);+ for (i=0; i<num; i++) {+ CPU_SET(cpus[i], &cs);+ }++ return sched_setaffinity(0, size, &cs);+}
+ include/pin.h view
@@ -0,0 +1,2 @@+int pin_pthread(int cpu);+int pin_process_multi(int cpus[], int num);
+ meta-par.cabal view
@@ -0,0 +1,87 @@+Name: meta-par+Version: 0.3+Synopsis: Provides the monad-par interface, but based on modular scheduler "mix-ins".++-- Version history:++-- 0.3: Initial release.++Description: ++ This library provides a composable abstraction for /Resources/ which+ can be mixed and matched to build /Schedulers/.+++ A `Resource` typically corresponds to a specific kind of hardware or+ to a style of scheduling. This package contains implementations of+ basic `Resource`s that implement parallel scheduling on the CPU.++ + To use a complete meta-par Scheduler, import Control.Monad.Par.SMP for + example, which will provide a `runPar` as well as instances for the + relevant classes that enable `Par` programming (e.g. `ParFuture`).+++Homepage: https://github.com/simonmar/monad-par+License: BSD3+License-file: LICENSE+Author: Adam Foltzer, Ryan Newton+Maintainer: Adam Foltzer <acfoltzer@gmail.com>+Copyright: (c) Adam Foltzer 2011-2012+Stability: Experimental+Category: Control,Parallelism,Monads+Build-type: Simple+Cabal-version: >=1.8++extra-source-files:+ include/pin.h++Library+ Exposed-modules: + -- Meta scheduler and friends+ Control.Monad.Par.Meta+ -- A complete scheduler:+ , Control.Monad.Par.Meta.SMP+ , Control.Monad.Par.Meta.Serial++ -- Mostly internal modules for use to build other schedulers:+ , Control.Monad.Par.Meta.Resources.SingleThreaded + , Control.Monad.Par.Meta.Resources.SMP+ , Control.Monad.Par.Meta.Resources.Backoff+ , Control.Monad.Par.Meta.Resources.Debugging+ , Control.Monad.Par.Meta.HotVar.IORef ++ -- NOTE: NUMA not released yet.++ Build-depends: base >= 4 && < 5+ -- This provides the interface for monad-par:+ , abstract-par + , abstract-deque >= 0.1.4+ , mwc-random >= 0.11+ , containers+ , bytestring >= 0.9+ , transformers >= 0.2.2.0+ , deepseq >= 1.2+ , vector+-- , cereal >= 0.3+ , mtl >= 2.0.1.0++ if flag(affinity) {+ CPP-options: -DAFFINITY+ }++ if flag(affinity) {+ Other-modules: System.Posix.Affinity+ }++ if flag(affinity) { + c-sources: cbits/pin.c+ include-dirs: include+ includes: pin.h+ }+ + ghc-options: -Wall++Flag affinity+ Description: Turn on pinning to CPUs via pthread/linux scheduler calls.+ Default: False