@@ -68,10 +68,10 @@ import Control.Monad.Trans.Recorder (MonadRecorder(..))
68-- Parent child thread communication types
68-- Parent child thread communication types
69------------------------------------------------------------------------------
69------------------------------------------------------------------------------
71−-- XXX for threads that yield a single value we can pack yield and done in
71+data ChildEvent a =
72−-- ChildDone (Either exception value) so that we do not use two messages over
72+ ChildYield a
73−-- the channel where we can use just one.
73+ | ChildDone ThreadId a
74−data ChildEvent a = ChildYield a | ChildDone ThreadId (Maybe SomeException)
74+ | ChildStop ThreadId (Maybe SomeException)
76------------------------------------------------------------------------------
76------------------------------------------------------------------------------
77-- State threaded around the monad for thread management
77-- State threaded around the monad for thread management
@@ -170,17 +170,20 @@ push context action = run (Just context) action
173− run ctx m = (runAsyncT m) ctx channelDone yield
173+ run ctx m = (runAsyncT m) ctx channelStop yield
175 -- 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
176 -- is not consumed yet.
176 -- is not consumed yet.
177 chan = childChannel context
177 chan = childChannel context
178 channelYield a = liftIO $ atomically $ writeTChan chan (ChildYield a)
178 channelYield a = liftIO $ atomically $ writeTChan chan (ChildYield a)
179− channelDone = do
179+ channelDone a = do
180 tid <- liftIO myThreadId
180 tid <- liftIO myThreadId
181− liftIO $ atomically $ writeTChan chan (ChildDone tid Nothing)
181+ liftIO $ atomically $ writeTChan chan (ChildDone tid a)
183+ tid <- liftIO myThreadId
184+ liftIO $ atomically $ writeTChan chan (ChildStop tid Nothing)
183− done a = channelYield a >> channelDone
186+ done a = channelDone a
184 continue a ctx m = channelYield a >> run ctx m
187 continue a ctx m = channelYield a >> run ctx m
185 yield a ctx r = maybe (done a) (\rx -> continue a ctx rx) r
188 yield a ctx r = maybe (done a) (\rx -> continue a ctx rx) r
@@ -191,7 +194,7 @@ push context action = run (Just context) action
191handlePushException :: TChan (ChildEvent a) -> SomeException -> IO ()
194handlePushException :: TChan (ChildEvent a) -> SomeException -> IO ()
192handlePushException pchan e = do
195handlePushException pchan e = do
193 tid <- myThreadId
196 tid <- myThreadId
194− atomically $ writeTChan pchan (ChildDone tid (Just e))
197+ atomically $ writeTChan pchan (ChildStop tid (Just e))
196-- | run m1 in a new thread, pushing its results to a pull channel and then run
199-- | run m1 in a new thread, pushing its results to a pull channel and then run
197-- m2 in the parent thread. Any exceptions are also pushed to the channel.
200-- m2 in the parent thread. Any exceptions are also pushed to the channel.
@@ -219,7 +222,8 @@ pull ctx = AsyncT $ \_ stp yld -> do
219 Right ev ->
222 Right ev ->
220 case ev of
223 case ev of
221 ChildYield a -> yld a Nothing (Just (pull ctx))
224 ChildYield a -> yld a Nothing (Just (pull ctx))
222− ChildDone _ e -> do
225+ ChildDone _ a -> yld a Nothing (Just (pull ctx))
223 let continue = (runAsyncT (pull ctx)) (Just ctx) stp yld
227 let continue = (runAsyncT (pull ctx)) (Just ctx) stp yld
224 maybe continue throwM e
228 maybe continue throwM e