9cb2ac0 · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Add concurrent generation and transformation capabilities

9cb2ac0Harendra Kumar8 years ago

.travis.yml+5−3

compare with master

unifiedsplit

@@ -58,7 +58,9 @@ matrix:

58 # (Linux) stack builds

59 # --------------------------------------------------------------------------

61+ # For 7.10.3 we disable -Werror because of some bogus SPECIALIZE pragma

62+ # warnings.

61− - env: BUILD=stack CABALVER=1.22 GHCVER=7.10.3 RESOLVER=lts-6 STACK_YAML=stack-7.10.yaml STACK_BUILD_OPTIONS="--flag streamly:examples-sdl"

63+ - env: BUILD=stack CABALVER=1.22 GHCVER=7.10.3 GHC_OPTIONS="" RESOLVER=lts-6 STACK_YAML=stack-7.10.yaml STACK_BUILD_OPTIONS="--flag streamly:examples-sdl"

62 addons: {apt: {packages: [cabal-install-1.22,ghc-7.10.3,libsdl1.2-dev], sources: [hvr-ghc]}}

64 addons: {apt: {packages: [cabal-install-1.22,ghc-7.10.3,libsdl1.2-dev], sources: [hvr-ghc]}}

64 - env: BUILD=stack CABALVER=1.24 GHCVER=8.0.2 RESOLVER=lts-9 STACK_YAML=stack-8.0.yaml STACK_BUILD_OPTIONS="--flag streamly:examples-sdl"

66 - env: BUILD=stack CABALVER=1.24 GHCVER=8.0.2 RESOLVER=lts-9 STACK_YAML=stack-8.0.yaml STACK_BUILD_OPTIONS="--flag streamly:examples-sdl"

@@ -121,8 +123,8 @@ matrix:

121

123

122 # Note COVERALLS (hpc-coveralls) works only with cabal build.

124 # Note COVERALLS (hpc-coveralls) works only with cabal build.

123 # For this to succeed you have to add your project to coveralls.io first

125 # For this to succeed you have to add your project to coveralls.io first

124− #- env: BUILD=cabal CABALVER=2.0 GHCVER=8.2.2 COVERALLS_OPTIONS="--coverage-mode=StrictlyFullLines --exclude-dir=test test properties loops nested-loops parallel-loops"

126+ - env: BUILD=cabal CABALVER=2.0 GHCVER=8.2.2 COVERALLS_OPTIONS="--coverage-mode=StrictlyFullLines --exclude-dir=test test properties loops nested-loops parallel-loops"

125− # addons: {apt: {packages: [cabal-install-2.0,ghc-8.2.2], sources: [hvr-ghc]}}

127+ addons: {apt: {packages: [cabal-install-2.0,ghc-8.2.2], sources: [hvr-ghc]}}

126

128

127 # --------------------------------------------------------------------------

129 # --------------------------------------------------------------------------

128 # Builds that are allowed to fail

130 # Builds that are allowed to fail

Changelog.md+22−2

@@ -1,7 +1,27 @@

1## Unreleased

3+### Breaking changes

5+* Some prelude functions, to which concurrency capability has been added, will

6+ now require a `MonadAsync` constraint.

8+### Bug Fixes

10+* Fixed a race due to which, in a rare case, we might block indefinitely on

11+ an MVar due to a lost wakeup.

12+* Fixed an issue in adaptive concurrency. The issue caused us to stop creating

13+ more worker threads in some cases due to a race. This would not cause any

14+ functional issue but would reduce concurrency in some cases.

15+

16+### Enhancements

3−* Add concurrent lookahead stream type `Ahead`

17+* Added a concurrent lookahead stream type `Ahead`

4−* Add `fromFoldableM` to create a stream from a container of monadic actions

18+* Added `fromFoldableM` API that creates a stream from a container of monadic

19+ actions

20+* Monadic stream generation functions `consM`, `|:`, `unfoldrM`, `replicateM`,

21+ `repeatM`, `iterateM` and `fromFoldableM` can now generate streams

22+ concurrently when used with concurrent stream types.

23+* Monad transformation functions `mapM` and `sequence` can now map actions

24+ concurrently when used at appropriate stream types.

6## 0.2.0

26## 0.2.0

benchmark/LinearOps.hs+7−4

@@ -5,6 +5,8 @@

5-- License : MIT

6-- Maintainer : harendra.kumar@gmail.com

8+{-# LANGUAGE FlexibleContexts #-}

8module LinearOps where

10module LinearOps where

10import Prelude

12import Prelude

@@ -44,13 +46,14 @@ maxValue = value + 1000

44{-# INLINE composeAllInFilters #-}

46{-# INLINE composeAllInFilters #-}

45{-# INLINE composeAllOutFilters #-}

47{-# INLINE composeAllOutFilters #-}

46{-# INLINE composeMapAllInFilter #-}

48{-# INLINE composeMapAllInFilter #-}

47−toNull, scan, map, filterEven, mapM, filterAllOut,

49+toNull, scan, map, filterEven, filterAllOut,

48 filterAllIn, takeOne, takeAll, takeWhileTrue, dropAll, dropWhileTrue, zip,

50 filterAllIn, takeOne, takeAll, takeWhileTrue, dropAll, dropWhileTrue, zip,

49− concat, composeMapM, composeAllInFilters, composeAllOutFilters,

51+ concat, composeAllInFilters, composeAllOutFilters,

50 composeMapAllInFilter

52 composeMapAllInFilter

51 :: Monad m

53 :: Monad m

52 => Stream m Int -> m ()

54 => Stream m Int -> m ()

56+mapM, composeMapM :: S.MonadAsync m => Stream m Int -> m ()

54toList :: Monad m => Stream m Int -> m [Int]

57toList :: Monad m => Stream m Int -> m [Int]

55foldl :: Monad m => Stream m Int -> m Int

58foldl :: Monad m => Stream m Int -> m Int

56last :: Monad m => Stream m Int -> m (Maybe Int)

59last :: Monad m => Stream m Int -> m (Maybe Int)

@@ -74,14 +77,14 @@ sourceFromFoldable :: Int -> Stream m Int

74sourceFromFoldable n = S.fromFoldable [n..n+value]

77sourceFromFoldable n = S.fromFoldable [n..n+value]

76{-# INLINE sourceFromFoldableM #-}

79{-# INLINE sourceFromFoldableM #-}

77−sourceFromFoldableM :: Monad m => Int -> Stream m Int

80+sourceFromFoldableM :: S.MonadAsync m => Int -> Stream m Int

78sourceFromFoldableM n = S.fromFoldableM (Prelude.fmap return [n..n+value])

81sourceFromFoldableM n = S.fromFoldableM (Prelude.fmap return [n..n+value])

80sourceFoldMapWith :: Monad m => Int -> Stream m Int

83sourceFoldMapWith :: Monad m => Int -> Stream m Int

81sourceFoldMapWith n = S.foldMapWith S.serial return [n..n+value]

84sourceFoldMapWith n = S.foldMapWith S.serial return [n..n+value]

83{-# INLINE sourceUnfoldrM #-}

86{-# INLINE sourceUnfoldrM #-}

84−sourceUnfoldrM :: Monad m => Int -> Stream m Int

87+sourceUnfoldrM :: S.MonadAsync m => Int -> Stream m Int

85sourceUnfoldrM n = S.unfoldrM step n

88sourceUnfoldrM n = S.unfoldrM step n

86 where

89 where

87 step cnt =

90 step cnt =

src/Streamly.hs+1−1

@@ -91,7 +91,7 @@ module Streamly

92 -- * Stream Type Adapters

93 -- $adapters

94− , IsStream

94+ , IsStream ()

96 , serially

97 , wSerially

src/Streamly/Core.hs+201−37

@@ -1,3 +1,4 @@

1+{-# LANGUAGE CPP #-}

1{-# LANGUAGE ConstraintKinds #-}

2{-# LANGUAGE ConstraintKinds #-}

2{-# LANGUAGE FlexibleContexts #-}

3{-# LANGUAGE FlexibleContexts #-}

3{-# LANGUAGE FlexibleInstances #-}

4{-# LANGUAGE FlexibleInstances #-}

@@ -25,13 +26,19 @@ module Streamly.Core

25 -- * Streams

26 -- * Streams

26 , Stream (..)

27 , Stream (..)

28− -- * Construction

29+ -- * Construction (pure)

30+ , nil

31+ , cons

29 , singleton

32 , singleton

30 , once

33 , once

31− , cons

32− , consM

33 , repeat

34 , repeat

34− , nil

35+

36+ -- * Construction (monadic)

37+ , consM

38+ , consMAhead

39+ , consMAsync

40+ , consMWAsync

41+ , consMParallel

36 -- * Semigroup Style Composition

43 -- * Semigroup Style Composition

37 , serial

44 , serial

@@ -41,13 +48,11 @@ module Streamly.Core

41 , wAsync

48 , wAsync

42 , parallel

49 , parallel

44− -- * Alternative

45− , alt

46−

47 -- * zip

51 -- * zip

48 , zipWith

52 , zipWith

49 , zipAsyncWith

53 , zipAsyncWith

55+ -- XXX move to Streams.hs

51 -- * Transformers

56 -- * Transformers

52 , withLocal

57 , withLocal

53 , withCatchError

58 , withCatchError

@@ -55,18 +60,24 @@ module Streamly.Core

55 -- * Concurrent Stream Vars (SVars)

60 -- * Concurrent Stream Vars (SVars)

56 , SVar

61 , SVar

57 , SVarStyle (..)

62 , SVarStyle (..)

58− , newEmptySVar

59 , newStreamVar1

63 , newStreamVar1

60− , joinStreamVarAsync

61 , fromStreamVar

64 , fromStreamVar

62 , toStreamVar

65 , toStreamVar

63 )

66 )

64where

67where

69+-- MVar diagnostics has some overhead - around 5% on asyncly null benchmark,

70+-- but keep it on for now to debug problems quickly if and when they happen.

71+-- define DIAGNOSTICS

72+

66import Control.Concurrent (ThreadId, myThreadId, threadDelay)

73import Control.Concurrent (ThreadId, myThreadId, threadDelay)

67import Control.Concurrent.MVar (MVar, newEmptyMVar, tryTakeMVar,

74import Control.Concurrent.MVar (MVar, newEmptyMVar, tryTakeMVar,

68 tryPutMVar, takeMVar, putMVar)

75 tryPutMVar, takeMVar, putMVar)

69import Control.Exception (SomeException (..))

76import Control.Exception (SomeException (..))

77+#ifdef DIAGNOSTICS

78+import Control.Exception (catch,

79+ BlockedIndefinitelyOnMVar(..))

80+#endif

70import qualified Control.Exception.Lifted as EL

81import qualified Control.Exception.Lifted as EL

71import Control.Monad (when)

82import Control.Monad (when)

72import Control.Monad.Catch (MonadThrow, throwM)

83import Control.Monad.Catch (MonadThrow, throwM)

@@ -120,7 +131,7 @@ data SVarStyle =

120 | WAsyncVar -- breadth first concurrent

131 | WAsyncVar -- breadth first concurrent

121 | ParallelVar -- all parallel

132 | ParallelVar -- all parallel

122 | AheadVar -- Concurrent look ahead

133 | AheadVar -- Concurrent look ahead

123− deriving Eq

134+ deriving (Eq, Show)

124

135

125-- | An SVar or a Stream Var is a conduit to the output from multiple streams

136-- | An SVar or a Stream Var is a conduit to the output from multiple streams

126-- running concurrently and asynchronously. An SVar can be thought of as an

137-- running concurrently and asynchronously. An SVar can be thought of as an

@@ -170,6 +181,7 @@ data SVar m a =

170 , workQueue :: IORef ([Stream m a], Int)

181 , workQueue :: IORef ([Stream m a], Int)

171 , enqueue :: Stream m a -> IO ()

182 , enqueue :: Stream m a -> IO ()

172 , queueEmpty :: m Bool

183 , queueEmpty :: m Bool

184+ , waitingForWork :: IORef Bool

173 , runqueue :: m ()

185 , runqueue :: m ()

174

186

175 -- Shared, thread tracking

187 -- Shared, thread tracking

@@ -177,6 +189,50 @@ data SVar m a =

177 , activeWorkers :: IORef Int

189 , activeWorkers :: IORef Int

178 }

190 }

179

191

192+#ifdef DIAGNOSTICS

193+{-# NOINLINE dumpSVar #-}

194+dumpSVar :: SVar m a -> IO String

195+dumpSVar sv = do

196+ (oqList, oqLen) <- readIORef $ outputQueue sv

197+ db <- tryTakeMVar $ doorBell sv

198+ srn <- tryTakeMVar $ siren sv

199+ (oheap, oheapSize) <- readIORef $ outputHeap sv

200+ hsrn <- tryTakeMVar $ heapSiren sv

201+ (wq, wqSize) <- readIORef $ workQueue sv

202+ waiting <- readIORef $ waitingForWork sv

203+ rthread <- readIORef $ runningThreads sv

204+ workers <- readIORef $ activeWorkers sv

205+ -- XXX queueEmpty should be made IO return type

206+

207+ return $ unlines

208+ [ "style = " ++ show (svarStyle sv)

209+ , "outputQueue length computed = " ++ show (length oqList)

210+ , "outputQueue length maintained = " ++ show oqLen

211+ , "output doorBell = " ++ show db

212+ , "worker siren = " ++ show srn

213+ , "heap length computed = " ++ show (H.size oheap)

214+ , "heap length maintained = " ++ show oheapSize

215+ , "heap siren = " ++ show hsrn

216+ , "work queue length computed = " ++ show (length wq)

217+ , "work queue length maintained = " ++ show wqSize

218+ , "waitingForWork = " ++ show waiting

219+ , "running threads = " ++ show rthread

220+ , "running thread count = " ++ show workers

221+ ]

222+

223+{-# NOINLINE mvarExcHandler #-}

224+mvarExcHandler :: SVar m a -> String -> BlockedIndefinitelyOnMVar -> IO ()

225+mvarExcHandler sv label BlockedIndefinitelyOnMVar = do

226+ svInfo <- dumpSVar sv

227+ error $ label ++ " " ++ "BlockedIndefinitelyOnMVar\n" ++ svInfo

228+

229+withDBGMVar :: SVar m a -> String -> IO () -> IO ()

230+withDBGMVar sv label action = action `catch` mvarExcHandler sv label

231+#else

232+withDBGMVar :: SVar m a -> String -> IO () -> IO ()

233+withDBGMVar _ _ action = action

234+#endif

235+

180-- Slightly faster version of CAS. Gained some improvement by avoiding the use

236-- Slightly faster version of CAS. Gained some improvement by avoiding the use

181-- of "evaluate" because we know we do not have exceptions in fn.

237-- of "evaluate" because we know we do not have exceptions in fn.

182{-# INLINE atomicModifyIORefCAS #-}

238{-# INLINE atomicModifyIORefCAS #-}

@@ -350,7 +406,7 @@ send sv msg = do

350 -- we will be woken up by the consumer blowing the siren.

406 -- we will be woken up by the consumer blowing the siren.

351 -- Note: current length after adding our output is len + 1

407 -- Note: current length after adding our output is len + 1

352 else when (len + 1 >= 1500) $ do

408 else when (len + 1 >= 1500) $ do

353− takeMVar (siren sv)

409+ withDBGMVar sv "send: wait for siren" $ takeMVar (siren sv)

354 (_, n) <- readIORef (outputQueue sv)

410 (_, n) <- readIORef (outputQueue sv)

355 when (n <= 1500) $ void $ tryPutMVar (siren sv) ()

411 when (n <= 1500) $ void $ tryPutMVar (siren sv) ()

356

412

@@ -367,8 +423,22 @@ sendStop sv = myThreadId >>= \tid -> send sv (ChildStop tid Nothing)

367-- expressions. Large left associated compositions can grow this to a

423-- expressions. Large left associated compositions can grow this to a

368-- large size

424-- large size

369{-# INLINE enqueueLIFO #-}

425{-# INLINE enqueueLIFO #-}

370−enqueueLIFO :: IORef [Stream m a] -> Stream m a -> IO ()

426+enqueueLIFO :: SVar m a -> IORef [Stream m a] -> Stream m a -> IO ()

427+enqueueLIFO sv q m = do

428+ w <- readIORef $ waitingForWork sv

429+ if w

430+ then withDoorBell

371−enqueueLIFO q m = atomicModifyIORefCAS_ q $ \ ms -> m : ms

431+ else atomicModifyIORefCAS_ q $ \ms -> m : ms

432+

433+ where

434+

435+ withDoorBell = do

436+ v <- atomicModifyIORefCAS q $ \ ms ->

437+ case ms of

438+ [] -> (m : ms, True)

439+ _ -> (m : ms, False)

440+ when v $ void $ tryPutMVar (doorBell sv) ()

441+ atomicModifyIORefCAS_ (waitingForWork sv) (const False)

372

442

373runqueueLIFO :: MonadIO m => SVar m a -> IORef [Stream m a] -> m ()

443runqueueLIFO :: MonadIO m => SVar m a -> IORef [Stream m a] -> m ()

374runqueueLIFO sv q = run

444runqueueLIFO sv q = run

@@ -394,8 +464,19 @@ runqueueLIFO sv q = run

394-------------------------------------------------------------------------------

464-------------------------------------------------------------------------------

395

465

396{-# INLINE enqueueFIFO #-}

466{-# INLINE enqueueFIFO #-}

397−enqueueFIFO :: LinkedQueue (Stream m a) -> Stream m a -> IO ()

467+enqueueFIFO :: SVar m a -> LinkedQueue (Stream m a) -> Stream m a -> IO ()

398−enqueueFIFO = pushL

468+enqueueFIFO sv q m = do

469+ w <- readIORef $ waitingForWork sv

470+ if w

471+ then withDoorBell

472+ else pushL q m

473+

474+ where

475+

476+ withDoorBell = do

477+ emp <- nullQ q

478+ pushL q m

479+ when emp $ void $ tryPutMVar (doorBell sv) ()

399

480

400runqueueFIFO :: MonadIO m => SVar m a -> LinkedQueue (Stream m a) -> m ()

481runqueueFIFO :: MonadIO m => SVar m a -> LinkedQueue (Stream m a) -> m ()

401runqueueFIFO sv q = run

482runqueueFIFO sv q = run

@@ -411,7 +492,7 @@ runqueueFIFO sv q = run

411 dequeue = liftIO $ tryPopR q

492 dequeue = liftIO $ tryPopR q

412 sendit a = liftIO $ send sv (ChildYield a)

493 sendit a = liftIO $ send sv (ChildYield a)

413 single a = sendit a >> run

494 single a = sendit a >> run

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

495+ yield a r = sendit a >> liftIO (enqueueFIFO sv q r) >> run

415

496

416-------------------------------------------------------------------------------

497-------------------------------------------------------------------------------

417-- Parallel

498-- Parallel

@@ -499,7 +580,8 @@ sendToHeap sv msg seqNo = do

499 if added

580 if added

500 then do

581 then do

501 when (len + 1 >= maxHeapSize) $ do

582 when (len + 1 >= maxHeapSize) $ do

583+ withDBGMVar sv "sendToHeap: wait for heapSiren"

502− takeMVar (heapSiren sv)

584+ $ takeMVar (heapSiren sv)

503 (h, _) <- readIORef (outputHeap sv)

585 (h, _) <- readIORef (outputHeap sv)

504 when (H.size h <= maxHeapSize) $ void $ tryPutMVar (siren sv) ()

586 when (H.size h <= maxHeapSize) $ void $ tryPutMVar (siren sv) ()

505 return False

587 return False

@@ -508,7 +590,7 @@ sendToHeap sv msg seqNo = do

508 AheadYieldSingle a -> send sv (ChildYield a)

590 AheadYieldSingle a -> send sv (ChildYield a)

509 AheadYieldMany a mv -> do

591 AheadYieldMany a mv -> do

510 send sv (ChildYield a)

592 send sv (ChildYield a)

511− putMVar mv ()

593+ withDBGMVar sv "sendToHeap: wakeup stream" $ putMVar mv ()

512 return True

594 return True

513

595

514popFromHeap

596popFromHeap

@@ -547,7 +629,7 @@ drainHeap sv nextSeqNo = do

547 drainHeap sv nextSeqNo

629 drainHeap sv nextSeqNo

548 AheadYieldMany a mv -> do

630 AheadYieldMany a mv -> do

549 send sv (ChildYield a)

631 send sv (ChildYield a)

550− putMVar mv ()

632+ withDBGMVar sv "drainHeap: wakeup stream" $ putMVar mv ()

551

633

552-- | The thread that has the token has the responsibility to move any pending

634-- | The thread that has the token has the responsibility to move any pending

553-- entries in the heap to the outputQueue. When the token is assigned to a

635-- entries in the heap to the outputQueue. When the token is assigned to a

@@ -585,11 +667,15 @@ handleOutput sv haveToken prevSeqNo seqNo a ev = do

585-- associated expression, we queue the right expression and execute the left.

667-- associated expression, we queue the right expression and execute the left.

586-- Thererefore the queue never has more than on item in it.

668-- Thererefore the queue never has more than on item in it.

587{-# INLINE enqueueAhead #-}

669{-# INLINE enqueueAhead #-}

588−enqueueAhead :: IORef ([Stream m a], Int) -> Stream m a -> IO ()

670+enqueueAhead :: SVar m a -> IORef ([Stream m a], Int) -> Stream m a -> IO ()

589−enqueueAhead q m = do

671+enqueueAhead sv q m = do

672+ w <- readIORef $ waitingForWork sv

590 atomicModifyIORefCAS_ q $ \ case

673 atomicModifyIORefCAS_ q $ \ case

591 ([], n) -> ([m], n + 1) -- increment sequence

674 ([], n) -> ([m], n + 1) -- increment sequence

592 _ -> error "not empty"

675 _ -> error "not empty"

676+ when w $ do

677+ void $ tryPutMVar (doorBell sv) ()

678+ atomicModifyIORefCAS_ (waitingForWork sv) (const False)

593

679

594runqueueAhead :: MonadIO m => SVar m a -> IORef ([Stream m a], Int) -> m ()

680runqueueAhead :: MonadIO m => SVar m a -> IORef ([Stream m a], Int) -> m ()

595runqueueAhead sv q = run False (-1)

681runqueueAhead sv q = run False (-1)

@@ -637,7 +723,9 @@ runqueueAhead sv q = run False (-1)

637 gotToken <-

723 gotToken <-

638 handleOutput sv tStatus prevSeqNo seqNo a

724 handleOutput sv tStatus prevSeqNo seqNo a

639 (AheadYieldMany a mv)

725 (AheadYieldMany a mv)

640− when (not gotToken) $ void $ takeMVar mv

726+ when (not gotToken) $ void $

727+ withDBGMVar sv "runqueueAhead: stream waiting"

728+ $ takeMVar mv

641 (runStream r) (Just sv) stopNext singleNext yieldNext

729 (runStream r) (Just sv) stopNext singleNext yieldNext

642

730

643 (runStream m) (Just sv) stop single yield

731 (runStream m) (Just sv) stop single yield

@@ -725,32 +813,76 @@ sendWorkerWait sv = do

725 -- If there is no output pending to process and there is no worker to be

813 -- If there is no output pending to process and there is no worker to be

726 -- sent then we block, so that we do not keep looping fruitlessly.

814 -- sent then we block, so that we do not keep looping fruitlessly.

727

815

816+ -- Note that we are guaranteed to have at least one outstanding worker when

817+ -- we enter this function. So if we sleep we are guaranteed to be woken up.

818+ -- Note that the worker count is only decremented during event processing

819+ -- in fromStreamVar and therefore it is safe to read and use it without a

820+ -- lock.

821+

728 -- XXX we should wait in such a way so that we can be immediately

822 -- XXX we should wait in such a way so that we can be immediately

729− -- interrupted by a doorBell.

823+ -- interrupted by a doorBell. We can sleep on the doorbell and use a

824+ -- monitor thread to send us a doorbell.

730 liftIO $ threadDelay 200

825 liftIO $ threadDelay 200

731 (_, n) <- liftIO $ readIORef (outputQueue sv)

826 (_, n) <- liftIO $ readIORef (outputQueue sv)

732 when (n <= 0) $ do

827 when (n <= 0) $ do

733− -- XXX the queue may be empty temporarily if the worker has dequeued

828+ -- The queue may be empty temporarily if the worker has dequeued the

734− -- the work item but has not enqueued the remaining part yet. For the

829+ -- work item but has not enqueued the remaining part yet. For the same

735− -- same reason, a worker may come back if it tries to dequeue when the

830+ -- reason, a worker may come back if it tries to dequeue and finds the

736− -- queue is empty, even though the whole work has not finished yet.

831+ -- queue empty, even though the whole work has not finished yet.

737 done <- queueEmpty sv

832 done <- queueEmpty sv

738 if not done

833 if not done

739 then do

834 then do

740 cnt <- liftIO $ readIORef $ activeWorkers sv

835 cnt <- liftIO $ readIORef $ activeWorkers sv

836+ -- Note that we may deadlock if the previous workers (tasks in the

837+ -- stream) wait/depend on the future workers (tasks in the stream)

838+ -- executing. In that case we should either configure the maxWorker

839+ -- count to higher or use parallel style instead of ahead or async

840+ -- style.

741 if (cnt < 1500)

841 if (cnt < 1500)

742 then do

842 then do

743 pushWorker sv

843 pushWorker sv

744 sendWorkerWait sv

844 sendWorkerWait sv

845+ else liftIO $ withDBGMVar sv "sendWorkerWait: wait for workers"

745− else liftIO $ takeMVar (doorBell sv)

846+ $ takeMVar (doorBell sv)

847+ else do

848+ -- We found that the queue was empty, but it may be empty

849+ -- temporarily, when we checked it. If that's the case we might

850+ -- sleep indefinitely unless the active workers produce some

851+ -- output. We may deadlock specially if the otuput from the other

852+ -- workers depends on the future workers that we may never send.

853+ -- So in case the queue was temporarily empty set a flag to inform

854+ -- the enqueue to send us a doorbell.

855+

856+ liftIO $ atomicModifyIORefCAS_ (waitingForWork sv) $ const True

857+

858+ -- check again, this time we have set the waitingForWork flag so we

859+ -- are guaranteed to get a doorbell in case the status changed

860+ -- before we could sleep.

861+ --

862+ -- Note that this is just a best effort mechanism to avoid a

863+ -- deadlock. Deadlocks may still happen if for some weird reason

864+ -- the consuming computation shares an MVar or some other resource

865+ -- with the producing computation and gets blocked on that resource

866+ -- and therefore cannot do any pushworker to add more threads to

867+ -- the producer. In such cases the programmer should use a parallel

868+ -- style so that all the producers are scheduled immediately and

869+ -- unconditionally. We can also use a separate monitor thread to

870+ -- push workers instead of pushing them from the consumer, but then

871+ -- we are no longer using pull based concurrency rate adaptation.

872+ --

873+ -- XXX update this in the tutorial.

874+ emp <- queueEmpty sv

875+ when emp $

876+ liftIO $ withDBGMVar sv "sendWorkerWait: nothing to do"

746− else liftIO $ takeMVar (doorBell sv)

877+ $ takeMVar (doorBell sv)

747

878

748-- | Pull a stream from an SVar.

879-- | Pull a stream from an SVar.

749{-# NOINLINE fromStreamVar #-}

880{-# NOINLINE fromStreamVar #-}

750fromStreamVar :: MonadAsync m => SVar m a -> Stream m a

881fromStreamVar :: MonadAsync m => SVar m a -> Stream m a

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

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

752 if svarStyle sv == ParallelVar

883 if svarStyle sv == ParallelVar

884+ then liftIO $ withDBGMVar sv "fromStreamVar: doorbell"

753− then liftIO $ takeMVar (doorBell sv)

885+ $ takeMVar (doorBell sv)

754 else do

886 else do

755 res <- liftIO $ tryTakeMVar (doorBell sv)

887 res <- liftIO $ tryTakeMVar (doorBell sv)

756 when (isNothing res) $ sendWorkerWait sv

888 when (isNothing res) $ sendWorkerWait sv

@@ -798,6 +930,7 @@ getFifoSVar ctype = do

798 outQMv <- newEmptyMVar

930 outQMv <- newEmptyMVar

799 hooter <- newEmptyMVar

931 hooter <- newEmptyMVar

800 active <- newIORef 0

932 active <- newIORef 0

933+ wfw <- newIORef False

801 running <- newIORef S.empty

934 running <- newIORef S.empty

802 q <- newQ

935 q <- newQ

803 let sv =

936 let sv =

@@ -809,8 +942,9 @@ getFifoSVar ctype = do

809 , runningThreads = running

942 , runningThreads = running

810 , workQueue = undefined

943 , workQueue = undefined

811 , runqueue = runqueueFIFO sv q

944 , runqueue = runqueueFIFO sv q

812− , enqueue = pushL q

945+ , enqueue = enqueueFIFO sv q

813 , queueEmpty = liftIO $ nullQ q

946 , queueEmpty = liftIO $ nullQ q

947+ , waitingForWork = wfw

814 , svarStyle = ctype

948 , svarStyle = ctype

815 , activeWorkers = active

949 , activeWorkers = active

816 }

950 }

@@ -822,6 +956,7 @@ getLifoSVar ctype = do

822 outQMv <- newEmptyMVar

956 outQMv <- newEmptyMVar

823 hooter <- newEmptyMVar

957 hooter <- newEmptyMVar

824 active <- newIORef 0

958 active <- newIORef 0

959+ wfw <- newIORef False

825 running <- newIORef S.empty

960 running <- newIORef S.empty

826 q <- newIORef []

961 q <- newIORef []

827 let checkEmpty = null <$> liftIO (readIORef q)

962 let checkEmpty = null <$> liftIO (readIORef q)

@@ -834,8 +969,9 @@ getLifoSVar ctype = do

834 , runningThreads = running

969 , runningThreads = running

835 , workQueue = undefined

970 , workQueue = undefined

836 , runqueue = runqueueLIFO sv q

971 , runqueue = runqueueLIFO sv q

837− , enqueue = enqueueLIFO q

972+ , enqueue = enqueueLIFO sv q

838 , queueEmpty = checkEmpty

973 , queueEmpty = checkEmpty

974+ , waitingForWork = wfw

839 , svarStyle = ctype

975 , svarStyle = ctype

840 , activeWorkers = active

976 , activeWorkers = active

841 }

977 }

@@ -847,6 +983,7 @@ getParSVar style = do

847 outQMv <- newEmptyMVar

983 outQMv <- newEmptyMVar

848 hooter <- newEmptyMVar

984 hooter <- newEmptyMVar

849 active <- newIORef 0

985 active <- newIORef 0

986+ wfw <- newIORef False

850 running <- newIORef S.empty

987 running <- newIORef S.empty

851 let sv =

988 let sv =

852 SVar { outputQueue = outQ

989 SVar { outputQueue = outQ

@@ -859,6 +996,7 @@ getParSVar style = do

859 , runqueue = undefined

996 , runqueue = undefined

860 , enqueue = undefined

997 , enqueue = undefined

861 , queueEmpty = undefined

998 , queueEmpty = undefined

999+ , waitingForWork = wfw

862 , svarStyle = style

1000 , svarStyle = style

863 , activeWorkers = active

1001 , activeWorkers = active

864 }

1002 }

@@ -872,6 +1010,7 @@ getAheadSVar style = do

872 hooter <- newEmptyMVar

1010 hooter <- newEmptyMVar

873 hsiren <- newEmptyMVar

1011 hsiren <- newEmptyMVar

874 active <- newIORef 0

1012 active <- newIORef 0

1013+ wfw <- newIORef False

875 running <- newIORef S.empty

1014 running <- newIORef S.empty

876 q <- newIORef ([], -1)

1015 q <- newIORef ([], -1)

877 let checkEmpty = liftIO $ do

1016 let checkEmpty = liftIO $ do

@@ -888,6 +1027,7 @@ getAheadSVar style = do

888 , runqueue = runqueueAhead sv q

1027 , runqueue = runqueueAhead sv q

889 , enqueue = undefined

1028 , enqueue = undefined

890 , queueEmpty = checkEmpty

1029 , queueEmpty = checkEmpty

1030+ , waitingForWork = wfw

891 , svarStyle = style

1031 , svarStyle = style

892 , activeWorkers = active

1032 , activeWorkers = active

893 }

1033 }

@@ -921,7 +1061,7 @@ newStreamVarAhead m = do

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

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

922 -- pushworker, otherwise the pushworker may exit before we even get a

1062 -- pushworker, otherwise the pushworker may exit before we even get a

923 -- chance to push.

1063 -- chance to push.

924− liftIO $ enqueueAhead (workQueue sv) m

1064+ liftIO $ enqueueAhead sv (workQueue sv) m

925 pushWorker sv

1065 pushWorker sv

926 return sv

1066 return sv

927

1067

@@ -1033,7 +1173,7 @@ ahead :: MonadAsync m => Stream m a -> Stream m a -> Stream m a

1033ahead m1 m2 = Stream $ \svr stp sng yld -> do

1173ahead m1 m2 = Stream $ \svr stp sng yld -> do

1034 case svr of

1174 case svr of

1035 Just sv | svarStyle sv == AheadVar -> do

1175 Just sv | svarStyle sv == AheadVar -> do

1036− liftIO $ enqueueAhead (workQueue sv) m2

1176+ liftIO $ enqueueAhead sv (workQueue sv) m2

1037 -- Always run the left side on a new SVar to avoid complexity in

1177 -- Always run the left side on a new SVar to avoid complexity in

1038 -- sequencing results. This means the left side cannot further

1178 -- sequencing results. This means the left side cannot further

1039 -- split into more ahead computations on the same SVar.

1179 -- split into more ahead computations on the same SVar.

@@ -1043,9 +1183,15 @@ ahead m1 m2 = Stream $ \svr stp sng yld -> do

1043 (runStream (fromStreamVar sv)) Nothing stp sng yld

1183 (runStream (fromStreamVar sv)) Nothing stp sng yld

1044 where

1184 where

1045 concurrently ma mb = Stream $ \svr stp sng yld -> do

1185 concurrently ma mb = Stream $ \svr stp sng yld -> do

1046− liftIO $ enqueueAhead (workQueue (fromJust svr)) mb

1186+ liftIO $ enqueueAhead (fromJust svr) (workQueue (fromJust svr)) mb

1047 (runStream ma) Nothing stp sng yld

1187 (runStream ma) Nothing stp sng yld

1048

1188

1189+-- | XXX we can implement it more efficienty by directly implementing instead

1190+-- of combining streams using ahead.

1191+{-# INLINE consMAhead #-}

1192+consMAhead :: MonadAsync m => m a -> Stream m a -> Stream m a

1193+consMAhead m r = once m `ahead` r

1194+

1049------------------------------------------------------------------------------

1195------------------------------------------------------------------------------

1050-- Semigroup and Monoid style compositions for parallel actions

1196-- Semigroup and Monoid style compositions for parallel actions

1051------------------------------------------------------------------------------

1197------------------------------------------------------------------------------

@@ -1054,14 +1200,32 @@ ahead m1 m2 = Stream $ \svr stp sng yld -> do

1054async :: MonadAsync m => Stream m a -> Stream m a -> Stream m a

1200async :: MonadAsync m => Stream m a -> Stream m a -> Stream m a

1055async = joinStreamVarAsync AsyncVar

1201async = joinStreamVarAsync AsyncVar

1056

1202

1203+-- | XXX we can implement it more efficienty by directly implementing instead

1204+-- of combining streams using async.

1205+{-# INLINE consMAsync #-}

1206+consMAsync :: MonadAsync m => m a -> Stream m a -> Stream m a

1207+consMAsync m r = once m `async` r

1208+

1057{-# INLINE wAsync #-}

1209{-# INLINE wAsync #-}

1058wAsync :: MonadAsync m => Stream m a -> Stream m a -> Stream m a

1210wAsync :: MonadAsync m => Stream m a -> Stream m a -> Stream m a

1059wAsync = joinStreamVarAsync WAsyncVar

1211wAsync = joinStreamVarAsync WAsyncVar

1060

1212

1213+-- | XXX we can implement it more efficienty by directly implementing instead

1214+-- of combining streams using wAsync.

1215+{-# INLINE consMWAsync #-}

1216+consMWAsync :: MonadAsync m => m a -> Stream m a -> Stream m a

1217+consMWAsync m r = once m `wAsync` r

1218+

1061{-# INLINE parallel #-}

1219{-# INLINE parallel #-}

1062parallel :: MonadAsync m => Stream m a -> Stream m a -> Stream m a

1220parallel :: MonadAsync m => Stream m a -> Stream m a -> Stream m a

1063parallel = joinStreamVarPar ParallelVar

1221parallel = joinStreamVarPar ParallelVar

1064

1222

1223+-- | XXX we can implement it more efficienty by directly implementing instead

1224+-- of combining streams using parallel.

1225+{-# INLINE consMParallel #-}

1226+consMParallel :: MonadAsync m => m a -> Stream m a -> Stream m a

1227+consMParallel m r = once m `parallel` r

1228+

1065-------------------------------------------------------------------------------

1229-------------------------------------------------------------------------------

1066-- Functor instace is the same for all types

1230-- Functor instace is the same for all types

1067-------------------------------------------------------------------------------

1231-------------------------------------------------------------------------------

@@ -1076,8 +1240,8 @@ instance Monad m => Functor (Stream m) where

1076-- Alternative & MonadPlus

1240-- Alternative & MonadPlus

1077------------------------------------------------------------------------------

1241------------------------------------------------------------------------------

1078

1242

1079−alt :: Stream m a -> Stream m a -> Stream m a

1243+_alt :: Stream m a -> Stream m a -> Stream m a

1080−alt m1 m2 = Stream $ \_ stp sng yld ->

1244+_alt m1 m2 = Stream $ \_ stp sng yld ->

1081 let stop = runStream m2 Nothing stp sng yld

1245 let stop = runStream m2 Nothing stp sng yld

1082 in runStream m1 Nothing stop sng yld

1246 in runStream m1 Nothing stop sng yld

1083

1247

src/Streamly/Prelude.hs+31−33

@@ -26,6 +26,9 @@

26-- provided for convenience and for consistency with other pure APIs in the

27-- @base@ package.

28--

29+-- Functions having a 'MonadAsync' constraint work concurrently when used

30+-- with appropriate stream type combinator.

31+--

29-- Deconstruction and folds accept a 'SerialT' type instead of a polymorphic

32-- Deconstruction and folds accept a 'SerialT' type instead of a polymorphic

30-- type to ensure that streams always have a concrete monomorphic type by

33-- type to ensure that streams always have a concrete monomorphic type by

31-- default, reducing type errors. In case you want to use any other type of

34-- default, reducing type errors. In case you want to use any other type of

@@ -159,14 +162,15 @@ unfoldr step = fromStream . go

159--

162--

160-- @since 0.1.0

163-- @since 0.1.0

161{-# INLINE unfoldrM #-}

164{-# INLINE unfoldrM #-}

162−unfoldrM :: (IsStream t, Monad m) => (b -> m (Maybe (a, b))) -> b -> t m a

165+unfoldrM :: (IsStream t, MonadAsync m) => (b -> m (Maybe (a, b))) -> b -> t m a

163−unfoldrM step = fromStream . go

166+unfoldrM step = go

164 where

167 where

165− go s = Stream $ \_ stp _ yld -> do

168+ go s = fromStream $ Stream $ \svr stp sng yld -> do

166 mayb <- step s

169 mayb <- step s

167 case mayb of

170 case mayb of

168 Nothing -> stp

171 Nothing -> stp

169− Just (a, b) -> yld a (go b)

172+ Just (a, b) ->

173+ S.runStream (toStream (return a |: go b)) svr stp sng yld

170

174

171-- | Construct a stream from a 'Foldable' containing pure values.

175-- | Construct a stream from a 'Foldable' containing pure values.

172--

176--

@@ -179,7 +183,7 @@ fromFoldable = Prelude.foldr cons nil

179--

183--

180-- @since 0.3.0

184-- @since 0.3.0

181{-# INLINE fromFoldableM #-}

185{-# INLINE fromFoldableM #-}

182−fromFoldableM :: (IsStream t, Monad m, Foldable f) => f (m a) -> t m a

186+fromFoldableM :: (IsStream t, MonadAsync m, Foldable f) => f (m a) -> t m a

183fromFoldableM = Prelude.foldr consM nil

187fromFoldableM = Prelude.foldr consM nil

184

188

185-- | Same as 'fromFoldable'.

189-- | Same as 'fromFoldable'.

@@ -206,22 +210,17 @@ once = fromStream . S.once

206-- | Generate a stream by performing a monadic action @n@ times.

210-- | Generate a stream by performing a monadic action @n@ times.

207--

211--

208-- @since 0.1.1

212-- @since 0.1.1

209−replicateM :: (IsStream t, Monad m) => Int -> m a -> t m a

213+replicateM :: (IsStream t, MonadAsync m) => Int -> m a -> t m a

210−replicateM n m = fromStream $ go n

214+replicateM n m = go n

211 where

215 where

212− go cnt = Stream $ \_ stp _ yld ->

213− if cnt <= 0

214− then stp

215− else m >>= \a -> yld a (go (cnt - 1))

216+ go cnt = if cnt <= 0 then nil else m |: go (cnt - 1)

216

217

217-- | Generate a stream by repeatedly executing a monadic action forever.

218-- | Generate a stream by repeatedly executing a monadic action forever.

218--

219--

219-- @since 0.2.0

220-- @since 0.2.0

220−repeatM :: (IsStream t, Monad m) => m a -> t m a

221+repeatM :: (IsStream t, MonadAsync m) => m a -> t m a

221−repeatM = fromStream . go

222+repeatM = go

222− where

223+ where go m = m |: go m

223− go m = Stream $ \_ _ _ yld ->

224− m >>= \a -> yld a (go m)

225

224

226-- | Iterate a pure function from a seed value, streaming the results forever.

225-- | Iterate a pure function from a seed value, streaming the results forever.

227--

226--

@@ -235,12 +234,12 @@ iterate step = fromStream . go

235-- forever.

234-- forever.

236--

235--

237-- @since 0.1.2

236-- @since 0.1.2

238−iterateM :: (IsStream t, Monad m) => (a -> m a) -> a -> t m a

237+iterateM :: (IsStream t, MonadAsync m) => (a -> m a) -> a -> t m a

239−iterateM step = fromStream . go

238+iterateM step = go

240 where

239 where

241− go s = Stream $ \_ _ _ yld -> do

240+ go s = fromStream $ Stream $ \svr stp sng yld -> do

242− a <- step s

241+ next <- step s

243− yld s (go a)

242+ S.runStream (toStream (return s |: go next)) svr stp sng yld

244

243

245-- | Read lines from an IO Handle into a stream of Strings.

244-- | Read lines from an IO Handle into a stream of Strings.

246--

245--

@@ -396,8 +395,8 @@ foldlM' :: Monad m => (b -> a -> m b) -> b -> SerialT m a -> m b

396foldlM' step begin m = foldxM step (return begin) return m

395foldlM' step begin m = foldxM step (return begin) return m

397

396

398-- | Decompose a stream into its head and tail. If the stream is empty, returns

397-- | Decompose a stream into its head and tail. If the stream is empty, returns

399−-- 'Nothing'. If the stream is non-empty, returns 'Just (a, ma)', where 'a' is

398+-- 'Nothing'. If the stream is non-empty, returns @Just (a, ma)@, where @a@ is

400−-- the head of the stream and 'ma' its tail.

399+-- the head of the stream and @ma@ its tail.

401--

400--

402-- @since 0.1.0

401-- @since 0.1.0

403uncons :: (IsStream t, Monad m) => SerialT m a -> m (Maybe (a, t m a))

402uncons :: (IsStream t, Monad m) => SerialT m a -> m (Maybe (a, t m a))

@@ -655,21 +654,20 @@ maximum m = go Nothing (toStream m)

655-- Transformation

654-- Transformation

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

655------------------------------------------------------------------------------

657

656

658−-- XXX Parallel variants of these? mapMWith et al. sequenceWith.

659−

660-- | Replace each element of the stream with the result of a monadic action

657-- | Replace each element of the stream with the result of a monadic action

661-- applied on the element.

658-- applied on the element.

662--

659--

663-- @since 0.1.0

660-- @since 0.1.0

664{-# INLINE mapM #-}

661{-# INLINE mapM #-}

665−mapM :: (IsStream t, Monad m) => (a -> m b) -> t m a -> t m b

662+mapM :: (IsStream t, MonadAsync m) => (a -> m b) -> t m a -> t m b

666−mapM f m = fromStream $ go (toStream m)

663+mapM f m = go (toStream m)

667 where

664 where

668− go m1 = Stream $ \_ stp sng yld ->

665+ go m1 = fromStream $ Stream $ \svr stp sng yld ->

669 let single a = f a >>= sng

666 let single a = f a >>= sng

670− yield a r = f a >>= \b -> yld b (go r)

667+ yield a r = S.runStream (toStream (f a |: (go r))) svr stp sng yld

671 in (S.runStream m1) Nothing stp single yield

668 in (S.runStream m1) Nothing stp single yield

672

669

670+-- XXX this can utilize parallel mapping if we implement it as runStream . mapM

673-- | Apply a monadic action to each element of the stream and discard the

671-- | Apply a monadic action to each element of the stream and discard the

674-- output of the action.

672-- output of the action.

675--

673--

@@ -687,12 +685,12 @@ mapM_ f m = go (toStream m)

687-- actions.

685-- actions.

688--

686--

689-- @since 0.1.0

687-- @since 0.1.0

690−sequence :: (IsStream t, Monad m) => t m (m a) -> t m a

688+sequence :: (IsStream t, MonadAsync m) => t m (m a) -> t m a

691−sequence m = fromStream $ go (toStream m)

689+sequence m = go (toStream m)

692 where

690 where

693− go m1 = Stream $ \_ stp sng yld ->

691+ go m1 = fromStream $ Stream $ \svr stp sng yld ->

694 let single ma = ma >>= sng

692 let single ma = ma >>= sng

695− yield ma r = ma >>= \b -> yld b (go r)

693+ yield ma r = S.runStream (toStream $ ma |: go r) svr stp sng yld

696 in (S.runStream m1) Nothing stp single yield

694 in (S.runStream m1) Nothing stp single yield

697

695

698------------------------------------------------------------------------------

696------------------------------------------------------------------------------

src/Streamly/Streams.hs+82−40

@@ -3,6 +3,7 @@

3{-# LANGUAGE FlexibleContexts #-}

4{-# LANGUAGE FlexibleInstances #-}

5{-# LANGUAGE GeneralizedNewtypeDeriving#-}

6+{-# LANGUAGE InstanceSigs #-}

6{-# LANGUAGE MultiParamTypeClasses #-}

7{-# LANGUAGE MultiParamTypeClasses #-}

7{-# LANGUAGE RankNTypes #-}

8{-# LANGUAGE RankNTypes #-}

8{-# LANGUAGE StandaloneDeriving #-}

9{-# LANGUAGE StandaloneDeriving #-}

@@ -24,15 +25,10 @@ module Streamly.Streams

24 , Streaming -- deprecated

25 , Streaming -- deprecated

25 , S.MonadAsync

26 , S.MonadAsync

27− -- * SVars

28− , SVarStyle (..)

29− , SVar

30− , S.newEmptySVar

31−

32 -- * Construction

28 -- * Construction

33 , nil

29 , nil

30+ , cons

34 , (.:)

31 , (.:)

35− , (|:)

36 , streamBuild

32 , streamBuild

37 , fromCallback

33 , fromCallback

38 , fromSVar

34 , fromSVar

@@ -125,6 +121,9 @@ import qualified Streamly.Core as S

125-- Types that can behave as a Stream

121-- Types that can behave as a Stream

126------------------------------------------------------------------------------

122------------------------------------------------------------------------------

127

123

124+infixr 5 `consM`

125+infixr 5 |:

126+

128-- | Class of types that can represent a stream of elements of some type 'a' in

127-- | Class of types that can represent a stream of elements of some type 'a' in

129-- some monad 'm'.

128-- some monad 'm'.

130--

129--

@@ -132,8 +131,29 @@ import qualified Streamly.Core as S

132class IsStream t where

131class IsStream t where

133 toStream :: t m a -> S.Stream m a

132 toStream :: t m a -> S.Stream m a

134 fromStream :: S.Stream m a -> t m a

133 fromStream :: S.Stream m a -> t m a

134+ -- | Constructs a stream by adding a monadic action at the head of an existing

135+ -- stream. For example:

136+ --

137+ -- @

138+ -- > toList $ getLine \`consM` getLine \`consM` nil

139+ -- hello

140+ -- world

141+ -- ["hello","world"]

142+ -- @

143+ --

144+ -- @since 0.2.0

135− cons :: a -> t m a -> t m a

145+ consM :: MonadAsync m => m a -> t m a -> t m a

146+ -- | Operator equivalent of 'consM'.

147+ --

148+ -- @

149+ -- > toList $ getLine |: getLine |: nil

150+ -- hello

151+ -- world

152+ -- ["hello","world"]

153+ -- @

154+ --

155+ -- @since 0.2.0

136− consM :: Monad m => m a -> t m a -> t m a

156+ (|:) :: MonadAsync m => m a -> t m a -> t m a

137

157

138-- | Same as 'IsStream'.

158-- | Same as 'IsStream'.

139--

159--

@@ -156,8 +176,6 @@ type Streaming = IsStream

156nil :: IsStream t => t m a

176nil :: IsStream t => t m a

157nil = fromStream S.nil

177nil = fromStream S.nil

158

178

159−infixr 5 `consM`

160−

161-- | Constructs a stream by adding a monadic action at the head of an existing

179-- | Constructs a stream by adding a monadic action at the head of an existing

162-- stream. For example:

180-- stream. For example:

163--

181--

@@ -169,25 +187,9 @@ infixr 5 `consM`

169-- @

187-- @

170--

188--

171-- @since 0.2.0

189-- @since 0.2.0

172−-- {-# INLINE consMSerial #-}

173consMSerial :: (IsStream t, Monad m) => m a -> t m a -> t m a

190consMSerial :: (IsStream t, Monad m) => m a -> t m a -> t m a

174consMSerial m r = fromStream $ S.consM m (toStream r)

191consMSerial m r = fromStream $ S.consM m (toStream r)

175

192

176−infixr 5 |:

177−

178−-- | Operator equivalent of 'consM'.

179−--

180−-- @

181−-- > toList $ getLine |: getLine |: nil

182−-- hello

183−-- world

184−-- ["hello","world"]

185−-- @

186−--

187−-- @since 0.2.0

188−(|:) :: (IsStream t, Monad m) => m a -> t m a -> t m a

189−(|:) = consM

190−

191infixr 5 `cons`

193infixr 5 `cons`

192

194

193-- | Construct a stream by adding a pure value at the head of an existing

195-- | Construct a stream by adding a pure value at the head of an existing

@@ -199,8 +201,8 @@ infixr 5 `cons`

199-- @

201-- @

200--

202--

201-- @since 0.1.0

203-- @since 0.1.0

202−consSerial :: IsStream t => a -> t m a -> t m a

204+cons :: IsStream t => a -> t m a -> t m a

203−consSerial a r = fromStream $ S.cons a (toStream r)

205+cons a r = fromStream $ S.cons a (toStream r)

204

206

205infixr 5 .:

207infixr 5 .:

206

208

@@ -412,11 +414,17 @@ type StreamT = SerialT

412instance IsStream SerialT where

414instance IsStream SerialT where

413 toStream = getSerialT

415 toStream = getSerialT

414 fromStream = SerialT

416 fromStream = SerialT

415− cons = consSerial

417+

416 {-# INLINE consM #-}

418 {-# INLINE consM #-}

417 {-# SPECIALIZE consM :: IO a -> SerialT IO a -> SerialT IO a #-}

419 {-# SPECIALIZE consM :: IO a -> SerialT IO a -> SerialT IO a #-}

420+ consM :: Monad m => m a -> SerialT m a -> SerialT m a

418 consM = consMSerial

421 consM = consMSerial

419

422

423+ {-# INLINE (|:) #-}

424+ {-# SPECIALIZE (|:) :: IO a -> SerialT IO a -> SerialT IO a #-}

425+ (|:) :: Monad m => m a -> SerialT m a -> SerialT m a

426+ (|:) = consMSerial

427+

420------------------------------------------------------------------------------

428------------------------------------------------------------------------------

421-- Semigroup

429-- Semigroup

422------------------------------------------------------------------------------

430------------------------------------------------------------------------------

@@ -501,11 +509,17 @@ type InterleavedT = WSerialT

501instance IsStream WSerialT where

509instance IsStream WSerialT where

502 toStream = getWSerialT

510 toStream = getWSerialT

503 fromStream = WSerialT

511 fromStream = WSerialT

504− cons = consSerial

512+

505 {-# INLINE consM #-}

513 {-# INLINE consM #-}

506 {-# SPECIALIZE consM :: IO a -> WSerialT IO a -> WSerialT IO a #-}

514 {-# SPECIALIZE consM :: IO a -> WSerialT IO a -> WSerialT IO a #-}

515+ consM :: Monad m => m a -> WSerialT m a -> WSerialT m a

507 consM = consMSerial

516 consM = consMSerial

508

517

518+ {-# INLINE (|:) #-}

519+ {-# SPECIALIZE (|:) :: IO a -> WSerialT IO a -> WSerialT IO a #-}

520+ (|:) :: Monad m => m a -> WSerialT m a -> WSerialT m a

521+ (|:) = consMSerial

522+

509------------------------------------------------------------------------------

523------------------------------------------------------------------------------

510-- Semigroup

524-- Semigroup

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

525------------------------------------------------------------------------------

@@ -612,10 +626,14 @@ newtype AheadT m a = AheadT {getAheadT :: S.Stream m a}

612instance IsStream AheadT where

626instance IsStream AheadT where

613 toStream = getAheadT

627 toStream = getAheadT

614 fromStream = AheadT

628 fromStream = AheadT

615− cons = consSerial

629+

616 {-# INLINE consM #-}

630 {-# INLINE consM #-}

617 {-# SPECIALIZE consM :: IO a -> AheadT IO a -> AheadT IO a #-}

631 {-# SPECIALIZE consM :: IO a -> AheadT IO a -> AheadT IO a #-}

618− consM = consMSerial

632+ consM m r = fromStream $ S.consMAhead m (toStream r)

633+

634+ {-# INLINE (|:) #-}

635+ {-# SPECIALIZE (|:) :: IO a -> AheadT IO a -> AheadT IO a #-}

636+ (|:) = consM

619

637

620------------------------------------------------------------------------------

638------------------------------------------------------------------------------

621-- Semigroup

639-- Semigroup

@@ -729,10 +747,14 @@ newtype AsyncT m a = AsyncT {getAsyncT :: S.Stream m a}

729instance IsStream AsyncT where

747instance IsStream AsyncT where

730 toStream = getAsyncT

748 toStream = getAsyncT

731 fromStream = AsyncT

749 fromStream = AsyncT

732− cons = consSerial

750+

733 {-# INLINE consM #-}

751 {-# INLINE consM #-}

734 {-# SPECIALIZE consM :: IO a -> AsyncT IO a -> AsyncT IO a #-}

752 {-# SPECIALIZE consM :: IO a -> AsyncT IO a -> AsyncT IO a #-}

735− consM = consMSerial

753+ consM m r = fromStream $ S.consMAsync m (toStream r)

754+

755+ {-# INLINE (|:) #-}

756+ {-# SPECIALIZE (|:) :: IO a -> AsyncT IO a -> AsyncT IO a #-}

757+ (|:) = consM

736

758

737------------------------------------------------------------------------------

759------------------------------------------------------------------------------

738-- Semigroup

760-- Semigroup

@@ -853,10 +875,14 @@ newtype WAsyncT m a = WAsyncT {getWAsyncT :: S.Stream m a}

853instance IsStream WAsyncT where

875instance IsStream WAsyncT where

854 toStream = getWAsyncT

876 toStream = getWAsyncT

855 fromStream = WAsyncT

877 fromStream = WAsyncT

856− cons = consSerial

878+

857 {-# INLINE consM #-}

879 {-# INLINE consM #-}

858 {-# SPECIALIZE consM :: IO a -> WAsyncT IO a -> WAsyncT IO a #-}

880 {-# SPECIALIZE consM :: IO a -> WAsyncT IO a -> WAsyncT IO a #-}

859− consM = consMSerial

881+ consM m r = fromStream $ S.consMWAsync m (toStream r)

882+

883+ {-# INLINE (|:) #-}

884+ {-# SPECIALIZE (|:) :: IO a -> WAsyncT IO a -> WAsyncT IO a #-}

885+ (|:) = consM

860

886

861------------------------------------------------------------------------------

887------------------------------------------------------------------------------

862-- Semigroup

888-- Semigroup

@@ -992,10 +1018,14 @@ newtype ParallelT m a = ParallelT {getParallelT :: S.Stream m a}

992instance IsStream ParallelT where

1018instance IsStream ParallelT where

993 toStream = getParallelT

1019 toStream = getParallelT

994 fromStream = ParallelT

1020 fromStream = ParallelT

995− cons = consSerial

1021+

996 {-# INLINE consM #-}

1022 {-# INLINE consM #-}

997 {-# SPECIALIZE consM :: IO a -> ParallelT IO a -> ParallelT IO a #-}

1023 {-# SPECIALIZE consM :: IO a -> ParallelT IO a -> ParallelT IO a #-}

998− consM = consMSerial

1024+ consM m r = fromStream $ S.consMParallel m (toStream r)

1025+

1026+ {-# INLINE (|:) #-}

1027+ {-# SPECIALIZE (|:) :: IO a -> ParallelT IO a -> ParallelT IO a #-}

1028+ (|:) = consM

999

1029

1000------------------------------------------------------------------------------

1030------------------------------------------------------------------------------

1001-- Semigroup

1031-- Semigroup

@@ -1068,11 +1098,17 @@ type ZipStream = ZipSerialM

1068instance IsStream ZipSerialM where

1098instance IsStream ZipSerialM where

1069 toStream = getZipSerialM

1099 toStream = getZipSerialM

1070 fromStream = ZipSerialM

1100 fromStream = ZipSerialM

1071− cons = consSerial

1101+

1072 {-# INLINE consM #-}

1102 {-# INLINE consM #-}

1073 {-# SPECIALIZE consM :: IO a -> ZipSerialM IO a -> ZipSerialM IO a #-}

1103 {-# SPECIALIZE consM :: IO a -> ZipSerialM IO a -> ZipSerialM IO a #-}

1104+ consM :: Monad m => m a -> ZipSerialM m a -> ZipSerialM m a

1074 consM = consMSerial

1105 consM = consMSerial

1075

1106

1107+ {-# INLINE (|:) #-}

1108+ {-# SPECIALIZE (|:) :: IO a -> ZipSerialM IO a -> ZipSerialM IO a #-}

1109+ (|:) :: Monad m => m a -> ZipSerialM m a -> ZipSerialM m a

1110+ (|:) = consMSerial

1111+

1076instance Monad m => Applicative (ZipSerialM m) where

1112instance Monad m => Applicative (ZipSerialM m) where

1077 pure = ZipSerialM . S.repeat

1113 pure = ZipSerialM . S.repeat

1078 m1 <*> m2 = fromStream $ S.zipWith id (toStream m1) (toStream m2)

1114 m1 <*> m2 = fromStream $ S.zipWith id (toStream m1) (toStream m2)

@@ -1104,11 +1140,17 @@ newtype ZipAsyncM m a = ZipAsyncM {getZipAsyncM :: S.Stream m a}

1104instance IsStream ZipAsyncM where

1140instance IsStream ZipAsyncM where

1105 toStream = getZipAsyncM

1141 toStream = getZipAsyncM

1106 fromStream = ZipAsyncM

1142 fromStream = ZipAsyncM

1107− cons = consSerial

1143+

1108 {-# INLINE consM #-}

1144 {-# INLINE consM #-}

1109 {-# SPECIALIZE consM :: IO a -> ZipAsyncM IO a -> ZipAsyncM IO a #-}

1145 {-# SPECIALIZE consM :: IO a -> ZipAsyncM IO a -> ZipAsyncM IO a #-}

1146+ consM :: Monad m => m a -> ZipAsyncM m a -> ZipAsyncM m a

1110 consM = consMSerial

1147 consM = consMSerial

1111

1148

1149+ {-# INLINE (|:) #-}

1150+ {-# SPECIALIZE (|:) :: IO a -> ZipAsyncM IO a -> ZipAsyncM IO a #-}

1151+ (|:) :: Monad m => m a -> ZipAsyncM m a -> ZipAsyncM m a

1152+ (|:) = consMSerial

1153+

1112instance MonadAsync m => Applicative (ZipAsyncM m) where

1154instance MonadAsync m => Applicative (ZipAsyncM m) where

1113 pure = ZipAsyncM . S.repeat

1155 pure = ZipAsyncM . S.repeat

1114 m1 <*> m2 = fromStream $ S.zipAsyncWith id (toStream m1) (toStream m2)

1156 m1 <*> m2 = fromStream $ S.zipAsyncWith id (toStream m1) (toStream m2)

src/Streamly/Tutorial.hs+17

@@ -307,6 +307,15 @@ import Control.Monad.Trans.Class (MonadTrans (lift))

307-- @

308--

309-- For more ways to construct a stream see the module "Streamly.Prelude".

310+--

311+-- Note that monadic generation functions e.g. 'consM', 'unfoldrM',

312+-- 'replicateM', 'repeatM', 'iterateM' and 'fromFoldableM' etc work

313+-- concurrently when used with appropriate stream type combinator. The pure

314+-- versions do not work concurrently, however you can use the monadic versions

315+-- even for pure computations to get the concurrent generation capability where

316+-- required.

317+--

318+-- TBD add more details and examples about concurrent generation.

310

319

311-- $eliminating

320-- $eliminating

312--

321--

@@ -353,6 +362,14 @@ import Control.Monad.Trans.Class (MonadTrans (lift))

353-- & S.'zipWith' (\\n s -> show n ++ " " ++ s) (S.'fromFoldable' [1..])

362-- & S.'zipWith' (\\n s -> show n ++ " " ++ s) (S.'fromFoldable' [1..])

354-- & S.'mapM' putStrLn

363-- & S.'mapM' putStrLn

355-- @

364-- @

365+--

366+-- Note that monadic transformation functions 'mapM' and 'sequence' work

367+-- concurrently when used with appropriate stream type combinator. The pure

368+-- versions do not work concurrently, however you can use the monadic versions

369+-- even for pure computations to get the concurrent transformation capability

370+-- where required.

371+--

372+-- TBD add more details and examples about concurrent transformation.

356

373

357-- $semigroup

374-- $semigroup

358--

375--

streamly.cabal+1

@@ -171,6 +171,7 @@ test-suite properties

171 hs-source-dirs: test

172 ghc-options: -O0 -Wall -threaded -with-rtsopts=-N4

173 if flag(dev)

174+ cpp-options: -DDEVBUILD

174 ghc-options: -Wmissed-specialisations

175 ghc-options: -Wmissed-specialisations

175 -Wall-missed-specialisations

176 -Wall-missed-specialisations

176 if impl(ghc >= 8.0)

177 if impl(ghc >= 8.0)

test/Prop.hs+158−26

@@ -2,14 +2,17 @@

3module Main (main) where

5+import Control.Exception (BlockedIndefinitelyOnMVar(..), catch)

5import Control.Monad (when)

6import Control.Monad (when)

6import Control.Applicative (ZipList(..))

7import Control.Applicative (ZipList(..))

8+import Control.Concurrent (MVar, takeMVar, putMVar, newEmptyMVar)

7−import Control.Monad (replicateM)

9+import Control.Monad (replicateM, replicateM_)

10+import Control.Monad.IO.Class (liftIO)

8import Data.List (sort, foldl', scanl')

11import Data.List (sort, foldl', scanl')

9import GHC.Word (Word8)

12import GHC.Word (Word8)

11import Test.Hspec.QuickCheck (prop)

14import Test.Hspec.QuickCheck (prop)

12−import Test.QuickCheck (counterexample, Property)

15+import Test.QuickCheck (counterexample, Property, withMaxSuccess)

13import Test.QuickCheck.Monadic (run, monadicIO, monitor, assert, PropertyM)

16import Test.QuickCheck.Monadic (run, monadicIO, monitor, assert, PropertyM)

15import Test.Hspec

18import Test.Hspec

@@ -18,6 +21,13 @@ import Streamly

18import Streamly.Prelude ((.:), nil)

21import Streamly.Prelude ((.:), nil)

19import qualified Streamly.Prelude as A

22import qualified Streamly.Prelude as A

24+maxTestCount :: Int

25+#ifdef DEVBUILD

26+maxTestCount = 100

27+#else

28+maxTestCount = 10

29+#endif

30+

21singleton :: IsStream t => a -> t m a

31singleton :: IsStream t => a -> t m a

22singleton a = a .: nil

32singleton a = a .: nil

@@ -59,6 +69,91 @@ transformFromList constr eq listOp op a =

59 let list = listOp a

69 let list = listOp a

60 equals eq stream list

70 equals eq stream list

72+dbgMVar :: String -> IO a -> IO a

73+dbgMVar label action =

74+ action `catch` \BlockedIndefinitelyOnMVar ->

75+ error $ label ++ " " ++ "BlockedIndefinitelyOnMVar"

76+

77+-- | first n actions takeMVar and the last action performs putMVar n times

78+mvarSequenceOp :: MVar () -> Word8 -> Word8 -> IO Word8

79+mvarSequenceOp mv n x = do

80+ let msg = show x ++ "/" ++ show n

81+ if x < n

82+ then dbgMVar ("take mvarSequenceOp " ++ msg) (takeMVar mv) >> return x

83+ else dbgMVar ("put mvarSequenceOp" ++ msg)

84+ (replicateM_ (fromIntegral n) (putMVar mv ())) >> return x

85+

86+concurrentMapM

87+ :: ([Word8] -> t IO Word8)

88+ -> ([Word8] -> [Word8] -> Bool)

89+ -> (Word8 -> MVar () -> t IO Word8 -> SerialT IO Word8)

90+ -> Word8

91+ -> Property

92+concurrentMapM constr eq op n =

93+ monadicIO $ do

94+ let list = [0..n]

95+ stream <- run $ do

96+ mv <- newEmptyMVar :: IO (MVar ())

97+ (A.toList . (op n mv)) (constr list)

98+ equals eq stream list

99+

100+concurrentFromFoldable

101+ :: IsStream t

102+ => ([Word8] -> [Word8] -> Bool)

103+ -> (t IO Word8 -> SerialT IO Word8)

104+ -> Word8

105+ -> Property

106+concurrentFromFoldable eq op n =

107+ monadicIO $ do

108+ let list = [0..n]

109+ stream <- run $ do

110+ mv <- newEmptyMVar :: IO (MVar ())

111+ (A.toList . op) (A.fromFoldableM (map (mvarSequenceOp mv n) list))

112+ equals eq stream list

113+

114+sourceUnfoldrM :: IsStream t => MVar () -> Word8 -> t IO Word8

115+sourceUnfoldrM mv n = A.unfoldrM step 0

116+ where

117+ -- argument must be integer to avoid overflow of word8 at 255

118+ step :: Int -> IO (Maybe (Word8, Int))

119+ step cnt = do

120+ let msg = show cnt ++ "/" ++ show n

121+ if cnt > fromIntegral n

122+ then return Nothing

123+ else do

124+ liftIO $ dbgMVar ("put sourceUnfoldrM " ++ msg) (putMVar mv ())

125+ return (Just (fromIntegral cnt, cnt + 1))

126+

127+concurrentUnfoldrM

128+ :: IsStream t

129+ => ([Word8] -> [Word8] -> Bool)

130+ -> (t IO Word8 -> SerialT IO Word8)

131+ -> Word8

132+ -> Property

133+concurrentUnfoldrM eq op n =

134+ monadicIO $ do

135+ let list = [0..n]

136+ stream <- run $ do

137+ -- putStrLn $ "concurrentUnfoldrM: " ++ show n

138+ mv <- newEmptyMVar :: IO (MVar ())

139+ -- since unfoldr happens in parallel with the stream processing we

140+ -- can do two takeMVar in one iteration. If it is not parallel then

141+ -- this will not work and the test will fail.

142+ A.toList $ do

143+ x <- op (sourceUnfoldrM mv n)

144+ let msg = show x ++ "/" ++ show n

145+ if even x

146+ then do

147+ liftIO $ dbgMVar ("first take concurrentUnfoldrM " ++ msg)

148+ (takeMVar mv)

149+ if n > x

150+ then liftIO $ dbgMVar ("second take concurrentUnfoldrM " ++ msg)

151+ (takeMVar mv)

152+ else return ()

153+ else return ()

154+ return x

155+ equals eq stream list

156+

62foldFromList

157foldFromList

63 :: ([Int] -> t IO Int)

158 :: ([Int] -> t IO Int)

64 -> (t IO Int -> SerialT IO Int)

159 -> (t IO Int -> SerialT IO Int)

@@ -147,7 +242,29 @@ transformOps constr desc t eq = do

147 prop (desc ++ " dropWhile > 0") $

242 prop (desc ++ " dropWhile > 0") $

148 transform (dropWhile (> 0)) $ t . (A.dropWhile (> 0))

243 transform (dropWhile (> 0)) $ t . (A.dropWhile (> 0))

149 prop (desc ++ " scan") $ transform (scanl' (+) 0) $ t . (A.scanl' (+) 0)

244 prop (desc ++ " scan") $ transform (scanl' (+) 0) $ t . (A.scanl' (+) 0)

150− prop (desc ++ "reverse") $ transform reverse $ t . A.reverse

245+ prop (desc ++ " reverse") $ transform reverse $ t . A.reverse

246+

247+concurrentOps

248+ :: IsStream t

249+ => ([Word8] -> t IO Word8)

250+ -> String

251+ -> (t IO Word8 -> SerialT IO Word8)

252+ -> ([Word8] -> [Word8] -> Bool)

253+ -> Spec

254+concurrentOps constr desc t eq = do

255+ prop (desc ++ " fromFoldableM") $ withMaxSuccess maxTestCount $

256+ concurrentFromFoldable eq t

257+ prop (desc ++ " unfoldrM") $ withMaxSuccess maxTestCount $

258+ concurrentUnfoldrM eq t

259+ -- we pass it the length of the stream n and an mvar mv.

260+ -- The stream is [0..n]. The threads communicate in such a way that the

261+ -- actions coming first in the stream are dependent on the last action. So

262+ -- if the stream is not processed concurrently it will block forever.

263+ -- Note that if the size of the stream is bigger than the thread limit

264+ -- then it will block even if it is concurrent.

265+ prop (desc ++ " mapM") $ withMaxSuccess maxTestCount $

266+ concurrentMapM constr eq $ \n mv stream ->

267+ t $ A.mapM (mvarSequenceOp mv n) stream

151

268

152wrapMaybe :: Eq a1 => ([a1] -> a2) -> [a1] -> Maybe a2

269wrapMaybe :: Eq a1 => ([a1] -> a2) -> [a1] -> Maybe a2

153wrapMaybe f =

270wrapMaybe f =

@@ -223,10 +340,11 @@ applicativeOps

223 -> ([(Int, Int)] -> [(Int, Int)] -> Bool)

340 -> ([(Int, Int)] -> [(Int, Int)] -> Bool)

224 -> ([Int], [Int])

341 -> ([Int], [Int])

225 -> Property

342 -> Property

226−applicativeOps constr t eq (a, b) = monadicIO $ do

343+applicativeOps constr t eq (a, b) = withMaxSuccess maxTestCount $

344+ monadicIO $ do

227− stream <- run ((A.toList . t) ((,) <$> (constr a) <*> (constr b)))

345+ stream <- run ((A.toList . t) ((,) <$> (constr a) <*> (constr b)))

228− let list = (,) <$> a <*> b

346+ let list = (,) <$> a <*> b

229− equals eq stream list

347+ equals eq stream list

230

348

231zipApplicative

349zipApplicative

232 :: (IsStream t, Applicative (t IO))

350 :: (IsStream t, Applicative (t IO))

@@ -235,14 +353,15 @@ zipApplicative

235 -> ([(Int, Int)] -> [(Int, Int)] -> Bool)

353 -> ([(Int, Int)] -> [(Int, Int)] -> Bool)

236 -> ([Int], [Int])

354 -> ([Int], [Int])

237 -> Property

355 -> Property

238−zipApplicative constr t eq (a, b) = monadicIO $ do

356+zipApplicative constr t eq (a, b) = withMaxSuccess maxTestCount $

357+ monadicIO $ do

239− stream1 <- run ((A.toList . t) ((,) <$> (constr a) <*> (constr b)))

358+ stream1 <- run ((A.toList . t) ((,) <$> (constr a) <*> (constr b)))

240− stream2 <- run ((A.toList . t) (pure (,) <*> (constr a) <*> (constr b)))

359+ stream2 <- run ((A.toList . t) (pure (,) <*> (constr a) <*> (constr b)))

241− stream3 <- run ((A.toList . t) (A.zipWith (,) (constr a) (constr b)))

360+ stream3 <- run ((A.toList . t) (A.zipWith (,) (constr a) (constr b)))

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

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

243− equals eq stream1 list

362+ equals eq stream1 list

244− equals eq stream2 list

363+ equals eq stream2 list

245− equals eq stream3 list

364+ equals eq stream3 list

246

365

247zipMonadic

366zipMonadic

248 :: (IsStream t, Monad (t IO))

367 :: (IsStream t, Monad (t IO))

@@ -251,7 +370,7 @@ zipMonadic

251 -> ([(Int, Int)] -> [(Int, Int)] -> Bool)

370 -> ([(Int, Int)] -> [(Int, Int)] -> Bool)

252 -> ([Int], [Int])

371 -> ([Int], [Int])

253 -> Property

372 -> Property

254−zipMonadic constr t eq (a, b) =

373+zipMonadic constr t eq (a, b) = withMaxSuccess maxTestCount $

255 monadicIO $ do

374 monadicIO $ do

256 stream1 <-

375 stream1 <-

257 run

376 run

@@ -272,7 +391,7 @@ monadThen

272 -> ([Int] -> [Int] -> Bool)

391 -> ([Int] -> [Int] -> Bool)

273 -> ([Int], [Int])

392 -> ([Int], [Int])

274 -> Property

393 -> Property

275−monadThen constr t eq (a, b) = monadicIO $ do

394+monadThen constr t eq (a, b) = withMaxSuccess maxTestCount $ monadicIO $ do

276 stream <- run ((A.toList . t) ((constr a) >> (constr b)))

395 stream <- run ((A.toList . t) ((constr a) >> (constr b)))

277 let list = a >> b

396 let list = a >> b

278 equals eq stream list

397 equals eq stream list

@@ -284,7 +403,7 @@ monadBind

284 -> ([Int] -> [Int] -> Bool)

403 -> ([Int] -> [Int] -> Bool)

285 -> ([Int], [Int])

404 -> ([Int], [Int])

286 -> Property

405 -> Property

287−monadBind constr t eq (a, b) =

406+monadBind constr t eq (a, b) = withMaxSuccess maxTestCount $

288 monadicIO $ do

407 monadicIO $ do

289 stream <-

408 stream <-

290 run

409 run

@@ -295,6 +414,12 @@ monadBind constr t eq (a, b) =

295

414

296main :: IO ()

415main :: IO ()

297main = hspec $ do

416main = hspec $ do

417+ let folded :: IsStream t => [a] -> t IO a

418+ folded = serially . (\xs ->

419+ case xs of

420+ [x] -> return x -- singleton stream case

421+ _ -> foldMapWith (<>) return xs

422+ )

298 describe "Construction" $ do

423 describe "Construction" $ do

299 -- XXX test for all types of streams

424 -- XXX test for all types of streams

300 prop "serially replicateM" $ constructWithReplicateM serially

425 prop "serially replicateM" $ constructWithReplicateM serially

@@ -307,12 +432,6 @@ main = hspec $ do

307 A.toList . serially . (A.take 100) $ A.iterateM addM (0 :: Int)

432 A.toList . serially . (A.take 100) $ A.iterateM addM (0 :: Int)

308 `shouldReturn` (take 100 $ iterate (+ 1) 0)

433 `shouldReturn` (take 100 $ iterate (+ 1) 0)

309

434

310− let folded :: IsStream t => [a] -> t IO a

311− folded = serially . (\xs ->

312− case xs of

313− [x] -> return x -- singleton stream case

314− _ -> foldMapWith (<>) return xs

315− )

316 describe "Functor operations" $ do

435 describe "Functor operations" $ do

317 functorOps A.fromFoldable "serially" serially (==)

436 functorOps A.fromFoldable "serially" serially (==)

318 functorOps folded "serially folded" serially (==)

437 functorOps folded "serially folded" serially (==)

@@ -358,8 +477,10 @@ main = hspec $ do

358

477

359 describe "Zip operations" $ do

478 describe "Zip operations" $ do

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

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

361− -- XXX this hangs

480+ prop "zipSerially applicative folded" $ zipApplicative folded zipSerially (==)

481+ prop "zipAsyncly applicative" $ zipApplicative A.fromFoldable zipAsyncly (==)

362− -- prop "zipAsyncly applicative" $ zipApplicative zipAsyncly (==)

482+ prop "zipAsyncly applicative folded" $ zipApplicative folded zipAsyncly (==)

483+

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

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

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

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

365 prop "zip monadic aheadly" $ zipMonadic A.fromFoldable aheadly (==)

486 prop "zip monadic aheadly" $ zipMonadic A.fromFoldable aheadly (==)

@@ -432,6 +553,17 @@ main = hspec $ do

432 transformOpsWord8 folded "wAsyncly folded" wAsyncly

553 transformOpsWord8 folded "wAsyncly folded" wAsyncly

433 transformOpsWord8 folded "parallely folded" parallely

554 transformOpsWord8 folded "parallely folded" parallely

434

555

556+ describe "Stream concurrent operations" $ do

557+ concurrentOps A.fromFoldable "aheadly" aheadly (==)

558+ concurrentOps A.fromFoldable "asyncly" asyncly sortEq

559+ concurrentOps A.fromFoldable "wAsyncly" wAsyncly sortEq

560+ concurrentOps A.fromFoldable "parallely" parallely sortEq

561+

562+ concurrentOps folded "aheadly folded" aheadly (==)

563+ concurrentOps folded "asyncly folded" asyncly sortEq

564+ concurrentOps folded "wAsyncly folded" wAsyncly sortEq

565+ concurrentOps folded "parallely folded" parallely sortEq

566+

435 describe "Stream elimination operations" $ do

567 describe "Stream elimination operations" $ do

436 eliminationOps A.fromFoldable "serially" serially

568 eliminationOps A.fromFoldable "serially" serially

437 eliminationOps A.fromFoldable "aheadly" aheadly

569 eliminationOps A.fromFoldable "aheadly" aheadly