@@ -22,7 +22,7 @@ import Control.Exception (ErrorCall (..),
2222 SomeException (..), catch)
2323import qualified Control.Exception.Lifted as EL
2424import Control.Monad.IO.Class (MonadIO (..))
25−import Control.Monad.State (StateT, get, gets, modify, put,
25+import Control.Monad.State (get, gets, modify, put,
2626 runStateT, when)
2727import Control.Monad.Trans.Class (MonadTrans (lift))
2828import Control.Monad.Trans.Control (MonadBaseControl, liftBaseWith)
@@ -179,9 +179,12 @@ forkContextWith runCtx context = do
179179 tid <- myThreadId
180180 liftIO $ atomically $ writeTChan p tid
181181
182−forkMaybe :: (MonadBaseControl IO m, MonadIO m)
182+-- | A wrapper to first decide whether to run the context in the same thread or
183+-- a new thread and then run the context in that thread.
184+--
185+runContext :: (MonadBaseControl IO m, MonadIO m)
183186 => (Context -> m ()) -> Context -> m ()
184−forkMaybe runCtx context = do
187+runContext runCtx context = do
185188 gotCredit <- liftIO $ waitQSemB (threadCredit context)
186189 pending <- liftIO $ readIORef $ pendingThreads context
187190 case gotCredit of
@@ -195,7 +198,7 @@ forkMaybe runCtx context = do
195198 -- async thread, cannot use sync for those.
196199 waitForOneChild (childChannel context)
197200 (pendingThreads context)
198− forkMaybe runCtx context
201+ runContext runCtx context
199202 True -> forkContextWith runCtx context
200203
201204-- | 'StreamData' represents a task in a task stream being generated.
@@ -209,7 +212,8 @@ data StreamData a =
209212-- The current model is to start a new thread for every task. The input is
210213-- provided at the time of the creation and therefore no synchronization is
211214-- needed compared to a pool of threads contending to get the input from a
212−-- channel. However the thread creation overhead may be more than that?
215+-- channel. However the thread creation overhead may be more than the
216+-- synchronization cost?
213217--
214218-- When the task is over the outputs need to be collected and that requires
215219-- synchronization irrespective of a thread pool model or per task new thread
@@ -230,7 +234,7 @@ loopContextWith ioaction ctx = do
230234 let ctx' = updateContextEvent ctx streamData
231235 case streamData of
232236 SMore _ -> do
233− forkMaybe resume ctx'
237+ runContext resume ctx'
234238 loopContextWith ioaction ctx
235239 _ -> resume ctx'
236240
@@ -259,7 +263,7 @@ parallel ioaction = AsyncT $ do
259263 -- we have to execute the io action and generate an event and then continue
260264 -- in this thread or a new thread.
261265 Nothing -> do
262− lift $ forkMaybe (loopContextWith ioaction) context
266+ lift $ runContext (loopContextWith ioaction) context
263267 loc <- getLocation
264268 when (loc /= RemoteNode) $ setLocation WaitingParent
265269 return Nothing