@@ -138,9 +138,9 @@ forkFinally1 action and_then =
138138 _ <- runInIO $ EL.try (restore action) >>= liftIO . and_then
139139 return ()
140140
141−forkContext :: (MonadBaseControl IO m, MonadIO m)
142− => Context -> (Context -> m ()) -> m ()
143−forkContext context runCtx = do
141+forkContextWith :: (MonadBaseControl IO m, MonadIO m)
142+ => (Context -> m ()) -> Context -> m ()
143+forkContextWith runCtx context = do
144144 child <- childContext context
145145 tid <- forkFinally1 (runCtx child) (beforeExit child)
146146 updatePendingThreads context tid
@@ -179,8 +179,9 @@ forkContext context runCtx = do
179179 tid <- myThreadId
180180 liftIO $ atomically $ writeTChan p tid
181181
182−forkMaybe :: (MonadBaseControl IO m, MonadIO m) => Context -> (Context -> m ()) -> m ()
183−forkMaybe context runCtx = do
182+forkMaybe :: (MonadBaseControl IO m, MonadIO m)
183+ => (Context -> m ()) -> Context -> m ()
184+forkMaybe runCtx context = do
184185 gotCredit <- liftIO $ waitQSemB (threadCredit context)
185186 pending <- liftIO $ readIORef $ pendingThreads context
186187 case gotCredit of
@@ -194,8 +195,8 @@ forkMaybe context runCtx = do
194195 -- async thread, cannot use sync for those.
195196 waitForOneChild (childChannel context)
196197 (pendingThreads context)
197− forkMaybe context runCtx
198− True -> forkContext context runCtx
198+ forkMaybe runCtx context
199+ True -> forkContextWith runCtx context
199200
200201-- | 'StreamData' represents a task in a task stream being generated.
201202data StreamData a =
@@ -220,21 +221,18 @@ data StreamData a =
220221-- they can go away rather than waiting indefinitely.
221222--
222223-- | Execute the IO action resume the saved context
223−genAsyncEvents :: (MonadIO m, MonadBaseControl IO m) => Context -> IO (StreamData t) -> m ()
224−genAsyncEvents context rec = forkMaybe context loop
225−
226− where
227−
228− loop ctx = do
229− streamData <- liftIO $ rec `catch`
230− \(e :: SomeException) -> return $ SError e
231−
232− let ctx' = updateContextEvent ctx streamData
233− case streamData of
234− SMore _ -> do
235− forkMaybe ctx' resume
236− loop ctx
237− _ -> resume ctx'
224+runContextWith :: (MonadIO m, MonadBaseControl IO m)
225+ => IO (StreamData t) -> Context -> m ()
226+runContextWith ioaction ctx = do
227+ streamData <- liftIO $ ioaction `catch`
228+ \(e :: SomeException) -> return $ SError e
229+
230+ let ctx' = updateContextEvent ctx streamData
231+ case streamData of
232+ SMore _ -> do
233+ forkMaybe resume ctx'
234+ runContextWith ioaction ctx
235+ _ -> resume ctx'
238236
239237-- | Run an IO action one or more times to generate a stream of tasks. The IO
240238-- action returns a 'StreamData'. When it returns an 'SMore' or 'SLast' a new
@@ -261,7 +259,7 @@ parallel ioaction = AsyncT $ do
261259 -- we have to execute the io action and generate an event and then continue
262260 -- in this thread or a new thread.
263261 Nothing -> do
264− lift $ genAsyncEvents context ioaction
262+ lift $ forkMaybe (runContextWith ioaction) context
265263 loc <- getLocation
266264 when (loc /= RemoteNode) $ setLocation WaitingParent
267265 return Nothing