9f2c3c2 · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

rename location attributes and context resumption

9f2c3c2Harendra Kumar9 years ago

src/Duct/Context.hs+19−5

compare with master

unifiedsplit

@@ -7,7 +7,7 @@ module Duct.Context

7 , initContext

8 , saveContext

9 , restoreContext

10− , continueContext

10+ , resumeContext

11 , Location(..)

12 , getLocation

13 , setLocation

@@ -108,7 +108,7 @@ initContext x childChan pending credit =

108 Context { event = mempty

109 , currentm = unsafeCoerce x

110 , fstack = []

111− , location = Local

111+ , location = Worker

112 , mfData = mempty

113 , parentChannel = Nothing

114 , childChannel = childChan

@@ -119,7 +119,13 @@ initContext x childChan pending credit =

119-- Where is the computation running?

120------------------------------------------------------------------------------

121

122+-- RemoteNode is used in the Alternative instance. We stop the computation when

123+-- the location is RemoteNode, we do not execute any alternatives.

124+--

125+-- WaitingParent is used in logging. When a parent forks a child and waits for

126+-- it, it is marked as WaitingParent and we log a "Wait" entry in the log.

127+--

122−data Location = Local | ForkedThread | RemoteNode

128+data Location = Worker | WaitingParent | RemoteNode

123 deriving (Typeable, Eq, Show)

129 deriving (Typeable, Eq, Show)

124

130

125getLocation :: Monad m => StateT Context m Location

131getLocation :: Monad m => StateT Context m Location

@@ -157,8 +163,16 @@ restoreContext x = do

157

163

158 return mres

164 return mres

159

165

166+-- | Resume execution of the computation from a given context. We can retrieve

167+-- a context at any point and resume that context at some later point, which

168+-- means start executing again from the same point where the context was

169+-- retrieved.

170+--

171+-- Run the stack of pending functions in fstack. The types don't match. We just

172+-- coerce the types here, we know that they actually match.

173+

160−continueContext :: MonadPlus m => Context -> m a

174+resumeContext :: MonadPlus m => Context -> m a

161−continueContext Context { currentm = m, fstack = fs } =

175+resumeContext Context { currentm = m, fstack = fs } =

162 unsafeCoerce m >>= runfStack (unsafeCoerce fs)

176 unsafeCoerce m >>= runfStack (unsafeCoerce fs)

163

177

164 where

178 where

src/Duct/Threads.hs+9−9

@@ -42,11 +42,10 @@ import Duct.Context

42------------------------------------------------------------------------------

44-- | Continue execution of the closure that we were executing when we migrated

45−-- to a new thread. Run the stack of pending functions in fstack. The types

45+-- to a new thread.

46−-- don't match. We just coerce the types here, we know that they actually match.

48−continue :: Monad m => Context -> StateT Context m (Maybe a)

47+resume :: Monad m => Context -> StateT Context m (Maybe a)

49−continue = runAsyncT . continueContext

48+resume = runAsyncT . resumeContext

51------------------------------------------------------------------------------

50------------------------------------------------------------------------------

52-- Thread Management (creation, reaping and killing)

51-- Thread Management (creation, reaping and killing)

@@ -211,7 +210,7 @@ genAsyncEvents parentc rec = forkMaybe parentc loop

211 -- pass on the io result and then run the continuation

210 -- pass on the io result and then run the continuation

212 runContWith dat cont = do

211 runContWith dat cont = do

213 let cont' = cont { event = Just $ unsafeCoerce dat }

212 let cont' = cont { event = Just $ unsafeCoerce dat }

214− _ <- runStateT (continue cont') cont'

213+ _ <- runStateT (resume cont') cont'

215 return ()

214 return ()

216

215

217-- | Run an IO action one or more times to generate a stream of tasks. The IO

216-- | Run an IO action one or more times to generate a stream of tasks. The IO

@@ -226,6 +225,7 @@ genAsyncEvents parentc rec = forkMaybe parentc loop

226parallel :: (Monad m, MonadIO m, MonadBaseControl IO m)

225parallel :: (Monad m, MonadIO m, MonadBaseControl IO m)

227 => IO (StreamData a) -> AsyncT m (StreamData a)

226 => IO (StreamData a) -> AsyncT m (StreamData a)

228parallel ioaction = AsyncT $ do

227parallel ioaction = AsyncT $ do

228+ -- receive the events via a channel instead

229 cont <- get

230 case event cont of

231 -- we have already exceuted the ioaction and now we are continuing in a new

@@ -238,7 +238,7 @@ parallel ioaction = AsyncT $ do

238 Nothing -> do

239 lift $ genAsyncEvents cont ioaction

240 loc <- getLocation

241− when (loc /= RemoteNode) $ setLocation ForkedThread

241+ when (loc /= RemoteNode) $ setLocation WaitingParent

242 return Nothing

243

244-- | An task stream generator that produces an infinite stream of tasks by

@@ -269,7 +269,7 @@ sync :: MonadIO m => AsyncT m a -> AsyncT m a

269sync x = AsyncT $ do

270 setLocation RemoteNode

271 r <- runAsyncT x

272− setLocation Local

272+ setLocation Worker

273 return r

274

275-- | An task stream generator that produces an infinite stream of tasks by

@@ -311,10 +311,10 @@ react setHandler iob = AsyncT $ do

311 case event cont of

312 Nothing -> do

313 lift $ setHandler $ \dat ->do

314− runStateT (continue cont) cont{event= Just $ unsafeCoerce dat}

314+ runStateT (resume cont) cont{event= Just $ unsafeCoerce dat}

315 liftIO iob

316 loc <- getLocation

317− when (loc /= RemoteNode) $ setLocation ForkedThread

317+ when (loc /= RemoteNode) $ setLocation WaitingParent

318 return Nothing

319

320 j@(Just _) -> do