9f2c3c2 · streamly · git.iko.soy

git.iko.soy/streamly

rename location attributes and context resumption

commit9f2c3c23c1bdf4a06b6a730a25eb5ba043e5472d authorHarendra Kumar <harendra.kumar@gmail.com> dateSun 11 Jun 2017, 17:07 (9 years ago) parent4adbddf

2 files changed+28−14

compare with master

unifiedsplit

src/Duct/Context.hs+19−5 src/Duct/Threads.hs+9−9

src/Duct/Context.hs+19−5

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

77 , initContext

88 , saveContext

99 , restoreContext

10− , continueContext

10+ , resumeContext

1111 , Location(..)

1212 , getLocation

1313 , setLocation

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

108108 Context { event = mempty

109109 , currentm = unsafeCoerce x

110110 , fstack = []

111− , location = Local

111+ , location = Worker

112112 , mfData = mempty

113113 , parentChannel = Nothing

114114 , childChannel = childChan

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

119119-- Where is the computation running?

120120------------------------------------------------------------------------------

121121

122−data Location = Local | ForkedThread | RemoteNode

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+--

128+data Location = Worker | WaitingParent | RemoteNode

123129 deriving (Typeable, Eq, Show)

124130

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

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

157163

158164 return mres

159165

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

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

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+

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

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

162176 unsafeCoerce m >>= runfStack (unsafeCoerce fs)

163177

164178 where

src/Duct/Threads.hs+9−9

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

4242------------------------------------------------------------------------------

4343

4444-- | 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

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

45+-- to a new thread.

4746

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

49−continue = runAsyncT . continueContext

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

48+resume = runAsyncT . resumeContext

5049

5150------------------------------------------------------------------------------

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

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

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

212211 runContWith dat cont = do

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

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

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

215214 return ()

216215

217216-- | 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

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

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

228227parallel ioaction = AsyncT $ do

228+ -- receive the events via a channel instead

229229 cont <- get

230230 case event cont of

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

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

238238 Nothing -> do

239239 lift $ genAsyncEvents cont ioaction

240240 loc <- getLocation

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

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

242242 return Nothing

243243

244244-- | 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

269269sync x = AsyncT $ do

270270 setLocation RemoteNode

271271 r <- runAsyncT x

272− setLocation Local

272+ setLocation Worker

273273 return r

274274

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

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

311311 case event cont of

312312 Nothing -> do

313313 lift $ setHandler $ \dat ->do

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

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

315315 liftIO iob

316316 loc <- getLocation

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

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

318318 return Nothing

319319

320320 j@(Just _) -> do