@@ -52,7 +52,7 @@ import Control.Monad.State.Class (MonadState(..))
5252import Control.Monad.Trans.Class (MonadTrans (lift))
5353import Control.Monad.Trans.Control (MonadBaseControl, liftBaseWith)
5454import Data.Concurrent.Queue.MichaelScott (LinkedQueue, newQ, pushL,
55− tryPopR, nullQ)
55+ tryPopR)
5656import Data.Functor (void)
5757import Data.IORef (IORef, modifyIORef, newIORef,
5858 writeIORef, readIORef)
@@ -100,9 +100,8 @@ data ChildEvent a =
100100data Context m a =
101101 Context { outputQueue :: IORef [ChildEvent a]
102102 , synchOutQ :: MVar Bool -- wakeup mechanism for outQ
103− , dequeueLoop :: AsyncT m a
104− , enqueueWork :: AsyncT m a -> IO ()
105− , hasNoWork :: IO Bool
103+ , enqueue :: AsyncT m a -> IO ()
104+ , runqueue :: m ()
106105 , runningThreads :: IORef (Set ThreadId)
107106 , doneThreads :: IORef (Set ThreadId)
108107 }
@@ -224,79 +223,56 @@ send ctx msg = liftIO $ do
224223 -- store has finished otherwise we can have lost wakeup problems.
225224 void $ tryPutMVar (synchOutQ ctx) True
226225
226+{-# INLINE sendStop #-}
227+sendStop :: MonadIO m => Context m a -> m ()
228+sendStop ctx = liftIO myThreadId >>= \tid -> send ctx (ChildStop tid Nothing)
229+
227230-- Note: Left associated operations can grow this queue to a large size
228231{-# INLINE enqueueLIFO #-}
229232enqueueLIFO :: IORef [AsyncT m a] -> AsyncT m a -> IO ()
230233enqueueLIFO q m = atomicModifyIORefCAS_ q $ \ ms -> m : ms
231234
232−{-# INLINE dequeueLIFO #-}
233−dequeueLIFO :: IORef [AsyncT m a] -> IO (Maybe (AsyncT m a))
234−dequeueLIFO q =
235− atomicModifyIORefCAS q $ \ ms ->
235+runqueueLIFO :: MonadIO m => Context m a -> IORef [AsyncT m a] -> m ()
236+runqueueLIFO ctx q = run
237+
238+ where
239+
240+ run = do
241+ work <- dequeue
242+ case work of
243+ Nothing -> sendStop ctx
244+ Just m -> (runAsyncT m) (Just ctx) run yield
245+
246+ sendit a = send ctx (ChildYield a)
247+ yield a _ Nothing = sendit a >> run
248+ yield a c (Just r) = sendit a >> (runAsyncT r) c run yield
249+
250+ dequeue = liftIO $ atomicModifyIORefCAS q $ \ ms ->
236251 case ms of
237252 [] -> ([], Nothing)
238253 x : xs -> (xs, Just x)
239254
240−{-# INLINE isEmptyLIFO #-}
241−isEmptyLIFO :: IORef [AsyncT m a] -> IO Bool
242−isEmptyLIFO q = do
243− work <- readIORef q
244− return $ case work of
245− [] -> True
246− _ -> False
255+{-# INLINE enqueueFIFO #-}
256+enqueueFIFO :: LinkedQueue (AsyncT m a) -> AsyncT m a -> IO ()
257+enqueueFIFO = pushL
247258
248−-- Note: This is performance sensitive code.
249−{-# INLINE dequeueLoopLIFO #-}
250−dequeueLoopLIFO :: MonadAsync m
251− => Maybe (Context m a)
252− -> IORef [AsyncT m a]
253− -> AsyncT m a
254−dequeueLoopLIFO ctx q = AsyncT $ \_ stp yld -> do
255− work <- liftIO $ dequeueLIFO q
256− case work of
257− Nothing -> stp
258− Just m -> do
259− let continue = dequeueLoopLIFO ctx q
260− stop = (runAsyncT continue) Nothing stp yld
261− yield a c Nothing = yld a c (Just continue)
262− yield a c r = yld a c r
263− (runAsyncT m) ctx stop yield
264−
265−{-# INLINE dequeueLoopFIFO #-}
266−dequeueLoopFIFO :: MonadAsync m
267− => Maybe (Context m a)
268− -> LinkedQueue (AsyncT m a)
269− -> AsyncT m a
270−dequeueLoopFIFO ctx q = AsyncT $ \_ stp yld -> do
271− work <- liftIO $ tryPopR q
272− case work of
273− Nothing -> stp
274− Just m -> do
275− let continue = dequeueLoopFIFO ctx q
276− stop = (runAsyncT continue) Nothing stp yld
277− yield a c Nothing = yld a c (Just continue)
278− yield a c (Just r) = do
279− liftIO $ pushL q r
280− yld a c (Just continue)
281− (runAsyncT m) ctx stop yield
282−
283−{-# NOINLINE push #-}
284−push :: MonadAsync m => Context m a -> m ()
285−push ctx = run (Just ctx) (dequeueLoop ctx)
259+runqueueFIFO :: MonadIO m => Context m a -> LinkedQueue (AsyncT m a) -> m ()
260+runqueueFIFO ctx q = run
286261
287262 where
288263
289− stop = do
290− done <- liftIO $ hasNoWork ctx
291− if done then do
292− tid <- liftIO $ myThreadId
293− send ctx (ChildStop tid Nothing)
294− else push ctx
295− yield a _ Nothing = do
296− tid <- liftIO myThreadId
297− send ctx (ChildDone tid a)
298− yield a c (Just r) = send ctx (ChildYield a) >> run c r
299− run c m = (runAsyncT m) c stop yield
264+ run = do
265+ work <- dequeue
266+ case work of
267+ Nothing -> sendStop ctx
268+ Just m -> (runAsyncT m) (Just ctx) run yield
269+
270+ dequeue = liftIO $ tryPopR q
271+ sendit a = send ctx (ChildYield a)
272+ yield a _ Nothing = sendit a >> run
273+ -- XXX do we need to save the context as well?
274+ -- we can enqueue "(runAsyncT m) ctx run yield" instead
275+ yield a c (Just r) = sendit a >> liftIO (enqueueFIFO q r) >> run
300276
301277-- Thread tracking has a significant performance overhead (~20% on empty
302278-- threads, it will be lower for heavy threads). It is needed for two reasons:
@@ -354,7 +330,7 @@ handleChildException ctx e = do
354330{-# NOINLINE pushWorker #-}
355331pushWorker :: MonadAsync m => Context m a -> m ()
356332pushWorker ctx = do
357− tid <- doFork (push ctx) (handleChildException ctx)
333+ tid <- doFork (runqueue ctx) (handleChildException ctx)
358334 liftIO $ modifyIORef (runningThreads ctx) $ (\s -> S.insert tid s)
359335
360336{-# INLINE sendWorkerWait #-}
@@ -407,7 +383,7 @@ pullWorker ctx = AsyncT $ \pctx stp yld -> do
407383-- computation pulls from a Channel while m1 and m2 push to the channel.
408384{-# NOINLINE pullFork #-}
409385pullFork :: MonadAsync m => AsyncT m a -> AsyncT m a -> Bool -> AsyncT m a
410−pullFork m1 m2 fair = AsyncT $ \_ stp yld -> do
386+pullFork m1 m2 fifo = AsyncT $ \_ stp yld -> do
411387 ctx <- liftIO $ newContext
412388 pushWorker ctx >> (runAsyncT (pullWorker ctx)) Nothing stp yld
413389
@@ -419,7 +395,7 @@ pullFork m1 m2 fair = AsyncT $ \_ stp yld -> do
419395 running <- newIORef S.empty
420396 done <- newIORef S.empty
421397
422− case fair of
398+ case fifo of
423399 True -> do
424400 q <- newQ
425401 pushL q m1 >> pushL q m2
@@ -428,9 +404,8 @@ pullFork m1 m2 fair = AsyncT $ \_ stp yld -> do
428404 , synchOutQ = outQMv
429405 , runningThreads = running
430406 , doneThreads = done
431− , dequeueLoop = dequeueLoopFIFO (Just ctx) q
432− , enqueueWork = pushL q
433− , hasNoWork = nullQ q
407+ , runqueue = runqueueFIFO ctx q
408+ , enqueue = pushL q
434409 }
435410 in return ctx
436411 False -> do
@@ -441,9 +416,8 @@ pullFork m1 m2 fair = AsyncT $ \_ stp yld -> do
441416 , synchOutQ = outQMv
442417 , runningThreads = running
443418 , doneThreads = done
444− , dequeueLoop = dequeueLoopLIFO (Just ctx) q
445− , enqueueWork = enqueueLIFO q
446− , hasNoWork = isEmptyLIFO q
419+ , runqueue = runqueueLIFO ctx q
420+ , enqueue = enqueueLIFO q
447421 }
448422 in return ctx
449423
@@ -482,15 +456,14 @@ pullFork m1 m2 fair = AsyncT $ \_ stp yld -> do
482456-- therefore always use a right fold for folding bigger structures.
483457{-# INLINE parallel #-}
484458parallel :: MonadAsync m => AsyncT m a -> AsyncT m a -> Bool -> AsyncT m a
485−parallel m1 m2 fair = AsyncT $ \ctx stp yld -> do
459+parallel m1 m2 fifo = AsyncT $ \ctx stp yld -> do
486460 case ctx of
487− Nothing -> (runAsyncT (pullFork m1 m2 fair)) Nothing stp yld
461+ Nothing -> (runAsyncT (pullFork m1 m2 fifo)) Nothing stp yld
488462 Just c -> do
489− let dqloop = (runAsyncT (dequeueLoop c)) Nothing stp yld
490463 -- XXX We can statically decide which function to call here, that
491464 -- will inline the call and improve perf a little bit.
492− liftIO $ (enqueueWork c) m2
493− (runAsyncT m1) ctx dqloop yld
465+ liftIO $ (enqueue c) m2
466+ (runAsyncT m1) ctx stp yld
494467
495468instance MonadAsync m => Alternative (AsyncT m) where
496469 empty = mempty