307fa9f · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Add direct style append/interleave/concatUnfoldInterleave etc

307fa9fHarendra Kumar7 years ago

benchmark/Linear.hs+49−17

compare with master

unifiedsplit

@@ -149,10 +149,6 @@ main =

149 -- These are essentially cons and consM

150 , benchIOSrc serially "fromFoldable" Ops.sourceFromFoldable

151 , benchIOSrc serially "fromFoldableM" Ops.sourceFromFoldableM

152− -- These are essentially appends

153− , benchIOSrc serially "foldMapWith" Ops.sourceFoldMapWith

154− , benchIOSrc serially "foldMapWithM" Ops.sourceFoldMapWithM

155− , benchIOSrc serially "foldMapM" Ops.sourceFoldMapM

156 ]

152 ]

157 , bgroup "elimination"

153 , bgroup "elimination"

158 [ bgroup "reduce"

154 [ bgroup "reduce"

@@ -257,6 +253,13 @@ main =

257 , benchIOSink "and" (\s -> S.runFold FL.and (S.map (<= Ops.maxValue) s))

253 , benchIOSink "and" (\s -> S.runFold FL.and (S.map (<= Ops.maxValue) s))

258 , benchIOSink "or" (\s -> S.runFold FL.or (S.map (> Ops.maxValue) s))

254 , benchIOSink "or" (\s -> S.runFold FL.or (S.map (> Ops.maxValue) s))

259 ]

255 ]

256+ , bgroup "fold-multi-stream"

257+ [ benchIOSink1 "eqBy" Ops.eqBy

258+ , benchIOSink1 "cmpBy" Ops.cmpBy

259+ , benchIOSink "isPrefixOf" Ops.isPrefixOf

260+ , benchIOSink "isSubsequenceOf" Ops.isSubsequenceOf

261+ , benchIOSink "stripPrefix" Ops.stripPrefix

262+ ]

260 , bgroup "folds-transforms"

263 , bgroup "folds-transforms"

261 [ benchIOSink "drain" (S.runFold FL.drain)

264 [ benchIOSink "drain" (S.runFold FL.drain)

262 , benchIOSink "lmap" (S.runFold (Internal.lmap (+1) FL.drain))

265 , benchIOSink "lmap" (S.runFold (Internal.lmap (+1) FL.drain))

@@ -345,19 +348,48 @@ main =

345 , benchIOSink "intersperse" (Ops.intersperse 4)

348 , benchIOSink "intersperse" (Ops.intersperse 4)

346 , benchIOSink "insertBy" (Ops.insertBy 4)

349 , benchIOSink "insertBy" (Ops.insertBy 4)

347 ]

350 ]

348− , bgroup "multi-stream"

351+ , bgroup "joining"

349− [ benchIOSink1 "eqBy" Ops.eqBy

352+ [ benchIOSrc1 "zip (2x50K)" (Ops.zip 50000)

350− , benchIOSink1 "cmpBy" Ops.cmpBy

353+ , benchIOSrc1 "zipM (2x50K)" (Ops.zipM 50000)

351− , benchIOSink "zip" Ops.zip

354+ , benchIOSrc1 "mergeBy (2x50K)" (Ops.mergeBy 50000)

352− , benchIOSink1 "zipM" Ops.zipM

355+ , benchIOSrc1 "serial (2x50K)" (Ops.serial2 50000)

353− , benchIOSink1 "mergeBy" Ops.mergeBy

356+ , benchIOSrc1 "append (2x50K)" (Ops.append2 50000)

354− , benchIOSink "isPrefixOf" Ops.isPrefixOf

357+ , benchIOSrc1 "serial (2x2x25K)" (Ops.serial4 25000)

355− , benchIOSink "isSubsequenceOf" Ops.isSubsequenceOf

358+ , benchIOSrc1 "append (2x2x25K)" (Ops.append4 25000)

356− , benchIOSink "stripPrefix" Ops.stripPrefix

359+ , benchIOSrc1 "wSerial (2x50K)" Ops.wSerial2

357− , benchIOSrc1 "concatMap1xN" Ops.concatMap

360+ , benchIOSrc1 "interleave (2x50K)" Ops.interleave2

358− , benchIOSrc1 "concatMapPure1xN" Ops.concatMapPure1xN

361+ , benchIOSrc1 "roundRobin (2x50K)" Ops.roundRobin2

359− , benchIOSrc1 "concatMapNxN" Ops.concatMapNxN

362+ ]

360− , benchIOSrc1 "concatMapRepl4xN" Ops.concatMapRepl4xN

363+ , bgroup "concat-foldable"

364+ [ benchIOSrc serially "foldMapWith (1x100K)" Ops.sourceFoldMapWith

365+ , benchIOSrc serially "foldMapWithM (1x100K)" Ops.sourceFoldMapWithM

366+ , benchIOSrc serially "foldMapM (1x100K)" Ops.sourceFoldMapM

367+ , benchIOSrc serially "foldWithConcatMapId (1x100K)" Ops.sourceConcatMapId

368+ ]

369+ , bgroup "concat-serial"

370+ [ benchIOSrc1 "concatMapPure (2x50K)" (Ops.concatMapPure 2 50000)

371+ , benchIOSrc1 "concatMap (2x50K)" (Ops.concatMap 2 50000)

372+ , benchIOSrc1 "concatMap (50Kx2)" (Ops.concatMap 50000 2)

373+ , benchIOSrc1 "concatMapRepl (25Kx4)" Ops.concatMapRepl4xN

374+ , benchIOSrc1 "concatUnfoldRepl (25Kx4)" Ops.concatUnfoldRepl4xN

375+

376+ , benchIOSrc1 "concatMapWithSerial (2x50K)"

377+ (Ops.concatMapWithSerial 2 50000)

378+ , benchIOSrc1 "concatMapWithSerial (50Kx2)"

379+ (Ops.concatMapWithSerial 50000 2)

380+

381+ , benchIOSrc1 "concatMapWithAppend (2x50K)"

382+ (Ops.concatMapWithAppend 2 50000)

383+ ]

384+ , bgroup "concat-interleave"

385+ [ benchIOSrc1 "concatMapWithWSerial (2x50K)"

386+ (Ops.concatMapWithWSerial 2 50000)

387+ , benchIOSrc1 "concatMapWithWSerial (50Kx2)"

388+ (Ops.concatMapWithWSerial 50000 2)

389+ , benchIOSrc1 "concatUnfoldInterleaveRepl (25Kx4)"

390+ Ops.concatUnfoldInterleaveRepl4xN

391+ , benchIOSrc1 "concatUnfoldRoundrobinRepl (25Kx4)"

392+ Ops.concatUnfoldRoundrobinRepl4xN

361 ]

393 ]

362 -- scanl-map and foldl-map are equivalent to the scan and fold in the foldl

394 -- scanl-map and foldl-map are equivalent to the scan and fold in the foldl

363 -- library. If scan/fold followed by a map is efficient enough we may not

395 -- library. If scan/fold followed by a map is efficient enough we may not

src/Streamly/Benchmark/Prelude.hs+179−50

@@ -38,13 +38,15 @@ import qualified Data.Foldable as F

38import qualified GHC.Exts as GHC

40#ifdef INSPECTION

41−import Streamly.Streams.StreamD.Type (Step(..))

42import Test.Inspection

41import Test.Inspection

42+

43+import qualified Streamly.Streams.StreamD as D

43#endif

44#endif

45import qualified Streamly as S hiding (foldMapWith, runStream)

46import qualified Streamly as S hiding (foldMapWith, runStream)

46import qualified Streamly.Prelude as S

47import qualified Streamly.Prelude as S

47import qualified Streamly.Internal as Internal

48import qualified Streamly.Internal as Internal

49+import qualified Streamly.Unfold as UF

48-- import qualified Streamly.Pipe as Pipe

50-- import qualified Streamly.Pipe as Pipe

50value, maxValue, value2 :: Int

52value, maxValue, value2 :: Int

@@ -132,6 +134,12 @@ sourceFoldMapM :: (S.IsStream t, Monad m, P.Monoid (t m Int))

132 => Int -> t m Int

134 => Int -> t m Int

133sourceFoldMapM n = F.foldMap (S.yieldM . return) [n..n+value]

135sourceFoldMapM n = F.foldMap (S.yieldM . return) [n..n+value]

134

136

137+{-# INLINE sourceConcatMapId #-}

138+sourceConcatMapId :: (S.IsStream t, Monad m)

139+ => Int -> t m Int

140+sourceConcatMapId n =

141+ S.concatMap P.id $ S.fromFoldable $ P.map (S.yieldM . return) [n..n+value]

142+

135{-# INLINE sourceUnfoldr #-}

143{-# INLINE sourceUnfoldr #-}

136sourceUnfoldr :: (Monad m, S.IsStream t) => Int -> t m Int

144sourceUnfoldr :: (Monad m, S.IsStream t) => Int -> t m Int

137sourceUnfoldr n = S.unfoldr step n

145sourceUnfoldr n = S.unfoldr step n

@@ -503,44 +511,105 @@ iterateDropWhileTrue = iterateSource (S.dropWhile (<= maxValue)) maxIters

503-------------------------------------------------------------------------------

511-------------------------------------------------------------------------------

504

512

505{-# INLINE zip #-}

513{-# INLINE zip #-}

506−zip :: Monad m => Stream m Int -> m ()

514+zip :: Int -> Int -> IO ()

507−zip src = do

515+zip count n =

508− r <- S.tail src

509− let src1 = fromJust r

510− transform (S.zipWith (,) src src1)

516+ S.drain $ S.zipWith (,)

517+ (sourceUnfoldrMN count n)

518+ (sourceUnfoldrMN count (n + 1))

511

519

512−{-# INLINE zipM' #-}

520+#ifdef INSPECTION

513−zipM' :: Monad m => Stream m Int -> m ()

521+inspect $ hasNoTypeClasses 'zip

514−zipM' src = do

522+inspect $ 'zip `hasNoType` ''D.Step

515− r <- S.tail src

523+#endif

516− let src1 = fromJust r

517− transform (S.zipWithM (curry return) src src1)

518

524

519{-# INLINE zipM #-}

525{-# INLINE zipM #-}

520−zipM :: Int -> IO ()

526+zipM :: Int -> Int -> IO ()

521−zipM n = zipM' $ source n

527+zipM count n =

528+ S.drain $ S.zipWithM (curry return)

529+ (sourceUnfoldrMN count n)

530+ (sourceUnfoldrMN count (n + 1))

522

531

523#ifdef INSPECTION

532#ifdef INSPECTION

524inspect $ hasNoTypeClasses 'zipM

533inspect $ hasNoTypeClasses 'zipM

525−-- inspect $ 'zipM `hasNoType` ''Step

534+inspect $ 'zipM `hasNoType` ''D.Step

526#endif

535#endif

527

536

528−{-# INLINE mergeBy' #-}

529−mergeBy' :: Monad m => Stream m Int -> m ()

530−mergeBy' src = do

531− r <- S.tail src

532− let src1 = fromJust r

533− transform (S.mergeBy P.compare src src1)

534−

535{-# INLINE mergeBy #-}

537{-# INLINE mergeBy #-}

536−mergeBy :: Int -> IO ()

538+mergeBy :: Int -> Int -> IO ()

537−mergeBy n = mergeBy' $ source n

539+mergeBy count n =

540+ S.drain $ S.mergeBy P.compare

541+ (sourceUnfoldrMN count n)

542+ (sourceUnfoldrMN count (n + 1))

538

543

539#ifdef INSPECTION

544#ifdef INSPECTION

540inspect $ hasNoTypeClasses 'mergeBy

545inspect $ hasNoTypeClasses 'mergeBy

541−-- inspect $ 'mergeBy `hasNoType` ''Step

546+inspect $ 'mergeBy `hasNoType` ''D.Step

547+#endif

548+

549+{-# INLINE serial2 #-}

550+serial2 :: Int -> Int -> IO ()

551+serial2 count n =

552+ S.drain $ S.serial

553+ (sourceUnfoldrMN count n)

554+ (sourceUnfoldrMN count (n + 1))

555+

556+{-# INLINE serial4 #-}

557+serial4 :: Int -> Int -> IO ()

558+serial4 count n =

559+ S.drain $ S.serial

560+ ((S.serial (sourceUnfoldrMN count n)

561+ (sourceUnfoldrMN count (n + 1))))

562+ ((S.serial (sourceUnfoldrMN count (n+2))

563+ (sourceUnfoldrMN count (n + 3))))

564+

565+{-# INLINE append2 #-}

566+append2 :: Int -> Int -> IO ()

567+append2 count n =

568+ S.drain $ Internal.append

569+ (sourceUnfoldrMN count n)

570+ (sourceUnfoldrMN count (n + 1))

571+

572+{-# INLINE append4 #-}

573+append4 :: Int -> Int -> IO ()

574+append4 count n =

575+ S.drain $ Internal.append

576+ ((Internal.append (sourceUnfoldrMN count n)

577+ (sourceUnfoldrMN count (n + 1))))

578+ ((Internal.append (sourceUnfoldrMN count (n+2))

579+ (sourceUnfoldrMN count (n + 3))))

580+

581+#ifdef INSPECTION

582+inspect $ hasNoTypeClasses 'append2

583+inspect $ 'append2 `hasNoType` ''D.AppendState

542#endif

584#endif

543

585

586+{-# INLINE wSerial2 #-}

587+wSerial2 :: Int -> IO ()

588+wSerial2 n = S.drain $ S.wSerial

589+ (sourceUnfoldrMN (value `div` 2) n)

590+ (sourceUnfoldrMN (value `div` 2) (n + 1))

591+

592+{-# INLINE interleave2 #-}

593+interleave2 :: Int -> IO ()

594+interleave2 n = S.drain $ Internal.interleave

595+ (sourceUnfoldrMN (value `div` 2) n)

596+ (sourceUnfoldrMN (value `div` 2) (n + 1))

597+

598+#ifdef INSPECTION

599+inspect $ hasNoTypeClasses 'interleave2

600+inspect $ 'interleave2 `hasNoType` ''D.InterleaveState

601+#endif

602+

603+{-# INLINE roundRobin2 #-}

604+roundRobin2 :: Int -> IO ()

605+roundRobin2 n = S.drain $ Internal.roundrobin

606+ (sourceUnfoldrMN (value `div` 2) n)

607+ (sourceUnfoldrMN (value `div` 2) (n + 1))

608+

609+#ifdef INSPECTION

610+inspect $ hasNoTypeClasses 'roundRobin2

611+inspect $ 'roundRobin2 `hasNoType` ''D.InterleaveState

612+#endif

544

613

545{-# INLINE isPrefixOf #-}

614{-# INLINE isPrefixOf #-}

546{-# INLINE isSubsequenceOf #-}

615{-# INLINE isSubsequenceOf #-}

@@ -586,7 +655,7 @@ eqBy n = eqBy' (source n)

586

655

587#ifdef INSPECTION

656#ifdef INSPECTION

588inspect $ hasNoTypeClasses 'eqBy

657inspect $ hasNoTypeClasses 'eqBy

589−inspect $ 'eqBy `hasNoType` ''Step

658+inspect $ 'eqBy `hasNoType` ''D.Step

590#endif

659#endif

591

660

592

661

@@ -596,7 +665,7 @@ eqByPure n = eqBy' (sourceUnfoldr n)

596

665

597#ifdef INSPECTION

666#ifdef INSPECTION

598inspect $ hasNoTypeClasses 'eqByPure

667inspect $ hasNoTypeClasses 'eqByPure

599−inspect $ 'eqByPure `hasNoType` ''Step

668+inspect $ 'eqByPure `hasNoType` ''D.Step

600#endif

669#endif

601

670

602{-# INLINE cmpBy' #-}

671{-# INLINE cmpBy' #-}

@@ -609,7 +678,7 @@ cmpBy n = cmpBy' (source n)

609

678

610#ifdef INSPECTION

679#ifdef INSPECTION

611inspect $ hasNoTypeClasses 'cmpBy

680inspect $ hasNoTypeClasses 'cmpBy

612−inspect $ 'cmpBy `hasNoType` ''Step

681+inspect $ 'cmpBy `hasNoType` ''D.Step

613#endif

682#endif

614

683

615{-# INLINE cmpByPure #-}

684{-# INLINE cmpByPure #-}

@@ -618,38 +687,29 @@ cmpByPure n = cmpBy' (sourceUnfoldr n)

618

687

619#ifdef INSPECTION

688#ifdef INSPECTION

620inspect $ hasNoTypeClasses 'cmpByPure

689inspect $ hasNoTypeClasses 'cmpByPure

621−inspect $ 'cmpByPure `hasNoType` ''Step

690+inspect $ 'cmpByPure `hasNoType` ''D.Step

622#endif

691#endif

623

692

624−-- The worst case for concatMap, concat a 1 element stream n times.

625{-# INLINE concatMap #-}

693{-# INLINE concatMap #-}

626−concatMap :: Int -> IO ()

694+concatMap :: Int -> Int -> Int -> IO ()

695+concatMap outer inner n =

627−concatMap n = S.drain $ S.concatMap (\_ -> sourceUnfoldrMN 1 n)

696+ S.drain $ S.concatMap

697+ (\_ -> sourceUnfoldrMN inner n)

628− (sourceUnfoldrMN value n)

698+ (sourceUnfoldrMN outer n)

629

699

630#ifdef INSPECTION

700#ifdef INSPECTION

631inspect $ hasNoTypeClasses 'concatMap

701inspect $ hasNoTypeClasses 'concatMap

632#endif

702#endif

633

703

634−{-# INLINE concatMapPure1xN #-}

704+{-# INLINE concatMapPure #-}

635−concatMapPure1xN :: Int -> IO ()

705+concatMapPure :: Int -> Int -> Int -> IO ()

636−concatMapPure1xN n = S.drain $ S.concatMap (\_ -> sourceUnfoldrN 1 n)

706+concatMapPure outer inner n =

637− (sourceUnfoldrN value n)

638−

639−#ifdef INSPECTION

640−inspect $ hasNoTypeClasses 'concatMapPure1xN

641−#endif

642−

643−-- We use a (sqrt n) element stream as source and then generate and concat a

644−-- (sqrt n) element stream for each element of the source to produce an n

645−-- element stream.

646−{-# INLINE concatMapNxN #-}

647−concatMapNxN :: Int -> IO ()

648−concatMapNxN n = S.drain $ S.concatMap (\_ -> sourceUnfoldrMN value2 n)

707+ S.drain $ S.concatMap

649− (sourceUnfoldrMN value2 n)

708+ (\_ -> sourceUnfoldrN inner n)

709+ (sourceUnfoldrN outer n)

650

710

651#ifdef INSPECTION

711#ifdef INSPECTION

652−inspect $ hasNoTypeClasses 'concatMapNxN

712+inspect $ hasNoTypeClasses 'concatMapPure

653#endif

713#endif

654

714

655{-# INLINE concatMapRepl4xN #-}

715{-# INLINE concatMapRepl4xN #-}

@@ -661,6 +721,75 @@ concatMapRepl4xN n = S.drain $ S.concatMap (S.replicate 4)

661inspect $ hasNoTypeClasses 'concatMapRepl4xN

721inspect $ hasNoTypeClasses 'concatMapRepl4xN

662#endif

722#endif

663

723

724+{-# INLINE concatUnfoldRepl4xN #-}

725+concatUnfoldRepl4xN :: Int -> IO ()

726+concatUnfoldRepl4xN n =

727+ S.drain $ Internal.concatMapU

728+ (UF.replicateM 4)

729+ (sourceUnfoldrMN (value `div` 4) n)

730+

731+#ifdef INSPECTION

732+inspect $ hasNoTypeClasses 'concatUnfoldRepl4xN

733+inspect $ 'concatUnfoldRepl4xN `hasNoType` ''D.ConcatMapUState

734+#endif

735+

736+{-# INLINE concatMapWithSerial #-}

737+concatMapWithSerial :: Int -> Int -> Int -> IO ()

738+concatMapWithSerial outer inner n =

739+ S.drain $ S.concatMapWith S.serial

740+ (\_ -> sourceUnfoldrMN inner n)

741+ (sourceUnfoldrMN outer n)

742+

743+#ifdef INSPECTION

744+inspect $ hasNoTypeClasses 'concatMapWithSerial

745+#endif

746+

747+{-# INLINE concatMapWithAppend #-}

748+concatMapWithAppend :: Int -> Int -> Int -> IO ()

749+concatMapWithAppend outer inner n =

750+ S.drain $ S.concatMapWith Internal.append

751+ (\_ -> sourceUnfoldrMN inner n)

752+ (sourceUnfoldrMN outer n)

753+

754+#ifdef INSPECTION

755+inspect $ hasNoTypeClasses 'concatMapWithAppend

756+#endif

757+

758+{-# INLINE concatMapWithWSerial #-}

759+concatMapWithWSerial :: Int -> Int -> Int -> IO ()

760+concatMapWithWSerial outer inner n =

761+ S.drain $ S.concatMapWith S.wSerial

762+ (\_ -> sourceUnfoldrMN inner n)

763+ (sourceUnfoldrMN outer n)

764+

765+#ifdef INSPECTION

766+inspect $ hasNoTypeClasses 'concatMapWithWSerial

767+#endif

768+

769+{-# INLINE concatUnfoldInterleaveRepl4xN #-}

770+concatUnfoldInterleaveRepl4xN :: Int -> IO ()

771+concatUnfoldInterleaveRepl4xN n =

772+ S.drain $ Internal.concatUnfoldInterleave

773+ (UF.replicateM 4)

774+ (sourceUnfoldrMN (value `div` 4) n)

775+

776+#ifdef INSPECTION

777+inspect $ hasNoTypeClasses 'concatUnfoldInterleaveRepl4xN

778+-- inspect $ 'concatUnfoldInterleaveRepl4xN `hasNoType` ''D.ConcatUnfoldInterleaveState

779+#endif

780+

781+{-# INLINE concatUnfoldRoundrobinRepl4xN #-}

782+concatUnfoldRoundrobinRepl4xN :: Int -> IO ()

783+concatUnfoldRoundrobinRepl4xN n =

784+ S.drain $ Internal.concatUnfoldRoundrobin

785+ (UF.replicateM 4)

786+ (sourceUnfoldrMN (value `div` 4) n)

787+

788+#ifdef INSPECTION

789+inspect $ hasNoTypeClasses 'concatUnfoldRoundrobinRepl4xN

790+-- inspect $ 'concatUnfoldRoundrobinRepl4xN `hasNoType` ''D.ConcatUnfoldInterleaveState

791+#endif

792+

664-------------------------------------------------------------------------------

793-------------------------------------------------------------------------------

665-- Mixed Composition

794-- Mixed Composition

666-------------------------------------------------------------------------------

795-------------------------------------------------------------------------------

src/Streamly/Internal.hs+15−2

@@ -33,7 +33,7 @@ module Streamly.Internal

33 , toStreamRev

35 , insertAfterEach

36− , intersperseByTime

36+ , interject

37 , reverse'

39 , splitAt -- spanN

@@ -53,7 +53,19 @@ module Streamly.Internal

53 , classifySessionsBy

54 , classifyKeepAliveSessions

56+ , append

57+ , interleave

58+ , interleaveFst

59+ , interleaveMin

60+ , roundrobin

61+ , wSerialFst

62+ , wSerialMin

63+ , parallelFst

64+ , parallelMin

65+

56 , concatMapU

66 , concatMapU

67+ , concatUnfoldInterleave

68+ , concatUnfoldRoundrobin

58 -- * Streamly.Fold Experimental Exports

70 -- * Streamly.Fold Experimental Exports

59 , Fold (..)

71 , Fold (..)

@@ -82,7 +94,8 @@ import Prelude hiding (break, span, splitAt)

83import Streamly.Memory.Array.Types (readU)

95import Streamly.Memory.Array.Types (readU)

84import Streamly.Streams.Combinators (inspectMode)

96import Streamly.Streams.Combinators (inspectMode)

85−import Streamly.Streams.Parallel (tapAsync)

97+import Streamly.Streams.Parallel (tapAsync, parallelFst, parallelMin)

98+import Streamly.Streams.Serial (wSerialFst, wSerialMin)

86import Streamly.Unfold.Types (Unfold(..))

99import Streamly.Unfold.Types (Unfold(..))

100

88import Streamly.Prelude.Internal

101import Streamly.Prelude.Internal

src/Streamly/Prelude/Internal.hs+104−6

@@ -205,7 +205,7 @@ module Streamly.Prelude.Internal

205 , intersperse

206 , insertAfterEach

207 -- , intersperseBySpan

208− , intersperseByTime

208+ , interject

209

210 -- ** Reordering

211 , reverse

@@ -214,6 +214,15 @@ module Streamly.Prelude.Internal

214 -- * Multi-Stream Operations

215

216 -- ** Appending

217+ , append

218+

219+ -- ** Interleaving

220+ , interleave

221+ , interleaveFst

222+ , interleaveMin

223+

224+ -- ** Scheduling

225+ , roundrobin

217

226

218 -- ** Merging

227 -- ** Merging

219

228

@@ -232,6 +241,8 @@ module Streamly.Prelude.Internal

232 -- ** Nested Streams

241 -- ** Nested Streams

233 , concatMapM

242 , concatMapM

234 , concatMapU

243 , concatMapU

244+ , concatUnfoldInterleave

245+ , concatUnfoldRoundrobin

235 , concatMap

246 , concatMap

236 , concatMapWith

247 , concatMapWith

237 -- , interposeBy

248 -- , interposeBy

@@ -1817,16 +1828,16 @@ intersperseBySpan _n _f _xs = undefined

1817-- seconds.

1828-- seconds.

1818--

1829--

1819-- @

1830-- @

1820−-- > S.drain $ S.intersperseByTime 1 (putChar ',') $ S.mapM (\\x -> threadDelay 1000000 >> putChar x) $ S.fromList "hello"

1831+-- > S.drain $ S.interject 1 (putChar ',') $ S.mapM (\\x -> threadDelay 1000000 >> putChar x) $ S.fromList "hello"

1821-- "h,e,l,l,o"

1832-- "h,e,l,l,o"

1822-- @

1833-- @

1823--

1834--

1824-- @since 0.7.0

1835-- @since 0.7.0

1825−{-# INLINE intersperseByTime #-}

1836+{-# INLINE interject #-}

1826−intersperseByTime

1837+interject

1827 :: (IsStream t, MonadAsync m)

1838 :: (IsStream t, MonadAsync m)

1828 => Double -> m a -> t m a -> t m a

1839 => Double -> m a -> t m a -> t m a

1829−intersperseByTime n f xs = xs `Par.parallelEndByFirst` repeatM timed

1840+interject n f xs = xs `Par.parallelFst` repeatM timed

1830 where timed = liftIO (threadDelay (round $ n * 1000000)) >> f

1841 where timed = liftIO (threadDelay (round $ n * 1000000)) >> f

1831

1842

1832-- | @insertBy cmp elem stream@ inserts @elem@ before the first element in

1843-- | @insertBy cmp elem stream@ inserts @elem@ before the first element in

@@ -2092,6 +2103,73 @@ concatMapWith = K.concatMapBy

2092concatMap ::(IsStream t, Monad m) => (a -> t m b) -> t m a -> t m b

2103concatMap ::(IsStream t, Monad m) => (a -> t m b) -> t m a -> t m b

2093concatMap f m = fromStreamD $ D.concatMap (toStreamD . f) (toStreamD m)

2104concatMap f m = fromStreamD $ D.concatMap (toStreamD . f) (toStreamD m)

2094

2105

2106+-- | Append the outputs of two streams, yielding all the elements from the

2107+-- first stream and then yielding all the elements from the second stream.

2108+--

2109+-- IMPORTANT NOTE: This could be 100x faster than @serial/<>@ for appending a

2110+-- few (say 100) streams because it can fuse via stream fusion. However, it

2111+-- does not scale for a large number of streams (say 1000s) and becomes

2112+-- qudartically slow. Therefore use this for custom appending of a few streams

2113+-- but use 'concatMap' or 'concatMapWith serial' for appending @n@ streams or

2114+-- infinite containers of streams.

2115+--

2116+-- @since 0.7.0

2117+{-# INLINE append #-}

2118+append ::(IsStream t, Monad m) => t m b -> t m b -> t m b

2119+append m1 m2 = fromStreamD $ D.append (toStreamD m1) (toStreamD m2)

2120+

2121+-- Same as 'wSerial'. We should perhaps rename wSerial. If named explicitly

2122+-- this would be interleaveMax.

2123+--

2124+-- | Interleaves the outputs of two streams, yielding elements from each stream

2125+-- alternately, starting from the first stream. If any of the streams finishes

2126+-- early the other stream continues alone until it too finishes.

2127+--

2128+-- Do not use at scale in concatMapWith.

2129+--

2130+-- @since 0.7.0

2131+{-# INLINE interleave #-}

2132+interleave ::(IsStream t, Monad m) => t m b -> t m b -> t m b

2133+interleave m1 m2 = fromStreamD $ D.interleave (toStreamD m1) (toStreamD m2)

2134+

2135+-- | Interleaves the outputs of two streams, yielding elements from each stream

2136+-- alternately, starting from the first stream. As soon as the first stream

2137+-- finishes the output stops discarding the second stream.

2138+--

2139+-- Do not use at scale in concatMapWith.

2140+--

2141+-- @since 0.7.0

2142+{-# INLINE interleaveFst #-}

2143+interleaveFst ::(IsStream t, Monad m) => t m b -> t m b -> t m b

2144+interleaveFst m1 m2 = fromStreamD $ D.interleaveFst (toStreamD m1) (toStreamD m2)

2145+

2146+-- | Interleaves the outputs of two streams, yielding elements from each stream

2147+-- alternately, starting from the first stream. The output stops as soon as any

2148+-- of the two streams finishes discarding the other.

2149+--

2150+-- Do not use at scale in concatMapWith.

2151+--

2152+-- @since 0.7.0

2153+{-# INLINE interleaveMin #-}

2154+interleaveMin ::(IsStream t, Monad m) => t m b -> t m b -> t m b

2155+interleaveMin m1 m2 = fromStreamD $ D.interleaveMin (toStreamD m1) (toStreamD m2)

2156+

2157+-- | Schedule the execution of two streams in a fair round-robin manner,

2158+-- executing each stream once, alternately. Execution of a stream may not

2159+-- necessarily result in an output, a stream may chose to @Skip@ producing an

2160+-- element until later giving the other stream a chance to run. Therefore, this

2161+-- combinator fairly interleaves the execution of two streams rather than

2162+-- fairly interleaving the output of the two streams. This can be useful in

2163+-- co-operative multitasking without using explicit threads. This can be used

2164+-- as an alternative to `async`.

2165+--

2166+-- Do not use at scale in concatMapWith.

2167+--

2168+-- @since 0.7.0

2169+{-# INLINE roundrobin #-}

2170+roundrobin ::(IsStream t, Monad m) => t m b -> t m b -> t m b

2171+roundrobin m1 m2 = fromStreamD $ D.roundRobin (toStreamD m1) (toStreamD m2)

2172+

2095-- | Map a stream producing monadic function on each element of the stream

2173-- | Map a stream producing monadic function on each element of the stream

2096-- and then flatten the results into a single stream. Since the stream

2174-- and then flatten the results into a single stream. Since the stream

2097-- generation function is monadic, unlike 'concatMap', it can produce an

2175-- generation function is monadic, unlike 'concatMap', it can produce an

@@ -2111,6 +2189,26 @@ concatMapM f m = fromStreamD $ D.concatMapM (fmap toStreamD . f) (toStreamD m)

2111concatMapU ::(IsStream t, Monad m) => Unfold m a b -> t m a -> t m b

2189concatMapU ::(IsStream t, Monad m) => Unfold m a b -> t m a -> t m b

2112concatMapU u m = fromStreamD $ D.concatMapU u (toStreamD m)

2190concatMapU u m = fromStreamD $ D.concatMapU u (toStreamD m)

2113

2191

2192+-- | Like 'concatUnfold' but interleaves the streams in the same way as

2193+-- 'interleave' behaves instead of appending them.

2194+--

2195+-- @since 0.7.0

2196+{-# INLINE concatUnfoldInterleave #-}

2197+concatUnfoldInterleave ::(IsStream t, Monad m)

2198+ => Unfold m a b -> t m a -> t m b

2199+concatUnfoldInterleave u m =

2200+ fromStreamD $ D.concatUnfoldInterleave u (toStreamD m)

2201+

2202+-- | Like 'concatUnfold' but executes the streams in the same way as

2203+-- 'roundrobin'.

2204+--

2205+-- @since 0.7.0

2206+{-# INLINE concatUnfoldRoundrobin #-}

2207+concatUnfoldRoundrobin ::(IsStream t, Monad m)

2208+ => Unfold m a b -> t m a -> t m b

2209+concatUnfoldRoundrobin u m =

2210+ fromStreamD $ D.concatUnfoldRoundrobin u (toStreamD m)

2211+

2114{-

2212{-

2115-- | A generalization of insertBy to inserting sequences.

2213-- | A generalization of insertBy to inserting sequences.

2116interposeBy :: (IsStream t, Monad m)

2214interposeBy :: (IsStream t, Monad m)

@@ -2276,7 +2374,7 @@ intervalsOf

2276 => Double -> Fold m a b -> t m a -> t m b

2374 => Double -> Fold m a b -> t m a -> t m b

2277intervalsOf n f xs =

2375intervalsOf n f xs =

2278 splitWithSuffix isNothing (FL.lcatMaybes f)

2376 splitWithSuffix isNothing (FL.lcatMaybes f)

2279− (intersperseByTime n (return Nothing) (Serial.map Just xs))

2377+ (interject n (return Nothing) (Serial.map Just xs))

2280

2378

2281------------------------------------------------------------------------------

2379------------------------------------------------------------------------------

2282-- Element Aware APIs

2380-- Element Aware APIs

src/Streamly/Streams/Parallel.hs+8−15

@@ -23,8 +23,8 @@ module Streamly.Streams.Parallel

23 , Parallel

24 , parallely

25 , parallel

26− , parallelEndByFirst

26+ , parallelFst

27− , parallelEndByAny

27+ , parallelMin

28 , tapAsync

30 -- * Function application

@@ -151,13 +151,6 @@ joinStreamVarPar style ss m1 m2 = mkStream $ \st yld sng stp ->

151consMParallel :: MonadAsync m => m a -> ParallelT m a -> ParallelT m a

152consMParallel m r = fromStream $ K.yieldM m `parallel` (toStream r)

153

154−-- TODO

155−-- Different variants of parallel differ in how the combined stream ends:

156−--

157−-- - - parallel - ends when all stream end

158−-- parallelEndByFirst - parallel1 - ends when the first stream ends

159−-- parallelEndByAny - parallel_ - ends when any stream ends

160−--

161-- | Polymorphic version of the 'Semigroup' operation '<>' of 'ParallelT'

154-- | Polymorphic version of the 'Semigroup' operation '<>' of 'ParallelT'

162-- Merges two streams concurrently.

155-- Merges two streams concurrently.

163--

156--

@@ -173,9 +166,9 @@ parallel = joinStreamVarPar ParallelVar StopNone

173-- | Like `parallel` but stops the output as soon as the first stream stops.

166-- | Like `parallel` but stops the output as soon as the first stream stops.

174--

167--

175-- @since 0.7.0

168-- @since 0.7.0

176−{-# INLINE parallelEndByFirst #-}

169+{-# INLINE parallelFst #-}

177−parallelEndByFirst :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a

170+parallelFst :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a

178−parallelEndByFirst = joinStreamVarPar ParallelVar StopBy

171+parallelFst = joinStreamVarPar ParallelVar StopBy

179

172

180-- This is a race like combinator for streams.

173-- This is a race like combinator for streams.

181--

174--

@@ -183,9 +176,9 @@ parallelEndByFirst = joinStreamVarPar ParallelVar StopBy

183-- stops.

176-- stops.

184--

177--

185-- @since 0.7.0

178-- @since 0.7.0

186−{-# INLINE parallelEndByAny #-}

179+{-# INLINE parallelMin #-}

187−parallelEndByAny :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a

180+parallelMin :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a

188−parallelEndByAny = joinStreamVarPar ParallelVar StopAny

181+parallelMin = joinStreamVarPar ParallelVar StopAny

189

182

190------------------------------------------------------------------------------

183------------------------------------------------------------------------------

191-- Convert a stream to parallel

184-- Convert a stream to parallel

src/Streamly/Streams/Serial.hs+8−17

@@ -32,8 +32,8 @@ module Streamly.Streams.Serial

32 , InterleavedT -- deprecated

33 , WSerial

34 , wSerial

35− , wSerialEndByFirst

35+ , wSerialFst

36− , wSerialEndByAny

36+ , wSerialMin

37 , (<=>) -- deprecated

38 , wSerially

39 , interleaving -- deprecated

@@ -307,15 +307,6 @@ instance IsStream WSerialT where

307-- Semigroup

308------------------------------------------------------------------------------

309

310−-- TODO

311−-- We always starts with the first stream yielding but there can be many ways

312−-- to end it. In general we can have the following variants of interleaving

313−-- streams:

314−-- wSerial - interleave - end when both streams end

315−-- wSerialEndByAny - interleave_ - end as soon as any stream ends

316−-- wSerialEndByFirst - interleave1 - make sure we end with first

317−-- - - interleave2 - make sure we end with second

318−--

319-- Additionally we can have m elements yield from the first stream and n

310-- Additionally we can have m elements yield from the first stream and n

320-- elements yeilding from the second stream. We can also have time slicing

311-- elements yeilding from the second stream. We can also have time slicing

321-- variants of positional interleaving, e.g. run first stream for m seconds and

312-- variants of positional interleaving, e.g. run first stream for m seconds and

@@ -340,9 +331,9 @@ wSerial m1 m2 = mkStream $ \st yld sng stp -> do

340-- | Like `wSerial` but stops interleaving as soon as the first stream stops.

331-- | Like `wSerial` but stops interleaving as soon as the first stream stops.

341--

332--

342-- @since 0.7.0

333-- @since 0.7.0

343−{-# INLINE wSerialEndByFirst #-}

334+{-# INLINE wSerialFst #-}

344−wSerialEndByFirst :: IsStream t => t m a -> t m a -> t m a

335+wSerialFst :: IsStream t => t m a -> t m a -> t m a

345−wSerialEndByFirst m1 m2 = mkStream $ \st yld sng stp -> do

336+wSerialFst m1 m2 = mkStream $ \st yld sng stp -> do

346 let yieldFirst a r = yld a (yieldSecond r m2)

337 let yieldFirst a r = yld a (yieldSecond r m2)

347 in foldStream st yieldFirst sng stp m1

338 in foldStream st yieldFirst sng stp m1

348

339

@@ -358,9 +349,9 @@ wSerialEndByFirst m1 m2 = mkStream $ \st yld sng stp -> do

358-- stops.

349-- stops.

359--

350--

360-- @since 0.7.0

351-- @since 0.7.0

361−{-# INLINE wSerialEndByAny #-}

352+{-# INLINE wSerialMin #-}

362−wSerialEndByAny :: IsStream t => t m a -> t m a -> t m a

353+wSerialMin :: IsStream t => t m a -> t m a -> t m a

363−wSerialEndByAny m1 m2 = mkStream $ \st yld sng stp -> do

354+wSerialMin m1 m2 = mkStream $ \st yld sng stp -> do

364 let stop = stp

355 let stop = stp

365 single a = sng a

356 single a = sng a

366 yieldk a r = yld a (wSerial m2 r)

357 yieldk a r = yld a (wSerial m2 r)

src/Streamly/Streams/StreamD.hs+291

@@ -137,6 +137,16 @@ module Streamly.Streams.StreamD

137 , concatMap

138 , ConcatMapUState (..)

139 , concatMapU

140+ , ConcatUnfoldInterleaveState (..)

141+ , concatUnfoldInterleave

142+ , concatUnfoldRoundrobin

143+ , AppendState(..)

144+ , append

145+ , InterleaveState(..)

146+ , interleave

147+ , interleaveMin

148+ , interleaveFst

149+ , roundRobin -- interleaveFair?/ParallelFair

140

150

141 -- ** Grouping

151 -- ** Grouping

142 , groupsOf

152 , groupsOf

@@ -1734,6 +1744,287 @@ concatMapU (Unfold istep inject) (Stream ostep ost) =

1734 Skip i' -> Skip (ConcatMapUInner o i')

1744 Skip i' -> Skip (ConcatMapUInner o i')

1735 Stop -> Skip (ConcatMapUOuter o)

1745 Stop -> Skip (ConcatMapUOuter o)

1736

1746

1747+data ConcatUnfoldInterleaveState o i =

1748+ ConcatUnfoldInterleaveOuter o [i]

1749+ | ConcatUnfoldInterleaveInner o [i]

1750+ | ConcatUnfoldInterleaveInnerL [i] [i]

1751+ | ConcatUnfoldInterleaveInnerR [i] [i]

1752+

1753+-- XXX use arrays to store state instead of lists.

1754+-- XXX In general we can use different scheduling strategies e.g. how to

1755+-- schedule the outer vs inner loop or assigning weights to different streams

1756+-- or outer and inner loops.

1757+

1758+-- After a yield, switch to the next stream. Do not switch streams on Skip.

1759+-- Yield from outer stream switches to the inner stream.

1760+--

1761+-- There are two choices here, (1) exhaust the outer stream first and then

1762+-- start yielding from the inner streams, this is much simpler to implement,

1763+-- (2) yield at least one element from an inner stream before going back to

1764+-- outer stream and opening the next stream from it.

1765+--

1766+-- Ideally, we need some scheduling bias to inner streams vs outer stream.

1767+-- Maybe we can configure the behavior.

1768+--

1769+{-# INLINE_NORMAL concatUnfoldInterleave #-}

1770+concatUnfoldInterleave :: Monad m => Unfold m a b -> Stream m a -> Stream m b

1771+concatUnfoldInterleave (Unfold istep inject) (Stream ostep ost) =

1772+ Stream step (ConcatUnfoldInterleaveOuter ost [])

1773+ where

1774+ {-# INLINE_LATE step #-}

1775+ step gst (ConcatUnfoldInterleaveOuter o ls) = do

1776+ r <- ostep (adaptState gst) o

1777+ case r of

1778+ Yield a o' -> do

1779+ i <- inject a

1780+ i `seq` return (Skip (ConcatUnfoldInterleaveInner o' (i : ls)))

1781+ Skip o' -> return $ Skip (ConcatUnfoldInterleaveOuter o' ls)

1782+ Stop -> return $ Skip (ConcatUnfoldInterleaveInnerL ls [])

1783+

1784+ step _ (ConcatUnfoldInterleaveInner _ []) = undefined

1785+ step _ (ConcatUnfoldInterleaveInner o (st:ls)) = do

1786+ r <- istep st

1787+ return $ case r of

1788+ Yield x s -> Yield x (ConcatUnfoldInterleaveOuter o (s:ls))

1789+ Skip s -> Skip (ConcatUnfoldInterleaveInner o (s:ls))

1790+ Stop -> Skip (ConcatUnfoldInterleaveOuter o ls)

1791+

1792+ step _ (ConcatUnfoldInterleaveInnerL [] []) = return Stop

1793+ step _ (ConcatUnfoldInterleaveInnerL [] rs) =

1794+ return $ Skip (ConcatUnfoldInterleaveInnerR [] rs)

1795+

1796+ step _ (ConcatUnfoldInterleaveInnerL (st:ls) rs) = do

1797+ r <- istep st

1798+ return $ case r of

1799+ Yield x s -> Yield x (ConcatUnfoldInterleaveInnerL ls (s:rs))

1800+ Skip s -> Skip (ConcatUnfoldInterleaveInnerL (s:ls) rs)

1801+ Stop -> Skip (ConcatUnfoldInterleaveInnerL ls rs)

1802+

1803+ step _ (ConcatUnfoldInterleaveInnerR [] []) = return Stop

1804+ step _ (ConcatUnfoldInterleaveInnerR ls []) =

1805+ return $ Skip (ConcatUnfoldInterleaveInnerL ls [])

1806+

1807+ step _ (ConcatUnfoldInterleaveInnerR ls (st:rs)) = do

1808+ r <- istep st

1809+ return $ case r of

1810+ Yield x s -> Yield x (ConcatUnfoldInterleaveInnerR (s:ls) rs)

1811+ Skip s -> Skip (ConcatUnfoldInterleaveInnerR ls (s:rs))

1812+ Stop -> Skip (ConcatUnfoldInterleaveInnerR ls rs)

1813+

1814+-- XXX In general we can use different scheduling strategies e.g. how to

1815+-- schedule the outer vs inner loop or assigning weights to different streams

1816+-- or outer and inner loops.

1817+--

1818+-- This could be inefficient if the tasks are too small.

1819+--

1820+-- Compared to concatUnfoldInterleave this one switches streams on Skips.

1821+--

1822+{-# INLINE_NORMAL concatUnfoldRoundrobin #-}

1823+concatUnfoldRoundrobin :: Monad m => Unfold m a b -> Stream m a -> Stream m b

1824+concatUnfoldRoundrobin (Unfold istep inject) (Stream ostep ost) =

1825+ Stream step (ConcatUnfoldInterleaveOuter ost [])

1826+ where

1827+ {-# INLINE_LATE step #-}

1828+ step gst (ConcatUnfoldInterleaveOuter o ls) = do

1829+ r <- ostep (adaptState gst) o

1830+ case r of

1831+ Yield a o' -> do

1832+ i <- inject a

1833+ i `seq` return (Skip (ConcatUnfoldInterleaveInner o' (i : ls)))

1834+ Skip o' -> return $ Skip (ConcatUnfoldInterleaveInner o' ls)

1835+ Stop -> return $ Skip (ConcatUnfoldInterleaveInnerL ls [])

1836+

1837+ step _ (ConcatUnfoldInterleaveInner o []) =

1838+ return $ Skip (ConcatUnfoldInterleaveOuter o [])

1839+

1840+ step _ (ConcatUnfoldInterleaveInner o (st:ls)) = do

1841+ r <- istep st

1842+ return $ case r of

1843+ Yield x s -> Yield x (ConcatUnfoldInterleaveOuter o (s:ls))

1844+ Skip s -> Skip (ConcatUnfoldInterleaveOuter o (s:ls))

1845+ Stop -> Skip (ConcatUnfoldInterleaveOuter o ls)

1846+

1847+ step _ (ConcatUnfoldInterleaveInnerL [] []) = return Stop

1848+ step _ (ConcatUnfoldInterleaveInnerL [] rs) =

1849+ return $ Skip (ConcatUnfoldInterleaveInnerR [] rs)

1850+

1851+ step _ (ConcatUnfoldInterleaveInnerL (st:ls) rs) = do

1852+ r <- istep st

1853+ return $ case r of

1854+ Yield x s -> Yield x (ConcatUnfoldInterleaveInnerL ls (s:rs))

1855+ Skip s -> Skip (ConcatUnfoldInterleaveInnerL ls (s:rs))

1856+ Stop -> Skip (ConcatUnfoldInterleaveInnerL ls rs)

1857+

1858+ step _ (ConcatUnfoldInterleaveInnerR [] []) = return Stop

1859+ step _ (ConcatUnfoldInterleaveInnerR ls []) =

1860+ return $ Skip (ConcatUnfoldInterleaveInnerL ls [])

1861+

1862+ step _ (ConcatUnfoldInterleaveInnerR ls (st:rs)) = do

1863+ r <- istep st

1864+ return $ case r of

1865+ Yield x s -> Yield x (ConcatUnfoldInterleaveInnerR (s:ls) rs)

1866+ Skip s -> Skip (ConcatUnfoldInterleaveInnerR (s:ls) rs)

1867+ Stop -> Skip (ConcatUnfoldInterleaveInnerR ls rs)

1868+

1869+data AppendState s1 s2 = AppendFirst s1 | AppendSecond s2

1870+

1871+-- Note that this could be much faster compared to the CPS stream. However, as

1872+-- the number of streams being composed increases this may become expensive.

1873+-- Need to see where the breaking point is between the two.

1874+--

1875+{-# INLINE_NORMAL append #-}

1876+append :: Monad m => Stream m a -> Stream m a -> Stream m a

1877+append (Stream step1 state1) (Stream step2 state2) =

1878+ Stream step (AppendFirst state1)

1879+

1880+ where

1881+

1882+ {-# INLINE_LATE step #-}

1883+ step gst (AppendFirst st) = do

1884+ r <- step1 gst st

1885+ return $ case r of

1886+ Yield a s -> Yield a (AppendFirst s)

1887+ Skip s -> Skip (AppendFirst s)

1888+ Stop -> Skip (AppendSecond state2)

1889+

1890+ step gst (AppendSecond st) = do

1891+ r <- step2 gst st

1892+ return $ case r of

1893+ Yield a s -> Yield a (AppendSecond s)

1894+ Skip s -> Skip (AppendSecond s)

1895+ Stop -> Stop

1896+

1897+data InterleaveState s1 s2 = InterleaveFirst s1 s2 | InterleaveSecond s1 s2

1898+ | InterleaveSecondOnly s2 | InterleaveFirstOnly s1

1899+

1900+{-# INLINE_NORMAL interleave #-}

1901+interleave :: Monad m => Stream m a -> Stream m a -> Stream m a

1902+interleave (Stream step1 state1) (Stream step2 state2) =

1903+ Stream step (InterleaveFirst state1 state2)

1904+

1905+ where

1906+

1907+ {-# INLINE_LATE step #-}

1908+ step gst (InterleaveFirst st1 st2) = do

1909+ r <- step1 gst st1

1910+ return $ case r of

1911+ Yield a s -> Yield a (InterleaveSecond s st2)

1912+ Skip s -> Skip (InterleaveFirst s st2)

1913+ Stop -> Skip (InterleaveSecondOnly st2)

1914+

1915+ step gst (InterleaveSecond st1 st2) = do

1916+ r <- step2 gst st2

1917+ return $ case r of

1918+ Yield a s -> Yield a (InterleaveFirst st1 s)

1919+ Skip s -> Skip (InterleaveSecond st1 s)

1920+ Stop -> Skip (InterleaveFirstOnly st1)

1921+

1922+ step gst (InterleaveFirstOnly st1) = do

1923+ r <- step1 gst st1

1924+ return $ case r of

1925+ Yield a s -> Yield a (InterleaveFirstOnly s)

1926+ Skip s -> Skip (InterleaveFirstOnly s)

1927+ Stop -> Stop

1928+

1929+ step gst (InterleaveSecondOnly st2) = do

1930+ r <- step2 gst st2

1931+ return $ case r of

1932+ Yield a s -> Yield a (InterleaveSecondOnly s)

1933+ Skip s -> Skip (InterleaveSecondOnly s)

1934+ Stop -> Stop

1935+

1936+{-# INLINE_NORMAL interleaveMin #-}

1937+interleaveMin :: Monad m => Stream m a -> Stream m a -> Stream m a

1938+interleaveMin (Stream step1 state1) (Stream step2 state2) =

1939+ Stream step (InterleaveFirst state1 state2)

1940+

1941+ where

1942+

1943+ {-# INLINE_LATE step #-}

1944+ step gst (InterleaveFirst st1 st2) = do

1945+ r <- step1 gst st1

1946+ return $ case r of

1947+ Yield a s -> Yield a (InterleaveSecond s st2)

1948+ Skip s -> Skip (InterleaveFirst s st2)

1949+ Stop -> Stop

1950+

1951+ step gst (InterleaveSecond st1 st2) = do

1952+ r <- step2 gst st2

1953+ return $ case r of

1954+ Yield a s -> Yield a (InterleaveFirst st1 s)

1955+ Skip s -> Skip (InterleaveSecond st1 s)

1956+ Stop -> Stop

1957+

1958+ step _ (InterleaveFirstOnly _) = undefined

1959+ step _ (InterleaveSecondOnly _) = undefined

1960+

1961+{-# INLINE_NORMAL interleaveFst #-}

1962+interleaveFst :: Monad m => Stream m a -> Stream m a -> Stream m a

1963+interleaveFst (Stream step1 state1) (Stream step2 state2) =

1964+ Stream step (InterleaveFirst state1 state2)

1965+

1966+ where

1967+

1968+ {-# INLINE_LATE step #-}

1969+ step gst (InterleaveFirst st1 st2) = do

1970+ r <- step1 gst st1

1971+ return $ case r of

1972+ Yield a s -> Yield a (InterleaveSecond s st2)

1973+ Skip s -> Skip (InterleaveFirst s st2)

1974+ Stop -> Stop

1975+

1976+ step gst (InterleaveSecond st1 st2) = do

1977+ r <- step2 gst st2

1978+ return $ case r of

1979+ Yield a s -> Yield a (InterleaveFirst st1 s)

1980+ Skip s -> Skip (InterleaveSecond st1 s)

1981+ Stop -> Skip (InterleaveFirstOnly st1)

1982+

1983+ step gst (InterleaveFirstOnly st1) = do

1984+ r <- step1 gst st1

1985+ return $ case r of

1986+ Yield a s -> Yield a (InterleaveFirstOnly s)

1987+ Skip s -> Skip (InterleaveFirstOnly s)

1988+ Stop -> Stop

1989+

1990+ step _ (InterleaveSecondOnly _) = undefined

1991+

1992+{-# INLINE_NORMAL roundRobin #-}

1993+roundRobin :: Monad m => Stream m a -> Stream m a -> Stream m a

1994+roundRobin (Stream step1 state1) (Stream step2 state2) =

1995+ Stream step (InterleaveFirst state1 state2)

1996+

1997+ where

1998+

1999+ {-# INLINE_LATE step #-}

2000+ step gst (InterleaveFirst st1 st2) = do

2001+ r <- step1 gst st1

2002+ return $ case r of

2003+ Yield a s -> Yield a (InterleaveSecond s st2)

2004+ Skip s -> Skip (InterleaveSecond s st2)

2005+ Stop -> Skip (InterleaveSecondOnly st2)

2006+

2007+ step gst (InterleaveSecond st1 st2) = do

2008+ r <- step2 gst st2

2009+ return $ case r of

2010+ Yield a s -> Yield a (InterleaveFirst st1 s)

2011+ Skip s -> Skip (InterleaveFirst st1 s)

2012+ Stop -> Skip (InterleaveFirstOnly st1)

2013+

2014+ step gst (InterleaveSecondOnly st2) = do

2015+ r <- step2 gst st2

2016+ return $ case r of

2017+ Yield a s -> Yield a (InterleaveSecondOnly s)

2018+ Skip s -> Skip (InterleaveSecondOnly s)

2019+ Stop -> Stop

2020+

2021+ step gst (InterleaveFirstOnly st1) = do

2022+ r <- step1 gst st1

2023+ return $ case r of

2024+ Yield a s -> Yield a (InterleaveFirstOnly s)

2025+ Skip s -> Skip (InterleaveFirstOnly s)

2026+ Stop -> Stop

2027+

1737------------------------------------------------------------------------------

2028------------------------------------------------------------------------------

1738-- Exceptions

2029-- Exceptions

1739------------------------------------------------------------------------------

2030------------------------------------------------------------------------------