{-# LANGUAGE FlexibleInstances
, BangPatterns
, MagicHash
, ScopedTypeVariables
, DeriveDataTypeable
, MultiParamTypeClasses
#-}
-- State monad transformer is needed for both step & graph:
#ifndef MODNAME
#define MODNAME Intel.Cnc8
#endif
#define CNC_SCHEDULER 8
#define STEPLIFT S.lift$
#define GRAPHLIFT id$
#define SUPPRESS_put
#define SUPPRESS_newItemCol
#define SUPPRESS_initialize
#define SUPPRESS_itemsToList
#include "Cnc.Header.hs"
--------------------------------------------------------------------
-- Version 8: This also uses the GHC scheduler directly (like 3) but
-- it uses sparks rather than forkIO.
--------------------------------------------------------------------
-- Note, the spark pool is lossy and can't be counted on. (It will
-- happily discard sparks when it overflows. In the future it may not
-- even serve as GC roots.)
-- Therefore this version does a litle book-keeping it both sparks a
-- step, and adds the step to a list so that after each step is
-- completed it can "sync" on its children. This this scheduler
-- behaves very much like a Cilk version of CnC.
-- Like Concurrent Collectins for C++, this version uses exceptions to
-- escape a step's execution upon a failed get. An alternative is to
-- use the ContT monad transformer.
type TagCol a = (IORef (Set a), IORef [Step a])
--type ItemCol a b = MutableMap a b
-- Here the hidden state keeps track of
--newtype StepCode a = StepCode (S.StateT HiddenState8 IO a)
type StepCode a = (S.StateT (HiddenState8) IO a)
type GraphCode = IO
-- The hidden stat stores two things:
-- (1) "Self": the current action, if needed for requeueing.
-- (2) A list of child tasks/thunks that were spawned in parallel.
newtype HiddenState8 = HiddenState8 (StepCode (), [()])
-- In this version we don't use MVars because gets don't block:
newtype ItemCol a b = ItemCol (IORef (Map a ((Maybe b), WaitingSteps)))
type WaitingSteps = [StepCode ()]
data EscapeStep = EscapeStep deriving (Show, Typeable)
instance Exception EscapeStep
--instance GHC.Exception.Exception EscapeStep
--------------------------------------------------------------------------------
-- Misc utility functions used by the version 8 API functions:
--------------------------------------------------------------------------------
liftHidden fn = (\ (HiddenState8 (self,ls)) -> HiddenState8 (self, fn ls))
atomicModifyIORef_ ref fn = atomicModifyIORef ref (\x -> (fn x, ()))
stepStats :: StepCode ()
stepStats =
do
tid <- S.lift myThreadId
HiddenState8 (_, ls) <- S.get
S.lift$ putStrLn (">>> Step state: list len: "++ show (length ls))
-- This is the high level interface for running several steps in
-- parallel and then blocking on the result.
launch_steps :: [StepCode ()] -> StepCode ()
launch_steps mls =
foldM (\ () m -> spawn (do try_stepcode m m; return ()))
() mls
-- This consumes the state thats threaded through step code by capping
-- the end of the step with a sync. It needs a retry action to tuck
-- into the state so that the step can store it if it needs to escape
-- with an exception.
--
-- DESIGN DECISION:
try_stepcode :: StepCode () -> StepCode a -> IO (Maybe a)
try_stepcode retry m = wrapped
where
-- If data is not ready yet, fizzle:
wrapped = do x <- try sync_action
case x of Left EscapeStep -> return Nothing
Right v -> return (Just v)
-- This is a Cilk-like sync. Run the action to accumulate the list of
-- spawned children-actions. Here we serially execute those children
-- if they haven't been done in parallel.
sync_action =
do -- First RUN the step code:
(v, HiddenState8 (_, ls)) <- S.runStateT m (HiddenState8 (retry,[]))
tid <- myThreadId
-- Second, sync all child computations that the step created.
-- We may be racing to fill these in with other threads.
#ifdef DEBUG_HASKELL_CNC
putStrLn (">>> "++show tid ++": Syncing children")
#endif
--return (foldr pseq v ls)
--return v
foldr pseq (return v) ls
-- Release an IO action for parallel execution, and squirrel it away
-- so we can sync as well.
spawn :: IO () -> StepCode ()
spawn ioaction =
do -- Add the new action to the list of actions for this step.
--let thunk = unsafeDupablePerformIO ioaction
let thunk = unsafePerformIO ioaction
#ifdef DEBUG_HASKELL_CNC
--let wrapped = unsafeDupablePerformIO$
let wrapped = unsafePerformIO$
do { tid <- myThreadId; putStrLn ("\n>>> "++show tid++": STOLE WORK!\n"); pseq thunk (return ()) }
let parthunk = wrapped
#else
let parthunk = thunk
#endif
--S.modify $ liftHidden (parthunk:)
S.modify $ liftHidden (thunk:)
id <- S.lift$ myThreadId
#ifdef DEBUG_HASKELL_CNC
S.lift$ putStrLn $ ">>> "++ show id ++ ": Spawning..."
stepStats
#endif
--return (parthunk `par` ())
parthunk `par` (do
#ifdef DEBUG_HASKELL_CNC
mid <- S.lift$ myThreadId
S.lift$putStrLn (">>> "++show mid++" (spawned parallel)")
#endif
return ())
--------------------------------------------------------------------------------
-- The core of the version 8 implementation:
--------------------------------------------------------------------------------
newItemCol = do ref <- newIORef Map.empty
return (ItemCol ref)
putt = proto_putt
(\ steps tag ->
-- Spark each downstream step, attempting to do it in parallel before a
-- subsequent sync (at the end of the containing step).
launch_steps (Prelude.map (\step -> step tag) steps))
get (ItemCol icol) tag =
do map <- S.lift$ readIORef icol
case Map.lookup tag map of
Nothing -> addquit []
Just (Nothing, waiting) -> addquit waiting
Just (Just v, []) -> return v
Just (Just v, a:b) -> error "CnC: internal invariant violated"
where
addquit ls =
do (HiddenState8 (retry ,_)) <- S.get
S.lift$ atomicModifyIORef_ icol (Map.insert tag (Nothing, retry:ls))
-- After adding ourself to the wait list, jump out of this step:
throw EscapeStep
initfin :: String -> StepCode a -> GraphCode a
initfin str m = do let err = error str
x <- try_stepcode err m
case x of Nothing -> err
Just v -> return v
initialize = initfin "Get failed within initialize action!"
finalize = initfin "Get failed within finalize action!"
-- Put must replay any steps that are waiting.
put (ItemCol icol) tag (!item) =
do waiting <- S.lift$ atomicModifyIORef icol mod
launch_steps waiting
return ()
where
mod map =
let new = (Just item, [])
f key _ (Nothing, _) = new
#ifdef REPEAT_PUT_ALLOWED
f key _ old@(Just v, ls) = old
#else
f key _ (Just v, _) = error ("Single assignment violated at tag: "++ show tag)
#endif
(old, map') = Map.insertLookupWithKey f tag new map
in case old of
Nothing -> (map', [])
Just (Nothing, waiting) -> (map', waiting)
#ifdef REPEAT_PUT_ALLOWED
Just (Just _, waiting) -> (map , waiting)
#else
Just (Just _, _) -> error ("Single assignment violated at tag: "++ show tag)
#endif
itemsToList (ItemCol icol) =
do if not quiescence_support
then error "need to use a scheduler with quiescence support for itemsToList"
else return ()
map <- S.lift$ readIORef icol
return $ Prelude.map (\ (key, (Just v, _)) -> (key,v))
$ Prelude.filter fil
$ (Map.toList map)
where
fil (key, (Nothing, _)) = False
fil _ = True
quiescence_support=True ;
--------------------------------------------------------------------------------
-- Version 9
-- TODO??? Get COULD explicitly capture the continuation to avoid replay from the beginning.
-- Combining continuation monad with IO:
-- import Control.Monad.Cont
-- import System.IO
-- main = do
-- hSetBuffering stdout NoBuffering
-- runContT (callCC askString) reportResult
-- askString :: (String -> ContT () IO String) -> ContT () IO String
-- askString next = do
-- liftIO $ putStrLn "Please enter a string"
-- s <- liftIO $ getLine
-- next s
-- reportResult :: String -> IO ()
-- reportResult s = do
-- putStrLn ("You entered: " ++ s)