3251739 · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Remove parentChannel from context

3251739Harendra Kumar9 years ago

src/Strands/Context.hs−4

compare with master

unifiedsplit

@@ -81,9 +81,6 @@ data Context = Context

81 -- XXX setup a cleanup computation to run rather than passing all these

82 -- params.

84− , parentChannel :: Maybe (TChan (ChildEvent Any))

85− -- ^ Parent thread's channel to communicate to when a child thread dies

86−

87 , childChannel :: TChan (ChildEvent Any)

84 , childChannel :: TChan (ChildEvent Any)

88 -- ^ A channel for the immediate children to communicate to us when they

85 -- ^ A channel for the immediate children to communicate to us when they

89 -- die. Each thread has its own dedicated channel for its children

86 -- die. Each thread has its own dedicated channel for its children

@@ -126,7 +123,6 @@ initContext x childChan pending credit finalizer =

126 , fstack = [unsafeCoerce finalizer]

123 , fstack = [unsafeCoerce finalizer]

127 , location = Worker

124 , location = Worker

128 -- , mfData = mempty

125 -- , mfData = mempty

129− , parentChannel = Nothing

130 , childChannel = unsafeCoerce childChan

126 , childChannel = unsafeCoerce childChan

131 , pendingThreads = pending

127 , pendingThreads = pending

132 , threadCredit = credit }

128 , threadCredit = credit }

src/Strands/Threads.hs+8−13

@@ -36,8 +36,7 @@ import Data.IORef (IORef, atomicModifyIORef,

36 modifyIORef, newIORef, readIORef,

37 writeIORef)

38import Data.List (delete)

39−import Data.Maybe (fromJust, isJust)

39+import Data.Maybe (isJust)

40−import Unsafe.Coerce (unsafeCoerce)

42import Strands.AsyncT

41import Strands.AsyncT

43import Strands.Context

42import Strands.Context

@@ -187,7 +186,7 @@ forkContext :: (MonadBaseControl IO m, MonadIO m, MonadThrow m)

187 => Context -> StateT Context m ()

186 => Context -> StateT Context m ()

188forkContext context = do

187forkContext context = do

189 child <- childContext context

188 child <- childContext context

190− tid <- forkFinally1 child (beforeExit child)

189+ tid <- forkFinally1 child (beforeExit child (childChannel context))

191 updatePendingThreads context tid

190 updatePendingThreads context tid

192

191

193 where

192 where

@@ -205,12 +204,11 @@ forkContext context = do

205 chan <- liftIO $ atomically newTChan

204 chan <- liftIO $ atomically newTChan

206 -- shares the threadCredit of the parent by default

205 -- shares the threadCredit of the parent by default

207 return $ ctx

206 return $ ctx

208− { parentChannel = Just (childChannel ctx)

209− , pendingThreads = pendingRef

207+ { pendingThreads = pendingRef

210 , childChannel = chan

208 , childChannel = chan

211 }

209 }

212

210

213− beforeExit ctx res = do

211+ beforeExit ctx pchan res = do

214 tid <- myThreadId

212 tid <- myThreadId

215 r <- case res of

213 r <- case res of

216 Left e -> do

214 Left e -> do

@@ -219,10 +217,8 @@ forkContext context = do

219 return (Just e)

217 return (Just e)

220 Right _ -> waitForChildren ctx

218 Right _ -> waitForChildren ctx

221

219

222− -- We are guaranteed to have a parent because we are forked.

223− let p = fromJust (parentChannel ctx)

224 signalQSemB (threadCredit ctx)

220 signalQSemB (threadCredit ctx)

225− liftIO $ atomically $ writeTChan p (ChildDone tid r)

221+ liftIO $ atomically $ writeTChan pchan (ChildDone tid r)

226

222

227-- | Decide whether to resume the context in the same thread or a new thread

223-- | Decide whether to resume the context in the same thread or a new thread

228--

224--

@@ -339,8 +335,7 @@ loopContextWith ioaction context = AsyncT $ do

339 -- parameters from the current context.

335 -- parameters from the current context.

340 curCtx <- get

336 curCtx <- get

341 loop context

337 loop context

342− { parentChannel = parentChannel curCtx

343− , pendingThreads = pendingThreads curCtx

338+ { pendingThreads = pendingThreads curCtx

344 , childChannel = childChannel curCtx

339 , childChannel = childChannel curCtx

345 }

340 }

346

341

@@ -500,7 +495,7 @@ noTrans x = AsyncT $ x >>= return . Just

500

495

501-- | Run an 'AsyncT m' computation and collect the results generated by each

496-- | Run an 'AsyncT m' computation and collect the results generated by each

502-- thread of the computation in a list.

497-- thread of the computation in a list.

503−waitAsync :: forall m a b. (MonadIO m, MonadCatch m)

498+waitAsync :: forall m a. (MonadIO m, MonadCatch m)

504 => (a -> AsyncT m a) -> AsyncT m a -> m ()

499 => (a -> AsyncT m a) -> AsyncT m a -> m ()

505waitAsync finalizer m = do

500waitAsync finalizer m = do

506 childChan <- liftIO $ atomically newTChan

501 childChan <- liftIO $ atomically newTChan

@@ -528,7 +523,7 @@ waitAsync finalizer m = do

528-- the computation when the computation is done

523-- the computation when the computation is done

529gatherResult :: MonadIO m => IORef [a] -> a -> AsyncT m a

524gatherResult :: MonadIO m => IORef [a] -> a -> AsyncT m a

530gatherResult ref r = do

525gatherResult ref r = do

531− liftIO $ atomicModifyIORef ref $ \rs -> (r : rs, rs)

526+ liftIO $ atomicModifyIORef ref $ \rs -> (r : rs, ())

532 mzero

527 mzero

533

528

534-- | Run an 'AsyncT m' computation and collect the results generated by each

529-- | Run an 'AsyncT m' computation and collect the results generated by each