@@ -208,6 +208,46 @@ doFork action exHandler =
208 -- XXX restore state here?
208 -- XXX restore state here?
209 return ()
209 return ()
211+{-# NOINLINE forkWorker #-}
212+forkWorker :: MonadAsync m => Maybe (Context m a) -> m ()
214+ let c = fromJust ctx
215+ let q = outputQueue c
216+ tid <- doFork (push c) (handleChildException q)
217+ liftIO $ atomically $ writeTBQueue q (ChildCreate tid)
219+-- Note: This is performance sensitive code.
220+{-# INLINE queueWork #-}
221+queueWork :: MonadAsync m
222+ => Maybe (Context m a) -> TBQueue (AsyncT m a) -> AsyncT m a -> m ()
223+queueWork ctx q m = do
224+ -- To guarantee deadlock avoidance we need to dispatch a worker when the
225+ -- workQueue goes full. Otherwise all the worker threads might be waiting
226+ -- on the queue and never wakeup.
228+ -- TBD If we run out of threads we can also evaluate the action completely
229+ -- right here, disallowing any further child workers and turning the
230+ -- parallel composition into interleaved serial composition.
231+ workQFull <- liftIO $ atomically $ isFullTBQueue q
232+ when (workQFull) $ forkWorker ctx
233+ liftIO $ atomically $ writeTBQueue q m
235+-- Note: This is performance sensitive code.
236+{-# INLINE dequeueLoop #-}
237+dequeueLoop :: MonadAsync m
238+ => Maybe (Context m a) -> TBQueue (AsyncT m a) -> Bool -> AsyncT m a
239+dequeueLoop ctx q fair = AsyncT $ \_ stp yld -> do
240+ work <- liftIO $ atomically $ tryReadTBQueue q
244+ let stop = (runAsyncT (dequeueLoop ctx q fair)) Nothing stp yld
245+ continue a c = yld a c (Just (dequeueLoop ctx q fair))
246+ yield a c Nothing = continue a c
247+ yield a c r | not fair = yld a c r
248+ yield a c (Just r) = queueWork ctx q r >> continue a c
249+ (runAsyncT m) ctx stop yield
211{-# NOINLINE push #-}
251{-# NOINLINE push #-}
212push :: MonadAsync m => Context m a -> m ()
252push :: MonadAsync m => Context m a -> m ()
213push ctx = run (Just ctx) (dequeueLoop (Just ctx) (workQueue ctx)
253push ctx = run (Just ctx) (dequeueLoop (Just ctx) (workQueue ctx)
@@ -378,45 +418,6 @@ pullFork m1 m2 fair = AsyncT $ \_ stp yld -> do
378-- residual work back to the dispatcher. It will also consume a lot of
418-- residual work back to the dispatcher. It will also consume a lot of
379-- memory due to queueing of all the work before execution starts.
419-- memory due to queueing of all the work before execution starts.
381−{-# NOINLINE forkWorker #-}
382−forkWorker :: MonadAsync m => Maybe (Context m a) -> m ()
384− let c = fromJust ctx
385− let q = outputQueue c
386− tid <- doFork (push c) (handleChildException q)
387− liftIO $ atomically $ writeTBQueue q (ChildCreate tid)
389−-- Note: This is performance sensitive code.
390−{-# INLINE queueWork #-}
391−queueWork :: MonadAsync m => Maybe (Context m a) -> TBQueue (AsyncT m a) -> AsyncT m a -> m ()
392−queueWork ctx q m = do
393− -- To guarantee deadlock avoidance we need to dispatch a worker when the
394− -- workQueue goes full. Otherwise all the worker threads might be waiting
395− -- on the queue and never wakeup.
397− -- TBD If we run out of threads we can also evaluate the action completely
398− -- right here, disallowing any further child workers and turning the
399− -- parallel composition into interleaved serial composition.
400− workQFull <- liftIO $ atomically $ isFullTBQueue q
401− when (workQFull) $ forkWorker ctx
402− liftIO $ atomically $ writeTBQueue q m
404−-- Note: This is performance sensitive code.
405−{-# INLINE dequeueLoop #-}
406−dequeueLoop :: MonadAsync m
407− => Maybe (Context m a) -> TBQueue (AsyncT m a) -> Bool -> AsyncT m a
408−dequeueLoop ctx q fair = AsyncT $ \_ stp yld -> do
409− work <- liftIO $ atomically $ tryReadTBQueue q
413− let stop = (runAsyncT (dequeueLoop ctx q fair)) Nothing stp yld
414− continue a c = yld a c (Just (dequeueLoop ctx q fair))
415− yield a c Nothing = continue a c
416− yield a c r | not fair = yld a c r
417− yield a c (Just r) = queueWork ctx q r >> continue a c
418− (runAsyncT m) ctx stop yield
420-- Note: This is designed to scale for right associated compositions,
421-- Note: This is designed to scale for right associated compositions,
421-- therefore always use a right fold for folding bigger structures.
422-- therefore always use a right fold for folding bigger structures.
422{-# INLINE parallel #-}
423{-# INLINE parallel #-}