071a90a · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

rename some zip operations and MonadAsync type

071a90aHarendra Kumar9 years ago

Changelog.md+8−3

compare with master

unifiedsplit

@@ -31,19 +31,24 @@

32### Deprecations

33* Deprecate and rename the following symbols:

34+ * `Streaming` to `IsStream`

35+ * `runStreaming` to `runStream`

34 * `StreamT` to `SerialT`

36 * `StreamT` to `SerialT`

35− * `ZipStream` to `ZipSerial`

36 * `AsyncT` to `CoparallelT`

37 * `AsyncT` to `CoparallelT`

38+ * `ZipStream` to `ZipSerial`

39+ * `ZipAsync` to `ZipParallel`

37 * `interleaving` to `coserially`

40 * `interleaving` to `coserially`

38 * `asyncly` to `coparallely`

41 * `asyncly` to `coparallely`

39− * `Streaming` to `IsStream`

42+ * `zipping` to `zipSerially`

40− * `runStreaming` to `runStream`

43+ * `zippingAsync` to `zipParallely`

41 * `<=>` to `coserial`

44 * `<=>` to `coserial`

42 * `<|` to `coparallel`

45 * `<|` to `coparallel`

43 * `each` to `fromFoldable`

46 * `each` to `fromFoldable`

44 * `scan` to `scanx`

47 * `scan` to `scanx`

45 * `foldl` to `foldx`

48 * `foldl` to `foldx`

46 * `foldlM` to `foldxM`

49 * `foldlM` to `foldxM`

50+ * `zipAsyncWith` to `zipParallelWith`

51+ * `zipAsyncWithM` to `zipParallelWithM`

47* Deprecate and remove the following symbols:

52* Deprecate and remove the following symbols:

48 * `runStreamT`

53 * `runStreamT`

49 * `runInterleavedT`

54 * `runInterleavedT`

src/Streamly.hs+53−33

@@ -15,30 +15,23 @@ module Streamly

15 -- * Overview

16 -- $overview

18− MonadAsync

18+ MonadParallel

19 , IsStream

21− -- * General Stream Styles

21+ -- * Serial Streams

22− -- $product

22+ -- $serial

23 , SerialT

24 , CoserialT

25+

26+ -- * Parallel Streams

27+ -- $parallel

25 , CoparallelT

28 , CoparallelT

26 , ParallelT

29 , ParallelT

28− -- * Zip Style Streams

31+ -- * Zipping Streams

29 -- $zipping

32 -- $zipping

30 , ZipSerial

33 , ZipSerial

31− , ZipAsync

34+ , ZipParallel

32−

33− -- * Polymorphic Sum Operations

34− -- $sum

35− , serial

36− , coserial

37− , coparallel

38− , parallel

39−

40− -- * Transformation

41− , async

43 -- * Stream Type Adapters

36 -- * Stream Type Adapters

44 -- $adapters

37 -- $adapters

@@ -46,14 +39,24 @@ module Streamly

46 , coserially

39 , coserially

47 , coparallely

40 , coparallely

48 , parallely

41 , parallely

49− , zipping

42+ , zipSerially

50− , zippingAsync

43+ , zipParallely

51 , adapt

44 , adapt

53 -- * Running Streams

46 -- * Running Streams

54 , runStream

47 , runStream

49+ -- * Transformation

50+ , async

51+

52+ -- * Polymorphic Sum Operations

53+ -- $sum

54+ , serial

55+ , coserial

56+ , coparallel

57+ , parallel

58+

56− -- * Fold Utilities

59+ -- * Polymorphic Fold Utilities

57 -- $foldutils

60 -- $foldutils

58 , foldWith

61 , foldWith

59 , foldMapWith

62 , foldMapWith

@@ -68,6 +71,7 @@ module Streamly

68 , MonadTrans (..)

71 , MonadTrans (..)

70 -- * Deprecated

73 -- * Deprecated

74+ , MonadAsync

71 , Streaming

75 , Streaming

72 , runStreaming

76 , runStreaming

73 , runStreamT

77 , runStreamT

@@ -80,8 +84,11 @@ module Streamly

80 , InterleavedT

84 , InterleavedT

81 , AsyncT

85 , AsyncT

82 , ZipStream

86 , ZipStream

87+ , ZipAsync

83 , interleaving

88 , interleaving

84 , asyncly

89 , asyncly

90+ , zipping

91+ , zippingAsync

85 , (<=>)

92 , (<=>)

86 , (<|)

93 , (<|)

87 )

94 )

@@ -129,7 +136,7 @@ import Control.Monad.Trans.Class (MonadTrans (..))

129-- $overview

136-- $overview

130--

137--

131-- Streamly provides six distinct stream types i.e. 'SerialT', 'CoserialT',

138-- Streamly provides six distinct stream types i.e. 'SerialT', 'CoserialT',

132−-- 'CoparallelT' and 'ParallelT', 'ZipStream' and 'ZipAsync', each representing a

139+-- 'CoparallelT' and 'ParallelT', 'ZipSerial' and 'ZipParallel', each representing a

133-- stream of elements. All these types have the same underlying representation

140-- stream of elements. All these types have the same underlying representation

134-- and can be adapted from one to another using type adaptor combinators

141-- and can be adapted from one to another using type adaptor combinators

135-- described later. Each of these types belongs to the 'IsStream' type class

142-- described later. Each of these types belongs to the 'IsStream' type class

@@ -146,10 +153,10 @@ import Control.Monad.Trans.Class (MonadTrans (..))

146-- whereas 'CoparallelT' and 'ParallelT' are their concurrent counterparts. See the

153-- whereas 'CoparallelT' and 'ParallelT' are their concurrent counterparts. See the

147-- documentation of the respective types for more details.

154-- documentation of the respective types for more details.

148--

155--

149−-- The types 'ZipStream' and 'ZipAsync' provide 'Applicative' instances to zip

156+-- The types 'ZipSerial' and 'ZipParallel' provide 'Applicative' instances to zip

150-- two streams together i.e. each element in one stream is combined with the

157-- two streams together i.e. each element in one stream is combined with the

151−-- corresponding element in the other stream. 'ZipStream' generates the streams

158+-- corresponding element in the other stream. 'ZipSerial' generates the streams

152−-- being zipped serially whereas 'ZipAsync' produces both the elements being

159+-- being zipped serially whereas 'ZipParallel' produces both the elements being

153-- zipped concurrently.

160-- zipped concurrently.

154--

161--

155-- Two streams of the same type can be merged using a sum style composition to

162-- Two streams of the same type can be merged using a sum style composition to

@@ -158,25 +165,37 @@ import Control.Monad.Trans.Class (MonadTrans (..))

158-- (concurrent or serial) in which the elements in the resulting stream are

165-- (concurrent or serial) in which the elements in the resulting stream are

159-- produced depends on the type of the stream.

166-- produced depends on the type of the stream.

160--

167--

161−-- Concurrent composition types 'CoparallelT', 'ParallelT', 'ZipAsync' require the

162−-- underlying monad of the streaming monad transformer to be 'MonadAsync'.

163−--

164-- For more details please see the "Streamly.Tutorial" and "Streamly.Examples"

168-- For more details please see the "Streamly.Tutorial" and "Streamly.Examples"

165-- (the latter is available only when built with the 'examples' build flag).

169-- (the latter is available only when built with the 'examples' build flag).

166

170

167-- A simple inline example here illustrating applicative, monad and alternative

171-- A simple inline example here illustrating applicative, monad and alternative

168-- compositions.

172-- compositions.

169

173

170−-- $product

174+-- $serial

175+--

176+-- Serial streams compose serially or non-concurrently. In a composed stream,

177+-- at any point of time only one constituent stream runs and yields an element.

178+-- The two serial stream types 'SerialT' and 'CoserialT' differ in how they

179+-- merge streams together in a 'Semigroup' or 'Monad' composition. As these

180+-- streams are serial, the sequence of items in a composed stream can be solely

181+-- determined by the position of elements in the consituent streams.

182+

183+-- $parallel

171--

184--

172−-- Streams that compose serially or non-concurrently come in two flavors i.e.

185+-- Parallel streams compose parallely or concurrently. In a composed stream, at

173−-- 'SerialT' and 'CoserialT'. Both of these serial flavors have

186+-- any point of time more than one stream can run concurrently and yield

174−-- corresponding concurrent equivalents, those are 'CoparallelT' and 'ParallelT'

187+-- elements. The two parallel types 'ParallelT' and 'CoparallelT' differ in how

175−-- respectively.

188+-- they merge streams together in a 'Semigroup' or 'Monad' composition. As

189+-- these streams compose concurrently, the sequence of items in a composed

190+-- stream cannot be determined by the position of elements in the constituent

191+-- streams. The elements are yielded by the composed stream as they are

192+-- generated by the constituent streams on a first come first serve basis.

193+-- Therefore, on each run the stream may yield elements in a different sequence

194+-- depending on the delays introduced by scheduling.

176

195

177-- $zipping

196-- $zipping

178--

197--

179−-- 'ZipStream' and 'ZipAsync', provide 'Applicative' instances for zipping the

198+-- 'ZipSerial' and 'ZipParallel', provide 'Applicative' instances for zipping the

180-- corresponding elements of two streams together. Note that these types are

199-- corresponding elements of two streams together. Note that these types are

181-- not monads.

200-- not monads.

182

201

@@ -204,5 +223,6 @@ import Control.Monad.Trans.Class (MonadTrans (..))

204

223

205-- $foldutils

224-- $foldutils

206--

225--

207−-- These are some convenience functions to fold any 'Foldable' container using

226+-- These are variants of standard 'Foldable' fold functions that use a

208−-- one of the sum composition operators to convert it into a streamly stream.

227+-- polymorphic stream sum operation (e.g. 'parallel' or 'coserial') to fold a

228+-- container of streams.

src/Streamly/Core.hs+21−17

@@ -18,7 +18,8 @@

18--

19module Streamly.Core

20 (

21+ MonadParallel

21− MonadAsync

22+ , MonadAsync -- deprecated

23 -- * Streams

24 -- * Streams

24 , Stream (..)

25 , Stream (..)

@@ -213,10 +214,13 @@ newtype Stream m a =

213 -> m r

214 -> m r

214 }

215 }

215

216

216−-- | A monad that can perform asynchronous/concurrent IO operations. Streams

217+-- | A monad that can perform concurrent or parallel IO operations. Streams

217-- that can be composed concurrently require the underlying monad to be

218-- that can be composed concurrently require the underlying monad to be

218−-- 'MonadAsync'.

219+-- 'MonadParallel'.

219−type MonadAsync m = (MonadIO m, MonadBaseControl IO m, MonadThrow m)

220+type MonadParallel m = (MonadIO m, MonadBaseControl IO m, MonadThrow m)

221+

222+{-# DEPRECATED MonadAsync "Please use MonadParallel instead." #-}

223+type MonadAsync m = MonadParallel m

220

224

221cons :: a -> Maybe (Stream m a) -> Stream m a

225cons :: a -> Maybe (Stream m a) -> Stream m a

222cons a r = Stream $ \_ _ yld -> yld a r

226cons a r = Stream $ \_ _ yld -> yld a r

@@ -377,7 +381,7 @@ handleChildException sv e = do

377 send sv (ChildStop tid (Just e))

381 send sv (ChildStop tid (Just e))

378

382

379{-# NOINLINE pushWorker #-}

383{-# NOINLINE pushWorker #-}

380−pushWorker :: MonadAsync m => SVar m a -> m ()

384+pushWorker :: MonadParallel m => SVar m a -> m ()

381pushWorker sv =

385pushWorker sv =

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

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

383

387

@@ -386,7 +390,7 @@ pushWorker sv =

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

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

387-- the type of the queue.

391-- the type of the queue.

388{-# INLINE sendWorkerWait #-}

392{-# INLINE sendWorkerWait #-}

389−sendWorkerWait :: MonadAsync m => SVar m a -> m ()

393+sendWorkerWait :: MonadParallel m => SVar m a -> m ()

390sendWorkerWait sv = do

394sendWorkerWait sv = do

391 case svarStyle sv of

395 case svarStyle sv of

392 SVarStyle _ LIFO -> liftIO $ threadDelay 200

396 SVarStyle _ LIFO -> liftIO $ threadDelay 200

@@ -401,7 +405,7 @@ sendWorkerWait sv = do

401

405

402-- | Pull a stream from an SVar.

406-- | Pull a stream from an SVar.

403{-# NOINLINE fromStreamVar #-}

407{-# NOINLINE fromStreamVar #-}

404−fromStreamVar :: MonadAsync m => SVar m a -> Stream m a

408+fromStreamVar :: MonadParallel m => SVar m a -> Stream m a

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

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

406 -- XXX if reading the IORef is costly we can use a flag in the SVar to

410 -- XXX if reading the IORef is costly we can use a flag in the SVar to

407 -- indicate we are done.

411 -- indicate we are done.

@@ -479,7 +483,7 @@ getLifoSVar ctype = do

479 in return sv

483 in return sv

480

484

481-- | Create a new empty SVar.

485-- | Create a new empty SVar.

482−newEmptySVar :: MonadAsync m => SVarStyle -> m (SVar m a)

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

483newEmptySVar style = do

487newEmptySVar style = do

484 liftIO $

488 liftIO $

485 case style of

489 case style of

@@ -487,7 +491,7 @@ newEmptySVar style = do

487 SVarStyle _ LIFO -> getLifoSVar style

491 SVarStyle _ LIFO -> getLifoSVar style

488

492

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

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

490−newStreamVar1 :: MonadAsync m => SVarStyle -> Stream m a -> m (SVar m a)

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

491newStreamVar1 style m = do

495newStreamVar1 style m = do

492 sv <- newEmptySVar style

496 sv <- newEmptySVar style

493 -- Note: We must have all the work on the queue before sending the

497 -- Note: We must have all the work on the queue before sending the

@@ -498,7 +502,7 @@ newStreamVar1 style m = do

498 return sv

502 return sv

499

503

500-- | Create a new SVar and enqueue two stream computations on it.

504-- | Create a new SVar and enqueue two stream computations on it.

501−newStreamVar2 :: MonadAsync m

505+newStreamVar2 :: MonadParallel m

502 => SVarStyle -> Stream m a -> Stream m a -> m (SVar m a)

506 => SVarStyle -> Stream m a -> Stream m a -> m (SVar m a)

503newStreamVar2 style m1 m2 = do

507newStreamVar2 style m1 m2 = do

504 -- Note: We must have all the work on the queue before sending the

508 -- Note: We must have all the work on the queue before sending the

@@ -519,7 +523,7 @@ newStreamVar2 style m1 m2 = do

519

523

520-- | Write a stream to an 'SVar' in a non-blocking manner. The stream can then

524-- | Write a stream to an 'SVar' in a non-blocking manner. The stream can then

521-- be read back from the SVar using 'fromSVar'.

525-- be read back from the SVar using 'fromSVar'.

522−toStreamVar :: MonadAsync m => SVar m a -> Stream m a -> m ()

526+toStreamVar :: MonadParallel m => SVar m a -> Stream m a -> m ()

523toStreamVar sv m = do

527toStreamVar sv m = do

524 liftIO $ (enqueue sv) m

528 liftIO $ (enqueue sv) m

525 done <- allThreadsDone sv

529 done <- allThreadsDone sv

@@ -561,7 +565,7 @@ toStreamVar sv m = do

561-- just the number of CPUs.

565-- just the number of CPUs.

562

566

563{-# NOINLINE withNewSVar2 #-}

567{-# NOINLINE withNewSVar2 #-}

564−withNewSVar2 :: MonadAsync m

568+withNewSVar2 :: MonadParallel m

565 => SVarStyle -> Stream m a -> Stream m a -> Stream m a

569 => SVarStyle -> Stream m a -> Stream m a -> Stream m a

566withNewSVar2 style m1 m2 = Stream $ \_ stp yld -> do

570withNewSVar2 style m1 m2 = Stream $ \_ stp yld -> do

567 sv <- newStreamVar2 style m1 m2

571 sv <- newStreamVar2 style m1 m2

@@ -596,7 +600,7 @@ withNewSVar2 style m1 m2 = Stream $ \_ stp yld -> do

596-- of the two.

600-- of the two.

597--

601--

598{-# INLINE joinStreamVar2 #-}

602{-# INLINE joinStreamVar2 #-}

599−joinStreamVar2 :: MonadAsync m

603+joinStreamVar2 :: MonadParallel m

600 => SVarStyle -> Stream m a -> Stream m a -> Stream m a

604 => SVarStyle -> Stream m a -> Stream m a -> Stream m a

601joinStreamVar2 style m1 m2 = Stream $ \st stp yld ->

605joinStreamVar2 style m1 m2 = Stream $ \st stp yld ->

602 case st of

606 case st of

@@ -609,11 +613,11 @@ joinStreamVar2 style m1 m2 = Stream $ \st stp yld ->

609------------------------------------------------------------------------------

613------------------------------------------------------------------------------

610

614

611{-# INLINE coparallel #-}

615{-# INLINE coparallel #-}

612−coparallel :: MonadAsync m => Stream m a -> Stream m a -> Stream m a

616+coparallel :: MonadParallel m => Stream m a -> Stream m a -> Stream m a

613coparallel = joinStreamVar2 (SVarStyle Disjunction LIFO)

617coparallel = joinStreamVar2 (SVarStyle Disjunction LIFO)

614

618

615{-# INLINE parallel #-}

619{-# INLINE parallel #-}

616−parallel :: MonadAsync m => Stream m a -> Stream m a -> Stream m a

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

617parallel = joinStreamVar2 (SVarStyle Disjunction FIFO)

621parallel = joinStreamVar2 (SVarStyle Disjunction FIFO)

618

622

619-------------------------------------------------------------------------------

623-------------------------------------------------------------------------------

@@ -647,11 +651,11 @@ alt m1 m2 = Stream $ \_ stp yld ->

647 yield = yld

651 yield = yld

648 in runStream m1 Nothing stop yield

652 in runStream m1 Nothing stop yield

649

653

650−instance MonadAsync m => Alternative (Stream m) where

654+instance MonadParallel m => Alternative (Stream m) where

651 empty = nil

655 empty = nil

652 (<|>) = alt

656 (<|>) = alt

653

657

654−instance MonadAsync m => MonadPlus (Stream m) where

658+instance MonadParallel m => MonadPlus (Stream m) where

655 mzero = nil

659 mzero = nil

656 mplus = (<|>)

660 mplus = (<|>)

657

661

src/Streamly/Examples/AcidRainGame.hs+1−1

@@ -26,7 +26,7 @@ userAction = cycle1 $ liftIO askUser

26acidRain :: MonadIO m => SerialT m Event

27acidRain = cycle1 $ liftIO (threadDelay 1000000) >> return (Harm 1)

29−game :: (MonadAsync m, MonadState Int m) => SerialT m ()

29+game :: (MonadParallel m, MonadState Int m) => SerialT m ()

30game = do

31 event <- userAction `parallel` acidRain

32 case event of

src/Streamly/Examples/MergeSortedStreams.hs+2−2

@@ -14,14 +14,14 @@ getSorted = do

14 let ls = take 100000 (randoms g) :: [Word16]

15 foldMapWith (<>) return (sort ls)

17−mergeAsync :: (Ord a, MonadAsync m)

17+mergeAsync :: (Ord a, MonadParallel m)

18 => SerialT m a -> SerialT m a -> SerialT m a

19mergeAsync a b = do

20 x <- lift $ async a

21 y <- lift $ async b

22 merge x y

24−merge :: (Ord a, MonadAsync m) => SerialT m a -> SerialT m a -> SerialT m a

24+merge :: (Ord a, MonadParallel m) => SerialT m a -> SerialT m a -> SerialT m a

25merge a b = do

26 a1 <- lift $ A.uncons a

27 case a1 of

src/Streamly/Examples/SearchEngineQuery.hs+1−1

@@ -10,7 +10,7 @@ searchEngineQuery = do

10 runStream . parallely $ google <> bing <> duckduckgo

12 putStrLn "\nUsing parallel applicative zip"

13− runStream . zippingAsync $ (,,) <$> google <*> bing <*> duckduckgo

13+ runStream . zipParallely $ (,,) <$> google <*> bing <*> duckduckgo

15 where

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

src/Streamly/Prelude.hs+11−4

@@ -72,8 +72,8 @@ module Streamly.Prelude

72 -- * Zipping

73 , zipWith

74 , zipWithM

75− , zipAsyncWith

75+ , zipParallelWith

76− , zipAsyncWithM

76+ , zipParallelWithM

78 -- * IO

79 , fromHandle

@@ -84,6 +84,8 @@ module Streamly.Prelude

84 , scan

85 , foldl

86 , foldlM

87+ , zipAsyncWith

88+ , zipAsyncWithM

87 )

89 )

88where

90where

@@ -569,9 +571,14 @@ zipWithM f m1 m2 = fromStream $ go (toStream m1) (toStream m2)

569

571

570-- | Zip two streams asyncly (i.e. both the elements being zipped are generated

572-- | Zip two streams asyncly (i.e. both the elements being zipped are generated

571-- concurrently) using a monadic zipping function.

573-- concurrently) using a monadic zipping function.

572−zipAsyncWithM :: (IsStream t, MonadAsync m)

574+zipParallelWithM :: (IsStream t, MonadParallel m)

573 => (a -> b -> t m c) -> t m a -> t m b -> t m c

575 => (a -> b -> t m c) -> t m a -> t m b -> t m c

574−zipAsyncWithM f m1 m2 = fromStream $ Stream $ \_ stp yld -> do

576+zipParallelWithM f m1 m2 = fromStream $ Stream $ \_ stp yld -> do

575 ma <- async m1

577 ma <- async m1

576 mb <- async m2

578 mb <- async m2

577 (S.runStream (toStream (zipWithM f ma mb))) Nothing stp yld

579 (S.runStream (toStream (zipWithM f ma mb))) Nothing stp yld

580+

581+{-# DEPRECATED zipAsyncWithM "Please use zipParallelWithM instead." #-}

582+zipAsyncWithM :: (IsStream t, MonadParallel m)

583+ => (a -> b -> t m c) -> t m a -> t m b -> t m c

584+zipAsyncWithM = zipParallelWithM

src/Streamly/Streams.hs+110−92

@@ -21,7 +21,8 @@ module Streamly.Streams

21 (

22 IsStream (..)

23 , Streaming -- deprecated

24+ , MonadParallel

24− , MonadAsync

25+ , S.MonadAsync -- deprecated

26 -- * SVars

27 -- * SVars

27 , SVarSched (..)

28 , SVarSched (..)

@@ -65,7 +66,8 @@ module Streamly.Streams

65 , ParallelT

66 , ParallelT

66 , ZipSerial

67 , ZipSerial

67 , ZipStream -- deprecated

68 , ZipStream -- deprecated

69+ , ZipParallel

68− , ZipAsync

70+ , ZipAsync -- deprecated

70 -- * Type Adapters

72 -- * Type Adapters

71 , serially

73 , serially

@@ -74,8 +76,10 @@ module Streamly.Streams

74 , coparallely

76 , coparallely

75 , asyncly -- deprecated

77 , asyncly -- deprecated

76 , parallely

78 , parallely

79+ , zipSerially

77− , zipping

80+ , zipping -- deprecated

81+ , zipParallely

78− , zippingAsync

82+ , zippingAsync -- deprecated

79 , adapt

83 , adapt

81 -- * Running Streams

85 -- * Running Streams

@@ -88,10 +92,10 @@ module Streamly.Streams

89 -- * Zipping

93 -- * Zipping

90 , zipWith

94 , zipWith

95+ , zipParallelWith

91− , zipAsyncWith

96+ , zipAsyncWith -- deprecated

93 -- * Fold Utilities

98 -- * Fold Utilities

94− -- $foldutils

95 , foldWith

99 , foldWith

96 , foldMapWith

100 , foldMapWith

97 , forEachWith

101 , forEachWith

@@ -109,7 +113,7 @@ import Control.Monad.State.Class (MonadState(..))

109import Control.Monad.Trans.Class (MonadTrans)

113import Control.Monad.Trans.Class (MonadTrans)

110import Data.Semigroup (Semigroup(..))

114import Data.Semigroup (Semigroup(..))

111import Prelude hiding (zipWith)

115import Prelude hiding (zipWith)

112−import Streamly.Core ( MonadAsync, Stream(Stream)

116+import Streamly.Core ( MonadParallel, Stream(Stream)

113 , SVar, SVarStyle(..)

117 , SVar, SVarStyle(..)

114 , SVarTag(..), SVarSched(..))

118 , SVarTag(..), SVarSched(..))

115import qualified Streamly.Core as S

119import qualified Streamly.Core as S

@@ -201,7 +205,7 @@ fromCallback :: IsStream t => (forall r. (a -> m r) -> m r) -> t m a

201fromCallback k = fromStream $ Stream $ \_ _ yld -> k (\a -> yld a Nothing)

205fromCallback k = fromStream $ Stream $ \_ _ yld -> k (\a -> yld a Nothing)

202

206

203-- | Read an SVar to get a stream.

207-- | Read an SVar to get a stream.

204−fromSVar :: (MonadAsync m, IsStream t) => SVar m a -> t m a

208+fromSVar :: (MonadParallel m, IsStream t) => SVar m a -> t m a

205fromSVar sv = fromStream $ S.fromStreamVar sv

209fromSVar sv = fromStream $ S.fromStreamVar sv

206

210

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

211------------------------------------------------------------------------------

@@ -218,7 +222,9 @@ streamFold sv step blank m =

218 yield a (Just x) = step a (Just (fromStream x))

222 yield a (Just x) = step a (Just (fromStream x))

219 in (S.runStream (toStream m)) sv blank yield

223 in (S.runStream (toStream m)) sv blank yield

220

224

221−-- | Run a streaming composition, discard the results.

225+-- | Run a streaming composition, discard the results. By default it interprets

226+-- the stream as 'SerialT', to run other types of streams use the type adapting

227+-- combinators for example @runStream . 'parallely'@.

222runStream :: Monad m => SerialT m a -> m ()

228runStream :: Monad m => SerialT m a -> m ()

223runStream m = go (toStream m)

229runStream m = go (toStream m)

224 where

230 where

@@ -235,7 +241,7 @@ runStreaming = runStream . adapt

235

241

236-- | Write a stream to an 'SVar' in a non-blocking manner. The stream can then

242-- | Write a stream to an 'SVar' in a non-blocking manner. The stream can then

237-- be read back from the SVar using 'fromSVar'.

243-- be read back from the SVar using 'fromSVar'.

238−toSVar :: (IsStream t, MonadAsync m) => SVar m a -> t m a -> m ()

244+toSVar :: (IsStream t, MonadParallel m) => SVar m a -> t m a -> m ()

239toSVar sv m = S.toStreamVar sv (toStream m)

245toSVar sv m = S.toStreamVar sv (toStream m)

240

246

241------------------------------------------------------------------------------

247------------------------------------------------------------------------------

@@ -249,7 +255,7 @@ toSVar sv m = S.toStreamVar sv (toStream m)

249-- not drained fully we may have a thread blocked forever and once exhausted it

255-- not drained fully we may have a thread blocked forever and once exhausted it

250-- will always return 'empty'.

256-- will always return 'empty'.

251

257

252−async :: (IsStream t, MonadAsync m) => t m a -> m (t m a)

258+async :: (IsStream t, MonadParallel m) => t m a -> m (t m a)

253async m = do

259async m = do

254 sv <- S.newStreamVar1 (SVarStyle Disjunction LIFO) (toStream m)

260 sv <- S.newStreamVar1 (SVarStyle Disjunction LIFO) (toStream m)

255 return $ fromSVar sv

261 return $ fromSVar sv

@@ -311,8 +317,8 @@ async m = do

311newtype SerialT m a = SerialT {getSerialT :: Stream m a}

317newtype SerialT m a = SerialT {getSerialT :: Stream m a}

312 deriving (Semigroup, Monoid, MonadTrans, MonadIO, MonadThrow)

318 deriving (Semigroup, Monoid, MonadTrans, MonadIO, MonadThrow)

313

319

314−deriving instance MonadAsync m => Alternative (SerialT m)

320+deriving instance MonadParallel m => Alternative (SerialT m)

315−deriving instance MonadAsync m => MonadPlus (SerialT m)

321+deriving instance MonadParallel m => MonadPlus (SerialT m)

316deriving instance (MonadBase b m, Monad m) => MonadBase b (SerialT m)

322deriving instance (MonadBase b m, Monad m) => MonadBase b (SerialT m)

317deriving instance MonadError e m => MonadError e (SerialT m)

323deriving instance MonadError e m => MonadError e (SerialT m)

318deriving instance MonadReader r m => MonadReader r (SerialT m)

324deriving instance MonadReader r m => MonadReader r (SerialT m)

@@ -454,8 +460,8 @@ instance (Monad m, Floating a) => Floating (SerialT m a) where

454newtype CoserialT m a = CoserialT {getCoserialT :: Stream m a}

460newtype CoserialT m a = CoserialT {getCoserialT :: Stream m a}

455 deriving (Monoid, MonadTrans, MonadIO, MonadThrow)

461 deriving (Monoid, MonadTrans, MonadIO, MonadThrow)

456

462

457−deriving instance MonadAsync m => Alternative (CoserialT m)

463+deriving instance MonadParallel m => Alternative (CoserialT m)

458−deriving instance MonadAsync m => MonadPlus (CoserialT m)

464+deriving instance MonadParallel m => MonadPlus (CoserialT m)

459deriving instance (MonadBase b m, Monad m) => MonadBase b (CoserialT m)

465deriving instance (MonadBase b m, Monad m) => MonadBase b (CoserialT m)

460deriving instance MonadError e m => MonadError e (CoserialT m)

466deriving instance MonadError e m => MonadError e (CoserialT m)

461deriving instance MonadReader r m => MonadReader r (CoserialT m)

467deriving instance MonadReader r m => MonadReader r (CoserialT m)

@@ -610,15 +616,15 @@ instance (Monad m, Floating a) => Floating (CoserialT m a) where

610newtype CoparallelT m a = CoparallelT {getCoparallelT :: Stream m a}

616newtype CoparallelT m a = CoparallelT {getCoparallelT :: Stream m a}

611 deriving (MonadTrans)

617 deriving (MonadTrans)

612

618

613−deriving instance MonadAsync m => Monoid (CoparallelT m a)

619+deriving instance MonadParallel m => Monoid (CoparallelT m a)

614−deriving instance MonadAsync m => Alternative (CoparallelT m)

620+deriving instance MonadParallel m => Alternative (CoparallelT m)

615−deriving instance MonadAsync m => MonadPlus (CoparallelT m)

621+deriving instance MonadParallel m => MonadPlus (CoparallelT m)

616−deriving instance MonadAsync m => MonadIO (CoparallelT m)

622+deriving instance MonadParallel m => MonadIO (CoparallelT m)

617−deriving instance MonadAsync m => MonadThrow (CoparallelT m)

623+deriving instance MonadParallel m => MonadThrow (CoparallelT m)

618−deriving instance (MonadBase b m, MonadAsync m) => MonadBase b (CoparallelT m)

624+deriving instance (MonadBase b m, MonadParallel m) => MonadBase b (CoparallelT m)

619−deriving instance (MonadError e m, MonadAsync m) => MonadError e (CoparallelT m)

625+deriving instance (MonadError e m, MonadParallel m) => MonadError e (CoparallelT m)

620−deriving instance (MonadReader r m, MonadAsync m) => MonadReader r (CoparallelT m)

626+deriving instance (MonadReader r m, MonadParallel m) => MonadReader r (CoparallelT m)

621−deriving instance (MonadState s m, MonadAsync m) => MonadState s (CoparallelT m)

627+deriving instance (MonadState s m, MonadParallel m) => MonadState s (CoparallelT m)

622

628

623{-# DEPRECATED AsyncT "Please use 'CoparallelT' instead." #-}

629{-# DEPRECATED AsyncT "Please use 'CoparallelT' instead." #-}

624type AsyncT = CoparallelT

630type AsyncT = CoparallelT

@@ -635,16 +641,16 @@ instance IsStream CoparallelT where

635-- polymorphic. Merges two streams possibly concurrently, preferring the

641-- polymorphic. Merges two streams possibly concurrently, preferring the

636-- elements from the left one when available.

642-- elements from the left one when available.

637{-# INLINE coparallel #-}

643{-# INLINE coparallel #-}

638−coparallel :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a

644+coparallel :: (IsStream t, MonadParallel m) => t m a -> t m a -> t m a

639coparallel m1 m2 = fromStream $ S.coparallel (toStream m1) (toStream m2)

645coparallel m1 m2 = fromStream $ S.coparallel (toStream m1) (toStream m2)

640

646

641−instance MonadAsync m => Semigroup (CoparallelT m a) where

647+instance MonadParallel m => Semigroup (CoparallelT m a) where

642 (<>) = coparallel

648 (<>) = coparallel

643

649

644-- | Same as 'coparallel'.

650-- | Same as 'coparallel'.

645{-# DEPRECATED (<|) "Please use 'coparallel' instead." #-}

651{-# DEPRECATED (<|) "Please use 'coparallel' instead." #-}

646{-# INLINE (<|) #-}

652{-# INLINE (<|) #-}

647−(<|) :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a

653+(<|) :: (IsStream t, MonadParallel m) => t m a -> t m a -> t m a

648(<|) = coparallel

654(<|) = coparallel

649

655

650------------------------------------------------------------------------------

656------------------------------------------------------------------------------

@@ -666,7 +672,7 @@ parbind par m f = go m

666 yield a (Just r) = run $ f a `par` go r

672 yield a (Just r) = run $ f a `par` go r

667 in g Nothing stp yield

673 in g Nothing stp yield

668

674

669−instance MonadAsync m => Monad (CoparallelT m) where

675+instance MonadParallel m => Monad (CoparallelT m) where

670 return = pure

676 return = pure

671 (CoparallelT m) >>= f = CoparallelT $ parbind par m g

677 (CoparallelT m) >>= f = CoparallelT $ parbind par m g

672 where g x = getCoparallelT (f x)

678 where g x = getCoparallelT (f x)

@@ -676,7 +682,7 @@ instance MonadAsync m => Monad (CoparallelT m) where

676-- Applicative

682-- Applicative

677------------------------------------------------------------------------------

683------------------------------------------------------------------------------

678

684

679−instance MonadAsync m => Applicative (CoparallelT m) where

685+instance MonadParallel m => Applicative (CoparallelT m) where

680 pure a = CoparallelT $ S.cons a Nothing

686 pure a = CoparallelT $ S.cons a Nothing

681 (<*>) = ap

687 (<*>) = ap

682

688

@@ -694,7 +700,7 @@ instance Monad m => Functor (CoparallelT m) where

694-- Num

700-- Num

695------------------------------------------------------------------------------

701------------------------------------------------------------------------------

696

702

697−instance (MonadAsync m, Num a) => Num (CoparallelT m a) where

703+instance (MonadParallel m, Num a) => Num (CoparallelT m a) where

698 fromInteger n = pure (fromInteger n)

704 fromInteger n = pure (fromInteger n)

699

705

700 negate = fmap negate

706 negate = fmap negate

@@ -705,14 +711,14 @@ instance (MonadAsync m, Num a) => Num (CoparallelT m a) where

705 (*) = liftA2 (*)

711 (*) = liftA2 (*)

706 (-) = liftA2 (-)

712 (-) = liftA2 (-)

707

713

708−instance (MonadAsync m, Fractional a) => Fractional (CoparallelT m a) where

714+instance (MonadParallel m, Fractional a) => Fractional (CoparallelT m a) where

709 fromRational n = pure (fromRational n)

715 fromRational n = pure (fromRational n)

710

716

711 recip = fmap recip

717 recip = fmap recip

712

718

713 (/) = liftA2 (/)

719 (/) = liftA2 (/)

714

720

715−instance (MonadAsync m, Floating a) => Floating (CoparallelT m a) where

721+instance (MonadParallel m, Floating a) => Floating (CoparallelT m a) where

716 pi = pure pi

722 pi = pure pi

717

723

718 exp = fmap exp

724 exp = fmap exp

@@ -778,15 +784,15 @@ instance (MonadAsync m, Floating a) => Floating (CoparallelT m a) where

778newtype ParallelT m a = ParallelT {getParallelT :: Stream m a}

784newtype ParallelT m a = ParallelT {getParallelT :: Stream m a}

779 deriving (MonadTrans)

785 deriving (MonadTrans)

780

786

781−deriving instance MonadAsync m => Monoid (ParallelT m a)

787+deriving instance MonadParallel m => Monoid (ParallelT m a)

782−deriving instance MonadAsync m => Alternative (ParallelT m)

788+deriving instance MonadParallel m => Alternative (ParallelT m)

783−deriving instance MonadAsync m => MonadPlus (ParallelT m)

789+deriving instance MonadParallel m => MonadPlus (ParallelT m)

784−deriving instance MonadAsync m => MonadIO (ParallelT m)

790+deriving instance MonadParallel m => MonadIO (ParallelT m)

785−deriving instance MonadAsync m => MonadThrow (ParallelT m)

791+deriving instance MonadParallel m => MonadThrow (ParallelT m)

786−deriving instance (MonadBase b m, MonadAsync m) => MonadBase b (ParallelT m)

792+deriving instance (MonadBase b m, MonadParallel m) => MonadBase b (ParallelT m)

787−deriving instance (MonadError e m, MonadAsync m) => MonadError e (ParallelT m)

793+deriving instance (MonadError e m, MonadParallel m) => MonadError e (ParallelT m)

788−deriving instance (MonadReader r m, MonadAsync m) => MonadReader r (ParallelT m)

794+deriving instance (MonadReader r m, MonadParallel m) => MonadReader r (ParallelT m)

789−deriving instance (MonadState s m, MonadAsync m) => MonadState s (ParallelT m)

795+deriving instance (MonadState s m, MonadParallel m) => MonadState s (ParallelT m)

790

796

791instance IsStream ParallelT where

797instance IsStream ParallelT where

792 toStream = getParallelT

798 toStream = getParallelT

@@ -799,17 +805,17 @@ instance IsStream ParallelT where

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

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

800-- Merges two streams concurrently choosing elements from both fairly.

806-- Merges two streams concurrently choosing elements from both fairly.

801{-# INLINE parallel #-}

807{-# INLINE parallel #-}

802−parallel :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a

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

803parallel m1 m2 = fromStream $ S.parallel (toStream m1) (toStream m2)

809parallel m1 m2 = fromStream $ S.parallel (toStream m1) (toStream m2)

804

810

805−instance MonadAsync m => Semigroup (ParallelT m a) where

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

806 (<>) = parallel

812 (<>) = parallel

807

813

808------------------------------------------------------------------------------

814------------------------------------------------------------------------------

809-- Monad

815-- Monad

810------------------------------------------------------------------------------

816------------------------------------------------------------------------------

811

817

812−instance MonadAsync m => Monad (ParallelT m) where

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

813 return = pure

819 return = pure

814 (ParallelT m) >>= f = ParallelT $ parbind par m g

820 (ParallelT m) >>= f = ParallelT $ parbind par m g

815 where g x = getParallelT (f x)

821 where g x = getParallelT (f x)

@@ -819,7 +825,7 @@ instance MonadAsync m => Monad (ParallelT m) where

819-- Applicative

825-- Applicative

820------------------------------------------------------------------------------

826------------------------------------------------------------------------------

821

827

822−instance MonadAsync m => Applicative (ParallelT m) where

828+instance MonadParallel m => Applicative (ParallelT m) where

823 pure a = ParallelT $ S.cons a Nothing

829 pure a = ParallelT $ S.cons a Nothing

824 (<*>) = ap

830 (<*>) = ap

825

831

@@ -838,7 +844,7 @@ instance Monad m => Functor (ParallelT m) where

838-- Num

844-- Num

839------------------------------------------------------------------------------

845------------------------------------------------------------------------------

840

846

841−instance (MonadAsync m, Num a) => Num (ParallelT m a) where

847+instance (MonadParallel m, Num a) => Num (ParallelT m a) where

842 fromInteger n = pure (fromInteger n)

848 fromInteger n = pure (fromInteger n)

843

849

844 negate = fmap negate

850 negate = fmap negate

@@ -849,14 +855,14 @@ instance (MonadAsync m, Num a) => Num (ParallelT m a) where

849 (*) = liftA2 (*)

855 (*) = liftA2 (*)

850 (-) = liftA2 (-)

856 (-) = liftA2 (-)

851

857

852−instance (MonadAsync m, Fractional a) => Fractional (ParallelT m a) where

858+instance (MonadParallel m, Fractional a) => Fractional (ParallelT m a) where

853 fromRational n = pure (fromRational n)

859 fromRational n = pure (fromRational n)

854

860

855 recip = fmap recip

861 recip = fmap recip

856

862

857 (/) = liftA2 (/)

863 (/) = liftA2 (/)

858

864

859−instance (MonadAsync m, Floating a) => Floating (ParallelT m a) where

865+instance (MonadParallel m, Floating a) => Floating (ParallelT m a) where

860 pi = pure pi

866 pi = pure pi

861

867

862 exp = fmap exp

868 exp = fmap exp

@@ -897,11 +903,10 @@ zipWith f m1 m2 = fromStream $ go (toStream m1) (toStream m2)

897

903

898-- | The applicative instance of 'ZipSerial' zips a number of streams serially

904-- | The applicative instance of 'ZipSerial' zips a number of streams serially

899-- i.e. it produces one element from each stream serially and then zips all

905-- i.e. it produces one element from each stream serially and then zips all

900−-- those elements. Note, for convenience we have used the 'zipping' combinator

906+-- those elements.

901−-- in the following example instead of using a 'ZipSerial' type annotation.

902--

907--

903-- @

908-- @

904−-- main = (toList . 'zipping' $ (,,) \<$\> s1 \<*\> s2 \<*\> s3) >>= print

909+-- main = (toList . 'zipSerially' $ (,,) \<$\> s1 \<*\> s2 \<*\> s3) >>= print

905-- where s1 = fromFoldable [1, 2]

910-- where s1 = fromFoldable [1, 2]

906-- s2 = fromFoldable [3, 4]

911-- s2 = fromFoldable [3, 4]

907-- s3 = fromFoldable [5, 6]

912-- s3 = fromFoldable [5, 6]

@@ -920,7 +925,7 @@ newtype ZipSerial m a = ZipSerial {getZipSerial :: Stream m a}

920-- | Same as ZipSerial.

925-- | Same as ZipSerial.

921type ZipStream = ZipSerial

926type ZipStream = ZipSerial

922

927

923−deriving instance MonadAsync m => Alternative (ZipSerial m)

928+deriving instance MonadParallel m => Alternative (ZipSerial m)

924

929

925instance Monad m => Functor (ZipSerial m) where

930instance Monad m => Functor (ZipSerial m) where

926 fmap f (ZipSerial (Stream m)) = ZipSerial $ Stream $ \_ stp yld ->

931 fmap f (ZipSerial (Stream m)) = ZipSerial $ Stream $ \_ stp yld ->

@@ -981,20 +986,25 @@ instance (Monad m, Floating a) => Floating (ZipSerial m a) where

981-- Parallely Zipping Streams

986-- Parallely Zipping Streams

982------------------------------------------------------------------------------

987------------------------------------------------------------------------------

983

988

984−-- | Zip two streams asyncly (i.e. both the elements being zipped are generated

989+-- | Zip two streams concurrently (i.e. both the elements being zipped are

985−-- concurrently) using a pure zipping function.

990+-- generated concurrently) using a pure zipping function.

986−zipAsyncWith :: (IsStream t, MonadAsync m)

991+zipParallelWith :: (IsStream t, MonadParallel m)

987 => (a -> b -> c) -> t m a -> t m b -> t m c

992 => (a -> b -> c) -> t m a -> t m b -> t m c

988−zipAsyncWith f m1 m2 = fromStream $ Stream $ \_ stp yld -> do

993+zipParallelWith f m1 m2 = fromStream $ Stream $ \_ stp yld -> do

989 ma <- async m1

994 ma <- async m1

990 mb <- async m2

995 mb <- async m2

991 (S.runStream (toStream (zipWith f ma mb))) Nothing stp yld

996 (S.runStream (toStream (zipWith f ma mb))) Nothing stp yld

992

997

998+{-# DEPRECATED zipAsyncWith "Please use zipParallelWith instead." #-}

999+zipAsyncWith :: (IsStream t, MonadParallel m)

1000+ => (a -> b -> c) -> t m a -> t m b -> t m c

1001+zipAsyncWith = zipParallelWith

1002+

993-- | Like 'ZipSerial' but zips in parallel, it generates all the elements to

1003-- | Like 'ZipSerial' but zips in parallel, it generates all the elements to

994-- be zipped concurrently.

1004-- be zipped concurrently.

995--

1005--

996-- @

1006-- @

997−-- main = (toList . 'zippingAsync' $ (,,) \<$\> s1 \<*\> s2 \<*\> s3) >>= print

1007+-- main = (toList . 'zipParallely' $ (,,) \<$\> s1 \<*\> s2 \<*\> s3) >>= print

998-- where s1 = fromFoldable [1, 2]

1008-- where s1 = fromFoldable [1, 2]

999-- s2 = fromFoldable [3, 4]

1009-- s2 = fromFoldable [3, 4]

1000-- s3 = fromFoldable [5, 6]

1010-- s3 = fromFoldable [5, 6]

@@ -1006,27 +1016,30 @@ zipAsyncWith f m1 m2 = fromStream $ Stream $ \_ stp yld -> do

1006-- The 'Semigroup' instance of this type works the same way as that of

1016-- The 'Semigroup' instance of this type works the same way as that of

1007-- 'SerialT'.

1017-- 'SerialT'.

1008--

1018--

1009−newtype ZipAsync m a = ZipAsync {getZipAsync :: Stream m a}

1019+newtype ZipParallel m a = ZipParallel {getZipParallel :: Stream m a}

1010 deriving (Semigroup, Monoid)

1020 deriving (Semigroup, Monoid)

1011

1021

1022+{-# DEPRECATED ZipAsync "Please use ZipParallel instead." #-}

1023+type ZipAsync = ZipParallel

1024+

1012−deriving instance MonadAsync m => Alternative (ZipAsync m)

1025+deriving instance MonadParallel m => Alternative (ZipParallel m)

1013

1026

1014−instance Monad m => Functor (ZipAsync m) where

1027+instance Monad m => Functor (ZipParallel m) where

1015− fmap f (ZipAsync (Stream m)) = ZipAsync $ Stream $ \_ stp yld ->

1028+ fmap f (ZipParallel (Stream m)) = ZipParallel $ Stream $ \_ stp yld ->

1016 let yield a Nothing = yld (f a) Nothing

1029 let yield a Nothing = yld (f a) Nothing

1017 yield a (Just r) = yld (f a)

1030 yield a (Just r) = yld (f a)

1018− (Just (getZipAsync (fmap f (ZipAsync r))))

1031+ (Just (getZipParallel (fmap f (ZipParallel r))))

1019 in m Nothing stp yield

1032 in m Nothing stp yield

1020

1033

1021−instance MonadAsync m => Applicative (ZipAsync m) where

1034+instance MonadParallel m => Applicative (ZipParallel m) where

1022− pure = ZipAsync . S.repeat

1035+ pure = ZipParallel . S.repeat

1023− (<*>) = zipAsyncWith id

1036+ (<*>) = zipParallelWith id

1024

1037

1025−instance IsStream ZipAsync where

1038+instance IsStream ZipParallel where

1026− toStream = getZipAsync

1039+ toStream = getZipParallel

1027− fromStream = ZipAsync

1040+ fromStream = ZipParallel

1028

1041

1029−instance (MonadAsync m, Num a) => Num (ZipAsync m a) where

1042+instance (MonadParallel m, Num a) => Num (ZipParallel m a) where

1030 fromInteger n = pure (fromInteger n)

1043 fromInteger n = pure (fromInteger n)

1031

1044

1032 negate = fmap negate

1045 negate = fmap negate

@@ -1037,14 +1050,14 @@ instance (MonadAsync m, Num a) => Num (ZipAsync m a) where

1037 (*) = liftA2 (*)

1050 (*) = liftA2 (*)

1038 (-) = liftA2 (-)

1051 (-) = liftA2 (-)

1039

1052

1040−instance (MonadAsync m, Fractional a) => Fractional (ZipAsync m a) where

1053+instance (MonadParallel m, Fractional a) => Fractional (ZipParallel m a) where

1041 fromRational n = pure (fromRational n)

1054 fromRational n = pure (fromRational n)

1042

1055

1043 recip = fmap recip

1056 recip = fmap recip

1044

1057

1045 (/) = liftA2 (/)

1058 (/) = liftA2 (/)

1046

1059

1047−instance (MonadAsync m, Floating a) => Floating (ZipAsync m a) where

1060+instance (MonadParallel m, Floating a) => Floating (ZipParallel m a) where

1048 pi = pure pi

1061 pi = pure pi

1049

1062

1050 exp = fmap exp

1063 exp = fmap exp

@@ -1101,12 +1114,22 @@ parallely :: IsStream t => ParallelT m a -> t m a

1101parallely = adapt

1114parallely = adapt

1102

1115

1103-- | Fix the type of a polymorphic stream as 'ZipSerial'.

1116-- | Fix the type of a polymorphic stream as 'ZipSerial'.

1117+zipSerially :: IsStream t => ZipSerial m a -> t m a

1118+zipSerially = adapt

1119+

1120+-- | Same as 'zipParallely'.

1121+{-# DEPRECATED zipping "Please use zipSerially instead." #-}

1104zipping :: IsStream t => ZipSerial m a -> t m a

1122zipping :: IsStream t => ZipSerial m a -> t m a

1105−zipping = adapt

1123+zipping = zipSerially

1124+

1125+-- | Fix the type of a polymorphic stream as 'ZipParallel'.

1126+zipParallely :: IsStream t => ZipParallel m a -> t m a

1127+zipParallely = adapt

1106

1128

1107−-- | Fix the type of a polymorphic stream as 'ZipAsync'.

1129+-- | Same as 'zipParallely'.

1130+{-# DEPRECATED zippingAsync "Please use zipParallely instead." #-}

1108−zippingAsync :: IsStream t => ZipAsync m a -> t m a

1131+zippingAsync :: IsStream t => ZipParallel m a -> t m a

1109−zippingAsync = adapt

1132+zippingAsync = zipParallely

1110

1133

1111-------------------------------------------------------------------------------

1134-------------------------------------------------------------------------------

1112-- Running Streams, convenience functions specialized to types

1135-- Running Streams, convenience functions specialized to types

@@ -1133,38 +1156,33 @@ runParallelT :: Monad m => ParallelT m a -> m ()

1133runParallelT = runStream . parallely

1156runParallelT = runStream . parallely

1134

1157

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

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

1136−{-# DEPRECATED runZipStream "Please use 'runStream . zipping instead." #-}

1159+{-# DEPRECATED runZipStream "Please use 'runStream . zipSerially instead." #-}

1137runZipStream :: Monad m => ZipSerial m a -> m ()

1160runZipStream :: Monad m => ZipSerial m a -> m ()

1138−runZipStream = runStream . zipping

1161+runZipStream = runStream . zipSerially

1139

1162

1140-- | Same as @runStream . zippingAsync@.

1163-- | Same as @runStream . zippingAsync@.

1141−{-# DEPRECATED runZipAsync "Please use 'runStream . zippingAsync instead." #-}

1164+{-# DEPRECATED runZipAsync "Please use 'runStream . zipParallely instead." #-}

1142−runZipAsync :: Monad m => ZipAsync m a -> m ()

1165+runZipAsync :: Monad m => ZipParallel m a -> m ()

1143−runZipAsync = runStream . zippingAsync

1166+runZipAsync = runStream . zipParallely

1144

1167

1145------------------------------------------------------------------------------

1168------------------------------------------------------------------------------

1146-- Fold Utilities

1169-- Fold Utilities

1147------------------------------------------------------------------------------

1170------------------------------------------------------------------------------

1148

1171

1149−-- $foldutils

1172+-- | A variant of 'Data.Foldable.fold' that allows you to fold a 'Foldable'

1150−-- These utilities are designed to pass the first argument as one of the sum

1173+-- container of streams using the specified stream sum operation.

1151−-- style composition operators (i.e. '<>', '<=>', '<|', '<|>') to conveniently

1152−-- fold a container using any style of stream composition.

1153−

1154−-- | Like the 'Prelude' 'fold' but allows you to specify a binary sum style

1155−-- stream composition operator to fold a container of streams.

1156--

1174--

1157−-- @foldWith (<>) $ map return [1..3]@

1175+-- @foldWith 'parallel' $ map return [1..3]@

1158{-# INLINABLE foldWith #-}

1176{-# INLINABLE foldWith #-}

1159foldWith :: (IsStream t, Foldable f)

1177foldWith :: (IsStream t, Foldable f)

1160 => (t m a -> t m a -> t m a) -> f (t m a) -> t m a

1178 => (t m a -> t m a -> t m a) -> f (t m a) -> t m a

1161foldWith f = foldr f nil

1179foldWith f = foldr f nil

1162

1180

1163−-- | Like 'foldMap' but allows you to specify a binary sum style composition

1181+-- | A variant of 'foldMap' that allows you to map a monadic streaming action

1164−-- operator to fold a container of streams. Maps a monadic streaming action on

1182+-- on a 'Foldable' container and then fold it using the specified stream sum

1165−-- the container before folding it.

1183+-- operation.

1166--

1184--

1167−-- @foldMapWith (<>) return [1..3]@

1185+-- @foldMapWith 'parallel' return [1..3]@

1168{-# INLINABLE foldMapWith #-}

1186{-# INLINABLE foldMapWith #-}

1169foldMapWith :: (IsStream t, Foldable f)

1187foldMapWith :: (IsStream t, Foldable f)

1170 => (t m b -> t m b -> t m b) -> (a -> t m b) -> f a -> t m b

1188 => (t m b -> t m b -> t m b) -> (a -> t m b) -> f a -> t m b

test/Main.hs+2−2

@@ -91,8 +91,8 @@ main = hspec $ do

91 describe "Interleaved Composition" $ compose coserially mempty sort

92 describe "Left biased parallel Composition" $ compose coparallely mempty sort

93 describe "Fair parallel Composition" $ compose parallely mempty sort

94− describe "Semigroup Composition for ZipSerial" $ compose zipping mempty id

94+ describe "Semigroup Composition for ZipSerial" $ compose zipSerially mempty id

95− describe "Semigroup Composition for ZipAsync" $ compose zippingAsync mempty id

95+ describe "Semigroup Composition for ZipAsync" $ compose zipParallely mempty id

96 -- XXX need to check alternative compositions as well

97 ---------------------------------------------------------------------------

98 -- Monoidal Composition ordering checks

test/Prop.hs+25−25

@@ -260,7 +260,7 @@ zipMonadic constr t eq (a, b) =

260 stream2 <-

261 run

262 ((A.toList . t)

263− (A.zipAsyncWithM (\x y -> return (x, y)) (constr a) (constr b)))

263+ (A.zipParallelWithM (\x y -> return (x, y)) (constr a) (constr b)))

264 let list = getZipList $ (,) <$> ZipList a <*> ZipList b

265 equals eq stream1 list

266 equals eq stream2 list

@@ -322,18 +322,18 @@ main = hspec $ do

322 functorOps folded "coparallely folded" coparallely sortEq

323 functorOps A.fromFoldable "parallely" parallely sortEq

324 functorOps folded "parallely folded" parallely sortEq

325− functorOps A.fromFoldable "zipping" zipping (==)

325+ functorOps A.fromFoldable "zipSerially" zipSerially (==)

326− functorOps folded "zipping folded" zipping (==)

326+ functorOps folded "zipSerially folded" zipSerially (==)

327− functorOps A.fromFoldable "zippingAsync" zippingAsync (==)

327+ functorOps A.fromFoldable "zipParallely" zipParallely (==)

328− functorOps folded "zippingAsync folded" zippingAsync (==)

328+ functorOps folded "zipParallely folded" zipParallely (==)

329

330 describe "Semigroup operations" $ do

331 semigroupOps "serially" serially (==)

332 semigroupOps "coserially" coserially (==)

333 semigroupOps "coparallely" coparallely sortEq

334 semigroupOps "parallely" parallely sortEq

335− semigroupOps "zipping" zipping (==)

335+ semigroupOps "zipSerially" zipSerially (==)

336− semigroupOps "zippingAsync" zippingAsync (==)

336+ semigroupOps "zipParallely" zipParallely (==)

337

338 describe "Applicative operations" $ do

339 -- The tests using sorted equality are weaker tests

@@ -348,9 +348,9 @@ main = hspec $ do

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

349

350 describe "Zip operations" $ do

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

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

352 -- XXX this hangs

353− -- prop "zippingAsync applicative" $ zipApplicative zippingAsync (==)

353+ -- prop "zipParallely applicative" $ zipApplicative zipParallely (==)

354 prop "zip monadic serially" $ zipMonadic A.fromFoldable serially (==)

355 prop "zip monadic serially folded" $ zipMonadic folded serially (==)

356 prop "zip monadic coserially" $ zipMonadic A.fromFoldable coserially (==)

@@ -379,54 +379,54 @@ main = hspec $ do

379 describe "Stream transform operations" $ do

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

381 transformOps A.fromFoldable "coserially" coserially (==)

382− transformOps A.fromFoldable "zipping" zipping (==)

382+ transformOps A.fromFoldable "zipSerially" zipSerially (==)

383− transformOps A.fromFoldable "zippingAsync" zippingAsync (==)

383+ transformOps A.fromFoldable "zipParallely" zipParallely (==)

384 transformOps A.fromFoldable "coparallely" coparallely sortEq

385 transformOps A.fromFoldable "parallely" parallely sortEq

386

387 transformOps folded "serially folded" serially (==)

388 transformOps folded "coserially folded" coserially (==)

389− transformOps folded "zipping folded" zipping (==)

389+ transformOps folded "zipSerially folded" zipSerially (==)

390− transformOps folded "zippingAsync folded" zippingAsync (==)

390+ transformOps folded "zipParallely folded" zipParallely (==)

391 transformOps folded "coparallely folded" coparallely sortEq

392 transformOps folded "parallely folded" parallely sortEq

393

394 transformOpsWord8 A.fromFoldable "serially" serially

395 transformOpsWord8 A.fromFoldable "coserially" coserially

396− transformOpsWord8 A.fromFoldable "zipping" zipping

396+ transformOpsWord8 A.fromFoldable "zipSerially" zipSerially

397− transformOpsWord8 A.fromFoldable "zippingAsync" zippingAsync

397+ transformOpsWord8 A.fromFoldable "zipParallely" zipParallely

398 transformOpsWord8 A.fromFoldable "coparallely" coparallely

399 transformOpsWord8 A.fromFoldable "parallely" parallely

400

401 transformOpsWord8 folded "serially folded" serially

402 transformOpsWord8 folded "coserially folded" coserially

403− transformOpsWord8 folded "zipping folded" zipping

403+ transformOpsWord8 folded "zipSerially folded" zipSerially

404− transformOpsWord8 folded "zippingAsync folded" zippingAsync

404+ transformOpsWord8 folded "zipParallely folded" zipParallely

405 transformOpsWord8 folded "coparallely folded" coparallely

406 transformOpsWord8 folded "parallely folded" parallely

407

408 describe "Stream elimination operations" $ do

409 eliminationOps A.fromFoldable "serially" serially

410 eliminationOps A.fromFoldable "coserially" coserially

411− eliminationOps A.fromFoldable "zipping" zipping

411+ eliminationOps A.fromFoldable "zipSerially" zipSerially

412− eliminationOps A.fromFoldable "zippingAsync" zippingAsync

412+ eliminationOps A.fromFoldable "zipParallely" zipParallely

413 eliminationOps A.fromFoldable "coparallely" coparallely

414 eliminationOps A.fromFoldable "parallely" parallely

415

416 eliminationOps folded "serially folded" serially

417 eliminationOps folded "coserially folded" coserially

418− eliminationOps folded "zipping folded" zipping

418+ eliminationOps folded "zipSerially folded" zipSerially

419− eliminationOps folded "zippingAsync folded" zippingAsync

419+ eliminationOps folded "zipParallely folded" zipParallely

420 eliminationOps folded "coparallely folded" coparallely

421 eliminationOps folded "parallely folded" parallely

422

423 describe "Stream elimination operations" $ do

424 serialEliminationOps A.fromFoldable "serially" serially

425 serialEliminationOps A.fromFoldable "coserially" coserially

426− serialEliminationOps A.fromFoldable "zipping" zipping

426+ serialEliminationOps A.fromFoldable "zipSerially" zipSerially

427− serialEliminationOps A.fromFoldable "zippingAsync" zippingAsync

427+ serialEliminationOps A.fromFoldable "zipParallely" zipParallely

428

429 serialEliminationOps folded "serially folded" serially

430 serialEliminationOps folded "coserially folded" coserially

431− serialEliminationOps folded "zipping folded" zipping

431+ serialEliminationOps folded "zipSerially folded" zipSerially

432− serialEliminationOps folded "zippingAsync folded" zippingAsync

432+ serialEliminationOps folded "zipParallely folded" zipParallely