{-# LANGUAGE RankNTypes, NamedFieldPuns, BangPatterns,
ExistentialQuantification, CPP, ScopedTypeVariables,
TypeSynonymInstances, MultiParamTypeClasses,
GeneralizedNewtypeDeriving, PackageImports,
ParallelListComp #-}
{-# LANGUAGE TypeFamilies #-}
module Control.Monad.Par.Scheds.Direct (
Sched(..),
Par,
IVar(..), IVarContents(..),
runPar, runParIO,
new, get, put_, fork,
newFull, newFull_, put,
spawn, spawn_, spawnP,
spawn1_, fixPar, FixParException (..)
) where
import Control.Applicative
import Control.Concurrent hiding (yield)
import Data.IORef (IORef,newIORef,readIORef,writeIORef,atomicModifyIORef)
import Text.Printf (printf)
import GHC.Conc (numCapabilities,yield)
import Control.Monad
import Control.Monad.IO.Class
import Control.Monad.Trans
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.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, 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)
#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 qualified Control.Exception as E
import Prelude hiding (null)
import qualified Prelude
#if __GLASGOW_HASKELL__ <= 700
import GHC.Conc (forkOnIO)
forkOn = forkOnIO
#endif
#define PARPUTS
#define FORKPARENT
#define IDLING_ON
#define WAKEIDLE
#ifdef DEBUG_DIRECT
#warning "DEBUG: Activating debugging for Direct.hs"
import Debug.Trace (trace)
import System.Environment (getEnvironment)
theEnv = unsafePerformIO $ getEnvironment
dbg = True
dbglvl = 1
#else
dbg :: Bool
dbg = Bool
False
dbglvl :: Int
dbglvl = Int
0
#endif
dbg :: Bool
dbglvl :: Int
_PARPUTS :: Bool
#ifdef PARPUTS
_PARPUTS :: Bool
_PARPUTS = Bool
True
#else
_PARPUTS = False
#endif
_FORKPARENT :: Bool
#ifdef FORKPARENT
_FORKPARENT :: Bool
_FORKPARENT = Bool
True
#else
#warning "FORKPARENT POLICY NOT USED; THIS IS GENERALLY WORSE"
_FORKPARENT = False
#endif
_IDLING_ON :: Bool
#ifdef IDLING_ON
_IDLING_ON :: Bool
_IDLING_ON = Bool
True
#else
_IDLING_ON = False
#endif
_WAIT_FOR_WORKERS :: Bool
#ifdef WAIT_FOR_WORKERS
_WAIT_FOR_WORKERS = True
#else
_WAIT_FOR_WORKERS :: Bool
_WAIT_FOR_WORKERS = Bool
False
#endif
type ROnly = RD.ReaderT Sched IO
newtype IVar a = IVar (IORef (IVarContents a))
data IVarContents a = Full a | Empty | Blocked [a -> IO ()]
unsafeParIO :: IO a -> Par a
unsafeParIO :: forall a. IO a -> Par a
unsafeParIO IO a
iom = forall a. ContT () (ReaderT Sched IO) a -> Par a
Par (forall (t :: (* -> *) -> * -> *) (m :: * -> *) a.
(MonadTrans t, Monad m) =>
m a -> t m a
liftforall a b. (a -> b) -> a -> b
$ forall (t :: (* -> *) -> * -> *) (m :: * -> *) a.
(MonadTrans t, Monad m) =>
m a -> t m a
lift IO a
iom)
io :: IO a -> Par a
io :: forall a. IO a -> Par a
io = forall a. IO a -> Par a
unsafeParIO
#ifdef NESTED_SCHEDS
globalWorkerPool :: IORef (M.Map ThreadId Sched)
globalWorkerPool = unsafePerformIO $ newIORef M.empty
#endif
{-# INLINE amINested #-}
{-# INLINE registerWorker #-}
{-# INLINE unregisterWorker #-}
amINested :: ThreadId -> IO (Maybe Sched)
registerWorker :: ThreadId -> Sched -> IO ()
unregisterWorker :: ThreadId -> IO ()
#ifdef NESTED_SCHEDS
amINested tid = do
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 :: ThreadId -> IO (Maybe Sched)
amINested ThreadId
_ = forall (m :: * -> *) a. Monad m => a -> m a
return forall a. Maybe a
Nothing
registerWorker :: ThreadId -> Sched -> IO ()
registerWorker ThreadId
_ Sched
_ = forall (m :: * -> *) a. Monad m => a -> m a
return ()
unregisterWorker :: ThreadId -> IO ()
unregisterWorker ThreadId
_tid = forall (m :: * -> *) a. Monad m => a -> m a
return ()
#endif
{-# INLINE popWork #-}
popWork :: Sched -> IO (Maybe (Par ()))
popWork :: Sched -> IO (Maybe (Par ()))
popWork Sched{ WSDeque (Par ())
workpool :: Sched -> WSDeque (Par ())
workpool :: WSDeque (Par ())
workpool, Int
no :: Sched -> Int
no :: Int
no } = do
Maybe (Par ())
mb <- forall a. SimpleDeque a -> IO (Maybe a)
R.tryPopL WSDeque (Par ())
workpool
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when Bool
dbg forall a b. (a -> b) -> a -> b
$ case Maybe (Par ())
mb of
Maybe (Par ())
Nothing -> forall (m :: * -> *) a. Monad m => a -> m a
return ()
Just Par ()
_ -> do StableName (Maybe (Par ()))
sn <- forall a. a -> IO (StableName a)
makeStableName Maybe (Par ())
mb
forall r. PrintfType r => String -> r
printf String
" [%d] -> POP work unit %d\n" Int
no (forall a. StableName a -> Int
hashStableName StableName (Maybe (Par ()))
sn)
forall (m :: * -> *) a. Monad m => a -> m a
return Maybe (Par ())
mb
{-# INLINE pushWork #-}
pushWork :: Sched -> Par () -> IO ()
pushWork :: Sched -> Par () -> IO ()
pushWork Sched { WSDeque (Par ())
workpool :: WSDeque (Par ())
workpool :: Sched -> WSDeque (Par ())
workpool, HotVar [MVar Bool]
idle :: Sched -> HotVar [MVar Bool]
idle :: HotVar [MVar Bool]
idle, Int
no :: Int
no :: Sched -> Int
no } Par ()
task = do
forall t. SimpleDeque t -> t -> IO ()
R.pushL WSDeque (Par ())
workpool Par ()
task
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when Bool
dbg forall a b. (a -> b) -> a -> b
$ do StableName (Par ())
sn <- forall a. a -> IO (StableName a)
makeStableName Par ()
task
forall r. PrintfType r => String -> r
printf String
" [%d] -> PUSH work unit %d\n" Int
no (forall a. StableName a -> Int
hashStableName StableName (Par ())
sn)
#if defined(IDLING_ON) && defined(WAKEIDLE)
HotVar [MVar Bool] -> IO ()
tryWakeIdle HotVar [MVar Bool]
idle
#endif
forall (m :: * -> *) a. Monad m => a -> m a
return ()
tryWakeIdle :: HotVar [MVar Bool] -> IO ()
tryWakeIdle :: HotVar [MVar Bool] -> IO ()
tryWakeIdle HotVar [MVar Bool]
idle = do
[MVar Bool]
idles <- forall a. HotVar a -> IO a
readHotVar HotVar [MVar Bool]
idle
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when (Bool -> Bool
not (forall (t :: * -> *) a. Foldable t => t a -> Bool
Prelude.null [MVar Bool]
idles)) forall a b. (a -> b) -> a -> b
$ do
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when Bool
dbgforall a b. (a -> b) -> a -> b
$ forall r. PrintfType r => String -> r
printf String
"Waking %d idle thread(s).\n" (forall (t :: * -> *) a. Foldable t => t a -> Int
length [MVar Bool]
idles)
IO ()
r <- forall a b. HotVar a -> (a -> (a, b)) -> IO b
modifyHotVar HotVar [MVar Bool]
idle (\[MVar Bool]
is -> case [MVar Bool]
is of
[] -> ([], forall (m :: * -> *) a. Monad m => a -> m a
return ())
(MVar Bool
i:[MVar Bool]
ils) -> ([MVar Bool]
ils, forall a. MVar a -> a -> IO ()
putMVar MVar Bool
i Bool
False))
IO ()
r
rand :: HotVar Random.GenIO -> IO Int
rand :: HotVar GenIO -> IO Int
rand HotVar GenIO
ref = forall a (m :: * -> *).
(Variate a, PrimMonad m) =>
(a, a) -> Gen (PrimState m) -> m a
Random.uniformR (Int
0, Int
numCapabilitiesforall a. Num a => a -> a -> a
-Int
1) forall (m :: * -> *) a b. Monad m => (a -> m b) -> m a -> m b
=<< forall a. HotVar a -> IO a
readHotVar HotVar GenIO
ref
instance NFData (IVar a) where
rnf :: IVar a -> ()
rnf !IVar a
_ = ()
{-# NOINLINE runPar #-}
runPar :: forall a. Par a -> a
runPar = forall a. IO a -> a
unsafePerformIO forall b c a. (b -> c) -> (a -> b) -> a -> c
. forall a. Par a -> IO a
runParIO
runNewSessionAndWait :: String -> Sched -> Par b -> IO b
runNewSessionAndWait :: forall b. String -> Sched -> Par b -> IO b
runNewSessionAndWait String
name Sched
sched Par b
userComp = do
ThreadId
tid <- IO ThreadId
myThreadId
SessionID
sid <- forall a b. HotVar a -> (a -> (a, b)) -> IO b
modifyHotVar (Sched -> HotVar SessionID
sessionCounter Sched
sched) (\ SessionID
x -> (SessionID
xforall a. Num a => a -> a -> a
+SessionID
1,SessionID
x))
()
_ <- forall a b. HotVar a -> (a -> (a, b)) -> IO b
modifyHotVar (Sched -> HotVar (Set SessionID)
activeSessions Sched
sched) (\ Set SessionID
set -> (forall a. Ord a => a -> Set a -> Set a
S.insert SessionID
sid Set SessionID
set, ()))
IORef b
ref <- forall a. a -> IO (IORef a)
newIORef (forall a. HasCallStack => String -> a
errorforall a b. (a -> b) -> a -> b
$ String
"Empty session-result ref ("forall a. [a] -> [a] -> [a]
++String
nameforall a. [a] -> [a] -> [a]
++String
") should never be touched (sid "forall a. [a] -> [a] -> [a]
++ forall a. Show a => a -> String
show SessionID
sidforall a. [a] -> [a] -> [a]
++String
", "forall a. [a] -> [a] -> [a]
++forall a. Show a => a -> String
show ThreadId
tid forall a. [a] -> [a] -> [a]
++String
")")
HotVar Bool
newFlag <- forall a. a -> IO (IORef a)
newHotVar Bool
False
()
_ <- forall a b. HotVar a -> (a -> (a, b)) -> IO b
modifyHotVar (Sched -> HotVar [Session]
sessions Sched
sched) (\ [Session]
ls -> ((SessionID -> HotVar Bool -> Session
Session SessionID
sid HotVar Bool
newFlag) forall a. a -> [a] -> [a]
: [Session]
ls, ()))
let userComp' :: Par ()
userComp' = do forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when Bool
dbgforall a b. (a -> b) -> a -> b
$ forall a. IO a -> Par a
ioforall a b. (a -> b) -> a -> b
$ do
ThreadId
tid2 <- IO ThreadId
myThreadId
forall r. PrintfType r => String -> r
printf String
" [%d %s] Starting Par computation on %s.\n" (Sched -> Int
no Sched
sched) (forall a. Show a => a -> String
show ThreadId
tid2) String
name
b
ans <- Par b
userComp
forall a. IO a -> Par a
ioforall a b. (a -> b) -> a -> b
$ do forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when (Int
dbglvlforall a. Ord a => a -> a -> Bool
>=Int
1) forall a b. (a -> b) -> a -> b
$ do
ThreadId
tid3 <- IO ThreadId
myThreadId
forall r. PrintfType r => String -> r
printf String
" [%d %s] Continuation for %s called, finishing it up (%d)...\n" (Sched -> Int
no Sched
sched) (forall a. Show a => a -> String
show ThreadId
tid3) String
name SessionID
sid
forall a. IORef a -> a -> IO ()
writeIORef IORef b
ref b
ans
forall a. IORef a -> a -> IO ()
writeHotVarRaw HotVar Bool
newFlag Bool
True
forall a b. HotVar a -> (a -> (a, b)) -> IO b
modifyHotVar (Sched -> HotVar (Set SessionID)
activeSessions Sched
sched) (\ Set SessionID
set -> (forall a. Ord a => a -> Set a -> Set a
S.delete SessionID
sid Set SessionID
set, ()))
kont :: Word64 -> a -> ROnly ()
kont :: forall a. SessionID -> a -> ROnly ()
kont SessionID
n = forall a. String -> a -> ROnly ()
trivialContforall a b. (a -> b) -> a -> b
$ String
"("forall a. [a] -> [a] -> [a]
++String
nameforall a. [a] -> [a] -> [a]
++String
", sid "forall a. [a] -> [a] -> [a]
++forall a. Show a => a -> String
show SessionID
sidforall a. [a] -> [a] -> [a]
++String
", round "forall a. [a] -> [a] -> [a]
++forall a. Show a => a -> String
show SessionID
nforall a. [a] -> [a] -> [a]
++String
")"
loop :: Word64 -> ROnly ()
loop :: SessionID -> ROnly ()
loop SessionID
n = do Bool
flg <- forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIOforall a b. (a -> b) -> a -> b
$ forall a. HotVar a -> IO a
readIORef HotVar Bool
newFlag
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
unless Bool
flg forall a b. (a -> b) -> a -> b
$ do
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when Bool
dbg forall a b. (a -> b) -> a -> b
$ forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIOforall a b. (a -> b) -> a -> b
$ do
ThreadId
tid4 <- IO ThreadId
myThreadId
forall r. PrintfType r => String -> r
printf String
" [%d %s] BOUNCE %d... going into reschedule until finished.\n" (Sched -> Int
no Sched
sched) (forall a. Show a => a -> String
show ThreadId
tid4) SessionID
n
forall a. SessionID -> (a -> ROnly ()) -> ROnly ()
rescheduleR SessionID
0 forall a b. (a -> b) -> a -> b
$ forall a. String -> a -> ROnly ()
trivialContforall a b. (a -> b) -> a -> b
$ String
"("forall a. [a] -> [a] -> [a]
++String
nameforall a. [a] -> [a] -> [a]
++String
", sid "forall a. [a] -> [a] -> [a]
++forall a. Show a => a -> String
show SessionID
sidforall a. [a] -> [a] -> [a]
++String
")"
SessionID -> ROnly ()
loop (SessionID
nforall a. Num a => a -> a -> a
+SessionID
1)
forall r (m :: * -> *) a. r -> ReaderT r m a -> m a
runReaderWith Sched
sched (forall {k} (r :: k) (m :: k -> *) a.
ContT r m a -> (a -> m r) -> m r
C.runContT (forall a. Par a -> ContT () (ReaderT Sched IO) a
unPar Par ()
userComp') (forall a. SessionID -> a -> ROnly ()
kont SessionID
0))
forall r (m :: * -> *) a. r -> ReaderT r m a -> m a
runReaderWith Sched
sched (SessionID -> ROnly ()
loop SessionID
1)
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when (Int
dbglvlforall a. Ord a => a -> a -> Bool
>=Int
1)forall a b. (a -> b) -> a -> b
$ do
Set SessionID
active <- forall a. HotVar a -> IO a
readHotVar (Sched -> HotVar (Set SessionID)
activeSessions Sched
sched)
sess :: Bool
sess@Bool
True <- forall a. HotVar a -> IO a
readHotVar HotVar Bool
newFlag
forall r. PrintfType r => String -> r
printf String
" [%d %s] RETURN from %s (sessFin %s) runContT (%d) active set %s\n"
(Sched -> Int
no Sched
sched) (forall a. Show a => a -> String
show ThreadId
tid) String
name (forall a. Show a => a -> String
show Bool
sess) SessionID
sid (forall a. Show a => a -> String
show Set SessionID
active)
forall a. HotVar a -> (a -> a) -> IO ()
modifyHotVar_ (Sched -> HotVar [Session]
sessions Sched
sched) forall a b. (a -> b) -> a -> b
$ \ (Session SessionID
sid2 HotVar Bool
_ : [Session]
tl) ->
if SessionID
sid forall a. Eq a => a -> a -> Bool
== SessionID
sid2
then [Session]
tl
else forall a. HasCallStack => String -> a
errorforall a b. (a -> b) -> a -> b
$ String
"Tried to pop the session stack and found we ("forall a. [a] -> [a] -> [a]
++forall a. Show a => a -> String
show SessionID
sid
forall a. [a] -> [a] -> [a]
++String
") were not on the top! (instead "forall a. [a] -> [a] -> [a]
++forall a. Show a => a -> String
show SessionID
sid2forall a. [a] -> [a] -> [a]
++String
")"
forall a. HotVar a -> IO a
readIORef IORef b
ref
{-# NOINLINE runParIO #-}
runParIO :: forall a. Par a -> IO a
runParIO Par a
userComp = do
ThreadId
tid <- IO ThreadId
myThreadId
#if __GLASGOW_HASKELL__ >= 701 /* 20110301 */
(Int
main_cpu, Bool
_) <- ThreadId -> IO (Int, Bool)
threadCapability ThreadId
tid
#else
let main_cpu = 0
#endif
Maybe Sched
maybSched <- ThreadId -> IO (Maybe Sched)
amINested ThreadId
tid
ThreadId
tidorig <- IO ThreadId
myThreadId
case Maybe Sched
maybSched of
Just (Sched
sched) -> do
SessionID
sid0 <- forall a. HotVar a -> IO a
readHotVar (Sched -> HotVar SessionID
sessionCounter Sched
sched)
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when (Int
dbglvlforall a. Ord a => a -> a -> Bool
>=Int
1)forall a b. (a -> b) -> a -> b
$ forall r. PrintfType r => String -> r
printf String
" [%d %s] runPar called from existing worker thread, new session (%d)....\n" (Sched -> Int
no Sched
sched) (forall a. Show a => a -> String
show ThreadId
tid) (SessionID
sid0 forall a. Num a => a -> a -> a
+ SessionID
1)
forall b. String -> Sched -> Par b -> IO b
runNewSessionAndWait String
"nested runPar" Sched
sched Par a
userComp
Maybe Sched
Nothing -> do
[Sched]
allscheds <- Int -> IO [Sched]
makeScheds Int
main_cpu
[Session SessionID
_ HotVar Bool
topSessFlag] <- forall a. HotVar a -> IO a
readHotVarforall a b. (a -> b) -> a -> b
$ Sched -> HotVar [Session]
sessionsforall a b. (a -> b) -> a -> b
$ forall a. [a] -> a
head [Sched]
allscheds
MVar a
mfin <- forall a. IO (MVar a)
newEmptyMVar
[Maybe (MVar Int)]
doneFlags <- forall (t :: * -> *) (m :: * -> *) a b.
(Traversable t, Monad m) =>
t a -> (a -> m b) -> m (t b)
forM (forall a b. [a] -> [b] -> [(a, b)]
zip [Int
0..] [Sched]
allscheds) forall a b. (a -> b) -> a -> b
$ \(Int
cpu,Sched
sched) -> do
MVar Int
workerDone <- forall a. IO (MVar a)
newEmptyMVar
let wname :: String
wname = (String
"(worker "forall a. [a] -> [a] -> [a]
++forall a. Show a => a -> String
show Int
cpuforall a. [a] -> [a] -> [a]
++String
" of originator "forall a. [a] -> [a] -> [a]
++forall a. Show a => a -> String
show ThreadId
tidorigforall a. [a] -> [a] -> [a]
++String
")")
ThreadId
_ <- (IO () -> IO ThreadId) -> String -> IO () -> IO ThreadId
forkWithExceptions (Int -> IO () -> IO ThreadId
forkOn Int
cpu) String
wname forall a b. (a -> b) -> a -> b
$ do
ThreadId
tid2 <- IO ThreadId
myThreadId
ThreadId -> Sched -> IO ()
registerWorker ThreadId
tid2 Sched
sched
if (Int
cpu forall a. Eq a => a -> a -> Bool
/= Int
main_cpu)
then do forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when Bool
dbgforall a b. (a -> b) -> a -> b
$ forall r. PrintfType r => String -> r
printf String
" [%d %s] Anonymous worker entering scheduling loop.\n" Int
cpu (forall a. Show a => a -> String
show ThreadId
tid2)
forall r (m :: * -> *) a. r -> ReaderT r m a -> m a
runReaderWith Sched
sched forall a b. (a -> b) -> a -> b
$ forall a. SessionID -> (a -> ROnly ()) -> ROnly ()
rescheduleR SessionID
0 (forall a. String -> a -> ROnly ()
trivialCont (String
wnameforall a. [a] -> [a] -> [a]
++forall a. Show a => a -> String
show ThreadId
tid2))
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when Bool
dbgforall a b. (a -> b) -> a -> b
$ forall r. PrintfType r => String -> r
printf String
" [%d] Anonymous worker exited scheduling loop. FINISHED.\n" Int
cpu
forall a. MVar a -> a -> IO ()
putMVar MVar Int
workerDone Int
cpu
forall (m :: * -> *) a. Monad m => a -> m a
return ()
else do a
x <- forall b. String -> Sched -> Par b -> IO b
runNewSessionAndWait String
"top-lvl main worker" Sched
sched Par a
userComp
forall a. IORef a -> a -> IO ()
writeIORef HotVar Bool
topSessFlag Bool
True
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when Bool
dbgforall a b. (a -> b) -> a -> b
$ do forall r. PrintfType r => String -> r
printf String
" *** Out of entire runContT user computation on main thread %s.\n" (forall a. Show a => a -> String
show ThreadId
tid2)
forall a. MVar a -> a -> IO ()
putMVar MVar a
mfin a
x
ThreadId -> IO ()
unregisterWorker ThreadId
tid
forall (m :: * -> *) a. Monad m => a -> m a
return (if Int
cpu forall a. Eq a => a -> a -> Bool
== Int
main_cpu then forall a. Maybe a
Nothing else forall a. a -> Maybe a
Just MVar Int
workerDone)
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when Bool
_WAIT_FOR_WORKERS forall a b. (a -> b) -> a -> b
$ do
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when Bool
dbgforall a b. (a -> b) -> a -> b
$ forall r. PrintfType r => String -> r
printf String
" *** [%s] Originator thread: waiting for workers to complete." (forall a. Show a => a -> String
show ThreadId
tidorig)
forall (t :: * -> *) (m :: * -> *) a b.
(Foldable t, Monad m) =>
t a -> (a -> m b) -> m ()
forM_ (forall a. [Maybe a] -> [a]
catMaybes [Maybe (MVar Int)]
doneFlags) forall a b. (a -> b) -> a -> b
$ \ MVar Int
mv -> do
Int
n <- forall a. MVar a -> IO a
readMVar MVar Int
mv
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when Bool
dbgforall a b. (a -> b) -> a -> b
$ forall r. PrintfType r => String -> r
printf String
" * [%s] Worker %s completed\n" (forall a. Show a => a -> String
show ThreadId
tidorig) (forall a. Show a => a -> String
show Int
n)
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when Bool
dbgforall a b. (a -> b) -> a -> b
$ do forall r. PrintfType r => String -> r
printf String
" *** [%s] Reading final MVar on originator thread.\n" (forall a. Show a => a -> String
show ThreadId
tidorig)
#ifdef DEBUG_DIRECT
busyTakeMVar (" The global wait "++ show tidorig) mfin
#else
forall a. MVar a -> IO a
takeMVar MVar a
mfin
#endif
makeScheds :: Int -> IO [Sched]
makeScheds :: Int -> IO [Sched]
makeScheds Int
main = do
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when Bool
dbgforall a b. (a -> b) -> a -> b
$ do ThreadId
tid <- IO ThreadId
myThreadId
forall r. PrintfType r => String -> r
printf String
"[initialization] Creating %d worker threads, currently on %s\n" Int
numCapabilities (forall a. Show a => a -> String
show ThreadId
tid)
[SimpleDeque (Par ())]
workpools <- forall (m :: * -> *) a. Applicative m => Int -> m a -> m [a]
replicateM Int
numCapabilities forall a b. (a -> b) -> a -> b
$ forall elt. IO (SimpleDeque elt)
R.newQ
[HotVar (Gen RealWorld)]
rngs <- forall (m :: * -> *) a. Applicative m => Int -> m a -> m [a]
replicateM Int
numCapabilities forall a b. (a -> b) -> a -> b
$ forall (m :: * -> *). PrimMonad m => m (Gen (PrimState m))
Random.create forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= forall a. a -> IO (IORef a)
newHotVar
HotVar [MVar Bool]
idle <- forall a. a -> IO (IORef a)
newHotVar []
HotVar Bool
sessionFinished <- forall a. a -> IO (IORef a)
newHotVar Bool
False
[HotVar [Session]]
sessionStacks <- forall (t :: * -> *) (m :: * -> *) a b.
(Traversable t, Monad m) =>
(a -> m b) -> t a -> m (t b)
mapM forall a. a -> IO (IORef a)
newHotVar (forall a. Int -> a -> [a]
replicate Int
numCapabilities [SessionID -> HotVar Bool -> Session
Session SessionID
baseSessionID HotVar Bool
sessionFinished])
HotVar (Set SessionID)
activeSessions <- forall a. a -> IO (IORef a)
newHotVar forall a. Set a
S.empty
HotVar SessionID
sessionCounter <- forall a. a -> IO (IORef a)
newHotVar (SessionID
baseSessionID forall a. Num a => a -> a -> a
+ SessionID
1)
let allscheds :: [Sched]
allscheds = [ Sched { no :: Int
no=Int
x, HotVar [MVar Bool]
idle :: HotVar [MVar Bool]
idle :: HotVar [MVar Bool]
idle, isMain :: Bool
isMain= (Int
xforall a. Eq a => a -> a -> Bool
==Int
main),
workpool :: WSDeque (Par ())
workpool=SimpleDeque (Par ())
wp, scheds :: [Sched]
scheds=[Sched]
allscheds, rng :: HotVar GenIO
rng=HotVar (Gen RealWorld)
rng,
sessions :: HotVar [Session]
sessions = HotVar [Session]
stck,
activeSessions :: HotVar (Set SessionID)
activeSessions=HotVar (Set SessionID)
activeSessions,
sessionCounter :: HotVar SessionID
sessionCounter=HotVar SessionID
sessionCounter
}
| Int
x <- [Int
0 .. Int
numCapabilitiesforall a. Num a => a -> a -> a
-Int
1]
| SimpleDeque (Par ())
wp <- [SimpleDeque (Par ())]
workpools
| HotVar (Gen RealWorld)
rng <- [HotVar (Gen RealWorld)]
rngs
| HotVar [Session]
stck <- [HotVar [Session]]
sessionStacks
]
forall (m :: * -> *) a. Monad m => a -> m a
return [Sched]
allscheds
baseSessionID :: SessionID
baseSessionID :: SessionID
baseSessionID = SessionID
1000
{-# INLINE new #-}
new :: Par (IVar a)
new :: forall a. Par (IVar a)
new = forall a. IO a -> Par a
ioforall a b. (a -> b) -> a -> b
$ do IORef (IVarContents a)
r <- forall a. a -> IO (IORef a)
newIORef forall a. IVarContents a
Empty
forall (m :: * -> *) a. Monad m => a -> m a
return (forall a. IORef (IVarContents a) -> IVar a
IVar IORef (IVarContents a)
r)
{-# INLINE get #-}
get :: forall a. IVar a -> Par a
get (IVar IORef (IVarContents a)
vr) = do
forall (m :: * -> *) a b. MonadCont m => ((a -> m b) -> m a) -> m a
callCC forall a b. (a -> b) -> a -> b
$ \a -> Par ()
kont ->
do
IVarContents a
e <- forall a. IO a -> Par a
ioforall a b. (a -> b) -> a -> b
$ forall a. HotVar a -> IO a
readIORef IORef (IVarContents a)
vr
case IVarContents a
e of
Full a
a -> forall (m :: * -> *) a. Monad m => a -> m a
return a
a
IVarContents a
_ -> do
Sched
sch <- forall r (m :: * -> *). MonadReader r m => m r
RD.ask
# ifdef DEBUG_DIRECT
sn <- io$ makeStableName vr
let resched = trace (" ["++ show (no sch) ++ "] - Rescheduling on unavailable ivar "++show (hashStableName sn)++"!")
#else
let resched :: Par a
resched =
# endif
forall a. Par a
longjmpSched
Par a
r <- forall a. IO a -> Par a
ioforall a b. (a -> b) -> a -> b
$ forall a b. HotVar a -> (a -> (a, b)) -> IO b
atomicModifyIORef IORef (IVarContents a)
vr forall a b. (a -> b) -> a -> b
$ \IVarContents a
x -> case IVarContents a
x of
IVarContents a
Empty -> (forall a. [a -> IO ()] -> IVarContents a
Blocked [Sched -> Par () -> IO ()
pushWork Sched
sch forall b c a. (b -> c) -> (a -> b) -> a -> c
. a -> Par ()
kont], forall a. Par a
resched)
Full a
a -> (forall a. a -> IVarContents a
Full a
a, forall (m :: * -> *) a. Monad m => a -> m a
return a
a)
Blocked [a -> IO ()]
ks -> (forall a. [a -> IO ()] -> IVarContents a
Blocked (Sched -> Par () -> IO ()
pushWork Sched
sch forall b c a. (b -> c) -> (a -> b) -> a -> c
. a -> Par ()
kontforall a. a -> [a] -> [a]
:[a -> IO ()]
ks), forall a. Par a
resched)
Par a
r
{-# INLINE unsafePeek #-}
unsafePeek :: IVar a -> Par (Maybe a)
unsafePeek :: forall a. IVar a -> Par (Maybe a)
unsafePeek (IVar IORef (IVarContents a)
v) = do
IVarContents a
e <- forall a. IO a -> Par a
ioforall a b. (a -> b) -> a -> b
$ forall a. HotVar a -> IO a
readIORef IORef (IVarContents a)
v
case IVarContents a
e of
Full a
a -> forall (m :: * -> *) a. Monad m => a -> m a
return (forall a. a -> Maybe a
Just a
a)
IVarContents a
_ -> forall (m :: * -> *) a. Monad m => a -> m a
return forall a. Maybe a
Nothing
{-# INLINE put_ #-}
put_ :: forall a. IVar a -> a -> Par ()
put_ (IVar IORef (IVarContents a)
vr) !a
content = do
Sched
sched <- forall r (m :: * -> *). MonadReader r m => m r
RD.ask
[a -> IO ()]
ks <- forall a. IO a -> Par a
ioforall a b. (a -> b) -> a -> b
$ do
[a -> IO ()]
ks <- forall a b. HotVar a -> (a -> (a, b)) -> IO b
atomicModifyIORef IORef (IVarContents a)
vr forall a b. (a -> b) -> a -> b
$ \IVarContents a
e -> case IVarContents a
e of
IVarContents a
Empty -> (forall a. a -> IVarContents a
Full a
content, [])
Full a
_ -> forall a. HasCallStack => String -> a
error String
"multiple put"
Blocked [a -> IO ()]
ks -> (forall a. a -> IVarContents a
Full a
content, [a -> IO ()]
ks)
#ifdef DEBUG_DIRECT
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
forall (m :: * -> *) a. Monad m => a -> m a
return [a -> IO ()]
ks
forall a. Sched -> [a -> IO ()] -> a -> Par ()
wakeUp Sched
sched [a -> IO ()]
ks a
content
{-# INLINE unsafeTryPut #-}
unsafeTryPut :: forall b. IVar b -> b -> Par b
unsafeTryPut (IVar IORef (IVarContents b)
vr) !b
content = do
Sched
sched <- forall r (m :: * -> *). MonadReader r m => m r
RD.ask
([b -> IO ()]
ks,b
res) <- forall a. IO a -> Par a
ioforall a b. (a -> b) -> a -> b
$ do
([b -> IO ()], b)
pr <- forall a b. HotVar a -> (a -> (a, b)) -> IO b
atomicModifyIORef IORef (IVarContents b)
vr forall a b. (a -> b) -> a -> b
$ \IVarContents b
e -> case IVarContents b
e of
IVarContents b
Empty -> (forall a. a -> IVarContents a
Full b
content, ([], b
content))
Full b
x -> (forall a. a -> IVarContents a
Full b
x, ([], b
x))
Blocked [b -> IO ()]
ks -> (forall a. a -> IVarContents a
Full b
content, ([b -> IO ()]
ks, b
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))
#endif
forall (m :: * -> *) a. Monad m => a -> m a
return ([b -> IO ()], b)
pr
forall a. Sched -> [a -> IO ()] -> a -> Par ()
wakeUp Sched
sched [b -> IO ()]
ks b
content
forall (m :: * -> *) a. Monad m => a -> m a
return b
res
{-# INLINE wakeUp #-}
wakeUp :: Sched -> [a -> IO ()]-> a -> Par ()
wakeUp :: forall a. Sched -> [a -> IO ()] -> a -> Par ()
wakeUp Sched
_sched [a -> IO ()]
ks a
arg = [a -> IO ()] -> Par ()
loop [a -> IO ()]
ks
where
loop :: [a -> IO ()] -> Par ()
loop [] = forall (m :: * -> *) a. Monad m => a -> m a
return ()
loop (a -> IO ()
kont:[a -> IO ()]
rest) = do
if Bool
_PARPUTS then
do IVar ()
_ <- forall a. Par a -> Par (IVar a)
spawn_forall a b. (a -> b) -> a -> b
$ (a -> IO ()) -> [a -> IO ()] -> Par ()
pMap a -> IO ()
kont [a -> IO ()]
rest
forall (m :: * -> *) a. Monad m => a -> m a
return ()
else
do forall a. IO a -> Par a
ioforall a b. (a -> b) -> a -> b
$ a -> IO ()
kont a
arg
[a -> IO ()] -> Par ()
loop [a -> IO ()]
rest
forall (m :: * -> *) a. Monad m => a -> m a
return ()
pMap :: (a -> IO ()) -> [a -> IO ()] -> Par ()
pMap a -> IO ()
kont [] = forall a. IO a -> Par a
ioforall a b. (a -> b) -> a -> b
$ a -> IO ()
kont a
arg
pMap a -> IO ()
kont (a -> IO ()
more:[a -> IO ()]
rest) =
do IVar ()
_ <- forall a. Par a -> Par (IVar a)
spawn_forall a b. (a -> b) -> a -> b
$ forall a. IO a -> Par a
ioforall a b. (a -> b) -> a -> b
$ a -> IO ()
kont a
arg
(a -> IO ()) -> [a -> IO ()] -> Par ()
pMap a -> IO ()
more [a -> IO ()]
rest
{-# INLINE fork #-}
fork :: Par () -> Par ()
fork :: Par () -> Par ()
fork Par ()
task =
case Bool
_FORKPARENT of
Bool
True -> do
Sched
sched <- forall r (m :: * -> *). MonadReader r m => m r
RD.ask
forall (m :: * -> *) a b. MonadCont m => ((a -> m b) -> m a) -> m a
callCCforall a b. (a -> b) -> a -> b
$ \() -> Par ()
parent -> do
let wrapped :: Par ()
wrapped = () -> Par ()
parent ()
forall a. IO a -> Par a
ioforall a b. (a -> b) -> a -> b
$ Sched -> Par () -> IO ()
pushWork Sched
sched Par ()
wrapped
Par ()
task
Any
_ <- forall a. Par a
longjmpSched
forall a. IO a -> Par a
ioforall a b. (a -> b) -> a -> b
$ forall r. PrintfType r => String -> r
printf String
" !!! ERROR: Should never reach this point #1\n"
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when Bool
dbgforall a b. (a -> b) -> a -> b
$ do
Sched
sched2 <- forall r (m :: * -> *). MonadReader r m => m r
RD.ask
forall a. IO a -> Par a
ioforall a b. (a -> b) -> a -> b
$ forall r. PrintfType r => String -> r
printf String
" - called parent continuation... was on worker [%d] now on worker [%d]\n" (Sched -> Int
no Sched
sched) (Sched -> Int
no Sched
sched2)
forall (m :: * -> *) a. Monad m => a -> m a
return ()
Bool
False -> do
Sched
sch <- forall r (m :: * -> *). MonadReader r m => m r
RD.ask
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when Bool
dbgforall a b. (a -> b) -> a -> b
$ forall a. IO a -> Par a
ioforall a b. (a -> b) -> a -> b
$ forall r. PrintfType r => String -> r
printf String
" [%d] forking task...\n" (Sched -> Int
no Sched
sch)
forall a. IO a -> Par a
ioforall a b. (a -> b) -> a -> b
$ Sched -> Par () -> IO ()
pushWork Sched
sch Par ()
task
longjmpSched :: Par a
longjmpSched :: forall a. Par a
longjmpSched = forall a. ContT () (ReaderT Sched IO) a -> Par a
Par forall a b. (a -> b) -> a -> b
$ forall {k} (r :: k) (m :: k -> *) a.
((a -> m r) -> m r) -> ContT r m a
C.ContT (\ a -> ROnly ()
_k -> forall a. SessionID -> (a -> ROnly ()) -> ROnly ()
rescheduleR SessionID
0 (forall a. String -> a -> ROnly ()
trivialCont String
"longjmpSched"))
rescheduleR :: Word64 -> (a -> ROnly ()) -> ROnly ()
rescheduleR :: forall a. SessionID -> (a -> ROnly ()) -> ROnly ()
rescheduleR SessionID
cnt a -> ROnly ()
kont = do
Sched
mysched <- forall r (m :: * -> *). MonadReader r m => m r
RD.ask
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when Bool
dbgforall a b. (a -> b) -> a -> b
$ forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIOforall a b. (a -> b) -> a -> b
$ do ThreadId
tid <- IO ThreadId
myThreadId
[(SessionID, Bool)]
sess <- Sched -> IO [(SessionID, Bool)]
readSessions Sched
mysched
Bool
null <- forall elt. SimpleDeque elt -> IO Bool
R.nullQ (Sched -> WSDeque (Par ())
workpool Sched
mysched)
forall r. PrintfType r => String -> r
printf String
" [%d %s] - Reschedule #%d... sessions %s, pool empty %s\n"
(Sched -> Int
no Sched
mysched) (forall a. Show a => a -> String
show ThreadId
tid) SessionID
cnt (forall a. Show a => a -> String
show [(SessionID, Bool)]
sess) (forall a. Show a => a -> String
show Bool
null)
Maybe (Par ())
mtask <- forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIOforall a b. (a -> b) -> a -> b
$ Sched -> IO (Maybe (Par ()))
popWork Sched
mysched
case Maybe (Par ())
mtask of
Maybe (Par ())
Nothing -> do
(Session SessionID
_ HotVar Bool
finRef):[Session]
_ <- forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIOforall a b. (a -> b) -> a -> b
$ forall a. HotVar a -> IO a
readIORef forall a b. (a -> b) -> a -> b
$ Sched -> HotVar [Session]
sessions Sched
mysched
Bool
fin <- forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIOforall a b. (a -> b) -> a -> b
$ forall a. HotVar a -> IO a
readIORef HotVar Bool
finRef
if Bool
fin
then do forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when (Int
dbglvl forall a. Ord a => a -> a -> Bool
>= Int
1) forall a b. (a -> b) -> a -> b
$ forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO forall a b. (a -> b) -> a -> b
$ do
ThreadId
tid <- IO ThreadId
myThreadId
[(SessionID, Bool)]
sess <- Sched -> IO [(SessionID, Bool)]
readSessions Sched
mysched
forall r. PrintfType r => String -> r
printf String
" [%d %s] - DROP out of reschedule loop, sessionFinished=%s, all sessions %s\n"
(Sched -> Int
no Sched
mysched) (forall a. Show a => a -> String
show ThreadId
tid) (forall a. Show a => a -> String
show Bool
fin) (forall a. Show a => a -> String
show [(SessionID, Bool)]
sess)
Bool
empt <- forall elt. SimpleDeque elt -> IO Bool
R.nullQforall a b. (a -> b) -> a -> b
$ Sched -> WSDeque (Par ())
workpool Sched
mysched
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when (Bool -> Bool
not Bool
empt) forall a b. (a -> b) -> a -> b
$ do
forall r. PrintfType r => String -> r
printf String
" [%d %s] - WARNING - leaving rescheduleR while local workpoll is nonempty\n"
(Sched -> Int
no Sched
mysched) (forall a. Show a => a -> String
show ThreadId
tid)
a -> ROnly ()
kont (forall a. HasCallStack => String -> a
error String
"Direct.hs: The result value from rescheduleR should not be used.")
else do
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIOforall a b. (a -> b) -> a -> b
$ Sched -> IO ()
steal Sched
mysched
#ifdef WAKEIDLE
#endif
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO IO ()
yield
forall a. SessionID -> (a -> ROnly ()) -> ROnly ()
rescheduleR (SessionID
cntforall a. Num a => a -> a -> a
+SessionID
1) a -> ROnly ()
kont
Just Par ()
task -> do
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when Bool
dbg forall a b. (a -> b) -> a -> b
$ do StableName (Par ())
sn <- forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIOforall a b. (a -> b) -> a -> b
$ forall a. a -> IO (StableName a)
makeStableName Par ()
task
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIOforall a b. (a -> b) -> a -> b
$ forall r. PrintfType r => String -> r
printf String
" [%d] popped work %d from own queue\n" (Sched -> Int
no Sched
mysched) (forall a. StableName a -> Int
hashStableName StableName (Par ())
sn)
let C.ContT (() -> ROnly ()) -> ROnly ()
fn = forall a. Par a -> ContT () (ReaderT Sched IO) a
unPar Par ()
task
(() -> ROnly ()) -> ROnly ()
fn (\ ()
_ -> do
Sched
sch <- forall r (m :: * -> *). MonadReader r m => m r
RD.ask
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when Bool
dbgforall a b. (a -> b) -> a -> b
$ forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIOforall a b. (a -> b) -> a -> b
$ forall r. PrintfType r => String -> r
printf String
" + task finished successfully on cpu %d, calling reschedule continuation..\n" (Sched -> Int
no Sched
sch)
forall a. SessionID -> (a -> ROnly ()) -> ROnly ()
rescheduleR SessionID
0 a -> ROnly ()
kont)
steal :: Sched -> IO ()
steal :: Sched -> IO ()
steal mysched :: Sched
mysched@Sched{ HotVar [MVar Bool]
idle :: HotVar [MVar Bool]
idle :: Sched -> HotVar [MVar Bool]
idle, [Sched]
scheds :: [Sched]
scheds :: Sched -> [Sched]
scheds, HotVar GenIO
rng :: HotVar GenIO
rng :: Sched -> HotVar GenIO
rng, no :: Sched -> Int
no=Int
my_no } = do
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when (Int
dbglvlforall a. Ord a => a -> a -> Bool
>=Int
2)forall a b. (a -> b) -> a -> b
$ do ThreadId
tid <- IO ThreadId
myThreadId
forall r. PrintfType r => String -> r
printf String
" [%d %s] + stealing\n" Int
my_no (forall a. Show a => a -> String
show ThreadId
tid)
Int
i <- Int -> IO Int
getnext (-Int
1 :: Int)
Int -> Int -> IO ()
go Int
maxtries Int
i
where
maxtries :: Int
maxtries = Int
20 forall a. Num a => a -> a -> a
* Int
numCapabilities
getnext :: Int -> IO Int
getnext Int
_ = HotVar GenIO -> IO Int
rand HotVar GenIO
rng
go :: Int -> Int -> IO ()
go Int
0 Int
_ | Bool
_IDLING_ON =
do MVar Bool
m <- forall a. IO (MVar a)
newEmptyMVar
[MVar Bool]
r <- forall a b. HotVar a -> (a -> (a, b)) -> IO b
modifyHotVar HotVar [MVar Bool]
idle forall a b. (a -> b) -> a -> b
$ \[MVar Bool]
is -> (MVar Bool
mforall a. a -> [a] -> [a]
:[MVar Bool]
is, [MVar Bool]
is)
if forall (t :: * -> *) a. Foldable t => t a -> Int
length [MVar Bool]
r forall a. Eq a => a -> a -> Bool
== Int
numCapabilities forall a. Num a => a -> a -> a
- Int
1
then do
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when Bool
dbgforall a b. (a -> b) -> a -> b
$ forall r. PrintfType r => String -> r
printf String
" [%d] | waking up all threads\n" Int
my_no
forall a. IORef a -> a -> IO ()
writeHotVarRaw HotVar [MVar Bool]
idle []
forall (t :: * -> *) (m :: * -> *) a b.
(Foldable t, Monad m) =>
(a -> m b) -> t a -> m ()
mapM_ (\MVar Bool
vr -> forall a. MVar a -> a -> IO ()
putMVar MVar Bool
vr Bool
True) [MVar Bool]
r
else do
(Session SessionID
_ HotVar Bool
finRef):[Session]
_ <- forall a. HotVar a -> IO a
readIORef forall a b. (a -> b) -> a -> b
$ Sched -> HotVar [Session]
sessions Sched
mysched
Bool
fin <- forall a. HotVar a -> IO a
readIORef HotVar Bool
finRef
Bool
done <- if Bool
fin then forall (f :: * -> *) a. Applicative f => a -> f a
pure Bool
True else forall a. MVar a -> IO a
takeMVar MVar Bool
m
if Bool
done
then do
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when Bool
dbgforall a b. (a -> b) -> a -> b
$ forall r. PrintfType r => String -> r
printf String
" [%d] | shutting down\n" Int
my_no
forall (m :: * -> *) a. Monad m => a -> m a
return ()
else do
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when Bool
dbgforall a b. (a -> b) -> a -> b
$ forall r. PrintfType r => String -> r
printf String
" [%d] | woken up\n" Int
my_no
Int
i <- Int -> IO Int
getnext (-Int
1::Int)
Int -> Int -> IO ()
go Int
maxtries Int
i
go Int
0 Int
_i | Bool
_IDLING_ON forall a. Eq a => a -> a -> Bool
== Bool
False = IO ()
yield
go Int
tries Int
i
| Int
i forall a. Eq a => a -> a -> Bool
== Int
my_no = do Int
i' <- Int -> IO Int
getnext Int
i
Int -> Int -> IO ()
go (Int
triesforall a. Num a => a -> a -> a
-Int
1) Int
i'
| Bool
otherwise = do
let schd :: Sched
schd = [Sched]
schedsforall a. [a] -> Int -> a
!!Int
i
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when (Int
dbglvlforall a. Ord a => a -> a -> Bool
>=Int
2)forall a b. (a -> b) -> a -> b
$ forall r. PrintfType r => String -> r
printf String
" [%d] | trying steal from %d\n" Int
my_no (Sched -> Int
no Sched
schd)
let dq :: WSDeque (Par ())
dq = Sched -> WSDeque (Par ())
workpool Sched
schd
Maybe (Par ())
r <- forall a. SimpleDeque a -> IO (Maybe a)
R.tryPopR WSDeque (Par ())
dq
case Maybe (Par ())
r of
Just Par ()
task -> do
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when Bool
dbgforall a b. (a -> b) -> a -> b
$ do StableName (Par ())
sn <- forall a. a -> IO (StableName a)
makeStableName Par ()
task
forall r. PrintfType r => String -> r
printf String
" [%d] | stole work (unit %d) from cpu %d\n" Int
my_no (forall a. StableName a -> Int
hashStableName StableName (Par ())
sn) (Sched -> Int
no Sched
schd)
forall r (m :: * -> *) a. r -> ReaderT r m a -> m a
runReaderWith Sched
mysched forall a b. (a -> b) -> a -> b
$
forall {k} (r :: k) (m :: k -> *) a.
ContT r m a -> (a -> m r) -> m r
C.runContT (forall a. Par a -> ContT () (ReaderT Sched IO) a
unPar Par ()
task)
(\()
_ -> do
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when Bool
dbgforall a b. (a -> b) -> a -> b
$ do StableName (Par ())
sn <- forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIOforall a b. (a -> b) -> a -> b
$ forall a. a -> IO (StableName a)
makeStableName Par ()
task
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIOforall a b. (a -> b) -> a -> b
$ forall r. PrintfType r => String -> r
printf String
" [%d] | DONE running stolen work (unit %d) from %d\n" Int
my_no (forall a. StableName a -> Int
hashStableName StableName (Par ())
sn) (Sched -> Int
no Sched
schd)
forall (m :: * -> *) a. Monad m => a -> m a
return ())
Maybe (Par ())
Nothing -> do Int
i' <- Int -> IO Int
getnext Int
i
Int -> Int -> IO ()
go (Int
triesforall a. Num a => a -> a -> a
-Int
1) Int
i'
_errK :: t
_errK :: forall t. t
_errK = forall a. HasCallStack => String -> a
error String
"Error cont: this closure shouldn't be used"
trivialCont :: String -> a -> ROnly ()
#ifdef DEBUG_DIRECT
trivialCont str _ = do
liftIO$ printf " !! trivialCont evaluated, msg: %s\n" str
#else
trivialCont :: forall a. String -> a -> ROnly ()
trivialCont String
_str a
_ = do
#endif
forall (m :: * -> *) a. Monad m => a -> m a
return ()
{-# INLINE spawn1_ #-}
spawn1_ :: forall a b. (a -> Par b) -> a -> Par (IVar b)
spawn1_ a -> Par b
f a
x =
#ifdef DEBUG_DIRECT
do sn <- io$ makeStableName f
sch <- RD.ask; when dbg$ io$ printf " [%d] spawning fn %d with arg %s\n" (no sch) (hashStableName sn) (show x)
#endif
forall a. Par a -> Par (IVar a)
spawn_ (a -> Par b
f a
x)
newFull_ :: forall a. a -> Par (IVar a)
newFull_ a
a = do IVar a
v <- forall a. Par (IVar a)
new
forall a. IVar a -> a -> Par ()
put_ IVar a
v a
a
forall (m :: * -> *) a. Monad m => a -> m a
return IVar a
v
newFull :: forall a. NFData a => a -> Par (IVar a)
newFull a
a = forall a b. NFData a => a -> b -> b
deepseq a
a (forall a. a -> Par (IVar a)
newFull_ a
a)
{-# INLINE put #-}
put :: forall a. NFData a => IVar a -> a -> Par ()
put IVar a
v a
a = forall a b. NFData a => a -> b -> b
deepseq a
a (forall a. IVar a -> a -> Par ()
put_ IVar a
v a
a)
spawn :: forall a. NFData a => Par a -> Par (IVar a)
spawn Par a
p = do IVar a
r <- forall a. Par (IVar a)
new; Par () -> Par ()
fork (Par a
p forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= forall a. NFData a => IVar a -> a -> Par ()
put IVar a
r); forall (m :: * -> *) a. Monad m => a -> m a
return IVar a
r
spawn_ :: forall a. Par a -> Par (IVar a)
spawn_ Par a
p = do IVar a
r <- forall a. Par (IVar a)
new; Par () -> Par ()
fork (Par a
p forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= forall a. IVar a -> a -> Par ()
put_ IVar a
r); forall (m :: * -> *) a. Monad m => a -> m a
return IVar a
r
spawnP :: forall a. NFData a => a -> Par (IVar a)
spawnP a
a = forall a. NFData a => Par a -> Par (IVar a)
spawn (forall (m :: * -> *) a. Monad m => a -> m a
return a
a)
#ifdef DEBUG_DIRECT
put :: (Show a, NFData a) => IVar a -> a -> Par ()
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
instance PC.ParFuture IVar Par where
get :: forall a. IVar a -> Par a
get = forall a. IVar a -> Par a
get
spawn :: forall a. NFData a => Par a -> Par (IVar a)
spawn = forall a. NFData a => Par a -> Par (IVar a)
spawn
spawn_ :: forall a. Par a -> Par (IVar a)
spawn_ = forall a. Par a -> Par (IVar a)
spawn_
spawnP :: forall a. NFData a => a -> Par (IVar a)
spawnP = forall a. NFData a => a -> Par (IVar a)
spawnP
instance PC.ParIVar IVar Par where
fork :: Par () -> Par ()
fork = Par () -> Par ()
fork
new :: forall a. Par (IVar a)
new = forall a. Par (IVar a)
new
put_ :: forall a. IVar a -> a -> Par ()
put_ = forall a. IVar a -> a -> Par ()
put_
newFull :: forall a. NFData a => a -> Par (IVar a)
newFull = forall a. NFData a => a -> Par (IVar a)
newFull
newFull_ :: forall a. a -> Par (IVar a)
newFull_ = forall a. a -> Par (IVar a)
newFull_
instance UN.ParUnsafe IVar Par where
unsafePeek :: forall a. IVar a -> Par (Maybe a)
unsafePeek = forall a. IVar a -> Par (Maybe a)
unsafePeek
unsafeTryPut :: forall b. IVar b -> b -> Par b
unsafeTryPut = forall b. IVar b -> b -> Par b
unsafeTryPut
unsafeParIO :: forall a. IO a -> Par a
unsafeParIO = forall a. IO a -> Par a
unsafeParIO
#endif
#ifdef NEW_GENERIC
instance PU.ParMonad Par where
fork = fork
internalLiftIO io = Par (lift $ lift io)
instance PU.ParThreadSafe Par where
unsafeParIO io = Par (lift $ lift io)
instance PN.ParFuture Par where
type Future Par = IVar
type FutContents Par a = ()
get = get
spawn = spawn
spawn_ = spawn_
spawnP = spawnP
instance PN.ParIVar Par where
new = new
put_ = put_
newFull = newFull
newFull_ = newFull_
#endif
{-# INLINE runReaderWith #-}
runReaderWith :: r -> RD.ReaderT r m a -> m a
runReaderWith :: forall r (m :: * -> *) a. r -> ReaderT r m a -> m a
runReaderWith r
state ReaderT r m a
m = forall r (m :: * -> *) a. ReaderT r m a -> r -> m a
RD.runReaderT ReaderT r m a
m r
state
_sanityCheck :: [Sched] -> IO ()
_sanityCheck :: [Sched] -> IO ()
_sanityCheck [Sched]
allscheds = do
forall (t :: * -> *) (m :: * -> *) a b.
(Foldable t, Monad m) =>
t a -> (a -> m b) -> m ()
forM_ [Sched]
allscheds forall a b. (a -> b) -> a -> b
$ \ Sched{Int
no :: Int
no :: Sched -> Int
no, WSDeque (Par ())
workpool :: WSDeque (Par ())
workpool :: Sched -> WSDeque (Par ())
workpool} -> do
Bool
b <- forall elt. SimpleDeque elt -> IO Bool
R.nullQ WSDeque (Par ())
workpool
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when (Bool -> Bool
not Bool
b) forall a b. (a -> b) -> a -> b
$ do
() <- forall r. PrintfType r => String -> r
printf String
"WARNING: After main thread exited non-empty queue remains for worker %d\n" Int
no
forall (m :: * -> *) a. Monad m => a -> m a
return ()
forall r. PrintfType r => String -> r
printf String
"Sanity check complete.\n"
_dbgTakeMVar :: String -> MVar a -> IO a
_dbgTakeMVar :: forall a. String -> MVar a -> IO a
_dbgTakeMVar String
msg MVar a
mv =
forall e a. Exception e => IO a -> (e -> IO a) -> IO a
E.catch (forall a. MVar a -> IO a
takeMVar MVar a
mv) (\(IOError
_::IOError) -> IO a
doDebugStuff)
where
doDebugStuff :: IO a
doDebugStuff = do forall r. PrintfType r => String -> r
printf String
"This takeMVar blocked indefinitely!: %s\n" String
msg
forall a. HasCallStack => String -> a
error String
"failed"
_forkIO_Suppress :: Int -> IO () -> IO ThreadId
_forkIO_Suppress :: Int -> IO () -> IO ThreadId
_forkIO_Suppress Int
whre IO ()
action =
Int -> IO () -> IO ThreadId
forkOn Int
whre forall a b. (a -> b) -> a -> b
$
forall e a. Exception e => (e -> IO a) -> IO a -> IO a
E.handle (\BlockedIndefinitelyOnMVar
e ->
case (BlockedIndefinitelyOnMVar
e :: E.BlockedIndefinitelyOnMVar) of
BlockedIndefinitelyOnMVar
_ -> do
String -> IO ()
putStrLnforall a b. (a -> b) -> a -> b
$String
"CAUGHT child thread exception: "forall a. [a] -> [a] -> [a]
++forall a. Show a => a -> String
show BlockedIndefinitelyOnMVar
e
forall (m :: * -> *) a. Monad m => a -> m a
return ()
)
IO ()
action
forkWithExceptions :: (IO () -> IO ThreadId) -> String -> IO () -> IO ThreadId
forkWithExceptions :: (IO () -> IO ThreadId) -> String -> IO () -> IO ThreadId
forkWithExceptions IO () -> IO ThreadId
forkit String
descr IO ()
action = do
ThreadId
parent <- IO ThreadId
myThreadId
IO () -> IO ThreadId
forkit forall a b. (a -> b) -> a -> b
$ do
ThreadId
tid <- IO ThreadId
myThreadId
forall e a. Exception e => IO a -> (e -> IO a) -> IO a
E.catch IO ()
action
(\ SomeException
e ->
case forall e. Exception e => SomeException -> Maybe e
E.fromException SomeException
e of
Just AsyncException
E.ThreadKilled -> forall r. PrintfType r => String -> r
printf
String
"\nThreadKilled exception inside child thread, %s (not propagating!): %s\n" (forall a. Show a => a -> String
show ThreadId
tid) (forall a. Show a => a -> String
show String
descr)
Maybe AsyncException
_ -> do forall r. PrintfType r => String -> r
printf
String
"\nException inside child thread %s, %s: %s\n" (forall a. Show a => a -> String
show String
descr) (forall a. Show a => a -> String
show ThreadId
tid) (forall a. Show a => a -> String
show SomeException
e)
forall e. Exception e => ThreadId -> e -> IO ()
E.throwTo ThreadId
parent (SomeException
e :: E.SomeException)
)
readSessions :: Sched -> IO [(SessionID, Bool)]
readSessions :: Sched -> IO [(SessionID, Bool)]
readSessions Sched
sched = do
[Session]
ls <- forall a. HotVar a -> IO a
readIORef (Sched -> HotVar [Session]
sessions Sched
sched)
[Bool]
bools <- forall (t :: * -> *) (m :: * -> *) a b.
(Traversable t, Monad m) =>
(a -> m b) -> t a -> m (t b)
mapM (\ (Session SessionID
_ HotVar Bool
r) -> forall a. HotVar a -> IO a
readIORef HotVar Bool
r) [Session]
ls
forall (m :: * -> *) a. Monad m => a -> m a
return (forall a b. [a] -> [b] -> [(a, b)]
zip (forall a b. (a -> b) -> [a] -> [b]
map (\ (Session SessionID
sid HotVar Bool
_) -> SessionID
sid) [Session]
ls) [Bool]
bools)