091f4b6 · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Add the strictly parallel stream type ParallelT

091f4b6Harendra Kumar8 years ago

benchmark/Linear.hs+5−4

compare with master

unifiedsplit

@@ -50,10 +50,11 @@ main = do

50 ]

51 , benchIO "zip" Ops.zip

52 , bgroup "append"

53− [ benchIOAppend "asStream" $ Ops.append asStream

53+ [ benchIOAppend "stream" $ Ops.append asStream

54− , benchIOAppend "asCostream" $ Ops.append asCostream

54+ , benchIOAppend "costream" $ Ops.append asCostream

55− , benchIOAppend "asParAhead" $ Ops.append asParAhead

55+ , benchIOAppend "parAhead" $ Ops.append asParAhead

56− , benchIOAppend "asCoparAhead" $ Ops.append asCoparAhead

56+ , benchIOAppend "coparAhead" $ Ops.append asCoparAhead

57+ , benchIOAppend "parallel" $ Ops.append asParallel

57 ]

58 ]

58 , bgroup "compose"

59 , bgroup "compose"

59 [ benchIO "mapM" Ops.composeMapM

60 [ benchIO "mapM" Ops.composeMapM

benchmark/LinearOps.hs+3−2

@@ -14,8 +14,9 @@ import Prelude

14import qualified Streamly as S

15import qualified Streamly.Prelude as S

17−value, maxValue :: Int

17+value, appendValue, maxValue :: Int

18value = 1000000

19+appendValue = 100000

19maxValue = value + 1000

20maxValue = value + 1000

21-------------------------------------------------------------------------------

22-------------------------------------------------------------------------------

@@ -111,7 +112,7 @@ concat _n = return ()

111append

112append

112 :: (Monoid (t m Int), Monad m, Monad (t m))

113 :: (Monoid (t m Int), Monad m, Monad (t m))

113 => (t m Int -> S.StreamT m Int) -> Int -> m ()

114 => (t m Int -> S.StreamT m Int) -> Int -> m ()

114−append t n = runStream $ t $ foldMap return [n..n+value]

115+append t n = runStream $ t $ foldMap return [n..n+appendValue]

115

116

116-------------------------------------------------------------------------------

117-------------------------------------------------------------------------------

117-- Composition

118-- Composition

examples/ListDir.hs+3−3

@@ -1,17 +1,17 @@

1import Control.Monad.IO.Class (liftIO)

2import Path.IO (listDir, getCurrentDir)

3import System.IO (stdout, hSetBuffering, BufferMode(LineBuffering))

4−import Streamly (runStream, coparallely)

4+import Streamly (runStream, asParAhead)

6-- | List the current directory recursively using concurrent processing

7--

8-- This example demonstrates that there is little difference between regular

9-- IO code and concurrent streamly code. You can just remove

10−-- 'runStream . coparallely' and this becomes your regular IO code.

10+-- 'runStream . asParAhead' and this becomes your regular IO code.

11main :: IO ()

12main = do

13 hSetBuffering stdout LineBuffering

14− runStream . coparallely $ getCurrentDir >>= readdir

14+ runStream . asParAhead $ getCurrentDir >>= readdir

15 where readdir d = do

16 (ds, fs) <- liftIO $ listDir d

17 liftIO $ mapM_ putStrLn $ map show fs ++ map show ds

examples/SearchQuery.hs+2−2

@@ -10,10 +10,10 @@ import Network.HTTP.Simple

10main :: IO ()

11main = do

12 putStrLn "Using parallel semigroup composition"

13− runStream . parallely $ google <> bing <> duckduckgo

13+ runStream . asParallel $ google <> bing <> duckduckgo

15 putStrLn "\nUsing parallel applicative zip"

16− runStream . zipParallely $ (,,) <$> google <*> bing <*> duckduckgo

16+ runStream . asZipParallel $ (,,) <$> google <*> bing <*> duckduckgo

18 where

19 get :: IsStream t => String -> t IO ()

src/Streamly.hs+5−1

@@ -65,10 +65,11 @@ module Streamly

65 , StreamT

66 , CostreamT

68− -- ** Parallel Ahead Streams

68+ -- ** Parallel Streams

69 -- $parAhead

70 , ParAheadT

71 , CoparAheadT

72+ , ParallelT

73 -- ** Zipping Streams

74 -- ** Zipping Streams

74 -- $zipping

75 -- $zipping

@@ -81,6 +82,7 @@ module Streamly

81 , cosplice

82 , cosplice

82 , parAhead

83 , parAhead

83 , coparAhead

84 , coparAhead

85+ , parallel

85 -- * Stream Type Adapters

87 -- * Stream Type Adapters

86 -- $adapters

88 -- $adapters

@@ -90,6 +92,7 @@ module Streamly

90 , asCostream

92 , asCostream

91 , asParAhead

93 , asParAhead

92 , asCoparAhead

94 , asCoparAhead

95+ , asParallel

93 , asZipStream

96 , asZipStream

94 , asZipParallel

97 , asZipParallel

95 , adapt

98 , adapt

@@ -99,6 +102,7 @@ module Streamly

99 , Costream

102 , Costream

100 , ParAhead

103 , ParAhead

101 , CoparAhead

104 , CoparAhead

105+ , Parallel

102 , ZipStream

106 , ZipStream

103 , ZipParallel

107 , ZipParallel

104

108

src/Streamly/Core.hs+91−9

@@ -37,6 +37,7 @@ module Streamly.Core

37 , cosplice

38 , parAhead

39 , coparAhead

40+ , parallel

41 -- * Alternative

42 -- * Alternative

42 , alt

43 , alt

@@ -107,9 +108,11 @@ data ChildEvent a =

107-- SVars so that the scheduling of the two is independent.

108-- SVars so that the scheduling of the two is independent.

108data SVarTag = Conjunction | Disjunction deriving Eq

109data SVarTag = Conjunction | Disjunction deriving Eq

109

110

110−-- | For fairly interleaved parallel composition the sched policy is FIFO

111−-- whereas for left biased parallel composition it is LIFO.

112−data SVarSched = LIFO | FIFO deriving Eq

111+data SVarSched =

112+ LIFO -- depth first concurrent

113+ | FIFO -- breadth first concurrent

114+ | Par -- all parallel

115+ deriving Eq

113

116

114-- | Identify the type of the SVar. Two computations using the same style can

117-- | Identify the type of the SVar. Two computations using the same style can

115-- be scheduled on the same SVar.

118-- be scheduled on the same SVar.

@@ -324,24 +327,53 @@ runqueueFIFO sv q = run

324 single a = sendit a >> run

327 single a = sendit a >> run

325 yield a r = sendit a >> liftIO (enqueueFIFO q r) >> run

328 yield a r = sendit a >> liftIO (enqueueFIFO q r) >> run

326

329

330+{-# INLINE runOne #-}

331+runOne :: MonadIO m => SVar m a -> Stream m a -> m ()

332+runOne sv m = (runStream m) (Just sv) stop single yield

333+

334+ where

335+

336+ stop = sendStop sv

337+ sendit a = send sv (ChildYield a)

338+ single a = sendit a >> stop

339+ yield a r = sendit a >> runOne sv r

340+

327-- Thread tracking is needed for two reasons:

341-- Thread tracking is needed for two reasons:

328--

342--

329−-- 1) Killing threads on exceptions. Threads may not be allowed to go away by

343+-- 1) Killing threads on exceptions. Threads may not be left to go away by

330-- themselves because they may run for significant times before going away or

344-- themselves because they may run for significant times before going away or

331-- worse they may be stuck in IO and never go away.

345-- worse they may be stuck in IO and never go away.

332--

346--

333−-- 2) To know when all threads are done.

347+-- 2) To know when all threads are done and the stream has ended.

334

348

335{-# NOINLINE addThread #-}

349{-# NOINLINE addThread #-}

336addThread :: MonadIO m => SVar m a -> ThreadId -> m ()

350addThread :: MonadIO m => SVar m a -> ThreadId -> m ()

337addThread sv tid =

351addThread sv tid =

338 liftIO $ modifyIORef (runningThreads sv) (S.insert tid)

352 liftIO $ modifyIORef (runningThreads sv) (S.insert tid)

339

353

354+{-

340{-# INLINE delThread #-}

355{-# INLINE delThread #-}

341delThread :: MonadIO m => SVar m a -> ThreadId -> m ()

356delThread :: MonadIO m => SVar m a -> ThreadId -> m ()

342delThread sv tid =

357delThread sv tid =

343 liftIO $ modifyIORef (runningThreads sv) $ (\s -> S.delete tid s)

358 liftIO $ modifyIORef (runningThreads sv) $ (\s -> S.delete tid s)

359+-}

360+

361+-- If present then delete else add. This takes care of out of order add and

362+-- delete i.e. a delete arriving before we even added a thread.

363+{-# INLINE modifyThread #-}

364+modifyThread :: MonadIO m => SVar m a -> ThreadId -> m ()

365+modifyThread sv tid = do

366+ changed <- liftIO $ atomicModifyIORefCAS (runningThreads sv) $ \old ->

367+ if (S.member tid old)

368+ then let new = (S.delete tid old) in (new, null new)

369+ else let new = (S.insert tid old) in (new, null old)

370+ if changed

371+ then liftIO $ void $ tryPutMVar (doorBell sv) True

372+ else return ()

344

373

374+-- | This is safe even if we are adding more threads concurrently because if

375+-- a child thread is adding another thread then anyway 'runningThreads' will

376+-- not be empty.

345{-# INLINE allThreadsDone #-}

377{-# INLINE allThreadsDone #-}

346allThreadsDone :: MonadIO m => SVar m a -> m Bool

378allThreadsDone :: MonadIO m => SVar m a -> m Bool

347allThreadsDone sv = liftIO $ S.null <$> readIORef (runningThreads sv)

379allThreadsDone sv = liftIO $ S.null <$> readIORef (runningThreads sv)

@@ -357,6 +389,16 @@ pushWorker :: MonadParallel m => SVar m a -> m ()

357pushWorker sv =

389pushWorker sv =

358 doFork (runqueue sv) (handleChildException sv) >>= addThread sv

390 doFork (runqueue sv) (handleChildException sv) >>= addThread sv

359

391

392+-- | In contrast to pushWorker which always happens only from the consumer

393+-- thread, a pushWorkerPar can happen concurrently from multiple threads on the

394+-- producer side. So we need to use a thread safe modification of

395+-- runningThreads. Alternatively, we can use a CreateThread event to avoid

396+-- using a CAS based modification.

397+{-# NOINLINE pushWorkerPar #-}

398+pushWorkerPar :: MonadParallel m => SVar m a -> Stream m a -> m ()

399+pushWorkerPar sv m =

400+ doFork (runOne sv m) (handleChildException sv) >>= modifyThread sv

401+

360-- XXX When the queue is LIFO we can put a limit on the number of dispatches.

402-- XXX When the queue is LIFO we can put a limit on the number of dispatches.

361-- Also, if a worker blocks on the output queue we can decide if we want to

403-- Also, if a worker blocks on the output queue we can decide if we want to

362-- block or make it go away entirely, depending on the number of workers and

404-- block or make it go away entirely, depending on the number of workers and

@@ -382,8 +424,13 @@ sendWorkerWait sv = do

382{-# NOINLINE fromStreamVar #-}

424{-# NOINLINE fromStreamVar #-}

383fromStreamVar :: MonadParallel m => SVar m a -> Stream m a

425fromStreamVar :: MonadParallel m => SVar m a -> Stream m a

384fromStreamVar sv = Stream $ \_ stp sng yld -> do

426fromStreamVar sv = Stream $ \_ stp sng yld -> do

427+ let SVarStyle _ sched = svarStyle sv

428+ if sched == Par

429+ then void $ liftIO $ takeMVar (doorBell sv)

430+ else do

385− res <- liftIO $ tryTakeMVar (doorBell sv)

431+ res <- liftIO $ tryTakeMVar (doorBell sv)

386− when (isNothing res) $ sendWorkerWait sv

432+ when (isNothing res) $ sendWorkerWait sv

433+

387 list <- liftIO $ atomicModifyIORefCAS (outputQueue sv) $ \x -> ([], x)

434 list <- liftIO $ atomicModifyIORefCAS (outputQueue sv) $ \x -> ([], x)

388 -- Reversing the output is important to guarantee that we process the

435 -- Reversing the output is important to guarantee that we process the

389 -- outputs in the same order as they were generated by the constituent

436 -- outputs in the same order as they were generated by the constituent

@@ -393,7 +440,7 @@ fromStreamVar sv = Stream $ \_ stp sng yld -> do

393 where

440 where

394

441

395 handleException e tid = do

442 handleException e tid = do

396− delThread sv tid

443+ modifyThread sv tid

397 -- XXX implement kill async exception handling

444 -- XXX implement kill async exception handling

398 -- liftIO $ readIORef (runningThreads sv) >>= mapM_ killThread

445 -- liftIO $ readIORef (runningThreads sv) >>= mapM_ killThread

399 throwM e

446 throwM e

@@ -413,7 +460,7 @@ fromStreamVar sv = Stream $ \_ stp sng yld -> do

413 ChildYield a -> yield a

460 ChildYield a -> yield a

414 ChildStop tid e ->

461 ChildStop tid e ->

415 case e of

462 case e of

416− Nothing -> delThread sv tid >> continue

463+ Nothing -> modifyThread sv tid >> continue

417 Just ex -> handleException ex tid

464 Just ex -> handleException ex tid

418

465

419getFifoSVar :: MonadIO m => SVarStyle -> IO (SVar m a)

466getFifoSVar :: MonadIO m => SVarStyle -> IO (SVar m a)

@@ -451,6 +498,22 @@ getLifoSVar ctype = do

451 }

498 }

452 in return sv

499 in return sv

453

500

501+getParSVar :: SVarStyle -> IO (SVar m a)

502+getParSVar style = do

503+ outQ <- newIORef []

504+ outQMv <- newEmptyMVar

505+ running <- newIORef S.empty

506+ let sv =

507+ SVar { outputQueue = outQ

508+ , doorBell = outQMv

509+ , runningThreads = running

510+ , runqueue = undefined

511+ , enqueue = undefined

512+ , queueEmpty = undefined

513+ , svarStyle = style

514+ }

515+ in return sv

516+

454-- | Create a new empty SVar.

517-- | Create a new empty SVar.

455newEmptySVar :: MonadParallel m => SVarStyle -> m (SVar m a)

518newEmptySVar :: MonadParallel m => SVarStyle -> m (SVar m a)

456newEmptySVar style = do

519newEmptySVar style = do

@@ -458,6 +521,7 @@ newEmptySVar style = do

458 case style of

521 case style of

459 SVarStyle _ FIFO -> getFifoSVar style

522 SVarStyle _ FIFO -> getFifoSVar style

460 SVarStyle _ LIFO -> getLifoSVar style

523 SVarStyle _ LIFO -> getLifoSVar style

524+ SVarStyle _ Par -> getParSVar style

461

525

462-- | Create a new SVar and enqueue one stream computation on it.

526-- | Create a new SVar and enqueue one stream computation on it.

463newStreamVar1 :: MonadParallel m => SVarStyle -> Stream m a -> m (SVar m a)

527newStreamVar1 :: MonadParallel m => SVarStyle -> Stream m a -> m (SVar m a)

@@ -487,6 +551,7 @@ newStreamVar2 style m1 m2 = do

487 c <- getLifoSVar style

551 c <- getLifoSVar style

488 (enqueue c) m2 >> (enqueue c) m1

552 (enqueue c) m2 >> (enqueue c) m1

489 return c

553 return c

554+ SVarStyle _ Par -> undefined

490 pushWorker sv

555 pushWorker sv

491 return sv

556 return sv

492

557

@@ -581,6 +646,19 @@ joinStreamVar2 style m1 m2 = Stream $ \svr stp sng yld ->

581 liftIO ((enqueue sv) m2) >> (runStream m1) svr stp sng yld

646 liftIO ((enqueue sv) m2) >> (runStream m1) svr stp sng yld

582 _ -> (runStream (withNewSVar2 style m1 m2)) Nothing stp sng yld

647 _ -> (runStream (withNewSVar2 style m1 m2)) Nothing stp sng yld

583

648

649+{-# INLINE joinStreamVarPar #-}

650+joinStreamVarPar :: MonadParallel m

651+ => SVarStyle -> Stream m a -> Stream m a -> Stream m a

652+joinStreamVarPar style m1 m2 = Stream $ \svr stp sng yld ->

653+ case svr of

654+ Just sv | svarStyle sv == style -> do

655+ pushWorkerPar sv m1 >> (runStream m2) svr stp sng yld

656+ _ -> do

657+ sv <- newEmptySVar style

658+ pushWorkerPar sv m1

659+ pushWorkerPar sv m2

660+ (runStream (fromStreamVar sv)) Nothing stp sng yld

661+

584------------------------------------------------------------------------------

662------------------------------------------------------------------------------

585-- Semigroup and Monoid style compositions for parallel actions

663-- Semigroup and Monoid style compositions for parallel actions

586------------------------------------------------------------------------------

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

@@ -593,6 +671,10 @@ parAhead = joinStreamVar2 (SVarStyle Disjunction LIFO)

593coparAhead :: MonadParallel m => Stream m a -> Stream m a -> Stream m a

671coparAhead :: MonadParallel m => Stream m a -> Stream m a -> Stream m a

594coparAhead = joinStreamVar2 (SVarStyle Disjunction FIFO)

672coparAhead = joinStreamVar2 (SVarStyle Disjunction FIFO)

595

673

674+{-# INLINE parallel #-}

675+parallel :: MonadParallel m => Stream m a -> Stream m a -> Stream m a

676+parallel = joinStreamVarPar (SVarStyle Disjunction Par)

677+

596-------------------------------------------------------------------------------

678-------------------------------------------------------------------------------

597-- Functor instace is the same for all types

679-- Functor instace is the same for all types

598-------------------------------------------------------------------------------

680-------------------------------------------------------------------------------

src/Streamly/Streams.hs+160−12

@@ -56,6 +56,7 @@ module Streamly.Streams

56 , cosplice

57 , parAhead

58 , coparAhead

59+ , parallel

59 , (<=>) --deprecated

60 , (<=>) --deprecated

60 , (<|) --deprecated

61 , (<|) --deprecated

@@ -64,6 +65,7 @@ module Streamly.Streams

64 , Costream

65 , Costream

65 , ParAhead

66 , ParAhead

66 , CoparAhead

67 , CoparAhead

68+ , Parallel

67 , ZipStream

69 , ZipStream

68 , ZipParallel

70 , ZipParallel

@@ -74,6 +76,7 @@ module Streamly.Streams

74 , ParAheadT

76 , ParAheadT

75 , AsyncT -- deprecated

77 , AsyncT -- deprecated

76 , CoparAheadT

78 , CoparAheadT

79+ , ParallelT

77 , ZipStreamM

80 , ZipStreamM

78 , ZipParallelM

81 , ZipParallelM

79 , ZipAsync -- deprecated

82 , ZipAsync -- deprecated

@@ -86,6 +89,7 @@ module Streamly.Streams

86 , asParAhead

89 , asParAhead

87 , asyncly -- deprecated

90 , asyncly -- deprecated

88 , asCoparAhead

91 , asCoparAhead

92+ , asParallel

89 , parallely -- deprecated

93 , parallely -- deprecated

90 , asZipStream

94 , asZipStream

91 , zipping -- deprecated

95 , zipping -- deprecated

@@ -648,8 +652,7 @@ parbind par m f = go m

648

652

649instance MonadParallel m => Monad (ParAheadT m) where

653instance MonadParallel m => Monad (ParAheadT m) where

650 return = pure

654 return = pure

651− (ParAheadT m) >>= f = ParAheadT $ parbind par m (getParAheadT . f)

655+ (ParAheadT m) >>= f = ParAheadT $ parbind S.parAhead m (getParAheadT . f)

652− where par = S.joinStreamVar2 (SVarStyle Conjunction LIFO)

653

656

654------------------------------------------------------------------------------

657------------------------------------------------------------------------------

655-- Other instances

658-- Other instances

@@ -743,8 +746,8 @@ instance MonadParallel m => Monoid (CoparAheadT m a) where

743

746

744instance MonadParallel m => Monad (CoparAheadT m) where

747instance MonadParallel m => Monad (CoparAheadT m) where

745 return = pure

748 return = pure

746− (CoparAheadT m) >>= f = CoparAheadT $ parbind par m (getCoparAheadT . f)

749+ (CoparAheadT m) >>= f =

747− where par = S.joinStreamVar2 (SVarStyle Conjunction FIFO)

750+ CoparAheadT $ parbind S.coparAhead m (getCoparAheadT . f)

748

751

749------------------------------------------------------------------------------

752------------------------------------------------------------------------------

750-- Other instances

753-- Other instances

@@ -753,6 +756,139 @@ instance MonadParallel m => Monad (CoparAheadT m) where

753MONAD_APPLICATIVE_INSTANCE(CoparAheadT,MONADPARALLEL)

756MONAD_APPLICATIVE_INSTANCE(CoparAheadT,MONADPARALLEL)

754MONAD_COMMON_INSTANCES(CoparAheadT, MONADPARALLEL)

757MONAD_COMMON_INSTANCES(CoparAheadT, MONADPARALLEL)

755

758

759+------------------------------------------------------------------------------

760+-- ParallelT

761+------------------------------------------------------------------------------

762+

763+-- | Strictly parallel composition.

764+--

765+-- The Semigroup instance of 'ParallelT' concurrently /merges/ two streams,

766+-- running both strictly concurrently and yielding elements from both streams

767+-- as they arrive. When multiple streams are combined using 'ParallelT' each

768+-- one is evaluated in its own thread and the results produced are presented in

769+-- the combined stream on a first come first serve basis.

770+--

771+-- 'ParAheadT' and 'CoparAheadT' are /concurrent lookahead streams/ each with a

772+-- specific type of consumption pattern (depth first or breadth first). Since

773+-- they are lookahead, they may introduce certain default latency in starting

774+-- more concurrent tasks for efficiency reasons or may put a default limitation

775+-- on the resource consumption (e.g. number of concurrent threads for

776+-- lookahead). If we look at the implementation detail, they both can share a

777+-- pool of worker threads to evaluate the streams in the desired pattern and at

778+-- the desired rate. However, 'ParallelT' uses a separate runtime thread to

779+-- evaluate each stream.

780+--

781+-- 'CoparAheadT' is similar to 'ParallelT', as both of them evaluate the

782+-- constituent streams fairly in a round robin fashion.

783+-- However, the key difference is that 'CoparAheadT' is lazy or pull driven

784+-- whereas 'ParallelT' is strict or push driven. 'ParallelT' immediately

785+-- starts concurrent evaluation of both the streams (in separate threads) and

786+-- later picks the results whereas 'CoparAheadT' may wait for a certain latency

787+-- threshold before initiating concurrent evaluation of the next stream. The

788+-- concurrent scheduling of the next stream or the degree of concurrency is

789+-- driven by the feedback from the consumer. In case of 'ParallelT' each stream

790+-- is evaluated in a separate thread and results are /pushed/ to a shared

791+-- output buffer, the evaluation rate is controlled by blocking when the buffer

792+-- is full.

793+--

794+-- Concurrent lookahead streams are generally more efficient than

795+-- 'ParallelT' and can work pretty efficiently even for smaller tasks because

796+-- they do not necessarily use a separate thread for each task. So they should

797+-- be preferred over 'ParallelT' especially when efficiency is a concern and

798+-- simultaneous strict evaluation is not a requirement. 'ParallelT' is useful

799+-- for cases when the streams are required to be evaluated simultaneously

800+-- irrespective of how the consumer consumes them e.g. when we want to race

801+-- two tasks and want to start both strictly at the same time or if we have

802+-- timers in the parallel tasks and our results depend on the timers being

803+-- started at the same time. We can say that 'ParallelT' is almost the same

804+-- (modulo some implementation differences) as 'CoparAheadT' when the latter is

805+-- used with unlimited lookahead and zero latency in initiating lookahead.

806+--

807+-- @

808+-- main = ('toList' . 'asParallel' $ (fromFoldable [1,2]) \<> (fromFoldable [3,4])) >>= print

809+-- @

810+-- @

811+-- [1,3,2,4]

812+-- @

813+--

814+-- When streams with more than one element are merged, it yields whichever

815+-- stream yields first without any bias, unlike the 'ParAhead' style streams.

816+--

817+-- Any exceptions generated by a constituent stream are propagated to the

818+-- output stream. The output and exceptions from a single stream are guaranteed

819+-- to arrive in the same order in the resulting stream as they were generated

820+-- in the input stream. However, the relative ordering of elements from

821+-- different streams in the resulting stream can vary depending on scheduling

822+-- and generation delays.

823+--

824+-- Similarly, the 'Monad' instance of 'ParallelT' runs /all/ iterations

825+-- of the loop concurrently.

826+--

827+-- @

828+-- import "Streamly"

829+-- import Control.Concurrent

830+--

831+-- main = 'runStream' . 'asParallel' $ do

832+-- n <- return 3 \<\> return 2 \<\> return 1

833+-- liftIO $ do

834+-- threadDelay (n * 1000000)

835+-- myThreadId >>= \\tid -> putStrLn (show tid ++ ": Delay " ++ show n)

836+-- @

837+-- @

838+-- ThreadId 40: Delay 1

839+-- ThreadId 39: Delay 2

840+-- ThreadId 38: Delay 3

841+-- @

842+--

843+-- Note that parallel composition can only combine a finite number of

844+-- streams as it needs to retain state for each unfinished stream.

845+--

846+-- @since 0.1.0

847+newtype ParallelT m a = ParallelT {getParallelT :: S.Stream m a}

848+ deriving (Functor, MonadTrans)

849+

850+instance IsStream ParallelT where

851+ toStream = getParallelT

852+ fromStream = ParallelT

853+

854+------------------------------------------------------------------------------

855+-- Semigroup

856+------------------------------------------------------------------------------

857+

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

859+-- Merges two streams concurrently.

860+--

861+-- @since 0.2.0

862+{-# INLINE parallel #-}

863+parallel :: (IsStream t, MonadParallel m) => t m a -> t m a -> t m a

864+parallel m1 m2 = fromStream $ S.parallel (toStream m1) (toStream m2)

865+

866+instance MonadParallel m => Semigroup (ParallelT m a) where

867+ (<>) = parallel

868+

869+------------------------------------------------------------------------------

870+-- Monoid

871+------------------------------------------------------------------------------

872+

873+instance MonadParallel m => Monoid (ParallelT m a) where

874+ mempty = nil

875+ mappend = (<>)

876+

877+------------------------------------------------------------------------------

878+-- Monad

879+------------------------------------------------------------------------------

880+

881+instance MonadParallel m => Monad (ParallelT m) where

882+ return = pure

883+ (ParallelT m) >>= f = ParallelT $ parbind S.parallel m (getParallelT . f)

884+

885+------------------------------------------------------------------------------

886+-- Other instances

887+------------------------------------------------------------------------------

888+

889+MONAD_APPLICATIVE_INSTANCE(ParallelT,MONADPARALLEL)

890+MONAD_COMMON_INSTANCES(ParallelT, MONADPARALLEL)

891+

756------------------------------------------------------------------------------

892------------------------------------------------------------------------------

757-- Serially Zipping Streams

893-- Serially Zipping Streams

758------------------------------------------------------------------------------

894------------------------------------------------------------------------------

@@ -878,12 +1014,18 @@ asyncly = asParAhead

878asCoparAhead :: IsStream t => CoparAheadT m a -> t m a

1014asCoparAhead :: IsStream t => CoparAheadT m a -> t m a

879asCoparAhead = adapt

1015asCoparAhead = adapt

880

1016

881−-- | Same as 'asCoparAhead'.

1017+-- | Fix the type of a polymorphic stream as 'ParallelT'.

882--

1018--

883-- @since 0.1.0

1019-- @since 0.1.0

1020+asParallel :: IsStream t => ParallelT m a -> t m a

1021+asParallel = adapt

1022+

1023+-- | Same as 'asParallel'.

1024+--

1025+-- @since 0.1.0

884−{-# DEPRECATED parallely "Please use asCoparAhead instead." #-}

1026+{-# DEPRECATED parallely "Please use asParallel instead." #-}

885−parallely :: IsStream t => CoparAheadT m a -> t m a

1027+parallely :: IsStream t => ParallelT m a -> t m a

886−parallely = asCoparAhead

1028+parallely = asParallel

887

1029

888-- | Fix the type of a polymorphic stream as 'ZipStreamM'.

1030-- | Fix the type of a polymorphic stream as 'ZipStreamM'.

889--

1031--

@@ -936,12 +1078,12 @@ runInterleavedT = runStream . asCostream

936runAsyncT :: Monad m => ParAheadT m a -> m ()

1078runAsyncT :: Monad m => ParAheadT m a -> m ()

937runAsyncT = runStream . asParAhead

1079runAsyncT = runStream . asParAhead

938

1080

939−-- | Same as @runStream . asCoparAhead@.

1081+-- | Same as @runStream . asParallel@.

940--

1082--

941-- @since 0.1.0

1083-- @since 0.1.0

942−{-# DEPRECATED runParallelT "Please use 'runStream . asCoparAhead' instead." #-}

1084+{-# DEPRECATED runParallelT "Please use 'runStream . asParallel' instead." #-}

943−runParallelT :: Monad m => CoparAheadT m a -> m ()

1085+runParallelT :: Monad m => ParallelT m a -> m ()

944−runParallelT = runStream . asCoparAhead

1086+runParallelT = runStream . asParallel

945

1087

946-- | Same as @runStream . zipping@.

1088-- | Same as @runStream . zipping@.

947--

1089--

@@ -985,6 +1127,12 @@ type ParAhead a = ParAheadT IO a

985-- @since 0.2.0

1127-- @since 0.2.0

986type CoparAhead a = CoparAheadT IO a

1128type CoparAhead a = CoparAheadT IO a

987

1129

1130+-- | A parallely composing IO stream of elements of type @a@.

1131+-- See 'ParallelT' documentation for more details.

1132+--

1133+-- @since 0.2.0

1134+type Parallel a = ParallelT IO a

1135+

988-- | An IO stream whose applicative instance zips streams serially.

1136-- | An IO stream whose applicative instance zips streams serially.

989--

1137--

990-- @since 0.2.0

1138-- @since 0.2.0

test/Main.hs+87−32

@@ -102,10 +102,13 @@ main = hspec $ do

102 describe "Parallel interleaved (<>) ordering check" $ interleaveCheck asCoparAhead (<>)

103 describe "Parallel interleaved mappend ordering check" $ interleaveCheck asCoparAhead mappend

104

105+ -- describe "Parallel (<>) ordering check" $ interleaveCheck asParallel (<>)

106+ -- describe "Parallel mappend ordering check" $ interleaveCheck asParallel mappend

107+

105− describe "Coparallel (<>) time order check" $ parallelCheck asParAhead (<>)

108+ describe "ParAhead (<>) time order check" $ parallelCheck asParAhead (<>)

106− describe "Coparallel mappend time order check" $ parallelCheck asParAhead mappend

109+ describe "ParAhead mappend time order check" $ parallelCheck asParAhead mappend

107− describe "Parallel (<>) time order check" $ parallelCheck asCoparAhead (<>)

110+ describe "CoparAhead (<>) time order check" $ parallelCheck asCoparAhead (<>)

108− describe "Parallel mappend time order check" $ parallelCheck asCoparAhead mappend

111+ describe "CoparAhead mappend time order check" $ parallelCheck asCoparAhead mappend

109

112

110 ---------------------------------------------------------------------------

113 ---------------------------------------------------------------------------

111 -- Monoidal Compositions, multiset equality checks

114 -- Monoidal Compositions, multiset equality checks

@@ -113,8 +116,9 @@ main = hspec $ do

113

116

114 describe "Serial Composition" $ compose asStream mempty id

117 describe "Serial Composition" $ compose asStream mempty id

115 describe "Interleaved Composition" $ compose asCostream mempty sort

118 describe "Interleaved Composition" $ compose asCostream mempty sort

116− describe "Left biased parallel Composition" $ compose asParAhead mempty sort

119+ describe "ParAhead Composition" $ compose asParAhead mempty sort

117− describe "Fair parallel Composition" $ compose asCoparAhead mempty sort

120+ describe "CoparAhead Composition" $ compose asCoparAhead mempty sort

121+ describe "Parallel Composition" $ compose asParallel mempty sort

118 describe "Semigroup Composition for ZipSerial" $ compose asZipStream mempty id

122 describe "Semigroup Composition for ZipSerial" $ compose asZipStream mempty id

119 describe "Semigroup Composition for ZipAsync" $ compose asZipParallel mempty id

123 describe "Semigroup Composition for ZipAsync" $ compose asZipParallel mempty id

120 -- XXX need to check alternative compositions as well

124 -- XXX need to check alternative compositions as well

@@ -198,33 +202,44 @@ main = hspec $ do

198 -- Monoidal composition recursion loops

202 -- Monoidal composition recursion loops

199 ---------------------------------------------------------------------------

203 ---------------------------------------------------------------------------

200

204

201− describe "Serial loops (<>)" $ loops asStream id reverse

205+ describe "Serial loops" $ loops asStream id reverse

202− describe "Left biased parallel loops (<|)" $ loops asParAhead sort sort

206+ describe "ParAhead parallel loops" $ loops asParAhead sort sort

203− describe "Fair parallel loops (<|>)" $ loops asCoparAhead sort sort

207+ describe "coparAhead loops" $ loops asCoparAhead sort sort

208+ describe "parallel loops" $ loops asParallel sort sort

204

209

205 ---------------------------------------------------------------------------

210 ---------------------------------------------------------------------------

206 -- Bind and monoidal composition combinations

211 -- Bind and monoidal composition combinations

207 ---------------------------------------------------------------------------

212 ---------------------------------------------------------------------------

208

213

209− describe "Bind and compose1" $ bindAndComposeSimple asStream asStream

214+ describe "Bind and compose Stream 1" $ bindAndComposeSimple asStream asStream

210− describe "Bind and compose2" $ bindAndComposeSimple asStream asCostream

215+ describe "Bind and compose Stream 2" $ bindAndComposeSimple asStream asCostream

211− describe "Bind and compose3" $ bindAndComposeSimple asStream asParAhead

216+ describe "Bind and compose Stream 3" $ bindAndComposeSimple asStream asParAhead

212− describe "Bind and compose4" $ bindAndComposeSimple asStream asCoparAhead

217+ describe "Bind and compose Stream 4" $ bindAndComposeSimple asStream asCoparAhead

213−

218+ describe "Bind and compose Stream 5" $ bindAndComposeSimple asStream asParallel

219+

214− describe "Bind and compose1" $ bindAndComposeSimple asCostream asStream

220+ describe "Bind and compose Costream 1" $ bindAndComposeSimple asCostream asStream

215− describe "Bind and compose2" $ bindAndComposeSimple asCostream asCostream

221+ describe "Bind and compose Costream 2" $ bindAndComposeSimple asCostream asCostream

216− describe "Bind and compose3" $ bindAndComposeSimple asCostream asParAhead

222+ describe "Bind and compose Costream 3" $ bindAndComposeSimple asCostream asParAhead

217− describe "Bind and compose4" $ bindAndComposeSimple asCostream asCoparAhead

223+ describe "Bind and compose Costream 4" $ bindAndComposeSimple asCostream asCoparAhead

218−

224+ describe "Bind and compose Costream 5" $ bindAndComposeSimple asCostream asParallel

225+

219− describe "Bind and compose1" $ bindAndComposeSimple asParAhead asStream

226+ describe "Bind and compose ParAhead 1" $ bindAndComposeSimple asParAhead asStream

220− describe "Bind and compose2" $ bindAndComposeSimple asParAhead asCostream

227+ describe "Bind and compose ParAhead 2" $ bindAndComposeSimple asParAhead asCostream

221− describe "Bind and compose3" $ bindAndComposeSimple asParAhead asParAhead

228+ describe "Bind and compose ParAhead 3" $ bindAndComposeSimple asParAhead asParAhead

222− describe "Bind and compose4" $ bindAndComposeSimple asParAhead asCoparAhead

229+ describe "Bind and compose ParAhead 4" $ bindAndComposeSimple asParAhead asCoparAhead

223−

230+ describe "Bind and compose ParAhead 5" $ bindAndComposeSimple asParAhead asParallel

231+

224− describe "Bind and compose1" $ bindAndComposeSimple asCoparAhead asStream

232+ describe "Bind and compose CoparAhead 1" $ bindAndComposeSimple asCoparAhead asStream

225− describe "Bind and compose2" $ bindAndComposeSimple asCoparAhead asCostream

233+ describe "Bind and compose CoparAhead 2" $ bindAndComposeSimple asCoparAhead asCostream

226− describe "Bind and compose3" $ bindAndComposeSimple asCoparAhead asParAhead

234+ describe "Bind and compose CoparAhead 3" $ bindAndComposeSimple asCoparAhead asParAhead

235+ describe "Bind and compose CoparAhead 4" $ bindAndComposeSimple asCoparAhead asCoparAhead

227− describe "Bind and compose4" $ bindAndComposeSimple asCoparAhead asCoparAhead

236+ describe "Bind and compose CoparAhead 5" $ bindAndComposeSimple asCoparAhead asParallel

237+

238+ describe "Bind and compose Parallel 1" $ bindAndComposeSimple asParallel asStream

239+ describe "Bind and compose Parallel 2" $ bindAndComposeSimple asParallel asCostream

240+ describe "Bind and compose Parallel 3" $ bindAndComposeSimple asParallel asParAhead

241+ describe "Bind and compose Parallel 4" $ bindAndComposeSimple asParallel asCoparAhead

242+ describe "Bind and compose Parallel 5" $ bindAndComposeSimple asParallel asParallel

228

243

229 let fldr, fldl :: (IsStream t, Semigroup (t IO Int)) => [t IO Int] -> t IO Int

244 let fldr, fldl :: (IsStream t, Semigroup (t IO Int)) => [t IO Int] -> t IO Int

230 fldr = foldr (<>) nil

245 fldr = foldr (<>) nil

@@ -238,6 +253,8 @@ main = hspec $ do

238 describe "Bind and compose" $ bindAndComposeHierarchy asStream asParAhead k

253 describe "Bind and compose" $ bindAndComposeHierarchy asStream asParAhead k

239 forM_ [fldr, fldl] $ \k ->

254 forM_ [fldr, fldl] $ \k ->

240 describe "Bind and compose" $ bindAndComposeHierarchy asStream asCoparAhead k

255 describe "Bind and compose" $ bindAndComposeHierarchy asStream asCoparAhead k

256+ forM_ [fldr, fldl] $ \k ->

257+ describe "Bind and compose" $ bindAndComposeHierarchy asStream asParallel k

241

258

242 forM_ [fldr, fldl] $ \k ->

259 forM_ [fldr, fldl] $ \k ->

243 describe "Bind and compose" $ bindAndComposeHierarchy asCostream asStream k

260 describe "Bind and compose" $ bindAndComposeHierarchy asCostream asStream k

@@ -247,6 +264,8 @@ main = hspec $ do

247 describe "Bind and compose" $ bindAndComposeHierarchy asCostream asParAhead k

264 describe "Bind and compose" $ bindAndComposeHierarchy asCostream asParAhead k

248 forM_ [fldr, fldl] $ \k ->

265 forM_ [fldr, fldl] $ \k ->

249 describe "Bind and compose" $ bindAndComposeHierarchy asCostream asCoparAhead k

266 describe "Bind and compose" $ bindAndComposeHierarchy asCostream asCoparAhead k

267+ forM_ [fldr, fldl] $ \k ->

268+ describe "Bind and compose" $ bindAndComposeHierarchy asCostream asParallel k

250

269

251 forM_ [fldr, fldl] $ \k ->

270 forM_ [fldr, fldl] $ \k ->

252 describe "Bind and compose" $ bindAndComposeHierarchy asParAhead asStream k

271 describe "Bind and compose" $ bindAndComposeHierarchy asParAhead asStream k

@@ -256,6 +275,8 @@ main = hspec $ do

256 describe "Bind and compose" $ bindAndComposeHierarchy asParAhead asParAhead k

275 describe "Bind and compose" $ bindAndComposeHierarchy asParAhead asParAhead k

257 forM_ [fldr, fldl] $ \k ->

276 forM_ [fldr, fldl] $ \k ->

258 describe "Bind and compose" $ bindAndComposeHierarchy asParAhead asCoparAhead k

277 describe "Bind and compose" $ bindAndComposeHierarchy asParAhead asCoparAhead k

278+ forM_ [fldr, fldl] $ \k ->

279+ describe "Bind and compose" $ bindAndComposeHierarchy asParAhead asParallel k

259

280

260 forM_ [fldr, fldl] $ \k ->

281 forM_ [fldr, fldl] $ \k ->

261 describe "Bind and compose" $ bindAndComposeHierarchy asCoparAhead asStream k

282 describe "Bind and compose" $ bindAndComposeHierarchy asCoparAhead asStream k

@@ -265,16 +286,31 @@ main = hspec $ do

265 describe "Bind and compose" $ bindAndComposeHierarchy asCoparAhead asParAhead k

286 describe "Bind and compose" $ bindAndComposeHierarchy asCoparAhead asParAhead k

266 forM_ [fldr, fldl] $ \k ->

287 forM_ [fldr, fldl] $ \k ->

267 describe "Bind and compose" $ bindAndComposeHierarchy asCoparAhead asCoparAhead k

288 describe "Bind and compose" $ bindAndComposeHierarchy asCoparAhead asCoparAhead k

289+ forM_ [fldr, fldl] $ \k ->

290+ describe "Bind and compose" $ bindAndComposeHierarchy asCoparAhead asParallel k

291+

292+ forM_ [fldr, fldl] $ \k ->

293+ describe "Bind and compose" $ bindAndComposeHierarchy asParallel asStream k

294+ forM_ [fldr, fldl] $ \k ->

295+ describe "Bind and compose" $ bindAndComposeHierarchy asParallel asCostream k

296+ forM_ [fldr, fldl] $ \k ->

297+ describe "Bind and compose" $ bindAndComposeHierarchy asParallel asParAhead k

298+ forM_ [fldr, fldl] $ \k ->

299+ describe "Bind and compose" $ bindAndComposeHierarchy asParallel asCoparAhead k

300+ forM_ [fldr, fldl] $ \k ->

301+ describe "Bind and compose" $ bindAndComposeHierarchy asParallel asParallel k

268

302

269 -- Nest two lists using different styles of product compositions

303 -- Nest two lists using different styles of product compositions

270 it "Nests two streams using monadic serial composition" nestTwoSerial

304 it "Nests two streams using monadic serial composition" nestTwoSerial

271 it "Nests two streams using monadic interleaved composition" nestTwoInterleaved

305 it "Nests two streams using monadic interleaved composition" nestTwoInterleaved

272− it "Nests two streams using monadic async composition" nestTwoAsync

306+ it "Nests two streams using monadic parAhead composition" nestTwoAsync

307+ it "Nests two streams using monadic CoparAhead composition" nestTwoCoparAhead

273 it "Nests two streams using monadic parallel composition" nestTwoParallel

308 it "Nests two streams using monadic parallel composition" nestTwoParallel

274

309

275 it "Nests two streams using applicative serial composition" nestTwoSerialApp

310 it "Nests two streams using applicative serial composition" nestTwoSerialApp

276 it "Nests two streams using applicative interleaved composition" nestTwoInterleavedApp

311 it "Nests two streams using applicative interleaved composition" nestTwoInterleavedApp

277− it "Nests two streams using applicative async composition" nestTwoAsyncApp

312+ it "Nests two streams using applicative parAhead composition" nestTwoAsyncApp

313+ it "Nests two streams using applicative CoparAhead composition" nestTwoCoparAheadApp

278 it "Nests two streams using applicative parallel composition" nestTwoParallelApp

314 it "Nests two streams using applicative parallel composition" nestTwoParallelApp

279

315

280 ---------------------------------------------------------------------------

316 ---------------------------------------------------------------------------

@@ -296,6 +332,7 @@ main = hspec $ do

296 describe "Composed MonadThrow asCostream" $ composeWithMonadThrow asCostream

332 describe "Composed MonadThrow asCostream" $ composeWithMonadThrow asCostream

297 describe "Composed MonadThrow asParAhead" $ composeWithMonadThrow asParAhead

333 describe "Composed MonadThrow asParAhead" $ composeWithMonadThrow asParAhead

298 describe "Composed MonadThrow asCoparAhead" $ composeWithMonadThrow asCoparAhead

334 describe "Composed MonadThrow asCoparAhead" $ composeWithMonadThrow asCoparAhead

335+ describe "Composed MonadThrow asParallel" $ composeWithMonadThrow asParallel

299

336

300-- XXX need to test that we have promptly cleaned up everything after the error

337-- XXX need to test that we have promptly cleaned up everything after the error

301-- XXX We can also check the output that we are expected to get before the

338-- XXX We can also check the output that we are expected to get before the

@@ -432,22 +469,40 @@ nestTwoAsyncApp =

432 in (toListAsync ((+) <$> s1 <*> s2) >>= return . sort)

469 in (toListAsync ((+) <$> s1 <*> s2) >>= return . sort)

433 `shouldReturn` sort ([6,7,8,9,7,8,9,10,8,9,10,11,9,10,11,12] :: [Int])

470 `shouldReturn` sort ([6,7,8,9,7,8,9,10,8,9,10,11,9,10,11,12] :: [Int])

434

471

472+nestTwoCoparAhead :: Expectation

473+nestTwoCoparAhead =

474+ let s1 = foldMapWith (<>) return [1..4]

475+ s2 = foldMapWith (<>) return [5..8]

476+ in ((A.toList . asCoparAhead) (do

477+ x <- s1

478+ y <- s2

479+ return (x + y)

480+ ) >>= return . sort)

481+ `shouldReturn` sort ([6,7,7,8,8,8,9,9,9,9,10,10,10,11,11,12] :: [Int])

482+

435nestTwoParallel :: Expectation

483nestTwoParallel :: Expectation

436nestTwoParallel =

484nestTwoParallel =

437 let s1 = foldMapWith (<>) return [1..4]

485 let s1 = foldMapWith (<>) return [1..4]

438 s2 = foldMapWith (<>) return [5..8]

486 s2 = foldMapWith (<>) return [5..8]

439− in (toListParallel (do

487+ in ((A.toList . asParallel) (do

440 x <- s1

488 x <- s1

441 y <- s2

489 y <- s2

442 return (x + y)

490 return (x + y)

443 ) >>= return . sort)

491 ) >>= return . sort)

444 `shouldReturn` sort ([6,7,7,8,8,8,9,9,9,9,10,10,10,11,11,12] :: [Int])

492 `shouldReturn` sort ([6,7,7,8,8,8,9,9,9,9,10,10,10,11,11,12] :: [Int])

445

493

494+nestTwoCoparAheadApp :: Expectation

495+nestTwoCoparAheadApp =

496+ let s1 = foldMapWith (<>) return [1..4]

497+ s2 = foldMapWith (<>) return [5..8]

498+ in ((A.toList . asCoparAhead) ((+) <$> s1 <*> s2) >>= return . sort)

499+ `shouldReturn` sort ([6,7,7,8,8,8,9,9,9,9,10,10,10,11,11,12] :: [Int])

500+

446nestTwoParallelApp :: Expectation

501nestTwoParallelApp :: Expectation

447nestTwoParallelApp =

502nestTwoParallelApp =

448 let s1 = foldMapWith (<>) return [1..4]

503 let s1 = foldMapWith (<>) return [1..4]

449 s2 = foldMapWith (<>) return [5..8]

504 s2 = foldMapWith (<>) return [5..8]

450− in (toListParallel ((+) <$> s1 <*> s2) >>= return . sort)

505+ in ((A.toList . asParallel) ((+) <$> s1 <*> s2) >>= return . sort)

451 `shouldReturn` sort ([6,7,7,8,8,8,9,9,9,9,10,10,10,11,11,12] :: [Int])

506 `shouldReturn` sort ([6,7,7,8,8,8,9,9,9,9,10,10,10,11,11,12] :: [Int])

452

507

453timed :: (IsStream t, Monad (t IO)) => Int -> t IO Int

508timed :: (IsStream t, Monad (t IO)) => Int -> t IO Int

test/Prop.hs+15

@@ -322,6 +322,8 @@ main = hspec $ do

322 functorOps folded "asParAhead folded" asParAhead sortEq

323 functorOps A.fromFoldable "asCoparAhead" asCoparAhead sortEq

324 functorOps folded "asCoparAhead folded" asCoparAhead sortEq

325+ functorOps A.fromFoldable "asParallel" asParallel sortEq

326+ functorOps folded "asParallel folded" asParallel sortEq

325 functorOps A.fromFoldable "asZipStream" asZipStream (==)

327 functorOps A.fromFoldable "asZipStream" asZipStream (==)

326 functorOps folded "asZipStream folded" asZipStream (==)

328 functorOps folded "asZipStream folded" asZipStream (==)

327 functorOps A.fromFoldable "asZipParallel" asZipParallel (==)

329 functorOps A.fromFoldable "asZipParallel" asZipParallel (==)

@@ -332,6 +334,7 @@ main = hspec $ do

332 semigroupOps "asCostream" asCostream (==)

334 semigroupOps "asCostream" asCostream (==)

333 semigroupOps "asParAhead" asParAhead sortEq

335 semigroupOps "asParAhead" asParAhead sortEq

334 semigroupOps "asCoparAhead" asCoparAhead sortEq

336 semigroupOps "asCoparAhead" asCoparAhead sortEq

337+ semigroupOps "asParallel" asParallel sortEq

335 semigroupOps "asZipStream" asZipStream (==)

338 semigroupOps "asZipStream" asZipStream (==)

336 semigroupOps "asZipParallel" asZipParallel (==)

339 semigroupOps "asZipParallel" asZipParallel (==)

337

340

@@ -346,6 +349,7 @@ main = hspec $ do

346 prop "asParAhead applicative" $ applicativeOps A.fromFoldable asParAhead sortEq

349 prop "asParAhead applicative" $ applicativeOps A.fromFoldable asParAhead sortEq

347 prop "asParAhead applicative folded" $ applicativeOps folded asParAhead sortEq

350 prop "asParAhead applicative folded" $ applicativeOps folded asParAhead sortEq

348 prop "asCoparAhead applicative folded" $ applicativeOps folded asCoparAhead sortEq

351 prop "asCoparAhead applicative folded" $ applicativeOps folded asCoparAhead sortEq

352+ prop "asParallel applicative folded" $ applicativeOps folded asParallel sortEq

349

353

350 describe "Zip operations" $ do

354 describe "Zip operations" $ do

351 prop "asZipStream applicative" $ zipApplicative A.fromFoldable asZipStream (==)

355 prop "asZipStream applicative" $ zipApplicative A.fromFoldable asZipStream (==)

@@ -359,22 +363,27 @@ main = hspec $ do

359 prop "zip monadic asParAhead folded" $ zipMonadic folded asParAhead (==)

363 prop "zip monadic asParAhead folded" $ zipMonadic folded asParAhead (==)

360 prop "zip monadic asCoparAhead" $ zipMonadic A.fromFoldable asCoparAhead (==)

364 prop "zip monadic asCoparAhead" $ zipMonadic A.fromFoldable asCoparAhead (==)

361 prop "zip monadic asCoparAhead folded" $ zipMonadic folded asCoparAhead (==)

365 prop "zip monadic asCoparAhead folded" $ zipMonadic folded asCoparAhead (==)

366+ prop "zip monadic asParallel" $ zipMonadic A.fromFoldable asParallel (==)

367+ prop "zip monadic asParallel folded" $ zipMonadic folded asParallel (==)

362

368

363 describe "Monad operations" $ do

369 describe "Monad operations" $ do

364 prop "asStream monad then" $ monadThen A.fromFoldable asStream (==)

370 prop "asStream monad then" $ monadThen A.fromFoldable asStream (==)

365 prop "asCostream monad then" $ monadThen A.fromFoldable asCostream sortEq

371 prop "asCostream monad then" $ monadThen A.fromFoldable asCostream sortEq

366 prop "asParAhead monad then" $ monadThen A.fromFoldable asParAhead sortEq

372 prop "asParAhead monad then" $ monadThen A.fromFoldable asParAhead sortEq

367 prop "asCoparAhead monad then" $ monadThen A.fromFoldable asCoparAhead sortEq

373 prop "asCoparAhead monad then" $ monadThen A.fromFoldable asCoparAhead sortEq

374+ prop "asParallel monad then" $ monadThen A.fromFoldable asParallel sortEq

368

375

369 prop "asStream monad then folded" $ monadThen folded asStream (==)

376 prop "asStream monad then folded" $ monadThen folded asStream (==)

370 prop "asCostream monad then folded" $ monadThen folded asCostream sortEq

377 prop "asCostream monad then folded" $ monadThen folded asCostream sortEq

371 prop "asParAhead monad then folded" $ monadThen folded asParAhead sortEq

378 prop "asParAhead monad then folded" $ monadThen folded asParAhead sortEq

372 prop "asCoparAhead monad then folded" $ monadThen folded asCoparAhead sortEq

379 prop "asCoparAhead monad then folded" $ monadThen folded asCoparAhead sortEq

380+ prop "asParallel monad then folded" $ monadThen folded asParallel sortEq

373

381

374 prop "asStream monad bind" $ monadBind A.fromFoldable asStream (==)

382 prop "asStream monad bind" $ monadBind A.fromFoldable asStream (==)

375 prop "asCostream monad bind" $ monadBind A.fromFoldable asCostream sortEq

383 prop "asCostream monad bind" $ monadBind A.fromFoldable asCostream sortEq

376 prop "asParAhead monad bind" $ monadBind A.fromFoldable asParAhead sortEq

384 prop "asParAhead monad bind" $ monadBind A.fromFoldable asParAhead sortEq

377 prop "asCoparAhead monad bind" $ monadBind A.fromFoldable asCoparAhead sortEq

385 prop "asCoparAhead monad bind" $ monadBind A.fromFoldable asCoparAhead sortEq

386+ prop "asParallel monad bind" $ monadBind A.fromFoldable asParallel sortEq

378

387

379 describe "Stream transform operations" $ do

388 describe "Stream transform operations" $ do

380 transformOps A.fromFoldable "asStream" asStream (==)

389 transformOps A.fromFoldable "asStream" asStream (==)

@@ -383,6 +392,7 @@ main = hspec $ do

383 transformOps A.fromFoldable "asZipParallel" asZipParallel (==)

392 transformOps A.fromFoldable "asZipParallel" asZipParallel (==)

384 transformOps A.fromFoldable "asParAhead" asParAhead sortEq

393 transformOps A.fromFoldable "asParAhead" asParAhead sortEq

385 transformOps A.fromFoldable "asCoparAhead" asCoparAhead sortEq

394 transformOps A.fromFoldable "asCoparAhead" asCoparAhead sortEq

395+ transformOps A.fromFoldable "asParallel" asParallel sortEq

386

396

387 transformOps folded "asStream folded" asStream (==)

397 transformOps folded "asStream folded" asStream (==)

388 transformOps folded "asCostream folded" asCostream (==)

398 transformOps folded "asCostream folded" asCostream (==)

@@ -390,6 +400,7 @@ main = hspec $ do

390 transformOps folded "asZipParallel folded" asZipParallel (==)

400 transformOps folded "asZipParallel folded" asZipParallel (==)

391 transformOps folded "asParAhead folded" asParAhead sortEq

401 transformOps folded "asParAhead folded" asParAhead sortEq

392 transformOps folded "asCoparAhead folded" asCoparAhead sortEq

402 transformOps folded "asCoparAhead folded" asCoparAhead sortEq

403+ transformOps folded "asParallel folded" asParallel sortEq

393

404

394 transformOpsWord8 A.fromFoldable "asStream" asStream

405 transformOpsWord8 A.fromFoldable "asStream" asStream

395 transformOpsWord8 A.fromFoldable "asCostream" asCostream

406 transformOpsWord8 A.fromFoldable "asCostream" asCostream

@@ -397,6 +408,7 @@ main = hspec $ do

397 transformOpsWord8 A.fromFoldable "asZipParallel" asZipParallel

408 transformOpsWord8 A.fromFoldable "asZipParallel" asZipParallel

398 transformOpsWord8 A.fromFoldable "asParAhead" asParAhead

409 transformOpsWord8 A.fromFoldable "asParAhead" asParAhead

399 transformOpsWord8 A.fromFoldable "asCoparAhead" asCoparAhead

410 transformOpsWord8 A.fromFoldable "asCoparAhead" asCoparAhead

411+ transformOpsWord8 A.fromFoldable "asParallel" asParallel

400

412

401 transformOpsWord8 folded "asStream folded" asStream

413 transformOpsWord8 folded "asStream folded" asStream

402 transformOpsWord8 folded "asCostream folded" asCostream

414 transformOpsWord8 folded "asCostream folded" asCostream

@@ -404,6 +416,7 @@ main = hspec $ do

404 transformOpsWord8 folded "asZipParallel folded" asZipParallel

416 transformOpsWord8 folded "asZipParallel folded" asZipParallel

405 transformOpsWord8 folded "asParAhead folded" asParAhead

417 transformOpsWord8 folded "asParAhead folded" asParAhead

406 transformOpsWord8 folded "asCoparAhead folded" asCoparAhead

418 transformOpsWord8 folded "asCoparAhead folded" asCoparAhead

419+ transformOpsWord8 folded "asParallel folded" asParallel

407

420

408 describe "Stream elimination operations" $ do

421 describe "Stream elimination operations" $ do

409 eliminationOps A.fromFoldable "asStream" asStream

422 eliminationOps A.fromFoldable "asStream" asStream

@@ -412,6 +425,7 @@ main = hspec $ do

412 eliminationOps A.fromFoldable "asZipParallel" asZipParallel

425 eliminationOps A.fromFoldable "asZipParallel" asZipParallel

413 eliminationOps A.fromFoldable "asParAhead" asParAhead

426 eliminationOps A.fromFoldable "asParAhead" asParAhead

414 eliminationOps A.fromFoldable "asCoparAhead" asCoparAhead

427 eliminationOps A.fromFoldable "asCoparAhead" asCoparAhead

428+ eliminationOps A.fromFoldable "asParallel" asParallel

415

429

416 eliminationOps folded "asStream folded" asStream

430 eliminationOps folded "asStream folded" asStream

417 eliminationOps folded "asCostream folded" asCostream

431 eliminationOps folded "asCostream folded" asCostream

@@ -419,6 +433,7 @@ main = hspec $ do

419 eliminationOps folded "asZipParallel folded" asZipParallel

433 eliminationOps folded "asZipParallel folded" asZipParallel

420 eliminationOps folded "asParAhead folded" asParAhead

434 eliminationOps folded "asParAhead folded" asParAhead

421 eliminationOps folded "asCoparAhead folded" asCoparAhead

435 eliminationOps folded "asCoparAhead folded" asCoparAhead

436+ eliminationOps folded "asParallel folded" asParallel

422

437

423 describe "Stream elimination operations" $ do

438 describe "Stream elimination operations" $ do

424 serialEliminationOps A.fromFoldable "asStream" asStream

439 serialEliminationOps A.fromFoldable "asStream" asStream