@@ -125,8 +125,8 @@ instance Monad m => Monoid (AsyncT m a) where
125instance Monad m => Monad (AsyncT m) where
125instance Monad m => Monad (AsyncT m) where
126 return a = AsyncT $ \ctx _ yld -> yld a ctx Nothing
126 return a = AsyncT $ \ctx _ yld -> yld a ctx Nothing
128− -- | On bind the new computation does not inherit the parent thread
128+ -- | A thread context is valid only until the next bind. Upon a bind we
129− -- context, it starts a new node in the tree.
129+ -- reset the context to Nothing.
130 AsyncT m >>= f = AsyncT $ \_ stp yld ->
130 AsyncT m >>= f = AsyncT $ \_ stp yld ->
131 m Nothing stp $ \a _ r ->
131 m Nothing stp $ \a _ r ->
132 let run x = (runAsyncT x) Nothing stp yld
132 let run x = (runAsyncT x) Nothing stp yld
@@ -166,34 +166,23 @@ doFork action preExit =
167{-# NOINLINE push #-}
167{-# NOINLINE push #-}
168push :: MonadIO m => Context a -> AsyncT m a -> m ()
168push :: MonadIO m => Context a -> AsyncT m a -> m ()
169−push ctx m = go (Just ctx) m
169+push context action = run (Just context) action
173+ run ctx m = (runAsyncT m) ctx channelDone yield
173 -- XXX make this a bounded channel so that we block if the previous value
175 -- XXX make this a bounded channel so that we block if the previous value
174 -- is not consumed yet.
176 -- is not consumed yet.
175− {-# INLINE channelDone #-}
177+ chan = childChannel context
176− channelDone chan = do
178+ channelYield a = liftIO $ atomically $ writeTChan chan (ChildYield a)
177 tid <- liftIO myThreadId
180 tid <- liftIO myThreadId
178 liftIO $ atomically $ writeTChan chan (ChildDone tid Nothing)
181 liftIO $ atomically $ writeTChan chan (ChildDone tid Nothing)
180− {-# INLINE channelYield #-}
181− channelYield chan a = liftIO $ atomically $ writeTChan chan (ChildYield a)
183− go c mx = (runAsyncT mx) c (channelDone pchan) yield
184− pchan = childChannel ctx
187− channelYield pchan a
183+ done a = channelYield a >> channelDone
190− {-# INLINE continue #-}
191− continue a c mx = do
192− channelYield pchan a
184+ continue a ctx m = channelYield a >> run ctx m
195− {-# INLINE yield #-}
196− yield a c r = maybe (done a) (\x -> continue a c x) r
185+ yield a ctx r = maybe (done a) (\rx -> continue a ctx rx) r
198-- If an exception occurs we push it to the channel so that it can handled by
187-- If an exception occurs we push it to the channel so that it can handled by
199-- the parent. 'Paused' exceptions are to be collected at the top level.
188-- the parent. 'Paused' exceptions are to be collected at the top level.
@@ -255,10 +244,6 @@ pullFork m1 m2 = AsyncT $ \_ stp yld -> do
255 -- XXX push from m2 only when the channel becomes empty
244 -- XXX push from m2 only when the channel becomes empty
256 push c mb
245 push c mb
258−-- If threads suspend due to IO we can start more of them to utilize concurrent
259−-- IO, but only if we have more IO bandwidth. If we have more parallel CPU
260−-- bandwidth we can start more threads appropriately, but we have to see that
261−-- the thread creation overhead is less than the actual work to be done.
262canFork :: IO Bool
247canFork :: IO Bool
263canFork = return True
248canFork = return True