@@ -69,74 +69,20 @@ import Strands.Context
6969-- | Continue execution of the closure that we were executing when we migrated
7070-- to a new thread.
7171
72−resume :: MonadIO m => Context -> StateT Context m ()
73−resume ctx = do
72+runContext :: MonadIO m => Context -> StateT Context m ()
73+runContext ctx = do
7474 -- XXX rename to buildContext or buildState?
7575 let s = runAsyncT (resumeContext ctx)
76− -- The returned value is always 'Nothing', we just discard it
77− (_, c) <- lift $ runStateT s ctx
78−
79− -- XXX can we pass the result directly to the root thread instead of
80− -- passing through all the parents? We can let the parent go away and
81− -- handle the ChildDone events as well in the root thread.
82− case parentChannel c of
83− Nothing -> modify $ \x ->
84− x { accumResults = accumResults c ++ accumResults x }
85− Just chan -> do
86− let r = accumResults c
87− when (length r /= 0) $
88− -- there is only one result in case of a non-root thread
89− -- XXX change the return type to 'a' instead of '[a]'
90− liftIO $ atomically $ writeTChan chan (PassOnResult (Right r))
76+ _ <- lift $ runStateT s ctx
77+ return ()
9178
9279------------------------------------------------------------------------------
9380-- Thread Management (creation, reaping and killing)
9481------------------------------------------------------------------------------
9582
96−-- XXX We are using unbounded channels so this will not block on writing to
97−-- pchan. We can use bounded channels to throttle the creation of threads based
98−-- on consumption rate.
99−processOneEvent :: MonadIO m
100− => ChildEvent a
101− -> TChan (ChildEvent a)
102− -> [ThreadId]
103− -> Maybe SomeException
104− -> m ([ThreadId], Maybe SomeException)
105−processOneEvent ev pchan pending exc = do
106− e <- case exc of
107− Nothing ->
108− case ev of
109− ChildDone tid res -> do
110− dbg $ "processOneEvent ChildDone: " ++ show tid
111− handlePass res
112− PassOnResult res -> do
113− dbg $ "processOneEvent PassOnResult"
114− handlePass res
115− Just _ -> return exc
116− let p = case ev of
117− ChildDone tid _ -> delete tid pending
118− _ -> pending
119− return (p, e)
120−
121− where
122−
123− handlePass :: MonadIO m
124− => Either SomeException [a] -> m (Maybe SomeException)
125− handlePass res =
126− case res of
127− Left e -> do
128− dbg $ "handlePass: caught exception"
129− liftIO $ mapM_ killThread pending
130− return (Just e)
131− Right [] -> return Nothing
132− Right _ -> do
133− liftIO $ atomically $ writeTChan pchan
134− (PassOnResult (unsafeCoerce res))
135− return Nothing
136−
13783tryReclaimZombies :: (MonadIO m, MonadThrow m) => Context -> m ()
13884tryReclaimZombies ctx = do
139− let pchan = fromJust (parentChannel ctx)
85+ let dest = getCtxResultDest ctx
14086 cchan = childChannel ctx
14187 pendingRef = pendingThreads ctx
14288
@@ -148,7 +94,7 @@ tryReclaimZombies ctx = do
14894 case mev of
14995 Nothing -> return ()
15096 Just ev -> do
151− (p, e) <- processOneEvent ev pchan pending Nothing
97+ (p, e) <- processOneEvent ev dest pending
15298 liftIO $ writeIORef pendingRef p
15399 maybe (return ()) throwM e
154100 tryReclaimZombies ctx
@@ -157,40 +103,16 @@ waitForOneEvent :: (MonadIO m, MonadThrow m) => Context -> m ()
157103waitForOneEvent ctx = do
158104 -- XXX assert pending must have at least one element
159105 -- assert that the tid is found in our list
160− let pchan = fromJust (parentChannel ctx)
106+ let dest = getCtxResultDest ctx
161107 cchan = childChannel ctx
162108 pendingRef = pendingThreads ctx
163109
164110 ev <- liftIO $ atomically $ readTChan cchan
165111 pending <- liftIO $ readIORef pendingRef
166− (p, e) <- processOneEvent ev pchan pending Nothing
112+ (p, e) <- processOneEvent ev dest pending
167113 liftIO $ writeIORef pendingRef p
168114 maybe (return ()) throwM e
169115
170−drainChildren :: MonadIO m
171− => TChan (ChildEvent a)
172− -> TChan (ChildEvent a)
173− -> [ThreadId]
174− -> Maybe SomeException
175− -> m (Maybe SomeException)
176−drainChildren pchan cchan pending exc =
177− if pending == []
178− then return exc
179− else do
180− ev <- liftIO $ atomically $ readTChan cchan
181− (p, e) <- processOneEvent ev pchan pending exc
182− drainChildren pchan cchan p e
183−
184−waitForChildren :: MonadIO m
185− => Context -> Maybe SomeException -> m (Maybe SomeException)
186−waitForChildren ctx exc = do
187− let pendingRef = pendingThreads ctx
188− pchan = fromJust (parentChannel ctx)
189− pending <- liftIO $ readIORef pendingRef
190− e <- drainChildren pchan (childChannel ctx) pending exc
191− liftIO $ writeIORef pendingRef []
192− return e
193−
194116-- | kill all the child threads associated with the continuation context
195117killChildren :: Context -> IO ()
196118killChildren ctx = do
@@ -233,7 +155,8 @@ forkFinally1 :: (MonadIO m, MonadBaseControl IO m)
233155forkFinally1 ctx preExit =
234156 EL.mask $ \restore ->
235157 liftBaseWith $ \runInIO -> forkIO $ do
236− _ <- runInIO $ EL.try (restore (resume ctx)) >>= liftIO . preExit
158+ _ <- runInIO $ EL.try (restore (runContext ctx))
159+ >>= liftIO . preExit
237160 return ()
238161
239162-- | Run a given context in a new thread.
@@ -263,27 +186,22 @@ forkContext context = do
263186 { parentChannel = Just (childChannel ctx)
264187 , pendingThreads = pendingRef
265188 , childChannel = chan
266− , accumResults = []
189+ , accumResults = Nothing
267190 }
268191
269192 beforeExit ctx res = do
270193 tid <- myThreadId
271− exc <- case res of
272− Left e -> do
194+ r <- case res of
195+ Left e -> do
273196 dbg $ "beforeExit: " ++ show tid ++ " caught exception"
274197 liftIO $ killChildren ctx
275198 return (Just e)
276− Right _ -> return Nothing
277−
278− e <- waitForChildren ctx exc
199+ Right _ -> waitForChildren ctx
279200
280− -- We are guaranteed to have a parent because we have been explicitly
281− -- forked by some parent.
201+ -- We are guaranteed to have a parent because we are forked.
282202 let p = fromJust (parentChannel ctx)
283203 signalQSemB (threadCredit ctx)
284− -- XXX change the return value type to Maybe SomeException
285− liftIO $ atomically $ writeTChan p
286− (ChildDone tid (maybe (Right []) Left e))
204+ liftIO $ atomically $ writeTChan p (ChildDone tid r)
287205
288206-- | Decide whether to resume the context in the same thread or a new thread
289207--
@@ -318,7 +236,7 @@ resumeContextWith context synch action = do
318236 let ctx = setContextMailBox context (action context)
319237 can <- liftIO $ canFork context
320238 case can && (not synch) of
321− False -> resume ctx -- run synchronously
239+ False -> runContext ctx -- run synchronously
322240 True -> forkContext ctx
323241
324242-- | 'StreamData' represents a task in a task stream being generated.