091f4b6 · streamly · git.iko.soy

git.iko.soy/streamly

Add the strictly parallel stream type ParallelT

commit091f4b69fe409a52bc1dc26f3e7edeae495b6410 authorHarendra Kumar <harendra.kumar@gmail.com> dateTue 8 May 2018, 07:49 (8 years ago) parent74dd5a0

9 files changed+371−65

compare with master

unifiedsplit

benchmark/Linear.hs+5−4 benchmark/LinearOps.hs+3−2 examples/ListDir.hs+3−3 examples/SearchQuery.hs+2−2 src/Streamly.hs+5−1 src/Streamly/Core.hs+91−9 src/Streamly/Streams.hs+160−12 test/Main.hs+87−32 test/Prop.hs+15

benchmark/Linear.hs+5−4

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

5050 ]

5151 , benchIO "zip" Ops.zip

5252 , bgroup "append"

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

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

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

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

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

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

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

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

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

5758 ]

5859 , bgroup "compose"

5960 [ benchIO "mapM" Ops.composeMapM

benchmark/LinearOps.hs+3−2

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

1414import qualified Streamly as S

1515import qualified Streamly.Prelude as S

1616

17−value, maxValue :: Int

17+value, appendValue, maxValue :: Int

1818value = 1000000

19+appendValue = 100000

1920maxValue = value + 1000

2021

2122-------------------------------------------------------------------------------

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

111112append

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

113114 => (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]

115116

116117-------------------------------------------------------------------------------

117118-- Composition

examples/ListDir.hs+3−3

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

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

22import Path.IO (listDir, getCurrentDir)

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

4−import Streamly (runStream, coparallely)

4+import Streamly (runStream, asParAhead)

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

77--

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

99-- 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.

1111main :: IO ()

1212main = do

1313 hSetBuffering stdout LineBuffering

14− runStream . coparallely $ getCurrentDir >>= readdir

14+ runStream . asParAhead $ getCurrentDir >>= readdir

1515 where readdir d = do

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

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

examples/SearchQuery.hs+2−2

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

1010main :: IO ()

1111main = do

1212 putStrLn "Using parallel semigroup composition"

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

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

1414

1515 putStrLn "\nUsing parallel applicative zip"

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

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

1717

1818 where

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

src/Streamly.hs+5−1

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

6565 , StreamT

6666 , CostreamT

6767

68− -- ** Parallel Ahead Streams

68+ -- ** Parallel Streams

6969 -- $parAhead

7070 , ParAheadT

7171 , CoparAheadT

72+ , ParallelT

7273

7374 -- ** Zipping Streams

7475 -- $zipping

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

8182 , cosplice

8283 , parAhead

8384 , coparAhead

85+ , parallel

8486

8587 -- * Stream Type Adapters

8688 -- $adapters

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

9092 , asCostream

9193 , asParAhead

9294 , asCoparAhead

95+ , asParallel

9396 , asZipStream

9497 , asZipParallel

9598 , adapt

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

99102 , Costream

100103 , ParAhead

101104 , CoparAhead

105+ , Parallel

102106 , ZipStream

103107 , ZipParallel

104108

src/Streamly/Core.hs+91−9

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

3737 , cosplice

3838 , parAhead

3939 , coparAhead

40+ , parallel

4041

4142 -- * Alternative

4243 , alt

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

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

108109data SVarTag = Conjunction | Disjunction deriving Eq

109110

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

113116

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

115118-- be scheduled on the same SVar.

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

324327 single a = sendit a >> run

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

326329

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+

327341-- Thread tracking is needed for two reasons:

328342--

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

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

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

332346--

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

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

334348

335349{-# NOINLINE addThread #-}

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

337351addThread sv tid =

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

339353

354+{-

340355{-# INLINE delThread #-}

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

342357delThread sv tid =

343358 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 ()

344373

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.

345377{-# INLINE allThreadsDone #-}

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

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

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

357389pushWorker sv =

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

359391

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+

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

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

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

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

382424{-# NOINLINE fromStreamVar #-}

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

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

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

386− when (isNothing res) $ sendWorkerWait sv

427+ let SVarStyle _ sched = svarStyle sv

428+ if sched == Par

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

430+ else do

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

432+ when (isNothing res) $ sendWorkerWait sv

433+

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

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

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

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

393440 where

394441

395442 handleException e tid = do

396− delThread sv tid

443+ modifyThread sv tid

397444 -- XXX implement kill async exception handling

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

399446 throwM e

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

413460 ChildYield a -> yield a

414461 ChildStop tid e ->

415462 case e of

416− Nothing -> delThread sv tid >> continue

463+ Nothing -> modifyThread sv tid >> continue

417464 Just ex -> handleException ex tid

418465

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

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

451498 }

452499 in return sv

453500

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+

454517-- | Create a new empty SVar.

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

456519newEmptySVar style = do

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

458521 case style of

459522 SVarStyle _ FIFO -> getFifoSVar style

460523 SVarStyle _ LIFO -> getLifoSVar style

524+ SVarStyle _ Par -> getParSVar style

461525

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

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

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

487551 c <- getLifoSVar style

488552 (enqueue c) m2 >> (enqueue c) m1

489553 return c

554+ SVarStyle _ Par -> undefined

490555 pushWorker sv

491556 return sv

492557

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

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

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

583648

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+

584662------------------------------------------------------------------------------

585663-- Semigroup and Monoid style compositions for parallel actions

586664------------------------------------------------------------------------------

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

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

594672coparAhead = joinStreamVar2 (SVarStyle Disjunction FIFO)

595673

674+{-# INLINE parallel #-}

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

676+parallel = joinStreamVarPar (SVarStyle Disjunction Par)

677+

596678-------------------------------------------------------------------------------

597679-- Functor instace is the same for all types

598680-------------------------------------------------------------------------------

src/Streamly/Streams.hs+160−12

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

5656 , cosplice

5757 , parAhead

5858 , coparAhead

59+ , parallel

5960 , (<=>) --deprecated

6061 , (<|) --deprecated

6162

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

6465 , Costream

6566 , ParAhead

6667 , CoparAhead

68+ , Parallel

6769 , ZipStream

6870 , ZipParallel

6971

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

7476 , ParAheadT

7577 , AsyncT -- deprecated

7678 , CoparAheadT

79+ , ParallelT

7780 , ZipStreamM

7881 , ZipParallelM

7982 , ZipAsync -- deprecated

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

8689 , asParAhead

8790 , asyncly -- deprecated

8891 , asCoparAhead

92+ , asParallel

8993 , parallely -- deprecated

9094 , asZipStream

9195 , zipping -- deprecated

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

648652

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

650654 return = pure

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

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

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

653656

654657------------------------------------------------------------------------------

655658-- Other instances

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

743746

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

745748 return = pure

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

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

749+ (CoparAheadT m) >>= f =

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

748751

749752------------------------------------------------------------------------------

750753-- Other instances

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

753756MONAD_APPLICATIVE_INSTANCE(CoparAheadT,MONADPARALLEL)

754757MONAD_COMMON_INSTANCES(CoparAheadT, MONADPARALLEL)

755758

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+

756892------------------------------------------------------------------------------

757893-- Serially Zipping Streams

758894------------------------------------------------------------------------------

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

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

8791015asCoparAhead = adapt

8801016

881−-- | Same as 'asCoparAhead'.

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

8821018--

8831019-- @since 0.1.0

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

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

886−parallely = asCoparAhead

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

1021+asParallel = adapt

1022+

1023+-- | Same as 'asParallel'.

1024+--

1025+-- @since 0.1.0

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

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

1028+parallely = asParallel

8871029

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

8891031--

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

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

9371079runAsyncT = runStream . asParAhead

9381080

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

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

9401082--

9411083-- @since 0.1.0

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

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

944−runParallelT = runStream . asCoparAhead

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

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

1086+runParallelT = runStream . asParallel

9451087

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

9471089--

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

9851127-- @since 0.2.0

9861128type CoparAhead a = CoparAheadT IO a

9871129

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+

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

9891137--

9901138-- @since 0.2.0

test/Main.hs+87−32

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

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

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

104104

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

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

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

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

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

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

107+

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

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

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

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

109112

110113 ---------------------------------------------------------------------------

111114 -- Monoidal Compositions, multiset equality checks

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

113116

114117 describe "Serial Composition" $ compose asStream mempty id

115118 describe "Interleaved Composition" $ compose asCostream mempty sort

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

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

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

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

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

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

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

120124 -- XXX need to check alternative compositions as well

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

198202 -- Monoidal composition recursion loops

199203 ---------------------------------------------------------------------------

200204

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

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

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

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

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

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

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

204209

205210 ---------------------------------------------------------------------------

206211 -- Bind and monoidal composition combinations

207212 ---------------------------------------------------------------------------

208213

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

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

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

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

213−

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

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

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

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

218−

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

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

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

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

223−

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

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

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

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

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

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

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

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

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

219+

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

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

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

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

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

225+

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

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

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

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

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

231+

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

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

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

235+ describe "Bind and compose CoparAhead 4" $ 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

228243

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

230245 fldr = foldr (<>) nil

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

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

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

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

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

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

241258

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

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

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

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

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

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

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

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

250269

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

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

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

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

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

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

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

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

259280

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

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

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

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

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

267288 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

268302

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

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

271305 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

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

274309

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

276311 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

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

279315

280316 ---------------------------------------------------------------------------

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

296332 describe "Composed MonadThrow asCostream" $ composeWithMonadThrow asCostream

297333 describe "Composed MonadThrow asParAhead" $ composeWithMonadThrow asParAhead

298334 describe "Composed MonadThrow asCoparAhead" $ composeWithMonadThrow asCoparAhead

335+ describe "Composed MonadThrow asParallel" $ composeWithMonadThrow asParallel

299336

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

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

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

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

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

434471

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+

435483nestTwoParallel :: Expectation

436484nestTwoParallel =

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

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

439− in (toListParallel (do

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

440488 x <- s1

441489 y <- s2

442490 return (x + y)

443491 ) >>= return . sort)

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

445493

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+

446501nestTwoParallelApp :: Expectation

447502nestTwoParallelApp =

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

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

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

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

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

452507

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

test/Prop.hs+15

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

322322 functorOps folded "asParAhead folded" asParAhead sortEq

323323 functorOps A.fromFoldable "asCoparAhead" asCoparAhead sortEq

324324 functorOps folded "asCoparAhead folded" asCoparAhead sortEq

325+ functorOps A.fromFoldable "asParallel" asParallel sortEq

326+ functorOps folded "asParallel folded" asParallel sortEq

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

326328 functorOps folded "asZipStream folded" asZipStream (==)

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

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

332334 semigroupOps "asCostream" asCostream (==)

333335 semigroupOps "asParAhead" asParAhead sortEq

334336 semigroupOps "asCoparAhead" asCoparAhead sortEq

337+ semigroupOps "asParallel" asParallel sortEq

335338 semigroupOps "asZipStream" asZipStream (==)

336339 semigroupOps "asZipParallel" asZipParallel (==)

337340

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

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

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

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

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

349353

350354 describe "Zip operations" $ do

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

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

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

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

361365 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 (==)

362368

363369 describe "Monad operations" $ do

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

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

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

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

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

368375

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

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

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

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

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

373381

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

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

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

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

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

378387

379388 describe "Stream transform operations" $ do

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

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

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

384393 transformOps A.fromFoldable "asParAhead" asParAhead sortEq

385394 transformOps A.fromFoldable "asCoparAhead" asCoparAhead sortEq

395+ transformOps A.fromFoldable "asParallel" asParallel sortEq

386396

387397 transformOps folded "asStream folded" asStream (==)

388398 transformOps folded "asCostream folded" asCostream (==)

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

390400 transformOps folded "asZipParallel folded" asZipParallel (==)

391401 transformOps folded "asParAhead folded" asParAhead sortEq

392402 transformOps folded "asCoparAhead folded" asCoparAhead sortEq

403+ transformOps folded "asParallel folded" asParallel sortEq

393404

394405 transformOpsWord8 A.fromFoldable "asStream" asStream

395406 transformOpsWord8 A.fromFoldable "asCostream" asCostream

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

397408 transformOpsWord8 A.fromFoldable "asZipParallel" asZipParallel

398409 transformOpsWord8 A.fromFoldable "asParAhead" asParAhead

399410 transformOpsWord8 A.fromFoldable "asCoparAhead" asCoparAhead

411+ transformOpsWord8 A.fromFoldable "asParallel" asParallel

400412

401413 transformOpsWord8 folded "asStream folded" asStream

402414 transformOpsWord8 folded "asCostream folded" asCostream

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

404416 transformOpsWord8 folded "asZipParallel folded" asZipParallel

405417 transformOpsWord8 folded "asParAhead folded" asParAhead

406418 transformOpsWord8 folded "asCoparAhead folded" asCoparAhead

419+ transformOpsWord8 folded "asParallel folded" asParallel

407420

408421 describe "Stream elimination operations" $ do

409422 eliminationOps A.fromFoldable "asStream" asStream

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

412425 eliminationOps A.fromFoldable "asZipParallel" asZipParallel

413426 eliminationOps A.fromFoldable "asParAhead" asParAhead

414427 eliminationOps A.fromFoldable "asCoparAhead" asCoparAhead

428+ eliminationOps A.fromFoldable "asParallel" asParallel

415429

416430 eliminationOps folded "asStream folded" asStream

417431 eliminationOps folded "asCostream folded" asCostream

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

419433 eliminationOps folded "asZipParallel folded" asZipParallel

420434 eliminationOps folded "asParAhead folded" asParAhead

421435 eliminationOps folded "asCoparAhead folded" asCoparAhead

436+ eliminationOps folded "asParallel folded" asParallel

422437

423438 describe "Stream elimination operations" $ do

424439 serialEliminationOps A.fromFoldable "asStream" asStream