{-# 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 "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 = ContT () ROnly a -> Par a
forall a. ContT () ROnly a -> Par a
Par (ReaderT Sched IO a -> ContT () ROnly a
forall (t :: (* -> *) -> * -> *) (m :: * -> *) a.
(MonadTrans t, Monad m) =>
m a -> t m a
lift(ReaderT Sched IO a -> ContT () ROnly a)
-> ReaderT Sched IO a -> ContT () ROnly a
forall a b. (a -> b) -> a -> b
$ IO a -> ReaderT Sched IO a
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 = IO a -> Par a
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
_ = Maybe Sched -> IO (Maybe Sched)
forall (m :: * -> *) a. Monad m => a -> m a
return Maybe Sched
forall a. Maybe a
Nothing
registerWorker :: ThreadId -> Sched -> IO ()
registerWorker ThreadId
_ Sched
_ = () -> IO ()
forall (m :: * -> *) a. Monad m => a -> m a
return ()
unregisterWorker :: ThreadId -> IO ()
unregisterWorker ThreadId
_tid = () -> IO ()
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 <- SimpleDeque (Par ()) -> IO (Maybe (Par ()))
forall a. SimpleDeque a -> IO (Maybe a)
R.tryPopL SimpleDeque (Par ())
WSDeque (Par ())
workpool
Bool -> IO () -> IO ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when Bool
dbg (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ case Maybe (Par ())
mb of
Maybe (Par ())
Nothing -> () -> IO ()
forall (m :: * -> *) a. Monad m => a -> m a
return ()
Just Par ()
_ -> do StableName (Maybe (Par ()))
sn <- Maybe (Par ()) -> IO (StableName (Maybe (Par ())))
forall a. a -> IO (StableName a)
makeStableName Maybe (Par ())
mb
String -> Int -> Int -> IO ()
forall r. PrintfType r => String -> r
printf String
" [%d] -> POP work unit %d\n" Int
no (StableName (Maybe (Par ())) -> Int
forall a. StableName a -> Int
hashStableName StableName (Maybe (Par ()))
sn)
Maybe (Par ()) -> IO (Maybe (Par ()))
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
SimpleDeque (Par ()) -> Par () -> IO ()
forall t. SimpleDeque t -> t -> IO ()
R.pushL SimpleDeque (Par ())
WSDeque (Par ())
workpool Par ()
task
Bool -> IO () -> IO ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when Bool
dbg (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ do StableName (Par ())
sn <- Par () -> IO (StableName (Par ()))
forall a. a -> IO (StableName a)
makeStableName Par ()
task
String -> Int -> Int -> IO ()
forall r. PrintfType r => String -> r
printf String
" [%d] -> PUSH work unit %d\n" Int
no (StableName (Par ()) -> Int
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
() -> IO ()
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 <- HotVar [MVar Bool] -> IO [MVar Bool]
forall a. HotVar a -> IO a
readHotVar HotVar [MVar Bool]
idle
Bool -> IO () -> IO ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when (Bool -> Bool
not ([MVar Bool] -> Bool
forall (t :: * -> *) a. Foldable t => t a -> Bool
Prelude.null [MVar Bool]
idles)) (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
Bool -> IO () -> IO ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when Bool
dbg(IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ String -> Int -> IO ()
forall r. PrintfType r => String -> r
printf String
"Waking %d idle thread(s).\n" ([MVar Bool] -> Int
forall (t :: * -> *) a. Foldable t => t a -> Int
length [MVar Bool]
idles)
IO ()
r <- HotVar [MVar Bool]
-> ([MVar Bool] -> ([MVar Bool], IO ())) -> IO (IO ())
forall a b. HotVar a -> (a -> (a, b)) -> IO b
modifyHotVar HotVar [MVar Bool]
idle (\[MVar Bool]
is -> case [MVar Bool]
is of
[] -> ([], () -> IO ()
forall (m :: * -> *) a. Monad m => a -> m a
return ())
(MVar Bool
i:[MVar Bool]
ils) -> ([MVar Bool]
ils, MVar Bool -> Bool -> IO ()
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 = (Int, Int) -> GenIO -> IO Int
forall a (m :: * -> *).
(Variate a, PrimMonad m) =>
(a, a) -> Gen (PrimState m) -> m a
Random.uniformR (Int
0, Int
numCapabilitiesInt -> Int -> Int
forall a. Num a => a -> a -> a
-Int
1) (Gen RealWorld -> IO Int) -> IO (Gen RealWorld) -> IO Int
forall (m :: * -> *) a b. Monad m => (a -> m b) -> m a -> m b
=<< HotVar (Gen RealWorld) -> IO (Gen RealWorld)
forall a. HotVar a -> IO a
readHotVar HotVar (Gen RealWorld)
HotVar GenIO
ref
instance NFData (IVar a) where
rnf :: IVar a -> ()
rnf !IVar a
_ = ()
{-# NOINLINE runPar #-}
runPar :: forall a. Par a -> a
runPar = IO a -> a
forall a. IO a -> a
unsafePerformIO (IO a -> a) -> (Par a -> IO a) -> Par a -> a
forall b c a. (b -> c) -> (a -> b) -> a -> c
. Par a -> IO a
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 <- HotVar SessionID
-> (SessionID -> (SessionID, SessionID)) -> IO SessionID
forall a b. HotVar a -> (a -> (a, b)) -> IO b
modifyHotVar (Sched -> HotVar SessionID
sessionCounter Sched
sched) (\ SessionID
x -> (SessionID
xSessionID -> SessionID -> SessionID
forall a. Num a => a -> a -> a
+SessionID
1,SessionID
x))
()
_ <- HotVar (Set SessionID)
-> (Set SessionID -> (Set SessionID, ())) -> IO ()
forall a b. HotVar a -> (a -> (a, b)) -> IO b
modifyHotVar (Sched -> HotVar (Set SessionID)
activeSessions Sched
sched) (\ Set SessionID
set -> (SessionID -> Set SessionID -> Set SessionID
forall a. Ord a => a -> Set a -> Set a
S.insert SessionID
sid Set SessionID
set, ()))
IORef b
ref <- b -> IO (IORef b)
forall a. a -> IO (IORef a)
newIORef (String -> b
forall a. HasCallStack => String -> a
error(String -> b) -> String -> b
forall a b. (a -> b) -> a -> b
$ String
"Empty session-result ref ("String -> String -> String
forall a. [a] -> [a] -> [a]
++String
nameString -> String -> String
forall a. [a] -> [a] -> [a]
++String
") should never be touched (sid "String -> String -> String
forall a. [a] -> [a] -> [a]
++ SessionID -> String
forall a. Show a => a -> String
show SessionID
sidString -> String -> String
forall a. [a] -> [a] -> [a]
++String
", "String -> String -> String
forall a. [a] -> [a] -> [a]
++ThreadId -> String
forall a. Show a => a -> String
show ThreadId
tid String -> String -> String
forall a. [a] -> [a] -> [a]
++String
")")
HotVar Bool
newFlag <- Bool -> IO (HotVar Bool)
forall a. a -> IO (IORef a)
newHotVar Bool
False
()
_ <- HotVar [Session] -> ([Session] -> ([Session], ())) -> IO ()
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) Session -> [Session] -> [Session]
forall a. a -> [a] -> [a]
: [Session]
ls, ()))
let userComp' :: Par ()
userComp' = do Bool -> Par () -> Par ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when Bool
dbg(Par () -> Par ()) -> Par () -> Par ()
forall a b. (a -> b) -> a -> b
$ IO () -> Par ()
forall a. IO a -> Par a
io(IO () -> Par ()) -> IO () -> Par ()
forall a b. (a -> b) -> a -> b
$ do
ThreadId
tid2 <- IO ThreadId
myThreadId
String -> Int -> String -> String -> IO ()
forall r. PrintfType r => String -> r
printf String
" [%d %s] Starting Par computation on %s.\n" (Sched -> Int
no Sched
sched) (ThreadId -> String
forall a. Show a => a -> String
show ThreadId
tid2) String
name
b
ans <- Par b
userComp
IO () -> Par ()
forall a. IO a -> Par a
io(IO () -> Par ()) -> IO () -> Par ()
forall a b. (a -> b) -> a -> b
$ do Bool -> IO () -> IO ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when (Int
dbglvlInt -> Int -> Bool
forall a. Ord a => a -> a -> Bool
>=Int
1) (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
ThreadId
tid3 <- IO ThreadId
myThreadId
String -> Int -> String -> String -> SessionID -> IO ()
forall r. PrintfType r => String -> r
printf String
" [%d %s] Continuation for %s called, finishing it up (%d)...\n" (Sched -> Int
no Sched
sched) (ThreadId -> String
forall a. Show a => a -> String
show ThreadId
tid3) String
name SessionID
sid
IORef b -> b -> IO ()
forall a. IORef a -> a -> IO ()
writeIORef IORef b
ref b
ans
HotVar Bool -> Bool -> IO ()
forall a. IORef a -> a -> IO ()
writeHotVarRaw HotVar Bool
newFlag Bool
True
HotVar (Set SessionID)
-> (Set SessionID -> (Set SessionID, ())) -> IO ()
forall a b. HotVar a -> (a -> (a, b)) -> IO b
modifyHotVar (Sched -> HotVar (Set SessionID)
activeSessions Sched
sched) (\ Set SessionID
set -> (SessionID -> Set SessionID -> Set SessionID
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 = String -> a -> ROnly ()
forall a. String -> a -> ROnly ()
trivialCont(String -> a -> ROnly ()) -> String -> a -> ROnly ()
forall a b. (a -> b) -> a -> b
$ String
"("String -> String -> String
forall a. [a] -> [a] -> [a]
++String
nameString -> String -> String
forall a. [a] -> [a] -> [a]
++String
", sid "String -> String -> String
forall a. [a] -> [a] -> [a]
++SessionID -> String
forall a. Show a => a -> String
show SessionID
sidString -> String -> String
forall a. [a] -> [a] -> [a]
++String
", round "String -> String -> String
forall a. [a] -> [a] -> [a]
++SessionID -> String
forall a. Show a => a -> String
show SessionID
nString -> String -> String
forall a. [a] -> [a] -> [a]
++String
")"
loop :: Word64 -> ROnly ()
loop :: SessionID -> ROnly ()
loop SessionID
n = do Bool
flg <- IO Bool -> ReaderT Sched IO Bool
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO(IO Bool -> ReaderT Sched IO Bool)
-> IO Bool -> ReaderT Sched IO Bool
forall a b. (a -> b) -> a -> b
$ HotVar Bool -> IO Bool
forall a. HotVar a -> IO a
readIORef HotVar Bool
newFlag
Bool -> ROnly () -> ROnly ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
unless Bool
flg (ROnly () -> ROnly ()) -> ROnly () -> ROnly ()
forall a b. (a -> b) -> a -> b
$ do
Bool -> ROnly () -> ROnly ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when Bool
dbg (ROnly () -> ROnly ()) -> ROnly () -> ROnly ()
forall a b. (a -> b) -> a -> b
$ IO () -> ROnly ()
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO(IO () -> ROnly ()) -> IO () -> ROnly ()
forall a b. (a -> b) -> a -> b
$ do
ThreadId
tid4 <- IO ThreadId
myThreadId
String -> Int -> String -> SessionID -> IO ()
forall r. PrintfType r => String -> r
printf String
" [%d %s] BOUNCE %d... going into reschedule until finished.\n" (Sched -> Int
no Sched
sched) (ThreadId -> String
forall a. Show a => a -> String
show ThreadId
tid4) SessionID
n
SessionID -> (Any -> ROnly ()) -> ROnly ()
forall a. SessionID -> (a -> ROnly ()) -> ROnly ()
rescheduleR SessionID
0 ((Any -> ROnly ()) -> ROnly ()) -> (Any -> ROnly ()) -> ROnly ()
forall a b. (a -> b) -> a -> b
$ String -> Any -> ROnly ()
forall a. String -> a -> ROnly ()
trivialCont(String -> Any -> ROnly ()) -> String -> Any -> ROnly ()
forall a b. (a -> b) -> a -> b
$ String
"("String -> String -> String
forall a. [a] -> [a] -> [a]
++String
nameString -> String -> String
forall a. [a] -> [a] -> [a]
++String
", sid "String -> String -> String
forall a. [a] -> [a] -> [a]
++SessionID -> String
forall a. Show a => a -> String
show SessionID
sidString -> String -> String
forall a. [a] -> [a] -> [a]
++String
")"
SessionID -> ROnly ()
loop (SessionID
nSessionID -> SessionID -> SessionID
forall a. Num a => a -> a -> a
+SessionID
1)
Sched -> ROnly () -> IO ()
forall r (m :: * -> *) a. r -> ReaderT r m a -> m a
runReaderWith Sched
sched (ContT () ROnly () -> (() -> ROnly ()) -> ROnly ()
forall {k} (r :: k) (m :: k -> *) a.
ContT r m a -> (a -> m r) -> m r
C.runContT (Par () -> ContT () ROnly ()
forall a. Par a -> ContT () ROnly a
unPar Par ()
userComp') (SessionID -> () -> ROnly ()
forall a. SessionID -> a -> ROnly ()
kont SessionID
0))
Sched -> ROnly () -> IO ()
forall r (m :: * -> *) a. r -> ReaderT r m a -> m a
runReaderWith Sched
sched (SessionID -> ROnly ()
loop SessionID
1)
Bool -> IO () -> IO ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when (Int
dbglvlInt -> Int -> Bool
forall a. Ord a => a -> a -> Bool
>=Int
1)(IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
Set SessionID
active <- HotVar (Set SessionID) -> IO (Set SessionID)
forall a. HotVar a -> IO a
readHotVar (Sched -> HotVar (Set SessionID)
activeSessions Sched
sched)
sess :: Bool
sess@Bool
True <- HotVar Bool -> IO Bool
forall a. HotVar a -> IO a
readHotVar HotVar Bool
newFlag
String
-> Int
-> String
-> String
-> String
-> SessionID
-> String
-> IO ()
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) (ThreadId -> String
forall a. Show a => a -> String
show ThreadId
tid) String
name (Bool -> String
forall a. Show a => a -> String
show Bool
sess) SessionID
sid (Set SessionID -> String
forall a. Show a => a -> String
show Set SessionID
active)
HotVar [Session] -> ([Session] -> [Session]) -> IO ()
forall a. HotVar a -> (a -> a) -> IO ()
modifyHotVar_ (Sched -> HotVar [Session]
sessions Sched
sched) (([Session] -> [Session]) -> IO ())
-> ([Session] -> [Session]) -> IO ()
forall a b. (a -> b) -> a -> b
$ \ (Session SessionID
sid2 HotVar Bool
_ : [Session]
tl) ->
if SessionID
sid SessionID -> SessionID -> Bool
forall a. Eq a => a -> a -> Bool
== SessionID
sid2
then [Session]
tl
else String -> [Session]
forall a. HasCallStack => String -> a
error(String -> [Session]) -> String -> [Session]
forall a b. (a -> b) -> a -> b
$ String
"Tried to pop the session stack and found we ("String -> String -> String
forall a. [a] -> [a] -> [a]
++SessionID -> String
forall a. Show a => a -> String
show SessionID
sid
String -> String -> String
forall a. [a] -> [a] -> [a]
++String
") were not on the top! (instead "String -> String -> String
forall a. [a] -> [a] -> [a]
++SessionID -> String
forall a. Show a => a -> String
show SessionID
sid2String -> String -> String
forall a. [a] -> [a] -> [a]
++String
")"
IORef b -> IO b
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 <- HotVar SessionID -> IO SessionID
forall a. HotVar a -> IO a
readHotVar (Sched -> HotVar SessionID
sessionCounter Sched
sched)
Bool -> IO () -> IO ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when (Int
dbglvlInt -> Int -> Bool
forall a. Ord a => a -> a -> Bool
>=Int
1)(IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ String -> Int -> String -> SessionID -> IO ()
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) (ThreadId -> String
forall a. Show a => a -> String
show ThreadId
tid) (SessionID
sid0 SessionID -> SessionID -> SessionID
forall a. Num a => a -> a -> a
+ SessionID
1)
String -> Sched -> Par a -> IO a
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] <- HotVar [Session] -> IO [Session]
forall a. HotVar a -> IO a
readHotVar(HotVar [Session] -> IO [Session])
-> HotVar [Session] -> IO [Session]
forall a b. (a -> b) -> a -> b
$ Sched -> HotVar [Session]
sessions(Sched -> HotVar [Session]) -> Sched -> HotVar [Session]
forall a b. (a -> b) -> a -> b
$ [Sched] -> Sched
forall a. [a] -> a
head [Sched]
allscheds
MVar a
mfin <- IO (MVar a)
forall a. IO (MVar a)
newEmptyMVar
[Maybe (MVar Int)]
doneFlags <- [(Int, Sched)]
-> ((Int, Sched) -> IO (Maybe (MVar Int))) -> IO [Maybe (MVar Int)]
forall (t :: * -> *) (m :: * -> *) a b.
(Traversable t, Monad m) =>
t a -> (a -> m b) -> m (t b)
forM ([Int] -> [Sched] -> [(Int, Sched)]
forall a b. [a] -> [b] -> [(a, b)]
zip [Int
0..] [Sched]
allscheds) (((Int, Sched) -> IO (Maybe (MVar Int))) -> IO [Maybe (MVar Int)])
-> ((Int, Sched) -> IO (Maybe (MVar Int))) -> IO [Maybe (MVar Int)]
forall a b. (a -> b) -> a -> b
$ \(Int
cpu,Sched
sched) -> do
MVar Int
workerDone <- IO (MVar Int)
forall a. IO (MVar a)
newEmptyMVar
let wname :: String
wname = (String
"(worker "String -> String -> String
forall a. [a] -> [a] -> [a]
++Int -> String
forall a. Show a => a -> String
show Int
cpuString -> String -> String
forall a. [a] -> [a] -> [a]
++String
" of originator "String -> String -> String
forall a. [a] -> [a] -> [a]
++ThreadId -> String
forall a. Show a => a -> String
show ThreadId
tidorigString -> String -> String
forall a. [a] -> [a] -> [a]
++String
")")
ThreadId
_ <- (IO () -> IO ThreadId) -> String -> IO () -> IO ThreadId
forkWithExceptions (Int -> IO () -> IO ThreadId
forkOn Int
cpu) String
wname (IO () -> IO ThreadId) -> IO () -> IO ThreadId
forall a b. (a -> b) -> a -> b
$ do
ThreadId
tid2 <- IO ThreadId
myThreadId
ThreadId -> Sched -> IO ()
registerWorker ThreadId
tid2 Sched
sched
if (Int
cpu Int -> Int -> Bool
forall a. Eq a => a -> a -> Bool
/= Int
main_cpu)
then do Bool -> IO () -> IO ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when Bool
dbg(IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ String -> Int -> String -> IO ()
forall r. PrintfType r => String -> r
printf String
" [%d %s] Anonymous worker entering scheduling loop.\n" Int
cpu (ThreadId -> String
forall a. Show a => a -> String
show ThreadId
tid2)
Sched -> ROnly () -> IO ()
forall r (m :: * -> *) a. r -> ReaderT r m a -> m a
runReaderWith Sched
sched (ROnly () -> IO ()) -> ROnly () -> IO ()
forall a b. (a -> b) -> a -> b
$ SessionID -> (Any -> ROnly ()) -> ROnly ()
forall a. SessionID -> (a -> ROnly ()) -> ROnly ()
rescheduleR SessionID
0 (String -> Any -> ROnly ()
forall a. String -> a -> ROnly ()
trivialCont (String
wnameString -> String -> String
forall a. [a] -> [a] -> [a]
++ThreadId -> String
forall a. Show a => a -> String
show ThreadId
tid2))
Bool -> IO () -> IO ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when Bool
dbg(IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ String -> Int -> IO ()
forall r. PrintfType r => String -> r
printf String
" [%d] Anonymous worker exited scheduling loop. FINISHED.\n" Int
cpu
MVar Int -> Int -> IO ()
forall a. MVar a -> a -> IO ()
putMVar MVar Int
workerDone Int
cpu
() -> IO ()
forall (m :: * -> *) a. Monad m => a -> m a
return ()
else do a
x <- String -> Sched -> Par a -> IO a
forall b. String -> Sched -> Par b -> IO b
runNewSessionAndWait String
"top-lvl main worker" Sched
sched Par a
userComp
HotVar Bool -> Bool -> IO ()
forall a. IORef a -> a -> IO ()
writeIORef HotVar Bool
topSessFlag Bool
True
Bool -> IO () -> IO ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when Bool
dbg(IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ do String -> String -> IO ()
forall r. PrintfType r => String -> r
printf String
" *** Out of entire runContT user computation on main thread %s.\n" (ThreadId -> String
forall a. Show a => a -> String
show ThreadId
tid2)
MVar a -> a -> IO ()
forall a. MVar a -> a -> IO ()
putMVar MVar a
mfin a
x
ThreadId -> IO ()
unregisterWorker ThreadId
tid
Maybe (MVar Int) -> IO (Maybe (MVar Int))
forall (m :: * -> *) a. Monad m => a -> m a
return (if Int
cpu Int -> Int -> Bool
forall a. Eq a => a -> a -> Bool
== Int
main_cpu then Maybe (MVar Int)
forall a. Maybe a
Nothing else MVar Int -> Maybe (MVar Int)
forall a. a -> Maybe a
Just MVar Int
workerDone)
Bool -> IO () -> IO ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when Bool
_WAIT_FOR_WORKERS (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
Bool -> IO () -> IO ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when Bool
dbg(IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ String -> String -> IO ()
forall r. PrintfType r => String -> r
printf String
" *** [%s] Originator thread: waiting for workers to complete." (ThreadId -> String
forall a. Show a => a -> String
show ThreadId
tidorig)
[MVar Int] -> (MVar Int -> IO ()) -> IO ()
forall (t :: * -> *) (m :: * -> *) a b.
(Foldable t, Monad m) =>
t a -> (a -> m b) -> m ()
forM_ ([Maybe (MVar Int)] -> [MVar Int]
forall a. [Maybe a] -> [a]
catMaybes [Maybe (MVar Int)]
doneFlags) ((MVar Int -> IO ()) -> IO ()) -> (MVar Int -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \ MVar Int
mv -> do
Int
n <- MVar Int -> IO Int
forall a. MVar a -> IO a
readMVar MVar Int
mv
Bool -> IO () -> IO ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when Bool
dbg(IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ String -> String -> String -> IO ()
forall r. PrintfType r => String -> r
printf String
" * [%s] Worker %s completed\n" (ThreadId -> String
forall a. Show a => a -> String
show ThreadId
tidorig) (Int -> String
forall a. Show a => a -> String
show Int
n)
Bool -> IO () -> IO ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when Bool
dbg(IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ do String -> String -> IO ()
forall r. PrintfType r => String -> r
printf String
" *** [%s] Reading final MVar on originator thread.\n" (ThreadId -> String
forall a. Show a => a -> String
show ThreadId
tidorig)
#ifdef DEBUG_DIRECT
busyTakeMVar (" The global wait "++ show tidorig) mfin
#else
MVar a -> IO a
forall a. MVar a -> IO a
takeMVar MVar a
mfin
#endif
makeScheds :: Int -> IO [Sched]
makeScheds :: Int -> IO [Sched]
makeScheds Int
main = do
Bool -> IO () -> IO ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when Bool
dbg(IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ do ThreadId
tid <- IO ThreadId
myThreadId
String -> Int -> String -> IO ()
forall r. PrintfType r => String -> r
printf String
"[initialization] Creating %d worker threads, currently on %s\n" Int
numCapabilities (ThreadId -> String
forall a. Show a => a -> String
show ThreadId
tid)
[SimpleDeque (Par ())]
workpools <- Int -> IO (SimpleDeque (Par ())) -> IO [SimpleDeque (Par ())]
forall (m :: * -> *) a. Applicative m => Int -> m a -> m [a]
replicateM Int
numCapabilities (IO (SimpleDeque (Par ())) -> IO [SimpleDeque (Par ())])
-> IO (SimpleDeque (Par ())) -> IO [SimpleDeque (Par ())]
forall a b. (a -> b) -> a -> b
$ IO (SimpleDeque (Par ()))
forall elt. IO (SimpleDeque elt)
R.newQ
[HotVar (Gen RealWorld)]
rngs <- Int -> IO (HotVar (Gen RealWorld)) -> IO [HotVar (Gen RealWorld)]
forall (m :: * -> *) a. Applicative m => Int -> m a -> m [a]
replicateM Int
numCapabilities (IO (HotVar (Gen RealWorld)) -> IO [HotVar (Gen RealWorld)])
-> IO (HotVar (Gen RealWorld)) -> IO [HotVar (Gen RealWorld)]
forall a b. (a -> b) -> a -> b
$ IO (Gen RealWorld)
forall (m :: * -> *). PrimMonad m => m (Gen (PrimState m))
Random.create IO (Gen RealWorld)
-> (Gen RealWorld -> IO (HotVar (Gen RealWorld)))
-> IO (HotVar (Gen RealWorld))
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= Gen RealWorld -> IO (HotVar (Gen RealWorld))
forall a. a -> IO (IORef a)
newHotVar
HotVar [MVar Bool]
idle <- [MVar Bool] -> IO (HotVar [MVar Bool])
forall a. a -> IO (IORef a)
newHotVar []
HotVar Bool
sessionFinished <- Bool -> IO (HotVar Bool)
forall a. a -> IO (IORef a)
newHotVar Bool
False
[HotVar [Session]]
sessionStacks <- ([Session] -> IO (HotVar [Session]))
-> [[Session]] -> IO [HotVar [Session]]
forall (t :: * -> *) (m :: * -> *) a b.
(Traversable t, Monad m) =>
(a -> m b) -> t a -> m (t b)
mapM [Session] -> IO (HotVar [Session])
forall a. a -> IO (IORef a)
newHotVar (Int -> [Session] -> [[Session]]
forall a. Int -> a -> [a]
replicate Int
numCapabilities [SessionID -> HotVar Bool -> Session
Session SessionID
baseSessionID HotVar Bool
sessionFinished])
HotVar (Set SessionID)
activeSessions <- Set SessionID -> IO (HotVar (Set SessionID))
forall a. a -> IO (IORef a)
newHotVar Set SessionID
forall a. Set a
S.empty
HotVar SessionID
sessionCounter <- SessionID -> IO (HotVar SessionID)
forall a. a -> IO (IORef a)
newHotVar (SessionID
baseSessionID SessionID -> SessionID -> SessionID
forall a. Num a => a -> a -> a
+ SessionID
1)
let allscheds :: [Sched]
allscheds = [ Sched :: Int
-> WSDeque (Par ())
-> HotVar GenIO
-> Bool
-> HotVar [Session]
-> HotVar [MVar Bool]
-> [Sched]
-> HotVar (Set SessionID)
-> HotVar SessionID
-> Sched
Sched { no :: Int
no=Int
x, HotVar [MVar Bool]
idle :: HotVar [MVar Bool]
idle :: HotVar [MVar Bool]
idle, isMain :: Bool
isMain= (Int
xInt -> Int -> Bool
forall a. Eq a => a -> a -> Bool
==Int
main),
workpool :: WSDeque (Par ())
workpool=SimpleDeque (Par ())
WSDeque (Par ())
wp, scheds :: [Sched]
scheds=[Sched]
allscheds, rng :: HotVar GenIO
rng=HotVar (Gen RealWorld)
HotVar GenIO
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
numCapabilitiesInt -> Int -> Int
forall 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
]
[Sched] -> IO [Sched]
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 = IO (IVar a) -> Par (IVar a)
forall a. IO a -> Par a
io(IO (IVar a) -> Par (IVar a)) -> IO (IVar a) -> Par (IVar a)
forall a b. (a -> b) -> a -> b
$ do IORef (IVarContents a)
r <- IVarContents a -> IO (IORef (IVarContents a))
forall a. a -> IO (IORef a)
newIORef IVarContents a
forall a. IVarContents a
Empty
IVar a -> IO (IVar a)
forall (m :: * -> *) a. Monad m => a -> m a
return (IORef (IVarContents a) -> IVar a
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
((a -> Par ()) -> Par a) -> Par a
forall (m :: * -> *) a b. MonadCont m => ((a -> m b) -> m a) -> m a
callCC (((a -> Par ()) -> Par a) -> Par a)
-> ((a -> Par ()) -> Par a) -> Par a
forall a b. (a -> b) -> a -> b
$ \a -> Par ()
kont ->
do
IVarContents a
e <- IO (IVarContents a) -> Par (IVarContents a)
forall a. IO a -> Par a
io(IO (IVarContents a) -> Par (IVarContents a))
-> IO (IVarContents a) -> Par (IVarContents a)
forall a b. (a -> b) -> a -> b
$ IORef (IVarContents a) -> IO (IVarContents a)
forall a. HotVar a -> IO a
readIORef IORef (IVarContents a)
vr
case IVarContents a
e of
Full a
a -> a -> Par a
forall (m :: * -> *) a. Monad m => a -> m a
return a
a
IVarContents a
_ -> do
Sched
sch <- Par Sched
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
Par a
forall a. Par a
longjmpSched
Par a
r <- IO (Par a) -> Par (Par a)
forall a. IO a -> Par a
io(IO (Par a) -> Par (Par a)) -> IO (Par a) -> Par (Par a)
forall a b. (a -> b) -> a -> b
$ IORef (IVarContents a)
-> (IVarContents a -> (IVarContents a, Par a)) -> IO (Par a)
forall a b. HotVar a -> (a -> (a, b)) -> IO b
atomicModifyIORef IORef (IVarContents a)
vr ((IVarContents a -> (IVarContents a, Par a)) -> IO (Par a))
-> (IVarContents a -> (IVarContents a, Par a)) -> IO (Par a)
forall a b. (a -> b) -> a -> b
$ \IVarContents a
x -> case IVarContents a
x of
IVarContents a
Empty -> ([a -> IO ()] -> IVarContents a
forall a. [a -> IO ()] -> IVarContents a
Blocked [Sched -> Par () -> IO ()
pushWork Sched
sch (Par () -> IO ()) -> (a -> Par ()) -> a -> IO ()
forall b c a. (b -> c) -> (a -> b) -> a -> c
. a -> Par ()
kont], Par a
forall a. Par a
resched)
Full a
a -> (a -> IVarContents a
forall a. a -> IVarContents a
Full a
a, a -> Par a
forall (m :: * -> *) a. Monad m => a -> m a
return a
a)
Blocked [a -> IO ()]
ks -> ([a -> IO ()] -> IVarContents a
forall a. [a -> IO ()] -> IVarContents a
Blocked (Sched -> Par () -> IO ()
pushWork Sched
sch (Par () -> IO ()) -> (a -> Par ()) -> a -> IO ()
forall b c a. (b -> c) -> (a -> b) -> a -> c
. a -> Par ()
kont(a -> IO ()) -> [a -> IO ()] -> [a -> IO ()]
forall a. a -> [a] -> [a]
:[a -> IO ()]
ks), Par a
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 <- IO (IVarContents a) -> Par (IVarContents a)
forall a. IO a -> Par a
io(IO (IVarContents a) -> Par (IVarContents a))
-> IO (IVarContents a) -> Par (IVarContents a)
forall a b. (a -> b) -> a -> b
$ IORef (IVarContents a) -> IO (IVarContents a)
forall a. HotVar a -> IO a
readIORef IORef (IVarContents a)
v
case IVarContents a
e of
Full a
a -> Maybe a -> Par (Maybe a)
forall (m :: * -> *) a. Monad m => a -> m a
return (a -> Maybe a
forall a. a -> Maybe a
Just a
a)
IVarContents a
_ -> Maybe a -> Par (Maybe a)
forall (m :: * -> *) a. Monad m => a -> m a
return Maybe a
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 <- Par Sched
forall r (m :: * -> *). MonadReader r m => m r
RD.ask
[a -> IO ()]
ks <- IO [a -> IO ()] -> Par [a -> IO ()]
forall a. IO a -> Par a
io(IO [a -> IO ()] -> Par [a -> IO ()])
-> IO [a -> IO ()] -> Par [a -> IO ()]
forall a b. (a -> b) -> a -> b
$ do
[a -> IO ()]
ks <- IORef (IVarContents a)
-> (IVarContents a -> (IVarContents a, [a -> IO ()]))
-> IO [a -> IO ()]
forall a b. HotVar a -> (a -> (a, b)) -> IO b
atomicModifyIORef IORef (IVarContents a)
vr ((IVarContents a -> (IVarContents a, [a -> IO ()]))
-> IO [a -> IO ()])
-> (IVarContents a -> (IVarContents a, [a -> IO ()]))
-> IO [a -> IO ()]
forall a b. (a -> b) -> a -> b
$ \IVarContents a
e -> case IVarContents a
e of
IVarContents a
Empty -> (a -> IVarContents a
forall a. a -> IVarContents a
Full a
content, [])
Full a
_ -> String -> (IVarContents a, [a -> IO ()])
forall a. HasCallStack => String -> a
error String
"multiple put"
Blocked [a -> IO ()]
ks -> (a -> IVarContents a
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
[a -> IO ()] -> IO [a -> IO ()]
forall (m :: * -> *) a. Monad m => a -> m a
return [a -> IO ()]
ks
Sched -> [a -> IO ()] -> a -> Par ()
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 <- Par Sched
forall r (m :: * -> *). MonadReader r m => m r
RD.ask
([b -> IO ()]
ks,b
res) <- IO ([b -> IO ()], b) -> Par ([b -> IO ()], b)
forall a. IO a -> Par a
io(IO ([b -> IO ()], b) -> Par ([b -> IO ()], b))
-> IO ([b -> IO ()], b) -> Par ([b -> IO ()], b)
forall a b. (a -> b) -> a -> b
$ do
([b -> IO ()], b)
pr <- IORef (IVarContents b)
-> (IVarContents b -> (IVarContents b, ([b -> IO ()], b)))
-> IO ([b -> IO ()], b)
forall a b. HotVar a -> (a -> (a, b)) -> IO b
atomicModifyIORef IORef (IVarContents b)
vr ((IVarContents b -> (IVarContents b, ([b -> IO ()], b)))
-> IO ([b -> IO ()], b))
-> (IVarContents b -> (IVarContents b, ([b -> IO ()], b)))
-> IO ([b -> IO ()], b)
forall a b. (a -> b) -> a -> b
$ \IVarContents b
e -> case IVarContents b
e of
IVarContents b
Empty -> (b -> IVarContents b
forall a. a -> IVarContents a
Full b
content, ([], b
content))
Full b
x -> (b -> IVarContents b
forall a. a -> IVarContents a
Full b
x, ([], b
x))
Blocked [b -> IO ()]
ks -> (b -> IVarContents b
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
([b -> IO ()], b) -> IO ([b -> IO ()], b)
forall (m :: * -> *) a. Monad m => a -> m a
return ([b -> IO ()], b)
pr
Sched -> [b -> IO ()] -> b -> Par ()
forall a. Sched -> [a -> IO ()] -> a -> Par ()
wakeUp Sched
sched [b -> IO ()]
ks b
content
b -> Par b
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 [] = () -> Par ()
forall (m :: * -> *) a. Monad m => a -> m a
return ()
loop (a -> IO ()
kont:[a -> IO ()]
rest) = do
if Bool
_PARPUTS then
do IVar ()
_ <- Par () -> Par (IVar ())
forall a. Par a -> Par (IVar a)
spawn_(Par () -> Par (IVar ())) -> Par () -> Par (IVar ())
forall a b. (a -> b) -> a -> b
$ (a -> IO ()) -> [a -> IO ()] -> Par ()
pMap a -> IO ()
kont [a -> IO ()]
rest
() -> Par ()
forall (m :: * -> *) a. Monad m => a -> m a
return ()
else
do IO () -> Par ()
forall a. IO a -> Par a
io(IO () -> Par ()) -> IO () -> Par ()
forall a b. (a -> b) -> a -> b
$ a -> IO ()
kont a
arg
[a -> IO ()] -> Par ()
loop [a -> IO ()]
rest
() -> Par ()
forall (m :: * -> *) a. Monad m => a -> m a
return ()
pMap :: (a -> IO ()) -> [a -> IO ()] -> Par ()
pMap a -> IO ()
kont [] = IO () -> Par ()
forall a. IO a -> Par a
io(IO () -> Par ()) -> IO () -> Par ()
forall a b. (a -> b) -> a -> b
$ a -> IO ()
kont a
arg
pMap a -> IO ()
kont (a -> IO ()
more:[a -> IO ()]
rest) =
do IVar ()
_ <- Par () -> Par (IVar ())
forall a. Par a -> Par (IVar a)
spawn_(Par () -> Par (IVar ())) -> Par () -> Par (IVar ())
forall a b. (a -> b) -> a -> b
$ IO () -> Par ()
forall a. IO a -> Par a
io(IO () -> Par ()) -> IO () -> Par ()
forall 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 <- Par Sched
forall r (m :: * -> *). MonadReader r m => m r
RD.ask
((() -> Par ()) -> Par ()) -> Par ()
forall (m :: * -> *) a b. MonadCont m => ((a -> m b) -> m a) -> m a
callCC(((() -> Par ()) -> Par ()) -> Par ())
-> ((() -> Par ()) -> Par ()) -> Par ()
forall a b. (a -> b) -> a -> b
$ \() -> Par ()
parent -> do
let wrapped :: Par ()
wrapped = () -> Par ()
parent ()
IO () -> Par ()
forall a. IO a -> Par a
io(IO () -> Par ()) -> IO () -> Par ()
forall a b. (a -> b) -> a -> b
$ Sched -> Par () -> IO ()
pushWork Sched
sched Par ()
wrapped
Par ()
task
Any
_ <- Par Any
forall a. Par a
longjmpSched
IO () -> Par ()
forall a. IO a -> Par a
io(IO () -> Par ()) -> IO () -> Par ()
forall a b. (a -> b) -> a -> b
$ String -> IO ()
forall r. PrintfType r => String -> r
printf String
" !!! ERROR: Should never reach this point #1\n"
Bool -> Par () -> Par ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when Bool
dbg(Par () -> Par ()) -> Par () -> Par ()
forall a b. (a -> b) -> a -> b
$ do
Sched
sched2 <- Par Sched
forall r (m :: * -> *). MonadReader r m => m r
RD.ask
IO () -> Par ()
forall a. IO a -> Par a
io(IO () -> Par ()) -> IO () -> Par ()
forall a b. (a -> b) -> a -> b
$ String -> Int -> Int -> IO ()
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)
() -> Par ()
forall (m :: * -> *) a. Monad m => a -> m a
return ()
Bool
False -> do
Sched
sch <- Par Sched
forall r (m :: * -> *). MonadReader r m => m r
RD.ask
Bool -> Par () -> Par ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when Bool
dbg(Par () -> Par ()) -> Par () -> Par ()
forall a b. (a -> b) -> a -> b
$ IO () -> Par ()
forall a. IO a -> Par a
io(IO () -> Par ()) -> IO () -> Par ()
forall a b. (a -> b) -> a -> b
$ String -> Int -> IO ()
forall r. PrintfType r => String -> r
printf String
" [%d] forking task...\n" (Sched -> Int
no Sched
sch)
IO () -> Par ()
forall a. IO a -> Par a
io(IO () -> Par ()) -> IO () -> Par ()
forall a b. (a -> b) -> a -> b
$ Sched -> Par () -> IO ()
pushWork Sched
sch Par ()
task
longjmpSched :: Par a
longjmpSched :: forall a. Par a
longjmpSched = ContT () ROnly a -> Par a
forall a. ContT () ROnly a -> Par a
Par (ContT () ROnly a -> Par a) -> ContT () ROnly a -> Par a
forall a b. (a -> b) -> a -> b
$ ((a -> ROnly ()) -> ROnly ()) -> ContT () ROnly a
forall {k} (r :: k) (m :: k -> *) a.
((a -> m r) -> m r) -> ContT r m a
C.ContT (\ a -> ROnly ()
_k -> SessionID -> (Any -> ROnly ()) -> ROnly ()
forall a. SessionID -> (a -> ROnly ()) -> ROnly ()
rescheduleR SessionID
0 (String -> Any -> ROnly ()
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 <- ReaderT Sched IO Sched
forall r (m :: * -> *). MonadReader r m => m r
RD.ask
Bool -> ROnly () -> ROnly ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when Bool
dbg(ROnly () -> ROnly ()) -> ROnly () -> ROnly ()
forall a b. (a -> b) -> a -> b
$ IO () -> ROnly ()
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO(IO () -> ROnly ()) -> IO () -> ROnly ()
forall a b. (a -> b) -> a -> b
$ do ThreadId
tid <- IO ThreadId
myThreadId
[(SessionID, Bool)]
sess <- Sched -> IO [(SessionID, Bool)]
readSessions Sched
mysched
Bool
null <- SimpleDeque (Par ()) -> IO Bool
forall elt. SimpleDeque elt -> IO Bool
R.nullQ (Sched -> WSDeque (Par ())
workpool Sched
mysched)
String -> Int -> String -> SessionID -> String -> String -> IO ()
forall r. PrintfType r => String -> r
printf String
" [%d %s] - Reschedule #%d... sessions %s, pool empty %s\n"
(Sched -> Int
no Sched
mysched) (ThreadId -> String
forall a. Show a => a -> String
show ThreadId
tid) SessionID
cnt ([(SessionID, Bool)] -> String
forall a. Show a => a -> String
show [(SessionID, Bool)]
sess) (Bool -> String
forall a. Show a => a -> String
show Bool
null)
Maybe (Par ())
mtask <- IO (Maybe (Par ())) -> ReaderT Sched IO (Maybe (Par ()))
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO(IO (Maybe (Par ())) -> ReaderT Sched IO (Maybe (Par ())))
-> IO (Maybe (Par ())) -> ReaderT Sched IO (Maybe (Par ()))
forall 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]
_ <- IO [Session] -> ReaderT Sched IO [Session]
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO(IO [Session] -> ReaderT Sched IO [Session])
-> IO [Session] -> ReaderT Sched IO [Session]
forall a b. (a -> b) -> a -> b
$ HotVar [Session] -> IO [Session]
forall a. HotVar a -> IO a
readIORef (HotVar [Session] -> IO [Session])
-> HotVar [Session] -> IO [Session]
forall a b. (a -> b) -> a -> b
$ Sched -> HotVar [Session]
sessions Sched
mysched
Bool
fin <- IO Bool -> ReaderT Sched IO Bool
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO(IO Bool -> ReaderT Sched IO Bool)
-> IO Bool -> ReaderT Sched IO Bool
forall a b. (a -> b) -> a -> b
$ HotVar Bool -> IO Bool
forall a. HotVar a -> IO a
readIORef HotVar Bool
finRef
if Bool
fin
then do Bool -> ROnly () -> ROnly ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when (Int
dbglvl Int -> Int -> Bool
forall a. Ord a => a -> a -> Bool
>= Int
1) (ROnly () -> ROnly ()) -> ROnly () -> ROnly ()
forall a b. (a -> b) -> a -> b
$ IO () -> ROnly ()
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO (IO () -> ROnly ()) -> IO () -> ROnly ()
forall a b. (a -> b) -> a -> b
$ do
ThreadId
tid <- IO ThreadId
myThreadId
[(SessionID, Bool)]
sess <- Sched -> IO [(SessionID, Bool)]
readSessions Sched
mysched
String -> Int -> String -> String -> String -> IO ()
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) (ThreadId -> String
forall a. Show a => a -> String
show ThreadId
tid) (Bool -> String
forall a. Show a => a -> String
show Bool
fin) ([(SessionID, Bool)] -> String
forall a. Show a => a -> String
show [(SessionID, Bool)]
sess)
Bool
empt <- SimpleDeque (Par ()) -> IO Bool
forall elt. SimpleDeque elt -> IO Bool
R.nullQ(SimpleDeque (Par ()) -> IO Bool)
-> SimpleDeque (Par ()) -> IO Bool
forall a b. (a -> b) -> a -> b
$ Sched -> WSDeque (Par ())
workpool Sched
mysched
Bool -> IO () -> IO ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when (Bool -> Bool
not Bool
empt) (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
String -> Int -> String -> IO ()
forall r. PrintfType r => String -> r
printf String
" [%d %s] - WARNING - leaving rescheduleR while local workpoll is nonempty\n"
(Sched -> Int
no Sched
mysched) (ThreadId -> String
forall a. Show a => a -> String
show ThreadId
tid)
a -> ROnly ()
kont (String -> a
forall a. HasCallStack => String -> a
error String
"Direct.hs: The result value from rescheduleR should not be used.")
else do
IO () -> ROnly ()
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO(IO () -> ROnly ()) -> IO () -> ROnly ()
forall a b. (a -> b) -> a -> b
$ Sched -> IO ()
steal Sched
mysched
#ifdef WAKEIDLE
#endif
IO () -> ROnly ()
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO IO ()
yield
SessionID -> (a -> ROnly ()) -> ROnly ()
forall a. SessionID -> (a -> ROnly ()) -> ROnly ()
rescheduleR (SessionID
cntSessionID -> SessionID -> SessionID
forall a. Num a => a -> a -> a
+SessionID
1) a -> ROnly ()
kont
Just Par ()
task -> do
Bool -> ROnly () -> ROnly ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when Bool
dbg (ROnly () -> ROnly ()) -> ROnly () -> ROnly ()
forall a b. (a -> b) -> a -> b
$ do StableName (Par ())
sn <- IO (StableName (Par ())) -> ReaderT Sched IO (StableName (Par ()))
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO(IO (StableName (Par ()))
-> ReaderT Sched IO (StableName (Par ())))
-> IO (StableName (Par ()))
-> ReaderT Sched IO (StableName (Par ()))
forall a b. (a -> b) -> a -> b
$ Par () -> IO (StableName (Par ()))
forall a. a -> IO (StableName a)
makeStableName Par ()
task
IO () -> ROnly ()
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO(IO () -> ROnly ()) -> IO () -> ROnly ()
forall a b. (a -> b) -> a -> b
$ String -> Int -> Int -> IO ()
forall r. PrintfType r => String -> r
printf String
" [%d] popped work %d from own queue\n" (Sched -> Int
no Sched
mysched) (StableName (Par ()) -> Int
forall a. StableName a -> Int
hashStableName StableName (Par ())
sn)
let C.ContT (() -> ROnly ()) -> ROnly ()
fn = Par () -> ContT () ROnly ()
forall a. Par a -> ContT () ROnly a
unPar Par ()
task
(() -> ROnly ()) -> ROnly ()
fn (\ ()
_ -> do
Sched
sch <- ReaderT Sched IO Sched
forall r (m :: * -> *). MonadReader r m => m r
RD.ask
Bool -> ROnly () -> ROnly ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when Bool
dbg(ROnly () -> ROnly ()) -> ROnly () -> ROnly ()
forall a b. (a -> b) -> a -> b
$ IO () -> ROnly ()
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO(IO () -> ROnly ()) -> IO () -> ROnly ()
forall a b. (a -> b) -> a -> b
$ String -> Int -> IO ()
forall r. PrintfType r => String -> r
printf String
" + task finished successfully on cpu %d, calling reschedule continuation..\n" (Sched -> Int
no Sched
sch)
SessionID -> (a -> ROnly ()) -> ROnly ()
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
Bool -> IO () -> IO ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when (Int
dbglvlInt -> Int -> Bool
forall a. Ord a => a -> a -> Bool
>=Int
2)(IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ do ThreadId
tid <- IO ThreadId
myThreadId
String -> Int -> String -> IO ()
forall r. PrintfType r => String -> r
printf String
" [%d %s] + stealing\n" Int
my_no (ThreadId -> String
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 Int -> Int -> Int
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 <- IO (MVar Bool)
forall a. IO (MVar a)
newEmptyMVar
[MVar Bool]
r <- HotVar [MVar Bool]
-> ([MVar Bool] -> ([MVar Bool], [MVar Bool])) -> IO [MVar Bool]
forall a b. HotVar a -> (a -> (a, b)) -> IO b
modifyHotVar HotVar [MVar Bool]
idle (([MVar Bool] -> ([MVar Bool], [MVar Bool])) -> IO [MVar Bool])
-> ([MVar Bool] -> ([MVar Bool], [MVar Bool])) -> IO [MVar Bool]
forall a b. (a -> b) -> a -> b
$ \[MVar Bool]
is -> (MVar Bool
mMVar Bool -> [MVar Bool] -> [MVar Bool]
forall a. a -> [a] -> [a]
:[MVar Bool]
is, [MVar Bool]
is)
if [MVar Bool] -> Int
forall (t :: * -> *) a. Foldable t => t a -> Int
length [MVar Bool]
r Int -> Int -> Bool
forall a. Eq a => a -> a -> Bool
== Int
numCapabilities Int -> Int -> Int
forall a. Num a => a -> a -> a
- Int
1
then do
Bool -> IO () -> IO ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when Bool
dbg(IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ String -> Int -> IO ()
forall r. PrintfType r => String -> r
printf String
" [%d] | waking up all threads\n" Int
my_no
HotVar [MVar Bool] -> [MVar Bool] -> IO ()
forall a. IORef a -> a -> IO ()
writeHotVarRaw HotVar [MVar Bool]
idle []
(MVar Bool -> IO ()) -> [MVar Bool] -> IO ()
forall (t :: * -> *) (m :: * -> *) a b.
(Foldable t, Monad m) =>
(a -> m b) -> t a -> m ()
mapM_ (\MVar Bool
vr -> MVar Bool -> Bool -> IO ()
forall a. MVar a -> a -> IO ()
putMVar MVar Bool
vr Bool
True) [MVar Bool]
r
else do
(Session SessionID
_ HotVar Bool
finRef):[Session]
_ <- HotVar [Session] -> IO [Session]
forall a. HotVar a -> IO a
readIORef (HotVar [Session] -> IO [Session])
-> HotVar [Session] -> IO [Session]
forall a b. (a -> b) -> a -> b
$ Sched -> HotVar [Session]
sessions Sched
mysched
Bool
fin <- HotVar Bool -> IO Bool
forall a. HotVar a -> IO a
readIORef HotVar Bool
finRef
Bool
done <- if Bool
fin then Bool -> IO Bool
forall (f :: * -> *) a. Applicative f => a -> f a
pure Bool
True else MVar Bool -> IO Bool
forall a. MVar a -> IO a
takeMVar MVar Bool
m
if Bool
done
then do
Bool -> IO () -> IO ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when Bool
dbg(IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ String -> Int -> IO ()
forall r. PrintfType r => String -> r
printf String
" [%d] | shutting down\n" Int
my_no
() -> IO ()
forall (m :: * -> *) a. Monad m => a -> m a
return ()
else do
Bool -> IO () -> IO ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when Bool
dbg(IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ String -> Int -> IO ()
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 Bool -> Bool -> Bool
forall a. Eq a => a -> a -> Bool
== Bool
False = IO ()
yield
go Int
tries Int
i
| Int
i Int -> Int -> Bool
forall a. Eq a => a -> a -> Bool
== Int
my_no = do Int
i' <- Int -> IO Int
getnext Int
i
Int -> Int -> IO ()
go (Int
triesInt -> Int -> Int
forall a. Num a => a -> a -> a
-Int
1) Int
i'
| Bool
otherwise = do
let schd :: Sched
schd = [Sched]
scheds[Sched] -> Int -> Sched
forall a. [a] -> Int -> a
!!Int
i
Bool -> IO () -> IO ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when (Int
dbglvlInt -> Int -> Bool
forall a. Ord a => a -> a -> Bool
>=Int
2)(IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ String -> Int -> Int -> IO ()
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 <- SimpleDeque (Par ()) -> IO (Maybe (Par ()))
forall a. SimpleDeque a -> IO (Maybe a)
R.tryPopR SimpleDeque (Par ())
WSDeque (Par ())
dq
case Maybe (Par ())
r of
Just Par ()
task -> do
Bool -> IO () -> IO ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when Bool
dbg(IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ do StableName (Par ())
sn <- Par () -> IO (StableName (Par ()))
forall a. a -> IO (StableName a)
makeStableName Par ()
task
String -> Int -> Int -> Int -> IO ()
forall r. PrintfType r => String -> r
printf String
" [%d] | stole work (unit %d) from cpu %d\n" Int
my_no (StableName (Par ()) -> Int
forall a. StableName a -> Int
hashStableName StableName (Par ())
sn) (Sched -> Int
no Sched
schd)
Sched -> ROnly () -> IO ()
forall r (m :: * -> *) a. r -> ReaderT r m a -> m a
runReaderWith Sched
mysched (ROnly () -> IO ()) -> ROnly () -> IO ()
forall a b. (a -> b) -> a -> b
$
ContT () ROnly () -> (() -> ROnly ()) -> ROnly ()
forall {k} (r :: k) (m :: k -> *) a.
ContT r m a -> (a -> m r) -> m r
C.runContT (Par () -> ContT () ROnly ()
forall a. Par a -> ContT () ROnly a
unPar Par ()
task)
(\()
_ -> do
Bool -> ROnly () -> ROnly ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when Bool
dbg(ROnly () -> ROnly ()) -> ROnly () -> ROnly ()
forall a b. (a -> b) -> a -> b
$ do StableName (Par ())
sn <- IO (StableName (Par ())) -> ReaderT Sched IO (StableName (Par ()))
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO(IO (StableName (Par ()))
-> ReaderT Sched IO (StableName (Par ())))
-> IO (StableName (Par ()))
-> ReaderT Sched IO (StableName (Par ()))
forall a b. (a -> b) -> a -> b
$ Par () -> IO (StableName (Par ()))
forall a. a -> IO (StableName a)
makeStableName Par ()
task
IO () -> ROnly ()
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO(IO () -> ROnly ()) -> IO () -> ROnly ()
forall a b. (a -> b) -> a -> b
$ String -> Int -> Int -> Int -> IO ()
forall r. PrintfType r => String -> r
printf String
" [%d] | DONE running stolen work (unit %d) from %d\n" Int
my_no (StableName (Par ()) -> Int
forall a. StableName a -> Int
hashStableName StableName (Par ())
sn) (Sched -> Int
no Sched
schd)
() -> ROnly ()
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
triesInt -> Int -> Int
forall a. Num a => a -> a -> a
-Int
1) Int
i'
_errK :: t
_errK :: forall t. t
_errK = String -> t
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
() -> ROnly ()
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
Par b -> Par (IVar b)
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 <- Par (IVar a)
forall a. Par (IVar a)
new
IVar a -> a -> Par ()
forall a. IVar a -> a -> Par ()
put_ IVar a
v a
a
IVar a -> Par (IVar 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 = a -> Par (IVar a) -> Par (IVar a)
forall a b. NFData a => a -> b -> b
deepseq a
a (a -> Par (IVar 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 = a -> Par () -> Par ()
forall a b. NFData a => a -> b -> b
deepseq a
a (IVar a -> a -> Par ()
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 <- Par (IVar a)
forall a. Par (IVar a)
new; Par () -> Par ()
fork (Par a
p Par a -> (a -> Par ()) -> Par ()
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= IVar a -> a -> Par ()
forall a. NFData a => IVar a -> a -> Par ()
put IVar a
r); IVar a -> Par (IVar a)
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 <- Par (IVar a)
forall a. Par (IVar a)
new; Par () -> Par ()
fork (Par a
p Par a -> (a -> Par ()) -> Par ()
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= IVar a -> a -> Par ()
forall a. IVar a -> a -> Par ()
put_ IVar a
r); IVar a -> Par (IVar a)
forall (m :: * -> *) a. Monad m => a -> m a
return IVar a
r
spawnP :: forall a. NFData a => a -> Par (IVar a)
spawnP a
a = Par a -> Par (IVar a)
forall a. NFData a => Par a -> Par (IVar a)
spawn (a -> Par a
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 = IVar a -> Par a
forall a. IVar a -> Par a
get
spawn :: forall a. NFData a => Par a -> Par (IVar a)
spawn = Par a -> Par (IVar a)
forall a. NFData a => Par a -> Par (IVar a)
spawn
spawn_ :: forall a. Par a -> Par (IVar a)
spawn_ = Par a -> Par (IVar a)
forall a. Par a -> Par (IVar a)
spawn_
spawnP :: forall a. NFData a => a -> Par (IVar a)
spawnP = a -> Par (IVar a)
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 = Par (IVar a)
forall a. Par (IVar a)
new
put_ :: forall a. IVar a -> a -> Par ()
put_ = IVar a -> a -> Par ()
forall a. IVar a -> a -> Par ()
put_
newFull :: forall a. NFData a => a -> Par (IVar a)
newFull = a -> Par (IVar a)
forall a. NFData a => a -> Par (IVar a)
newFull
newFull_ :: forall a. a -> Par (IVar a)
newFull_ = a -> Par (IVar a)
forall a. a -> Par (IVar a)
newFull_
instance UN.ParUnsafe IVar Par where
unsafePeek :: forall a. IVar a -> Par (Maybe a)
unsafePeek = IVar a -> Par (Maybe a)
forall a. IVar a -> Par (Maybe a)
unsafePeek
unsafeTryPut :: forall b. IVar b -> b -> Par b
unsafeTryPut = IVar a -> a -> Par a
forall b. IVar b -> b -> Par b
unsafeTryPut
unsafeParIO :: forall a. IO a -> Par a
unsafeParIO = IO a -> Par a
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 = ReaderT r m a -> r -> m a
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
[Sched] -> (Sched -> IO ()) -> IO ()
forall (t :: * -> *) (m :: * -> *) a b.
(Foldable t, Monad m) =>
t a -> (a -> m b) -> m ()
forM_ [Sched]
allscheds ((Sched -> IO ()) -> IO ()) -> (Sched -> IO ()) -> IO ()
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 <- SimpleDeque (Par ()) -> IO Bool
forall elt. SimpleDeque elt -> IO Bool
R.nullQ SimpleDeque (Par ())
WSDeque (Par ())
workpool
Bool -> IO () -> IO ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when (Bool -> Bool
not Bool
b) (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
() <- String -> Int -> IO ()
forall r. PrintfType r => String -> r
printf String
"WARNING: After main thread exited non-empty queue remains for worker %d\n" Int
no
() -> IO ()
forall (m :: * -> *) a. Monad m => a -> m a
return ()
String -> IO ()
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 =
IO a -> (IOError -> IO a) -> IO a
forall e a. Exception e => IO a -> (e -> IO a) -> IO a
E.catch (MVar a -> IO a
forall a. MVar a -> IO a
takeMVar MVar a
mv) (\(IOError
_::IOError) -> IO a
doDebugStuff)
where
doDebugStuff :: IO a
doDebugStuff = do String -> String -> IO ()
forall r. PrintfType r => String -> r
printf String
"This takeMVar blocked indefinitely!: %s\n" String
msg
String -> IO a
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 (IO () -> IO ThreadId) -> IO () -> IO ThreadId
forall a b. (a -> b) -> a -> b
$
(BlockedIndefinitelyOnMVar -> IO ()) -> IO () -> IO ()
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 ()
putStrLn(String -> IO ()) -> String -> IO ()
forall a b. (a -> b) -> a -> b
$String
"CAUGHT child thread exception: "String -> String -> String
forall a. [a] -> [a] -> [a]
++BlockedIndefinitelyOnMVar -> String
forall a. Show a => a -> String
show BlockedIndefinitelyOnMVar
e
() -> IO ()
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 (IO () -> IO ThreadId) -> IO () -> IO ThreadId
forall a b. (a -> b) -> a -> b
$ do
ThreadId
tid <- IO ThreadId
myThreadId
IO () -> (SomeException -> IO ()) -> IO ()
forall e a. Exception e => IO a -> (e -> IO a) -> IO a
E.catch IO ()
action
(\ SomeException
e ->
case SomeException -> Maybe AsyncException
forall e. Exception e => SomeException -> Maybe e
E.fromException SomeException
e of
Just AsyncException
E.ThreadKilled -> String -> String -> String -> IO ()
forall r. PrintfType r => String -> r
printf
String
"\nThreadKilled exception inside child thread, %s (not propagating!): %s\n" (ThreadId -> String
forall a. Show a => a -> String
show ThreadId
tid) (String -> String
forall a. Show a => a -> String
show String
descr)
Maybe AsyncException
_ -> do String -> String -> String -> String -> IO ()
forall r. PrintfType r => String -> r
printf
String
"\nException inside child thread %s, %s: %s\n" (String -> String
forall a. Show a => a -> String
show String
descr) (ThreadId -> String
forall a. Show a => a -> String
show ThreadId
tid) (SomeException -> String
forall a. Show a => a -> String
show SomeException
e)
ThreadId -> SomeException -> IO ()
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 <- HotVar [Session] -> IO [Session]
forall a. HotVar a -> IO a
readIORef (Sched -> HotVar [Session]
sessions Sched
sched)
[Bool]
bools <- (Session -> IO Bool) -> [Session] -> IO [Bool]
forall (t :: * -> *) (m :: * -> *) a b.
(Traversable t, Monad m) =>
(a -> m b) -> t a -> m (t b)
mapM (\ (Session SessionID
_ HotVar Bool
r) -> HotVar Bool -> IO Bool
forall a. HotVar a -> IO a
readIORef HotVar Bool
r) [Session]
ls
[(SessionID, Bool)] -> IO [(SessionID, Bool)]
forall (m :: * -> *) a. Monad m => a -> m a
return ([SessionID] -> [Bool] -> [(SessionID, Bool)]
forall a b. [a] -> [b] -> [(a, b)]
zip ((Session -> SessionID) -> [Session] -> [SessionID]
forall a b. (a -> b) -> [a] -> [b]
map (\ (Session SessionID
sid HotVar Bool
_) -> SessionID
sid) [Session]
ls) [Bool]
bools)