{-# 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).moduleControl.Monad.Par.Scheds.DirectInternalwhereimportControl.Applicativeimport"mtl"Control.Monad.ContasCimportqualified"mtl"Control.Monad.ReaderasRDimportqualifiedSystem.Random.MWCasRandomimportControl.Concurrenthiding(yield)importGHC.ConcimportData.IORefimportData.Concurrent.Deque.Class(WSDeque)-- import Data.Concurrent.Deque.Reference.DequeInstance-- import Data.Concurrent.Deque.Reference as RimportData.Concurrent.Deque.Class(WSDeque)importData.Concurrent.Deque.Reference.DequeInstanceimportData.Concurrent.Deque.ReferenceasRimportqualifiedData.SetasSimportData.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.--newtypePara=Par{unPar::C.ContT()ROnlya}deriving(Monad,MonadCont,RD.MonadReaderSched)typeROnly=RD.ReaderTSchedIOtypeSessionID=Word64-- An ID along with a flag to signal completion:dataSession=SessionSessionID(HotVarBool)dataSched=Sched{---- Per worker ----no::{-# UNPACK #-}!Int,workpool::WSDeque(Par()),rng::HotVarRandom.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[MVarBool],-- waiting idle workersscheds::[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.SetSessionID),-- A counter to support unique session IDs:sessionCounter::HotVarSessionID}---------------------------------------------------------------------------------- Helpers #1: Atomic Variables---------------------------------------------------------------------------------- TEMP: Experimental#ifndef HOTVAR#define HOTVAR 1#endifnewHotVar::a->IO(HotVara)modifyHotVar::HotVara->(a->(a,b))->IObmodifyHotVar_::HotVara->(a->a)->IO()writeHotVar::HotVara->a->IO()readHotVar::HotVara->IOa-- readHotVarRaw :: HotVar a -> m a-- writeHotVarRaw :: HotVar a -> m a{-# INLINE newHotVar #-}{-# INLINE modifyHotVar #-}{-# INLINE modifyHotVar_ #-}{-# INLINE readHotVar #-}{-# INLINE writeHotVar #-}#if HOTVAR == 1typeHotVara=IORefanewHotVar=newIORefmodifyHotVar=atomicModifyIORefmodifyHotVar_vfn=atomicModifyIORefv(\a->(fna,()))readHotVar=readIORefwriteHotVar=writeIORefinstanceShow(IORefa)whereshowref="<ioref>"-- hotVarTransaction = idhotVarTransaction=error"Transactions not currently possible for IO refs"readHotVarRaw=readHotVarwriteHotVarRaw=writeHotVar#elif HOTVAR == 2 #warning "Using MVars for hot atomic variables."-- This uses MVars that are always full with *something*typeHotVara=MVaranewHotVarx=dov<-newMVar;putMVarvx;returnvmodifyHotVarvfn=modifyMVarv(return.fn)modifyHotVar_vfn=modifyMVar_v(return.fn)readHotVar=readMVarwriteHotVarvx=doswapMVarvx;return()instanceShow(MVara)whereshowref="<mvar>"-- hotVarTransaction = id-- We could in theory do this by taking the mvar to grab the lock.-- But we'd need some temporary storage....hotVarTransaction=error"Transactions not currently possible for MVars"readHotVarRaw=readHotVarwriteHotVarRaw=writeHotVar#elif HOTVAR == 3#warning "Using TVars for hot atomic variables."-- Simon Marlow said he saw better scaling with TVars (surprise to me):typeHotVara=TVaranewHotVar=newTVarIOmodifyHotVartvfn=atomically(dox<-readTVartvlet(x2,b)=fnxwriteTVartvx2returnb)modifyHotVar_tvfn=atomically(dox<-readTVartv;writeTVartv(fnx))readHotVarx=atomically$readTVarxwriteHotVarvx=atomically$writeTVarvxinstanceShow(TVara)whereshowref="<tvar>"hotVarTransaction=atomicallyreadHotVarRaw=readTVarwriteHotVarRaw=writeTVar#endif