11deaa7 · streamly · git.iko.soy

22 SomeException (..), catch)

23import qualified Control.Exception.Lifted as EL

24import Control.Monad.IO.Class (MonadIO (..))

25−import Control.Monad.State (StateT, get, gets, modify, put,

25+import Control.Monad.State (get, gets, modify, put,

26 runStateT, when)

27import Control.Monad.Trans.Class (MonadTrans (lift))

28import Control.Monad.Trans.Control (MonadBaseControl, liftBaseWith)

179 tid <- myThreadId

180 liftIO $ atomically $ writeTChan p tid

181

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+--

182−forkMaybe :: (MonadBaseControl IO m, MonadIO m)

185+runContext :: (MonadBaseControl IO m, MonadIO m)

183 => (Context -> m ()) -> Context -> m ()

186 => (Context -> m ()) -> Context -> m ()

184−forkMaybe runCtx context = do

187+runContext runCtx context = do

185 gotCredit <- liftIO $ waitQSemB (threadCredit context)

188 gotCredit <- liftIO $ waitQSemB (threadCredit context)

186 pending <- liftIO $ readIORef $ pendingThreads context

189 pending <- liftIO $ readIORef $ pendingThreads context

187 case gotCredit of

190 case gotCredit of

195 -- async thread, cannot use sync for those.

198 -- async thread, cannot use sync for those.

196 waitForOneChild (childChannel context)

199 waitForOneChild (childChannel context)

197 (pendingThreads context)

200 (pendingThreads context)

198− forkMaybe runCtx context

201+ runContext runCtx context

199 True -> forkContextWith runCtx context

202 True -> forkContextWith runCtx context

200

203

201-- | 'StreamData' represents a task in a task stream being generated.

204-- | 'StreamData' represents a task in a task stream being generated.

209-- The current model is to start a new thread for every task. The input is

212-- The current model is to start a new thread for every task. The input is

210-- provided at the time of the creation and therefore no synchronization is

213-- provided at the time of the creation and therefore no synchronization is

211-- needed compared to a pool of threads contending to get the input from a

214-- 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?

213--

217--

214-- When the task is over the outputs need to be collected and that requires

218-- When the task is over the outputs need to be collected and that requires

215-- synchronization irrespective of a thread pool model or per task new thread

219-- synchronization irrespective of a thread pool model or per task new thread

230 let ctx' = updateContextEvent ctx streamData

234 let ctx' = updateContextEvent ctx streamData

231 case streamData of

235 case streamData of

232 SMore _ -> do

236 SMore _ -> do

233− forkMaybe resume ctx'

237+ runContext resume ctx'

234 loopContextWith ioaction ctx

238 loopContextWith ioaction ctx

235 _ -> resume ctx'

239 _ -> resume ctx'

236

240

259 -- we have to execute the io action and generate an event and then continue

263 -- we have to execute the io action and generate an event and then continue

260 -- in this thread or a new thread.

264 -- in this thread or a new thread.

261 Nothing -> do

265 Nothing -> do

262− lift $ forkMaybe (loopContextWith ioaction) context

266+ lift $ runContext (loopContextWith ioaction) context

263 loc <- getLocation

267 loc <- getLocation

264 when (loc /= RemoteNode) $ setLocation WaitingParent

268 when (loc /= RemoteNode) $ setLocation WaitingParent

265 return Nothing

269 return Nothing