diff --git a/Intel/Cnc5.hs b/Intel/Cnc5.hs --- a/Intel/Cnc5.hs +++ b/Intel/Cnc5.hs @@ -34,11 +34,12 @@ -- At finalize time we set up the workers and run them. finalize finalAction = - do joiner <- S.lift$ newChan + do (HiddenState5 (stack, _, _, _)) <- S.get + joiner <- GRAPHLIFT newChan let worker = - do x <- S.lift$ tryPop global_stack + do x <- STEPLIFT tryPop stack case x of - Nothing -> S.lift$ writeChan joiner () + Nothing -> STEPLIFT writeChan joiner () Just action -> do action worker ver5_6_core_finalize joiner finalAction worker diff --git a/Intel/Cnc6.hs b/Intel/Cnc6.hs --- a/Intel/Cnc6.hs +++ b/Intel/Cnc6.hs @@ -33,22 +33,25 @@ -- Then at finalize time we set up the workers and run them. finalize finalAction = do joiner <- GRAPHLIFT newChan + (HiddenState5 (stack, _, _, mortalthreads)) <- S.get let worker :: StepCode () = - do x <- STEPLIFT tryPop global_stack + do x <- STEPLIFT tryPop stack case x of Nothing -> STEPLIFT writeChan joiner () Just action -> do action myId <- STEPLIFT myThreadId - set <- STEPLIFT readIORef global_mortalthreads + set <- STEPLIFT readIORef mortalthreads if Set.notMember myId set then worker -- keep going else STEPLIFT writeChan joiner () ver5_6_core_finalize joiner finalAction worker -get col tag = ver5_6_core_get io col tag - where io = do myId <- myThreadId - atomicModifyIORef global_mortalthreads (\s -> (Set.insert myId s, ())) +get col tag = + do (HiddenState5 (stack, _, _, mortalthreads)) <- S.get + let io = do myId <- myThreadId + atomicModifyIORef mortalthreads (\s -> (Set.insert myId s, ())) + ver5_6_core_get io col tag quiescence_support = True diff --git a/Intel/shared_5_6.hs b/Intel/shared_5_6.hs --- a/Intel/shared_5_6.hs +++ b/Intel/shared_5_6.hs @@ -17,7 +17,7 @@ -- (2) the number of workers for this graph -- (3) the "make worker" function to spawn new threads -- (4) the set of "mortal threads" -newtype HiddenState5 = HiddenState5 (HotVar [StepCode ()], HotVar Int, IO (), Set ThreadId) +newtype HiddenState5 = HiddenState5 (HotVar [StepCode ()], HotVar Int, IO (), HotVar (Set ThreadId)) deriving Show instance Show (IORef a) where @@ -29,24 +29,6 @@ atomicDecr x = atomicModifyIORef x (\n -> (n-1, ())) --- This will be one hot IORef: -global_stack :: HotVar [StepCode ()] -global_stack = unsafePerformIO (newHotVar []) - -global_numworkers :: IORef Int -global_numworkers = unsafePerformIO (newIORef 0) - --- A computation that forks a new worker thread: -global_makeworker :: IORef (IO ()) -global_makeworker = unsafePerformIO$ newIORef (return ()) - - --- This is a bit silly, this emulates "thread local storage" to let --- each worker thread know whether it is recursive (True) or "oneshot". -global_mortalthreads :: IORef (Set ThreadId) -global_mortalthreads = unsafePerformIO (newIORef Set.empty) - - -- A simple stack interface: ---------------------------------------- push :: HotVar [a] -> a -> IO () @@ -71,10 +53,11 @@ -- should succeed. issueReplacement = - do STEPLIFT atomicIncr global_numworkers + do (HiddenState5 (stack, numworkers, makeworker, _)) <- S.get + STEPLIFT atomicIncr numworkers -- If this were CPS then we would just give our -- continuation to the forked thread. Alas, no. - makeworker <- STEPLIFT readIORef global_makeworker + --makeworker <- STEPLIFT readIORef global_makeworker STEPLIFT forkIO makeworker grabWithBackup hook mvar = @@ -93,26 +76,28 @@ ver5_6_core_get hook (col) tag = - do --(HiddenState5 (stack, numworkers, makeworker, _)) <- S.get + do (HiddenState5 (stack, numworkers, makeworker, _)) <- S.get mvar <- STEPLIFT assureMvar col tag grabWithBackup hook mvar --ver5_6_core_finalize :: Chan a -> IO b -> IO () -> StepCode b ver5_6_core_finalize :: Chan a -> StepCode b -> StepCode () -> GraphCode b ver5_6_core_finalize joiner finalAction worker = - do --(HiddenState5 (stack, numworkers, makeworker, _)) <- S.get - state <- S.get - let makeworker = do S.runStateT worker state; return () - S.lift$ writeIORef global_makeworker makeworker - S.lift$ atomicModifyIORef global_numworkers (\n -> (n + numCapabilities, ())) + do (HiddenState5 (stack, numworkers, _, _)) <- S.get + --state <- S.get + --GRAPHLIFT writeIORef global_makeworker makeworker + let mkwrkr = do S.runStateT worker state2; return () + state2 = HiddenState5 (stack, numworkers, mkwrkr, undefined) + + GRAPHLIFT atomicModifyIORef numworkers (\n -> (n + numCapabilities, ())) -- Fork one worker per thread: #ifdef DEBUG_HASKELL_CNC S.lift$ putStrLn$ "Forking "++ show numCapabilities ++" threads" #endif - S.lift$ mapM (\n -> forkIO makeworker) [0..numCapabilities-1] + S.lift$ mapM (\n -> forkIO mkwrkr) [0..numCapabilities-1] -- This waits for quiescense: - let waitloop = do num <- readIORef global_numworkers + let waitloop = do num <- readIORef numworkers if num == 0 then return () else do @@ -120,7 +105,7 @@ putStrLn ("=== Waiting on workers: "++ show num ++" left") #endif readChan joiner - atomicDecr global_numworkers + atomicDecr numworkers waitloop S.lift$ waitloop finalAction @@ -128,13 +113,14 @@ putt = proto_putt (\ steps tag -> - do --(HiddenState5 (stack, numworkers, makeworker, _)) <- S.get - foldM (\ () step -> S.lift$ push global_stack (step tag)) + do (HiddenState5 (stack, _, _, _)) <- S.get + foldM (\ () step -> STEPLIFT push stack (step tag)) () steps) runGraph x = unsafePerformIO (runState x) runState x = do hv <- newHotVar [] hv2 <- newHotVar 0 - (a,_) <- S.runStateT x (HiddenState5 (hv,hv2, undefined, Set.empty)) + hv3 <- newHotVar Set.empty + (a,_) <- S.runStateT x (HiddenState5 (hv,hv2, undefined, hv3)) return a diff --git a/examples/hello_world.hs b/examples/hello_world.hs --- a/examples/hello_world.hs +++ b/examples/hello_world.hs @@ -21,7 +21,7 @@ -- #include -- This demonstrates the normal (NOT #include) method of loading CnC: -import Intel.Cnc +import Intel.Cnc6 -- Here's an odd little hello world where we communicate the two words -- to a computational step which puts them together. diff --git a/haskell-cnc.cabal b/haskell-cnc.cabal --- a/haskell-cnc.cabal +++ b/haskell-cnc.cabal @@ -1,5 +1,5 @@ Name: haskell-cnc -Version: 0.1.1 +Version: 0.1.2 License: LGPL License-file: LICENSE Stability: Beta