f3e1d59 · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Modularize, separate stream impl specific code

f3e1d59Harendra Kumar8 years ago

src/Streamly/Core.hs+62−49

compare with master

unifiedsplit

@@ -465,6 +465,20 @@ enqueueLIFO sv q m = do

465 atomicModifyIORefCAS_ (needDoorBell sv) (const False)

466 void $ tryPutMVar (outputDoorBell sv) ()

467

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+

468workLoopLIFO :: MonadIO m => SVar m a -> IORef [Stream m a] -> m ()

482workLoopLIFO :: MonadIO m => SVar m a -> IORef [Stream m a] -> m ()

469workLoopLIFO sv q = run

483workLoopLIFO sv q = run

470

484

@@ -474,16 +488,7 @@ workLoopLIFO sv q = run

474 work <- dequeue

488 work <- dequeue

475 case work of

489 case work of

476 Nothing -> liftIO $ sendStop sv

490 Nothing -> liftIO $ sendStop sv

477− Just m -> (runStream m) (Just sv) run single yield

491+ Just m -> runStreamLIFO sv q m run

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

487

492

488 dequeue = liftIO $ atomicModifyIORefCAS q $ \case

493 dequeue = liftIO $ atomicModifyIORefCAS q $ \case

489 [] -> ([], Nothing)

494 [] -> ([], Nothing)

@@ -512,25 +517,29 @@ enqueueFIFO sv q m = do

512 atomicModifyIORefCAS_ (needDoorBell sv) (const False)

517 atomicModifyIORefCAS_ (needDoorBell sv) (const False)

513 void $ tryPutMVar (outputDoorBell sv) ()

518 void $ tryPutMVar (outputDoorBell sv) ()

514

519

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+

515workLoopFIFO :: MonadIO m => SVar m a -> LinkedQueue (Stream m a) -> m ()

533workLoopFIFO :: MonadIO m => SVar m a -> LinkedQueue (Stream m a) -> m ()

516workLoopFIFO sv q = run

534workLoopFIFO sv q = run

517

535

518 where

536 where

519

537

520 run = do

538 run = do

521− work <- dequeue

539+ work <- liftIO $ tryPopR q

522 case work of

540 case work of

523 Nothing -> liftIO $ sendStop sv

541 Nothing -> liftIO $ sendStop sv

524− Just m -> (runStream m) (Just sv) run single yield

542+ Just m -> runStreamFIFO sv q m run

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

534

543

535-------------------------------------------------------------------------------

544-------------------------------------------------------------------------------

536-- Parallel

545-- Parallel

@@ -644,6 +653,35 @@ enqueueAhead sv q m = do

644--

653--

645-- XXX review for livelock

654-- XXX review for livelock

646--

655--

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+

647workLoopAhead :: MonadIO m

685workLoopAhead :: MonadIO m

648 => SVar m a

686 => SVar m a

649 -> IORef ([Stream m a], Int)

687 -> IORef ([Stream m a], Int)

@@ -653,8 +691,6 @@ workLoopAhead sv q heap = runHeap

653

691

654 where

692 where

655

693

656− maxHeap = 1500

657−

658 toHeap seqNo ent = do

694 toHeap seqNo ent = do

659 hp <- liftIO $ atomicModifyIORefCAS heap $ \(h, snum) ->

695 hp <- liftIO $ atomicModifyIORefCAS heap $ \(h, snum) ->

660 ((H.insert (Entry seqNo ent) h, snum), h)

696 ((H.insert (Entry seqNo ent) h, snum), h)

@@ -686,7 +722,7 @@ workLoopAhead sv q heap = runHeap

686

722

687 {-# INLINE runQueueToken #-}

723 {-# INLINE runQueueToken #-}

688 runQueueToken prevSeqNo = do

724 runQueueToken prevSeqNo = do

689− work <- dequeue

725+ work <- dequeueAhead q

690 case work of

726 case work of

691 Nothing -> do

727 Nothing -> do

692 liftIO $ atomicModifyIORefCAS_ heap $ \(h, _) ->

728 liftIO $ atomicModifyIORefCAS_ heap $ \(h, _) ->

@@ -705,7 +741,7 @@ workLoopAhead sv q heap = runHeap

705 (singleToHeap seqNo)

741 (singleToHeap seqNo)

706 (yieldToHeap seqNo)

742 (yieldToHeap seqNo)

707 runQueueNoToken = do

743 runQueueNoToken = do

708− work <- dequeue

744+ work <- dequeueAhead q

709 case work of

745 case work of

710 Nothing -> runHeap

746 Nothing -> runHeap

711 Just (m, seqNo) -> do

747 Just (m, seqNo) -> do

@@ -730,7 +766,7 @@ workLoopAhead sv q heap = runHeap

730 ent <- liftIO $ dequeueFromHeap heap

766 ent <- liftIO $ dequeueFromHeap heap

731 case ent of

767 case ent of

732 Nothing -> do

768 Nothing -> do

733− done <- queueEmpty q

769+ done <- queueEmptyAhead q

734 if done

770 if done

735 then liftIO $ sendStop sv

771 then liftIO $ sendStop sv

736 else runQueueNoToken

772 else runQueueNoToken

@@ -742,29 +778,6 @@ workLoopAhead sv q heap = runHeap

742 (singleOutput seqNo)

778 (singleOutput seqNo)

743 (yieldOutput seqNo)

779 (yieldOutput seqNo)

744

780

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−

768-------------------------------------------------------------------------------

781-------------------------------------------------------------------------------

769-- WAhead

782-- WAhead

770-------------------------------------------------------------------------------

783-------------------------------------------------------------------------------