@@ -465,6 +465,20 @@ enqueueLIFO sv q m = do
465465 atomicModifyIORefCAS_ (needDoorBell sv) (const False)
466466 void $ tryPutMVar (outputDoorBell sv) ()
467467
468+{-# INLINE runStreamLIFO #-}
469+runStreamLIFO :: MonadIO m
470+ => SVar m a -> IORef [Stream m a] -> Stream m a -> m () -> m ()
471+runStreamLIFO sv q m stop = runStream m (Just sv) stop single yield
472+ where
473+ single a = do
474+ res <- liftIO $ send sv (ChildYield a)
475+ if res then stop else liftIO $ sendStop sv
476+ yield a r = do
477+ res <- liftIO $ send sv (ChildYield a)
478+ if res
479+ then (runStream r) (Just sv) stop single yield
480+ else liftIO $ enqueueLIFO sv q r >> sendStop sv
481+
468482workLoopLIFO :: MonadIO m => SVar m a -> IORef [Stream m a] -> m ()
469483workLoopLIFO sv q = run
470484
@@ -474,16 +488,7 @@ workLoopLIFO sv q = run
474488 work <- dequeue
475489 case work of
476490 Nothing -> liftIO $ sendStop sv
477− Just m -> (runStream m) (Just sv) run single yield
478−
479− single a = do
480− res <- liftIO $ send sv (ChildYield a)
481− if res then run else liftIO $ sendStop sv
482− yield a r = do
483− res <- liftIO $ send sv (ChildYield a)
484− if res
485− then (runStream r) (Just sv) run single yield
486− else liftIO $ enqueueLIFO sv q r >> sendStop sv
491+ Just m -> runStreamLIFO sv q m run
487492
488493 dequeue = liftIO $ atomicModifyIORefCAS q $ \case
489494 [] -> ([], Nothing)
@@ -512,25 +517,29 @@ enqueueFIFO sv q m = do
512517 atomicModifyIORefCAS_ (needDoorBell sv) (const False)
513518 void $ tryPutMVar (outputDoorBell sv) ()
514519
520+{-# INLINE runStreamFIFO #-}
521+runStreamFIFO :: MonadIO m
522+ => SVar m a -> LinkedQueue (Stream m a) -> Stream m a -> m () -> m ()
523+runStreamFIFO sv q m stop = runStream m (Just sv) stop single yield
524+ where
525+ single a = do
526+ res <- liftIO $ send sv (ChildYield a)
527+ if res then stop else liftIO $ sendStop sv
528+ yield a r = do
529+ res <- liftIO $ send sv (ChildYield a)
530+ liftIO (enqueueFIFO sv q r)
531+ if res then stop else liftIO $ sendStop sv
532+
515533workLoopFIFO :: MonadIO m => SVar m a -> LinkedQueue (Stream m a) -> m ()
516534workLoopFIFO sv q = run
517535
518536 where
519537
520538 run = do
521− work <- dequeue
539+ work <- liftIO $ tryPopR q
522540 case work of
523541 Nothing -> liftIO $ sendStop sv
524− Just m -> (runStream m) (Just sv) run single yield
525−
526− dequeue = liftIO $ tryPopR q
527− single a = do
528− res <- liftIO $ send sv (ChildYield a)
529− if res then run else liftIO $ sendStop sv
530− yield a r = do
531− res <- liftIO $ send sv (ChildYield a)
532− liftIO (enqueueFIFO sv q r)
533− if res then run else liftIO $ sendStop sv
542+ Just m -> runStreamFIFO sv q m run
534543
535544-------------------------------------------------------------------------------
536545-- Parallel
@@ -644,6 +653,35 @@ enqueueAhead sv q m = do
644653--
645654-- XXX review for livelock
646655--
656+maxHeap :: Int
657+maxHeap = 1500
658+
659+queueEmptyAhead :: MonadIO m => IORef ([Stream m a], Int) -> m Bool
660+queueEmptyAhead q = liftIO $ do
661+ (xs, _) <- readIORef q
662+ return $ null xs
663+
664+dequeueAhead :: MonadIO m
665+ => IORef ([Stream m a], Int) -> m (Maybe (Stream m a, Int))
666+dequeueAhead q = liftIO $ do
667+ atomicModifyIORefCAS q $ \case
668+ ([], n) -> (([], n), Nothing)
669+ (x : [], n) -> (([], n), Just (x, n))
670+ _ -> error "more than one item on queue"
671+
672+dequeueFromHeap
673+ :: IORef (Heap (Entry Int (AheadHeapEntry m a)), Int)
674+ -> IO (Maybe (Entry Int (AheadHeapEntry m a)))
675+dequeueFromHeap hpRef = do
676+ atomicModifyIORefCAS hpRef $ \hp@(h, snum) -> do
677+ let r = H.uncons h
678+ case r of
679+ Nothing -> (hp, Nothing)
680+ Just (ent@(Entry seqNo _ev), hp') ->
681+ if (seqNo == snum)
682+ then ((hp', seqNo), Just ent)
683+ else (hp, Nothing)
684+
647685workLoopAhead :: MonadIO m
648686 => SVar m a
649687 -> IORef ([Stream m a], Int)
@@ -653,8 +691,6 @@ workLoopAhead sv q heap = runHeap
653691
654692 where
655693
656− maxHeap = 1500
657−
658694 toHeap seqNo ent = do
659695 hp <- liftIO $ atomicModifyIORefCAS heap $ \(h, snum) ->
660696 ((H.insert (Entry seqNo ent) h, snum), h)
@@ -686,7 +722,7 @@ workLoopAhead sv q heap = runHeap
686722
687723 {-# INLINE runQueueToken #-}
688724 runQueueToken prevSeqNo = do
689− work <- dequeue
725+ work <- dequeueAhead q
690726 case work of
691727 Nothing -> do
692728 liftIO $ atomicModifyIORefCAS_ heap $ \(h, _) ->
@@ -705,7 +741,7 @@ workLoopAhead sv q heap = runHeap
705741 (singleToHeap seqNo)
706742 (yieldToHeap seqNo)
707743 runQueueNoToken = do
708− work <- dequeue
744+ work <- dequeueAhead q
709745 case work of
710746 Nothing -> runHeap
711747 Just (m, seqNo) -> do
@@ -730,7 +766,7 @@ workLoopAhead sv q heap = runHeap
730766 ent <- liftIO $ dequeueFromHeap heap
731767 case ent of
732768 Nothing -> do
733− done <- queueEmpty q
769+ done <- queueEmptyAhead q
734770 if done
735771 then liftIO $ sendStop sv
736772 else runQueueNoToken
@@ -742,29 +778,6 @@ workLoopAhead sv q heap = runHeap
742778 (singleOutput seqNo)
743779 (yieldOutput seqNo)
744780
745− queueEmpty qu = liftIO $ do
746− (xs, _) <- readIORef qu
747− return $ null xs
748−
749− dequeue = liftIO $ do
750− atomicModifyIORefCAS q $ \case
751− ([], n) -> (([], n), Nothing)
752− (x : [], n) -> (([], n), Just (x, n))
753− _ -> error "more than one item on queue"
754−
755− dequeueFromHeap
756− :: IORef (Heap (Entry Int (AheadHeapEntry m a)), Int)
757− -> IO (Maybe (Entry Int (AheadHeapEntry m a)))
758− dequeueFromHeap hpRef = do
759− atomicModifyIORefCAS hpRef $ \hp@(h, snum) -> do
760− let r = H.uncons h
761− case r of
762− Nothing -> (hp, Nothing)
763− Just (ent@(Entry seqNo _ev), hp') ->
764− if (seqNo == snum)
765− then ((hp', seqNo), Just ent)
766− else (hp, Nothing)
767−
768781-------------------------------------------------------------------------------
769782-- WAhead
770783-------------------------------------------------------------------------------