@@ -69,11 +69,12 @@ import Duct.Context
6969-- | Continue execution of the closure that we were executing when we migrated
7070-- to a new thread.
7171
72−resume :: Monad m => Context -> m (Maybe a)
72+resume :: MonadIO m => Context -> m Context
7373resume ctx = do
7474 let s = runAsyncT (resumeContext ctx)
75− (a, _) <- runStateT s ctx
76− return a
75+ -- The returned value is always 'Nothing', we just discard it
76+ (_, c) <- runStateT s ctx
77+ return c
7778
7879updateContextEvent :: Context -> a -> Context
7980updateContextEvent ctx evdata = ctx { event = Just $ unsafeCoerce evdata }
@@ -96,9 +97,11 @@ processOneEvent ev pchan pending exc = do
9697 Nothing ->
9798 case ev of
9899 ChildDone tid res -> do
99− dbg $ "Reaping child: " ++ show tid
100− handleDone res
101− PassOnResult res -> handlePass res
100+ dbg $ "processOneEvent ChildDone: " ++ show tid
101+ handlePass res
102+ PassOnResult res -> do
103+ dbg $ "processOneEvent PassOnResult"
104+ handlePass res
102105 Just _ -> return exc
103106 let p = case ev of
104107 ChildDone tid _ -> delete tid pending
@@ -107,29 +110,16 @@ processOneEvent ev pchan pending exc = do
107110
108111 where
109112
110− handleDone :: MonadIO m
111− => Either SomeException (Maybe a) -> m (Maybe SomeException)
112− handleDone res =
113− case res of
114− Left e -> do
115− liftIO $ mapM_ killThread pending
116− return (Just e)
117− Right r -> do
118− case r of
119− Nothing -> return ()
120− Just x -> liftIO $ atomically
121− $ writeTChan pchan
122− (PassOnResult (Right (unsafeCoerce x)))
123− return Nothing
124−
125113 handlePass :: MonadIO m
126− => Either SomeException a -> m (Maybe SomeException)
114+ => Either SomeException [a] -> m (Maybe SomeException)
127115 handlePass res =
128116 case res of
129117 Left e -> do
118+ dbg $ "handlePass: caught exception"
130119 liftIO $ mapM_ killThread pending
131120 return (Just e)
132− Right _ -> do
121+ Right [] -> return Nothing
122+ Right _ -> do
133123 liftIO $ atomically $ writeTChan pchan
134124 (PassOnResult (unsafeCoerce res))
135125 return Nothing
@@ -230,10 +220,10 @@ instance Read SomeException where
230220
231221forkFinally1 :: (MonadIO m, MonadBaseControl IO m) =>
232222 m a -> (Either SomeException a -> IO ()) -> m ThreadId
233−forkFinally1 action and_then =
223+forkFinally1 action preExit =
234224 EL.mask $ \restore ->
235225 liftBaseWith $ \runInIO -> forkIO $ do
236− _ <- runInIO $ EL.try (restore action) >>= liftIO . and_then
226+ _ <- runInIO $ EL.try (restore action) >>= liftIO . preExit
237227 return ()
238228
239229-- | Run a given context in a new thread.
@@ -258,33 +248,38 @@ forkContextWith runCtx context = do
258248 childContext ctx = do
259249 pendingRef <- liftIO $ newIORef []
260250 chan <- liftIO $ atomically newTChan
251+ resultsRef <- liftIO $ newIORef []
261252 -- shares the threadCredit of the parent by default
262253 return $ ctx
263254 { parentChannel = Just (childChannel ctx)
264255 , pendingThreads = pendingRef
265256 , childChannel = chan
257+ , accumResults = resultsRef
266258 }
267259
268260 beforeExit ctx res = do
261+ tid <- myThreadId
269262 exc <- case res of
270263 Left e -> do
264+ dbg $ "beforeExit: " ++ show tid ++ " caught exception"
271265 liftIO $ killChildren ctx
272266 return (Just e)
273267 Right _ -> return Nothing
274268
275269 e <- waitForChildren ctx exc
276270
277− tid <- myThreadId
278271 -- We are guaranteed to have a parent because we have been explicitly
279272 -- forked by some parent.
280273 let p = fromJust (parentChannel ctx)
274+ results <- liftIO $ readIORef (accumResults ctx)
281275 signalQSemB (threadCredit ctx)
282276 liftIO $ atomically $ writeTChan p
283− (ChildDone tid (maybe (unsafeCoerce res) Left e))
277+ (ChildDone tid (maybe (unsafeCoerce (Right results)) Left e))
284278
285279-- | A wrapper to first decide whether to run the context in the same thread or
286280-- a new thread and then run the context in that thread.
287281--
282+-- XXX create a canFork function
288283resumeContextWith :: (MonadBaseControl IO m, MonadIO m, MonadThrow m)
289284 => (Context -> m (Maybe a)) -> Context -> m ()
290285resumeContextWith runCtx context = do
@@ -339,9 +334,11 @@ loopContextWith ioaction ctx = do
339334 let ctx' = updateContextEvent ctx streamData
340335 case streamData of
341336 SMore _ -> do
342− resumeContextWith resume ctx'
337+ resumeContextWith (\x -> resume x >> return Nothing) ctx'
343338 loopContextWith ioaction ctx
344− _ -> resume ctx' -- run synchronously
339+ _ -> do
340+ _ <- resume ctx' -- run synchronously
341+ return Nothing -- we are done with the loop
345342
346343-- | Run an IO action one or more times to generate a stream of tasks. The IO
347344-- action returns a 'StreamData'. When it returns an 'SMore' or 'SLast' a new
@@ -360,7 +357,7 @@ parallel ioaction = AsyncT $ do
360357 -- generating the event data from the ioaction.
361358 context <- get
362359 case event context of
363− -- We have already executed the ioaction in the parent thread and put the
360+ -- We have already executed the ioaction in the parent thread and put its
364361 -- result in the context and now we are continuing the context in a new
365362 -- thread. Just return the result of the ioaction stored in the 'event'
366363 -- field.