{-# 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 = 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 :: IO a -> Par a
unsafeParIO iom :: 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 :: 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 _ = 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 _ _ = () -> IO ()
forall (m :: * -> *) a. Monad m => a -> m a
return ()
unregisterWorker :: ThreadId -> IO ()
unregisterWorker _tid :: 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
Nothing -> () -> IO ()
forall (m :: * -> *) a. Monad m => a -> m a
return ()
Just _ -> 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 " [%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 } task :: 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 " [%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 idle :: 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 "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 (\is :: [MVar Bool]
is -> case [MVar Bool]
is of
[] -> ([], () -> IO ()
forall (m :: * -> *) a. Monad m => a -> m a
return ())
(i :: MVar Bool
i:ils :: [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 ref :: HotVar GenIO
ref = (Int, Int) -> GenIO -> IO Int
forall a (m :: * -> *).
(Variate a, PrimMonad m) =>
(a, a) -> Gen (PrimState m) -> m a
Random.uniformR (0, Int
numCapabilitiesInt -> Int -> Int
forall a. Num a => a -> a -> a
-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 :: 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 :: String -> Sched -> Par b -> IO b
runNewSessionAndWait name :: String
name sched :: Sched
sched userComp :: 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) (\ x :: SessionID
x -> (SessionID
xSessionID -> SessionID -> SessionID
forall a. Num a => a -> a -> a
+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 :: 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
$ "Empty session-result ref ("String -> String -> String
forall a. [a] -> [a] -> [a]
++String
nameString -> String -> String
forall a. [a] -> [a] -> [a]
++") 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
forall a. [a] -> [a] -> [a]
++ThreadId -> String
forall a. Show a => a -> String
show ThreadId
tid String -> String -> String
forall a. [a] -> [a] -> [a]
++")")
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) (\ ls :: [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 " [%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
>=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 " [%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 :: 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 :: SessionID -> a -> ROnly ()
kont n :: 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
forall a. [a] -> [a] -> [a]
++String
nameString -> String -> String
forall a. [a] -> [a] -> [a]
++", 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]
++", 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]
++")"
loop :: Word64 -> ROnly ()
loop :: SessionID -> ROnly ()
loop n :: 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 " [%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 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
forall a. [a] -> [a] -> [a]
++String
nameString -> String -> String
forall a. [a] -> [a] -> [a]
++", 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]
++")"
SessionID -> ROnly ()
loop (SessionID
nSessionID -> SessionID -> SessionID
forall a. Num a => a -> a -> a
+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 0))
Sched -> ROnly () -> IO ()
forall r (m :: * -> *) a. r -> ReaderT r m a -> m a
runReaderWith Sched
sched (SessionID -> ROnly ()
loop 1)
Bool -> IO () -> IO ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when (Int
dbglvlInt -> Int -> Bool
forall a. Ord a => a -> a -> Bool
>=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 " [%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 sid2 :: SessionID
sid2 _ : tl :: [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
$ "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]
++") 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]
++")"
IORef b -> IO b
forall a. HotVar a -> IO a
readIORef IORef b
ref
{-# NOINLINE runParIO #-}
runParIO :: Par a -> IO a
runParIO userComp :: Par a
userComp = do
ThreadId
tid <- IO ThreadId
myThreadId
#if __GLASGOW_HASKELL__ >= 701 /* 20110301 */
(main_cpu :: Int
main_cpu, _) <- 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
>=1)(IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ String -> Int -> String -> SessionID -> IO ()
forall r. PrintfType r => String -> r
printf " [%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
+ 1)
String -> Sched -> Par a -> IO a
forall b. String -> Sched -> Par b -> IO b
runNewSessionAndWait "nested runPar" Sched
sched Par a
userComp
Nothing -> do
[Sched]
allscheds <- Int -> IO [Sched]
makeScheds Int
main_cpu
[Session _ topSessFlag :: 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 [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
$ \(cpu :: Int
cpu,sched :: Sched
sched) -> do
MVar Int
workerDone <- IO (MVar Int)
forall a. IO (MVar a)
newEmptyMVar
let wname :: String
wname = ("(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]
++" 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]
++")")
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 " [%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 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 " [%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 "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 " *** 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 " *** [%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
$ \ mv :: 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 " * [%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 " *** [%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 main :: 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 "[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
+ 1)
let allscheds :: [Sched]
allscheds = [ $WSched :: 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 <- [0 .. Int
numCapabilitiesInt -> Int -> Int
forall a. Num a => a -> a -> a
-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 = 1000
{-# INLINE new #-}
new :: Par (IVar a)
new :: 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 :: IVar a -> Par a
get (IVar vr :: 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
$ \kont :: 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 -> a -> Par a
forall (m :: * -> *) a. Monad m => a -> m a
return a
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
$ \x :: IVarContents a
x -> case IVarContents a
x of
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 -> (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 ks :: [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 :: IVar a -> Par (Maybe a)
unsafePeek (IVar v :: 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
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)
_ -> Maybe a -> Par (Maybe a)
forall (m :: * -> *) a. Monad m => a -> m a
return Maybe a
forall a. Maybe a
Nothing
{-# INLINE put_ #-}
put_ :: IVar a -> a -> Par ()
put_ (IVar vr :: 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
$ \e :: IVarContents a
e -> case IVarContents a
e of
Empty -> (a -> IVarContents a
forall a. a -> IVarContents a
Full a
content, [])
Full _ -> String -> (IVarContents a, [a -> IO ()])
forall a. HasCallStack => String -> a
error "multiple put"
Blocked ks :: [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 :: IVar b -> b -> Par b
unsafeTryPut (IVar vr :: IORef (IVarContents b)
vr) !b
content = do
Sched
sched <- Par Sched
forall r (m :: * -> *). MonadReader r m => m r
RD.ask
(ks :: [b -> IO ()]
ks,res :: 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
$ \e :: IVarContents b
e -> case IVarContents b
e of
Empty -> (b -> IVarContents b
forall a. a -> IVarContents a
Full b
content, ([], b
content))
Full x :: b
x -> (b -> IVarContents b
forall a. a -> IVarContents a
Full b
x, ([], b
x))
Blocked ks :: [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 :: Sched -> [a -> IO ()] -> a -> Par ()
wakeUp _sched :: Sched
_sched ks :: [a -> IO ()]
ks arg :: 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 (kont :: a -> IO ()
kont:rest :: [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 kont :: 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 kont :: a -> IO ()
kont (more :: a -> IO ()
more:rest :: [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 task :: Par ()
task =
case Bool
_FORKPARENT of
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
$ \parent :: () -> 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 " !!! 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 " - 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 ()
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 " [%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 :: 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 (\ _k :: a -> ROnly ()
_k -> SessionID -> (Any -> ROnly ()) -> ROnly ()
forall a. SessionID -> (a -> ROnly ()) -> ROnly ()
rescheduleR 0 (String -> Any -> ROnly ()
forall a. String -> a -> ROnly ()
trivialCont "longjmpSched"))
rescheduleR :: Word64 -> (a -> ROnly ()) -> ROnly ()
rescheduleR :: SessionID -> (a -> ROnly ()) -> ROnly ()
rescheduleR cnt :: SessionID
cnt kont :: 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 " [%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
Nothing -> do
(Session _ finRef :: HotVar Bool
finRef):_ <- 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
>= 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 " [%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 " [%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 "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
+1) a -> ROnly ()
kont
Just task :: 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 " [%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 fn :: (() -> 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 " + 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 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
>=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 " [%d %s] + stealing\n" Int
my_no (ThreadId -> String
forall a. Show a => a -> String
show ThreadId
tid)
Int
i <- Int -> IO Int
getnext (-1 :: Int)
Int -> Int -> IO ()
go Int
maxtries Int
i
where
maxtries :: Int
maxtries = 20 Int -> Int -> Int
forall a. Num a => a -> a -> a
* Int
numCapabilities
getnext :: Int -> IO Int
getnext _ = HotVar GenIO -> IO Int
rand HotVar GenIO
rng
go :: Int -> Int -> IO ()
go 0 _ | 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
$ \is :: [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
- 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 " [%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_ (\vr :: 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 _ finRef :: HotVar Bool
finRef):_ <- 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 " [%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 " [%d] | woken up\n" Int
my_no
Int
i <- Int -> IO Int
getnext (-1::Int)
Int -> Int -> IO ()
go Int
maxtries Int
i
go 0 _i :: Int
_i | Bool
_IDLING_ON Bool -> Bool -> Bool
forall a. Eq a => a -> a -> Bool
== Bool
False = IO ()
yield
go tries :: Int
tries i :: 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
-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
>=2)(IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ String -> Int -> Int -> IO ()
forall r. PrintfType r => String -> r
printf " [%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 task :: 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 " [%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 " [%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 ())
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
-1) Int
i'
_errK :: t
_errK :: t
_errK = String -> t
forall a. HasCallStack => String -> a
error "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 :: String -> a -> ROnly ()
trivialCont _str :: String
_str _ = do
#endif
() -> ROnly ()
forall (m :: * -> *) a. Monad m => a -> m a
return ()
{-# INLINE spawn1_ #-}
spawn1_ :: (a -> Par b) -> a -> Par (IVar b)
spawn1_ f :: a -> Par b
f x :: 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_ :: a -> Par (IVar a)
newFull_ a :: 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 :: a -> Par (IVar a)
newFull a :: 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 :: IVar a -> a -> Par ()
put v :: IVar a
v a :: 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 :: Par a -> Par (IVar a)
spawn p :: 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_ :: Par a -> Par (IVar a)
spawn_ p :: 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 :: a -> Par (IVar a)
spawnP a :: 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 :: IVar a -> Par a
get = IVar a -> Par a
forall a. IVar a -> Par a
get
spawn :: Par a -> Par (IVar a)
spawn = Par a -> Par (IVar a)
forall a. NFData a => Par a -> Par (IVar a)
spawn
spawn_ :: Par a -> Par (IVar a)
spawn_ = Par a -> Par (IVar a)
forall a. Par a -> Par (IVar a)
spawn_
spawnP :: 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 :: Par (IVar a)
new = Par (IVar a)
forall a. Par (IVar a)
new
put_ :: IVar a -> a -> Par ()
put_ = IVar a -> a -> Par ()
forall a. IVar a -> a -> Par ()
put_
newFull :: a -> Par (IVar a)
newFull = a -> Par (IVar a)
forall a. NFData a => a -> Par (IVar a)
newFull
newFull_ :: a -> Par (IVar a)
newFull_ = a -> Par (IVar a)
forall a. a -> Par (IVar a)
newFull_
instance UN.ParUnsafe IVar Par where
unsafePeek :: IVar a -> Par (Maybe a)
unsafePeek = IVar a -> Par (Maybe a)
forall a. IVar a -> Par (Maybe a)
unsafePeek
unsafeTryPut :: IVar a -> a -> Par a
unsafeTryPut = IVar a -> a -> Par a
forall a. IVar a -> a -> Par a
unsafeTryPut
unsafeParIO :: 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 :: r -> ReaderT r m a -> m a
runReaderWith state :: r
state m :: 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 allscheds :: [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 "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 "Sanity check complete.\n"
_dbgTakeMVar :: String -> MVar a -> IO a
_dbgTakeMVar :: String -> MVar a -> IO a
_dbgTakeMVar msg :: String
msg mv :: 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 "This takeMVar blocked indefinitely!: %s\n" String
msg
String -> IO a
forall a. HasCallStack => String -> a
error "failed"
_forkIO_Suppress :: Int -> IO () -> IO ThreadId
_forkIO_Suppress :: Int -> IO () -> IO ThreadId
_forkIO_Suppress whre :: Int
whre action :: 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 (\e :: BlockedIndefinitelyOnMVar
e ->
case (BlockedIndefinitelyOnMVar
e :: E.BlockedIndefinitelyOnMVar) of
_ -> do
String -> IO ()
putStrLn(String -> IO ()) -> String -> IO ()
forall a b. (a -> b) -> a -> b
$"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 forkit :: IO () -> IO ThreadId
forkit descr :: String
descr action :: 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
(\ e :: SomeException
e ->
case SomeException -> Maybe AsyncException
forall e. Exception e => SomeException -> Maybe e
E.fromException SomeException
e of
Just E.ThreadKilled -> String -> String -> String -> IO ()
forall r. PrintfType r => String -> r
printf
"\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)
_ -> do String -> String -> String -> String -> IO ()
forall r. PrintfType r => String -> r
printf
"\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
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 _ r :: 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 sid :: SessionID
sid _) -> SessionID
sid) [Session]
ls) [Bool]
bools)