{-# LANGUAGE RankNTypes, NamedFieldPuns, BangPatterns,
             ExistentialQuantification, CPP, ScopedTypeVariables,
             TypeSynonymInstances, MultiParamTypeClasses,
             GeneralizedNewtypeDeriving, PackageImports,
             ParallelListComp #-}


{- OPTIONS_GHC -Wall -fno-warn-name-shadowing -fno-warn-unused-do-bind -}

-- {- LANGUAGE Trustworthy -}
-- TODO: Before declaring this module TRUSTWORTHY/SAFE, we need to
-- make the IVar type abstract.

{-# LANGUAGE TypeFamilies #-}

-- | A scheduler for the Par monad based on directly performing IO
-- actions when Par methods are called (i.e. without using a lazy
-- trace data structure).

module Control.Monad.Par.Scheds.Direct (
   Sched(..),
   Par, -- abstract: Constructor not exported.
   IVar(..), IVarContents(..),
--    sched,
    runPar, runParIO,
    new, get, put_, fork,
    newFull, newFull_, put,
    spawn, spawn_, spawnP,
    spawn1_, fixPar, FixParException (..)
--   runParAsync, runParAsyncHelper,
--   yield,
 ) where

import Control.Applicative
import Control.Concurrent hiding (yield)
import Data.IORef         (IORef,newIORef,readIORef,writeIORef,atomicModifyIORef)
import Text.Printf        (printf)
import GHC.Conc           (numCapabilities,yield)
import Control.Monad
import Control.Monad.IO.Class
import Control.Monad.Trans
import           "mtl" Control.Monad.Cont as C
import qualified "mtl" Control.Monad.Reader as RD
import qualified       System.Random.MWC as Random
import                 System.IO.Unsafe (unsafePerformIO)
import                 System.Mem.StableName (makeStableName, hashStableName)
import qualified       Control.Monad.Par.Class  as PC
import qualified       Control.Monad.Par.Unsafe as UN
import                 Control.Monad.Par.Scheds.DirectInternal
                       (Par(..), Sched(..), HotVar, SessionID, Session(Session),
                        newHotVar, readHotVar, modifyHotVar, modifyHotVar_,
                        writeHotVarRaw, fixPar, FixParException (..))
#ifdef NEW_GENERIC
import qualified       Control.Par.Class as PN
import qualified       Control.Par.Class.Unsafe as PU
#endif
import Control.DeepSeq
#ifdef NESTED_SCHEDS
import qualified Data.Map as M
#endif
import qualified Data.Set as S
import Data.Maybe (catMaybes)
import Data.Word (Word64)

-- import Data.Concurrent.Deque.Class (WSDeque)
#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

--------------------------------------------------------------------------------
-- Configuration Toggles
--------------------------------------------------------------------------------

-- [2012.08.30] This shows a 10X improvement on nested parfib:
-- #define NESTED_SCHEDS
#define PARPUTS
#define FORKPARENT
#define IDLING_ON
   -- Next, IF idling is on, should we do wakeups?:
#define WAKEIDLE

-- #define WAIT_FOR_WORKERS

-------------------------------------------------------------------
-- Ifdefs for the above preprocessor defines.  Try to MINIMIZE code
-- that lives in this dangerous region, and instead do normal
-- conditionals and trust dead-code-elimination.
--------------------------------------------------------------------

#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



--------------------------------------------------------------------------------
-- Core type definitions
--------------------------------------------------------------------------------

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 () (ReaderT Sched IO) a -> Par a
forall a. ContT () (ReaderT Sched IO) a -> Par a
Par (ROnly a -> ContT () (ReaderT Sched IO) a
forall (m :: * -> *) a. Monad m => m a -> ContT () m a
forall (t :: (* -> *) -> * -> *) (m :: * -> *) a.
(MonadTrans t, Monad m) =>
m a -> t m a
lift(ROnly a -> ContT () (ReaderT Sched IO) a)
-> ROnly a -> ContT () (ReaderT Sched IO) a
forall a b. (a -> b) -> a -> b
$ IO a -> ROnly a
forall (m :: * -> *) a. Monad m => m a -> ReaderT Sched m 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 -- shorthand used below

--------------------------------------------------------------------------------
-- Global State
--------------------------------------------------------------------------------

-- This keeps track of ALL worker threads across all unreated
-- `runPar` instantiations.  This is used to detect nested invocations
-- of `runPar` and avoid reinitialization.
-- globalWorkerPool :: IORef (Data.IntMap ())
#ifdef NESTED_SCHEDS
globalWorkerPool :: IORef (M.Map ThreadId Sched)
globalWorkerPool = unsafePerformIO $ newIORef M.empty
#endif
-- TODO! Make this semi-local! (not shared between "top-level" runPars)

{-# INLINE amINested #-}
{-# INLINE registerWorker #-}
{-# INLINE unregisterWorker #-}
amINested :: ThreadId -> IO (Maybe Sched)
registerWorker :: ThreadId -> Sched -> IO ()
unregisterWorker :: ThreadId -> IO ()
#ifdef NESTED_SCHEDS
-- | If the current threadID is ALREADY a worker, return the corresponding Sched structure.
amINested tid = do
  -- There is no race here.  Each thread inserts itself before it
  -- becomes an active worker.
  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 a. a -> IO a
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 a. a -> IO a
forall (m :: * -> *) a. Monad m => a -> m a
return ()
unregisterWorker :: ThreadId -> IO ()
unregisterWorker ThreadId
_tid = () -> IO ()
forall a. a -> IO a
forall (m :: * -> *) a. Monad m => a -> m a
return ()
#endif

-----------------------------------------------------------------------------
-- Helpers #2:  Pushing and popping work.
-----------------------------------------------------------------------------

{-# INLINE popWork  #-}
popWork :: Sched -> IO (Maybe (Par ()))
popWork :: Sched -> IO (Maybe (Par ()))
popWork Sched{ WSDeque (Par ())
workpool :: WSDeque (Par ())
workpool :: Sched -> WSDeque (Par ())
workpool, Int
no :: Int
no :: Sched -> Int
no } = do
  Maybe (Par ())
mb <- SimpleDeque (Par ()) -> IO (Maybe (Par ()))
forall a. SimpleDeque a -> IO (Maybe a)
R.tryPopL WSDeque (Par ())
SimpleDeque (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 a. a -> IO a
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 a. a -> IO a
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 :: Sched -> WSDeque (Par ())
workpool :: WSDeque (Par ())
workpool, HotVar [MVar Bool]
idle :: HotVar [MVar Bool]
idle :: Sched -> HotVar [MVar Bool]
idle, Int
no :: Sched -> Int
no :: Int
no } Par ()
task = do
  SimpleDeque (Par ()) -> Par () -> IO ()
forall t. SimpleDeque t -> t -> IO ()
R.pushL WSDeque (Par ())
SimpleDeque (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)
  --when isMain$    -- Experimenting with reducing contention by doing this only from a single thread.
                    -- TODO: We need to have a proper binary wakeup-tree.
  HotVar [MVar Bool] -> IO ()
tryWakeIdle HotVar [MVar Bool]
idle
#endif
  () -> IO ()
forall a. a -> IO a
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
-- NOTE: I worry about having the idle var hammered by all threads on their spawn-path:
  -- If any worker is idle, wake one up and give it work to do.
  [MVar Bool]
idles <- HotVar [MVar Bool] -> IO [MVar Bool]
forall a. HotVar a -> IO a
readHotVar HotVar [MVar Bool]
idle -- Optimistically do a normal read first.
  Bool -> IO () -> IO ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when (Bool -> Bool
not ([MVar Bool] -> Bool
forall a. [a] -> 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 a. [a] -> 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 a. a -> IO a
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 -- wake an idle worker up by putting an MVar.

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
forall (m :: * -> *).
PrimMonad m =>
(Int, Int) -> Gen (PrimState m) -> m Int
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

--------------------------------------------------------------------------------
-- Running computations in the Par monad
--------------------------------------------------------------------------------

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


-- | This procedure creates a new worker on the current thread (with a
--   new session ID) and plugs it into the work-stealing environment.
--   This new worker extracts itself from the work stealing pool when
--   `userComp` has completed, thus freeing the current thread (this
--   procedure) to return normally.
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 -- TODO: remove when done debugging
    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, ()))

    -- Here we have an extra IORef... ugly.
    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
    -- Push the new session:
    ()
_ <- 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
                       -- This add-on to userComp will run only after userComp has completed successfully,
                       -- but that does NOT guarantee that userComp-forked computations have terminated:
                       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 a. IO a -> ReaderT Sched IO a
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 a. IO a -> ReaderT Sched IO a
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)

    -- THIS IS RETURNING TOO EARLY!!:
    Sched -> ROnly () -> IO ()
forall r (m :: * -> *) a. r -> ReaderT r m a -> m a
runReaderWith Sched
sched (ContT () (ReaderT Sched IO) () -> (() -> ROnly ()) -> ROnly ()
forall {k} (r :: k) (m :: k -> *) a.
ContT r m a -> (a -> m r) -> m r
C.runContT (Par () -> ContT () (ReaderT Sched IO) ()
forall a. Par a -> ContT () (ReaderT Sched IO) a
unPar Par ()
userComp') (SessionID -> () -> ROnly ()
forall a. SessionID -> a -> ROnly ()
kont SessionID
0))  -- Does this ASSUME child stealing?
    Sched -> ROnly () -> IO ()
forall r (m :: * -> *) a. r -> ReaderT r m a -> m a
runReaderWith Sched
sched (SessionID -> ROnly ()
loop SessionID
1)

    -- TODO: Ideally we would wait for ALL outstanding (stolen) work on this "team" to complete.

    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 -- ASSERT!
      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)

    -- Here we pop off the frame we added to the session stack:
    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
")"

    -- By returning here we ARE implicitly reengaging the scheduler, since we
    -- are already inside the rescheduleR loop on this thread
    -- (before runParIO was called in a nested fashion).
    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 */
    --
    -- We create a thread on each CPU with forkOn.  The CPU on which
    -- the current thread is running will host the main thread; the
    -- other CPUs will host worker threads.
    --
    -- Note: GHC 7.1.20110301 is required for this to work, because that
    -- is when threadCapability was added.
    --
   (Int
main_cpu, Bool
_) <- ThreadId -> IO (Int, Bool)
threadCapability ThreadId
tid
#else
    --
    -- Lacking threadCapability, we always pick CPU #0 to run the main
    -- thread.  If the current thread is not running on CPU #0, this
    -- will require some data to be shipped over the memory bus, and
    -- hence will be slightly slower than the version above.
    --
   let main_cpu = 0
#endif
   Maybe Sched
maybSched <- ThreadId -> IO (Maybe Sched)
amINested ThreadId
tid
   ThreadId
tidorig <- IO ThreadId
myThreadId -- TODO: remove when done debugging
   case Maybe Sched
maybSched of
     Just (Sched
sched) -> do
       -- Here the current thread is ALREADY a worker.  All we need to
       -- do is plug the users new computation in.

       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

     ------------------------------------------------------------
     -- Non-nested case, make a new set of worker threads:
     ------------------------------------------------------------
     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. HasCallStack => [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
")")
--            forkOn cpu $ do
            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
            ------------------------------------------------------------STRT WORKER THREAD
              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 a. a -> IO a
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
                         -- When the main worker finishes we can tell the anonymous "system" workers:
                         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)
                         --  sanityCheck allscheds
                         MVar a -> a -> IO ()
forall a. MVar a -> a -> IO ()
putMVar MVar a
mfin a
x

              ThreadId -> IO ()
unregisterWorker ThreadId
tid
            ------------------------------------------------------------END WORKER THREAD
            Maybe (MVar Int) -> IO (Maybe (MVar Int))
forall a. a -> IO a
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
    --         n <- A.wait 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)
       -- We don't directly use the thread we come in on.  Rather, that thread waits
       -- waits.  One reason for this is that the main/progenitor thread in
       -- GHC is expensive like a forkOS thread.
       ----------------------------------------
       --              DEBUGGING             --
#ifdef DEBUG_DIRECT
       busyTakeMVar (" The global wait "++ show tidorig) mfin -- Final value.
--       dbgTakeMVar "global waiting thread" mfin -- Final value.
#else
       MVar a -> IO a
forall a. MVar a -> IO a
takeMVar MVar a
mfin -- Final value.
#endif
       ----------------------------------------

-- Create the default scheduler(s) state:
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)
IO GenIO
forall (m :: * -> *). PrimMonad m => m (Gen (PrimState m))
Random.create IO (Gen RealWorld)
-> (Gen RealWorld -> IO (HotVar (Gen RealWorld)))
-> IO (HotVar (Gen RealWorld))
forall a b. IO a -> (a -> IO b) -> IO b
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 []
   -- The STACKs are per-worker.. but the root finished flag is shared between all anonymous system workers:
   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)
forall (m :: * -> *) a b. Monad m => (a -> m b) -> [a] -> m [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 { 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=WSDeque (Par ())
SimpleDeque (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
                           }
                   --  | (x,wp,rng,stck) <- zip4 [0..] workpools rngs sessionStacks
                   | 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 a. a -> IO a
forall (m :: * -> *) a. Monad m => a -> m a
return [Sched]
allscheds


-- The ID of top-level runPar sessions.
baseSessionID :: SessionID
baseSessionID :: SessionID
baseSessionID = SessionID
1000


--------------------------------------------------------------------------------
-- IVar operations
--------------------------------------------------------------------------------

{-# INLINE new  #-}
-- | Creates a new @IVar@
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 a. a -> IO 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  #-}
-- | Read the value in an @IVar@.  The 'get' operation can only return when the
-- value has been written by a prior or parallel @put@ to the same
-- @IVar@.
get :: forall a. IVar a -> Par a
get (IVar IORef (IVarContents a)
vr) =  do
  ((a -> Par ()) -> Par a) -> Par a
forall a b. ((a -> Par b) -> 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 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  -- Should probably do the MutVar inside...
            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 -- Invariant: kont must not be lost.
            -- Because we continue on the same processor the Sched stays the same:
            -- TODO: Try NOT using monadic values as first class.  Check for performance effect:
            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 a. a -> Par a
forall (m :: * -> *) a. Monad m => a -> m a
return a
a) -- kont is implicit here.
                      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

-- | NOTE unsafePeek is NOT exposed directly through this module.  (So
-- this module remains SAFE in the Safe Haskell sense.)  It can only
-- be accessed by importing Control.Monad.Par.Unsafe.
{-# 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 a. a -> Par 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 a. a -> Par a
forall (m :: * -> *) a. Monad m => a -> m a
return Maybe a
forall a. Maybe a
Nothing

------------------------------------------------------------
{-# INLINE put_ #-}
-- | @put_@ is a version of @put@ that is head-strict rather than fully-strict.
--   In this scheduler, puts immediately execute woken work in the current thread.
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 a. a -> IO a
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

-- | NOTE unsafeTryPut is NOT exposed directly through this module.  (So
-- this module remains SAFE in the Safe Haskell sense.)  It can only
-- be accessed by importing Control.Monad.Par.Unsafe.
{-# INLINE unsafeTryPut #-}
unsafeTryPut :: forall b. IVar b -> b -> Par b
unsafeTryPut (IVar IORef (IVarContents b)
vr) !b
content = do
   -- Head strict rather than fully strict.
   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 a. a -> IO a
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 a. a -> Par a
forall (m :: * -> *) a. Monad m => a -> m a
return b
res

-- | When an IVar is filled in, continuations wake up.
{-# 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 a. a -> Par a
forall (m :: * -> *) a. Monad m => a -> m a
return ()
   loop (a -> IO ()
kont:[a -> IO ()]
rest) = do
     -- FIXME -- without strict firewalls keeping ivars from moving
     -- between runPar sessions, if we allow nested scheduler use
     -- we could potentially wake up work belonging to a different
     -- runPar and thus bring it into our worker and delay our own
     -- continuation until its completion.
     if Bool
_PARPUTS then
       -- We do NOT force the putting thread to postpone its continuation.
       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 a. a -> Par a
forall (m :: * -> *) a. Monad m => a -> m a
return ()
       -- case rest of
       --   [] -> spawn_$ io$ kont arg
       --   _  -> spawn_$ do spawn_$ io$ kont arg
       --                    io$ parchain rest
       -- error$"FINISHME - wake "++show (length ks)++" conts"
      else
       -- This version sacrifices a parallelism opportunity and
       -- imposes additional serialization.
       --
       -- [2012.08.31] WARNING -- this serialzation CAN cause deadlock.
       -- This "optimization" should not be on the table.
       -- mapM_ ($arg) ks
       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 a. a -> Par a
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

   -- parchain [kont] = kont arg
   -- parchain (kont:rest) = do spawn$ io$ kont arg
   --                           parchain rest


------------------------------------------------------------
{-# INLINE fork #-}
fork :: Par () -> Par ()
fork :: Par () -> Par ()
fork Par ()
task =
  -- Forking the "parent" means offering up the continuation of the
  -- fork rather than the task argument for stealing:
  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 a b. ((a -> Par b) -> Par a) -> Par a
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
         -- Then execute the child task and return to the scheduler when it is complete:
         Par ()
task
         -- If we get to this point we have finished the child task:
         Any
_ <- Par Any
forall a. Par a
longjmpSched -- We reschedule to pop the cont we pushed.
         -- TODO... OPTIMIZATION: we could also try the pop directly, and if it succeeds return normally....
         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 a. a -> Par a
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

-- This routine "longjmp"s to the scheduler, throwing out its own continuation.
longjmpSched :: Par a
-- longjmpSched = Par $ C.ContT rescheduleR
longjmpSched :: forall a. Par a
longjmpSched = ContT () (ReaderT Sched IO) a -> Par a
forall a. ContT () (ReaderT Sched IO) a -> Par a
Par (ContT () (ReaderT Sched IO) a -> Par a)
-> ContT () (ReaderT Sched IO) a -> Par a
forall a b. (a -> b) -> a -> b
$ ((a -> ROnly ()) -> ROnly ()) -> ContT () (ReaderT Sched IO) 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"))

-- Reschedule the scheduler loop until it observes sessionFinished==True, and
-- then it finally invokes its continuation.
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 a. IO a -> ReaderT Sched IO a
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 a. IO a -> ReaderT Sched IO a
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 a. IO a -> ReaderT Sched IO a
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 a. IO a -> ReaderT Sched IO a
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 a. IO a -> ReaderT Sched IO a
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
                     -- when (dbglvl >= 1) $ liftIO $ do
                     --     tid <- myThreadId
                     --     sess <- readSessions mysched
                     --     printf " [%d %s]  -    Apparently NOT finished with head session... trying to steal, all sessions %s\n"
                     --            (no mysched) (show tid) (show sess)
                     IO () -> ROnly ()
forall a. IO a -> ReaderT Sched IO a
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
--                     io$ tryWakeIdle (idle mysched)
#endif
                     IO () -> ROnly ()
forall a. IO a -> ReaderT Sched IO a
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
       -- When popping work from our own queue the Sched (Reader value) stays the same:
       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 a. IO a -> ReaderT Sched IO a
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 a. IO a -> ReaderT Sched IO a
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 () (ReaderT Sched IO) ()
forall a. Par a -> ContT () (ReaderT Sched IO) a
unPar Par ()
task
       -- Run the stolen task with a continuation that returns to the scheduler if the task exits normally:
       (() -> 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 a. IO a -> ReaderT Sched IO a
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)


-- | Attempt to steal work or, failing that, give up and go idle.
--
--   The current policy is to do a burst of of N tries without
--   yielding or pausing in between.
steal :: Sched -> IO ()
steal :: Sched -> IO ()
steal mysched :: Sched
mysched@Sched{ HotVar [MVar Bool]
idle :: Sched -> HotVar [MVar Bool]
idle :: HotVar [MVar Bool]
idle, [Sched]
scheds :: Sched -> [Sched]
scheds :: [Sched]
scheds, HotVar GenIO
rng :: Sched -> HotVar GenIO
rng :: 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 = numCapabilities -- How many times should we attempt theft before going idle?
    maxtries :: Int
maxtries = Int
20 Int -> Int -> Int
forall a. Num a => a -> a -> a
* Int
numCapabilities -- How many times should we attempt theft before going idle?

    getnext :: Int -> IO Int
getnext Int
_ = HotVar GenIO -> IO Int
rand HotVar GenIO
rng

    ----------------------------------------
    -- IDLING behavior:
    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 a. [a] -> 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 a. a -> IO a
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 a. a -> IO a
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

    -- We need to return from this loop to check sessionFinished and exit the scheduler if necessary.
    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
         -- We ONLY go through the global sched array to access victims:
         let schd :: Sched
schd = [Sched]
scheds[Sched] -> Int -> Sched
forall a. HasCallStack => [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 = workpool schd :: WSDeque (Par ())
         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 WSDeque (Par ())
SimpleDeque (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 () (ReaderT Sched IO) () -> (() -> ROnly ()) -> ROnly ()
forall {k} (r :: k) (m :: k -> *) a.
ContT r m a -> (a -> m r) -> m r
C.runContT (Par () -> ContT () (ReaderT Sched IO) ()
forall a. Par a -> ContT () (ReaderT Sched IO) 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 a. IO a -> ReaderT Sched IO a
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 a. IO a -> ReaderT Sched IO a
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 a. a -> ReaderT Sched IO a
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'

-- | The continuation which should not be called.
_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
--                trace (str ++" trivialCont evaluated!")
                liftIO$ printf " !! trivialCont evaluated, msg: %s\n" str
#else
trivialCont :: forall a. String -> a -> ROnly ()
trivialCont String
_str a
_ = do
#endif
                () -> ROnly ()
forall a. a -> ReaderT Sched IO a
forall (m :: * -> *) a. Monad m => a -> m a
return ()

----------------------------------------------------------------------------------------------------


--------------------------------------------------------------------------------
-- <boilerplate>

-- TEMP: TODO: Factor out this boilerplate somehow.

{-# INLINE spawn1_ #-}
-- Spawn a one argument function instead of a thunk.  This is good for debugging if the value supports "Show".
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)

-- The following is usually inefficient!
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 a. a -> Par 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 a b. Par a -> (a -> Par b) -> Par b
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 a. a -> Par 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 a b. Par a -> (a -> Par b) -> Par b
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 a. a -> Par 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 a. a -> Par a
forall (m :: * -> *) a. Monad m => a -> m a
return a
a)

-- In Debug mode we require that IVar contents be Show-able:
#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

-- We can't make proper instances with the extra Show constraints:
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

-- </boilerplate>
--------------------------------------------------------------------------------


{-# INLINE runReaderWith #-}
-- | Arguments flipped for convenience.
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


--------------------------------------------------------------------------------
-- DEBUGGING TOOLs
--------------------------------------------------------------------------------

-- Make sure there is no work left in any deque after exiting.
_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 :: Sched -> Int
no :: Int
no, WSDeque (Par ())
workpool :: Sched -> WSDeque (Par ())
workpool :: WSDeque (Par ())
workpool} -> do
     Bool
b <- SimpleDeque (Par ()) -> IO Bool
forall elt. SimpleDeque elt -> IO Bool
R.nullQ WSDeque (Par ())
SimpleDeque (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 a. a -> IO a
forall (m :: * -> *) a. Monad m => a -> m a
return ()
  String -> IO ()
forall r. PrintfType r => String -> r
printf String
"Sanity check complete.\n"


-- | This tries to localize the blocked-indefinitely exception:
_dbgTakeMVar :: String -> MVar a -> IO a
_dbgTakeMVar :: forall a. String -> MVar a -> IO a
_dbgTakeMVar String
msg MVar a
mv =
--  catch (takeMVar mv) ((\_ -> doDebugStuff) :: BlockedIndefinitelyOnMVar -> IO a)
  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"

-- | For debugging purposes.  This can help us figure out (by an ugly
--   process of elimination) which MVar reads are leading to a "Thread
--   blocked indefinitely" exception.
{-
busyTakeMVar :: String -> MVar a -> IO a
busyTakeMVar msg mv = try (10 * 1000 * 1000)
 where
 try 0 = do
   when dbg $ do
     tid <- myThreadId
     -- After we've failed enough times, start complaining:
     printf "%s not getting anywhere, msg: %s\n" (show tid) msg
   try (100 * 1000)
 try n = do
   x <- tryTakeMVar mv
   case x of
     Just y  -> return y
     Nothing -> do yield; try (n-1)
-}

-- | Fork a thread but ALSO set up an error handler that suppresses
--   MVar exceptions.
_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 a. a -> IO a
forall (m :: * -> *) a. Monad m => a -> m a
return ()
                    )
           IO ()
action


-- | Exceptions that walk up the fork tree of threads:
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)
         )


-- Do all the memory reads to snapshot the current session stack:
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)
forall (m :: * -> *) a b. Monad m => (a -> m b) -> [a] -> m [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 a. a -> IO a
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)