9c85811 · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

rename "once" to "yieldM"

9c85811Harendra Kumar8 years ago

Changelog.md+6

compare with master

unifiedsplit

@@ -1,3 +1,9 @@

1+## Unreleased

3+### Deprecations

5+* `once` has been deprecated and renamed to `yieldM`

1## 0.3.0

7## 0.3.0

3### Breaking changes

9### Breaking changes

benchmark/LinearOps.hs+1−1

@@ -89,7 +89,7 @@ sourceFoldMapWith n = S.foldMapWith (S.<>) return [n..n+value]

89{-# INLINE sourceFoldMapWithM #-}

90sourceFoldMapWithM :: (S.IsStream t, Monad m, S.Semigroup (t m Int))

91 => Int -> t m Int

92−sourceFoldMapWithM n = S.foldMapWith (S.<>) (S.once . return) [n..n+value]

92+sourceFoldMapWithM n = S.foldMapWith (S.<>) (S.yieldM . return) [n..n+value]

94{-# INLINE sourceUnfoldr #-}

95sourceUnfoldr :: S.IsStream t => Int -> t m Int

examples/CirclingSquare.hs+3−3

@@ -9,7 +9,7 @@

9import Data.IORef

10import Graphics.UI.SDL as SDL

11import Streamly

12−import Streamly.Prelude (once)

12+import Streamly.Prelude (yieldM)

13import Streamly.Time

15------------------------------------------------------------------------------

@@ -87,5 +87,5 @@ main :: IO ()

87main = do

88 sdlInit

89 cref <- newIORef (0,0)

90− runStream $ once (updateController cref)

90+ runStream $ yieldM (updateController cref)

91− `parallel` once (updateDisplay cref)

91+ `parallel` yieldM (updateDisplay cref)

examples/MergeSort.hs+6−6

@@ -4,29 +4,29 @@ import Data.Word

4import System.Random (getStdGen, randoms)

5import Data.List (sort)

6import Streamly

7−import Streamly.Prelude (once)

7+import Streamly.Prelude (yieldM)

8import qualified Streamly.Prelude as A

10getSorted :: Serial Word16

11getSorted = do

12− g <- once getStdGen

12+ g <- yieldM getStdGen

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

14 foldMap return (sort ls)

16-- | merge two streams generating the elements from each in parallel

17mergeAsync :: Ord a => Serial a -> Serial a -> Serial a

18mergeAsync a b = do

19− x <- once $ mkAsync a

19+ x <- yieldM $ mkAsync a

20− y <- once $ mkAsync b

20+ y <- yieldM $ mkAsync b

21 merge x y

23merge :: Ord a => Serial a -> Serial a -> Serial a

24merge a b = do

25− a1 <- once $ A.uncons a

25+ a1 <- yieldM $ A.uncons a

26 case a1 of

27 Nothing -> b

28 Just (x, ma) -> do

29− b1 <- once $ A.uncons b

29+ b1 <- yieldM $ A.uncons b

30 case b1 of

31 Nothing -> return x <> ma

32 Just (y, mb) ->

examples/SearchQuery.hs+4−3

@@ -1,5 +1,5 @@

1import Streamly

2−import Streamly.Prelude (nil, once, (|:))

2+import Streamly.Prelude (nil, yieldM, (|:))

3import Network.HTTP.Simple

5-- | Runs three search engine queries in parallel and prints the search engine

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

13 runStream . parallely $ google |: bing |: duckduckgo |: nil

15 putStrLn "\nUsing parallel semigroup composition"

16− runStream . parallely $ once google <> once bing <> once duckduckgo

16+ runStream . parallely $ yieldM google <> yieldM bing <> yieldM duckduckgo

18 putStrLn "\nUsing parallel applicative zip"

19− runStream . zipAsyncly $ (,,) <$> once google <*> once bing <*> once duckduckgo

19+ runStream . zipAsyncly $

20+ (,,) <$> yieldM google <*> yieldM bing <*> yieldM duckduckgo

21 where

22 where

22 get :: String -> IO ()

23 get :: String -> IO ()

src/Streamly/Prelude.hs+62−61

@@ -53,7 +53,7 @@ module Streamly.Prelude

53 -- * Special Generation

54 -- | Generate a monadic stream from an input structure, a seed or a

55 -- generation function.

56− , once

56+ , K.yieldM

57 , replicateM

58 , repeatM

59 , iterate

@@ -121,6 +121,7 @@ module Streamly.Prelude

121 , toHandle

122

123 -- * Deprecated

124+ , once

124 , each

125 , each

125 , scan

126 , scan

126 , foldl

127 , foldl

@@ -217,7 +218,7 @@ unfoldrM step = go

217 case mayb of

218 case mayb of

218 Nothing -> stp

219 Nothing -> stp

219 Just (a, b) ->

220 Just (a, b) ->

220− K.runStream (toStream (return a |: go b)) svr stp sng yld

221+ K.unStream (toStream (return a |: go b)) svr stp sng yld

221

222

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

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

223--

224--

@@ -310,7 +311,7 @@ iterateM step = go

310 where

311 where

311 go s = fromStream $ K.Stream $ \svr stp sng yld -> do

312 go s = fromStream $ K.Stream $ \svr stp sng yld -> do

312 next <- step s

313 next <- step s

313− K.runStream (toStream (return s |: go next)) svr stp sng yld

314+ K.unStream (toStream (return s |: go next)) svr stp sng yld

314

315

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

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

316--

317--

@@ -344,8 +345,8 @@ foldr step acc m = go (toStream m)

344 go m1 =

345 go m1 =

345 let stop = return acc

346 let stop = return acc

346 single a = return (step a acc)

347 single a = return (step a acc)

347− yield a r = go r >>= \b -> return (step a b)

348+ yieldk a r = go r >>= \b -> return (step a b)

348− in (K.runStream m1) Nothing stop single yield

349+ in (K.unStream m1) Nothing stop single yieldk

349

350

350-- | Lazy right fold with a monadic step function. For example, to fold a

351-- | Lazy right fold with a monadic step function. For example, to fold a

351-- stream into a list:

352-- stream into a list:

@@ -363,8 +364,8 @@ foldrM step acc m = go (toStream m)

363 go m1 =

364 go m1 =

364 let stop = return acc

365 let stop = return acc

365 single a = step a acc

366 single a = step a acc

366− yield a r = go r >>= step a

367+ yieldk a r = go r >>= step a

367− in (K.runStream m1) Nothing stop single yield

368+ in (K.unStream m1) Nothing stop single yieldk

368

369

369-- | Strict left scan with an extraction function. Like 'scanl'', but applies a

370-- | Strict left scan with an extraction function. Like 'scanl'', but applies a

370-- user supplied extraction function (the third argument) at each step. This is

371-- user supplied extraction function (the third argument) at each step. This is

@@ -379,10 +380,10 @@ scanx step begin done m =

379 where

380 where

380 go m1 !acc = K.Stream $ \_ stp sng yld ->

381 go m1 !acc = K.Stream $ \_ stp sng yld ->

381 let single a = sng (done $ step acc a)

382 let single a = sng (done $ step acc a)

382− yield a r =

383+ yieldk a r =

383 let s = step acc a

384 let s = step acc a

384 in yld (done s) (go r s)

385 in yld (done s) (go r s)

385− in K.runStream m1 Nothing stp single yield

386+ in K.unStream m1 Nothing stp single yieldk

386

387

387-- |

388-- |

388-- @since 0.1.1

389-- @since 0.1.1

@@ -413,7 +414,7 @@ foldx step begin done m = get $ go (toStream m) begin

413 {-# NOINLINE get #-}

414 {-# NOINLINE get #-}

414 get m1 =

415 get m1 =

415 let single = return . done

416 let single = return . done

416− in (K.runStream m1) Nothing undefined single undefined

417+ in (K.unStream m1) Nothing undefined single undefined

417

418

418 -- Note, this can be implemented by making a recursive call to "go",

419 -- Note, this can be implemented by making a recursive call to "go",

419 -- however that is more expensive because of unnecessary recursion

420 -- however that is more expensive because of unnecessary recursion

@@ -422,10 +423,10 @@ foldx step begin done m = get $ go (toStream m) begin

422 go m1 !acc = K.Stream $ \_ _ sng yld ->

423 go m1 !acc = K.Stream $ \_ _ sng yld ->

423 let stop = sng acc

424 let stop = sng acc

424 single a = sng $ step acc a

425 single a = sng $ step acc a

425− yield a r =

426+ yieldk a r =

426 let stream = go r (step acc a)

427 let stream = go r (step acc a)

427− in (K.runStream stream) Nothing undefined sng yld

428+ in (K.unStream stream) Nothing undefined sng yld

428− in (K.runStream m1) Nothing stop single yield

429+ in (K.unStream m1) Nothing stop single yieldk

429

430

430-- |

431-- |

431-- @since 0.1.0

432-- @since 0.1.0

@@ -450,8 +451,8 @@ foldxM step begin done m = go begin (toStream m)

450 go !acc m1 =

451 go !acc m1 =

451 let stop = acc >>= done

452 let stop = acc >>= done

452 single a = acc >>= \b -> step b a >>= done

453 single a = acc >>= \b -> step b a >>= done

453− yield a r = acc >>= \b -> go (step b a) r

454+ yieldk a r = acc >>= \b -> go (step b a) r

454− in (K.runStream m1) Nothing stop single yield

455+ in (K.unStream m1) Nothing stop single yieldk

455

456

456-- |

457-- |

457-- @since 0.1.0

458-- @since 0.1.0

@@ -474,8 +475,8 @@ uncons :: (IsStream t, Monad m) => SerialT m a -> m (Maybe (a, t m a))

474uncons m =

475uncons m =

475 let stop = return Nothing

476 let stop = return Nothing

476 single a = return (Just (a, nil))

477 single a = return (Just (a, nil))

477− yield a r = return (Just (a, fromStream r))

478+ yieldk a r = return (Just (a, fromStream r))

478− in (K.runStream (toStream m)) Nothing stop single yield

479+ in (K.unStream (toStream m)) Nothing stop single yieldk

479

480

480-- | Write a stream of Strings to an IO Handle.

481-- | Write a stream of Strings to an IO Handle.

481--

482--

@@ -486,8 +487,8 @@ toHandle h m = go (toStream m)

486 go m1 =

487 go m1 =

487 let stop = return ()

488 let stop = return ()

488 single a = liftIO (IO.hPutStrLn h a)

489 single a = liftIO (IO.hPutStrLn h a)

489− yield a r = liftIO (IO.hPutStrLn h a) >> go r

490+ yieldk a r = liftIO (IO.hPutStrLn h a) >> go r

490− in (K.runStream m1) Nothing stop single yield

491+ in (K.unStream m1) Nothing stop single yieldk

491

492

492------------------------------------------------------------------------------

493------------------------------------------------------------------------------

493-- Special folds

494-- Special folds

@@ -508,8 +509,8 @@ take :: IsStream t => Int -> t m a -> t m a

508take n m = fromStream $ go n (toStream m)

509take n m = fromStream $ go n (toStream m)

509 where

510 where

510 go n1 m1 = K.Stream $ \_ stp sng yld ->

511 go n1 m1 = K.Stream $ \_ stp sng yld ->

511− let yield a r = yld a (go (n1 - 1) r)

512+ let yieldk a r = yld a (go (n1 - 1) r)

512− in if n1 <= 0 then stp else (K.runStream m1) Nothing stp sng yield

513+ in if n1 <= 0 then stp else (K.unStream m1) Nothing stp sng yieldk

513

514

514-- | Include only those elements that pass a predicate.

515-- | Include only those elements that pass a predicate.

515--

516--

@@ -521,9 +522,9 @@ filter p m = fromStream $ go (toStream m)

521 go m1 = K.Stream $ \_ stp sng yld ->

522 go m1 = K.Stream $ \_ stp sng yld ->

522 let single a | p a = sng a

523 let single a | p a = sng a

523 | otherwise = stp

524 | otherwise = stp

524− yield a r | p a = yld a (go r)

525+ yieldk a r | p a = yld a (go r)

525− | otherwise = (K.runStream r) Nothing stp single yield

526+ | otherwise = (K.unStream r) Nothing stp single yieldk

526− in (K.runStream m1) Nothing stp single yield

527+ in (K.unStream m1) Nothing stp single yieldk

527

528

528-- | End the stream as soon as the predicate fails on an element.

529-- | End the stream as soon as the predicate fails on an element.

529--

530--

@@ -535,9 +536,9 @@ takeWhile p m = fromStream $ go (toStream m)

535 go m1 = K.Stream $ \_ stp sng yld ->

536 go m1 = K.Stream $ \_ stp sng yld ->

536 let single a | p a = sng a

537 let single a | p a = sng a

537 | otherwise = stp

538 | otherwise = stp

538− yield a r | p a = yld a (go r)

539+ yieldk a r | p a = yld a (go r)

539 | otherwise = stp

540 | otherwise = stp

540− in (K.runStream m1) Nothing stp single yield

541+ in (K.unStream m1) Nothing stp single yieldk

541

542

542-- | Discard first 'n' elements from the stream and take the rest.

543-- | Discard first 'n' elements from the stream and take the rest.

543--

544--

@@ -547,11 +548,11 @@ drop n m = fromStream $ go n (toStream m)

547 where

548 where

548 go n1 m1 = K.Stream $ \_ stp sng yld ->

549 go n1 m1 = K.Stream $ \_ stp sng yld ->

549 let single _ = stp

550 let single _ = stp

550− yield _ r = (K.runStream $ go (n1 - 1) r) Nothing stp sng yld

551+ yieldk _ r = (K.unStream $ go (n1 - 1) r) Nothing stp sng yld

551 -- Somehow "<=" check performs better than a ">"

552 -- Somehow "<=" check performs better than a ">"

552 in if n1 <= 0

553 in if n1 <= 0

553− then (K.runStream m1) Nothing stp sng yld

554+ then (K.unStream m1) Nothing stp sng yld

554− else (K.runStream m1) Nothing stp single yield

555+ else (K.unStream m1) Nothing stp single yieldk

555

556

556-- | Drop elements in the stream as long as the predicate succeeds and then

557-- | Drop elements in the stream as long as the predicate succeeds and then

557-- take the rest of the stream.

558-- take the rest of the stream.

@@ -564,9 +565,9 @@ dropWhile p m = fromStream $ go (toStream m)

564 go m1 = K.Stream $ \_ stp sng yld ->

565 go m1 = K.Stream $ \_ stp sng yld ->

565 let single a | p a = stp

566 let single a | p a = stp

566 | otherwise = sng a

567 | otherwise = sng a

567− yield a r | p a = (K.runStream r) Nothing stp single yield

568+ yieldk a r | p a = (K.unStream r) Nothing stp single yieldk

568 | otherwise = yld a r

569 | otherwise = yld a r

569− in (K.runStream m1) Nothing stp single yield

570+ in (K.unStream m1) Nothing stp single yieldk

570

571

571-- | Determine whether all elements of a stream satisfy a predicate.

572-- | Determine whether all elements of a stream satisfy a predicate.

572--

573--

@@ -577,9 +578,9 @@ all p m = go (toStream m)

577 go m1 =

578 go m1 =

578 let single a | p a = return True

579 let single a | p a = return True

579 | otherwise = return False

580 | otherwise = return False

580− yield a r | p a = go r

581+ yieldk a r | p a = go r

581 | otherwise = return False

582 | otherwise = return False

582− in (K.runStream m1) Nothing (return True) single yield

583+ in (K.unStream m1) Nothing (return True) single yieldk

583

584

584-- | Determine whether any of the elements of a stream satisfy a predicate.

585-- | Determine whether any of the elements of a stream satisfy a predicate.

585--

586--

@@ -590,9 +591,9 @@ any p m = go (toStream m)

590 go m1 =

591 go m1 =

591 let single a | p a = return True

592 let single a | p a = return True

592 | otherwise = return False

593 | otherwise = return False

593− yield a r | p a = return True

594+ yieldk a r | p a = return True

594 | otherwise = go r

595 | otherwise = go r

595− in (K.runStream m1) Nothing (return False) single yield

596+ in (K.unStream m1) Nothing (return False) single yieldk

596

597

597-- | Determine the sum of all elements of a stream of numbers

598-- | Determine the sum of all elements of a stream of numbers

598--

599--

@@ -613,8 +614,8 @@ head :: Monad m => SerialT m a -> m (Maybe a)

613head m =

614head m =

614 let stop = return Nothing

615 let stop = return Nothing

615 single a = return (Just a)

616 single a = return (Just a)

616− yield a _ = return (Just a)

617+ yieldk a _ = return (Just a)

617− in (K.runStream (toStream m)) Nothing stop single yield

618+ in (K.unStream (toStream m)) Nothing stop single yieldk

618

619

619-- | Extract all but the first element of the stream, if any.

620-- | Extract all but the first element of the stream, if any.

620--

621--

@@ -623,8 +624,8 @@ tail :: (IsStream t, Monad m) => SerialT m a -> m (Maybe (t m a))

623tail m =

624tail m =

624 let stop = return Nothing

625 let stop = return Nothing

625 single _ = return $ Just nil

626 single _ = return $ Just nil

626− yield _ r = return $ Just $ fromStream r

627+ yieldk _ r = return $ Just $ fromStream r

627− in (K.runStream (toStream m)) Nothing stop single yield

628+ in (K.unStream (toStream m)) Nothing stop single yieldk

628

629

629-- | Extract the last element of the stream, if any.

630-- | Extract the last element of the stream, if any.

630--

631--

@@ -640,8 +641,8 @@ null :: Monad m => SerialT m a -> m Bool

640null m =

641null m =

641 let stop = return True

642 let stop = return True

642 single _ = return False

643 single _ = return False

643− yield _ _ = return False

644+ yieldk _ _ = return False

644− in (K.runStream (toStream m)) Nothing stop single yield

645+ in (K.unStream (toStream m)) Nothing stop single yieldk

645

646

646-- | Determine whether an element is present in the stream.

647-- | Determine whether an element is present in the stream.

647--

648--

@@ -652,8 +653,8 @@ elem e m = go (toStream m)

652 go m1 =

653 go m1 =

653 let stop = return False

654 let stop = return False

654 single a = return (a == e)

655 single a = return (a == e)

655− yield a r = if a == e then return True else go r

656+ yieldk a r = if a == e then return True else go r

656− in (K.runStream m1) Nothing stop single yield

657+ in (K.unStream m1) Nothing stop single yieldk

657

658

658-- | Determine whether an element is not present in the stream.

659-- | Determine whether an element is not present in the stream.

659--

660--

@@ -664,8 +665,8 @@ notElem e m = go (toStream m)

664 go m1 =

665 go m1 =

665 let stop = return True

666 let stop = return True

666 single a = return (a /= e)

667 single a = return (a /= e)

667− yield a r = if a == e then return False else go r

668+ yieldk a r = if a == e then return False else go r

668− in (K.runStream m1) Nothing stop single yield

669+ in (K.unStream m1) Nothing stop single yieldk

669

670

670-- | Determine the length of the stream.

671-- | Determine the length of the stream.

671--

672--

@@ -681,11 +682,11 @@ reverse :: (IsStream t) => t m a -> t m a

681reverse m = fromStream $ go K.nil (toStream m)

682reverse m = fromStream $ go K.nil (toStream m)

682 where

683 where

683 go rev rest = K.Stream $ \_ stp sng yld ->

684 go rev rest = K.Stream $ \_ stp sng yld ->

684− let runIt x = K.runStream x Nothing stp sng yld

685+ let runIt x = K.unStream x Nothing stp sng yld

685 stop = runIt rev

686 stop = runIt rev

686 single a = runIt $ a `K.cons` rev

687 single a = runIt $ a `K.cons` rev

687− yield a r = runIt $ go (a `K.cons` rev) r

688+ yieldk a r = runIt $ go (a `K.cons` rev) r

688− in K.runStream rest Nothing stop single yield

689+ in K.unStream rest Nothing stop single yieldk

689

690

690-- XXX replace the recursive "go" with continuation

691-- XXX replace the recursive "go" with continuation

691-- | Determine the minimum element in a stream.

692-- | Determine the minimum element in a stream.

@@ -697,8 +698,8 @@ minimum m = go Nothing (toStream m)

697 go res m1 =

698 go res m1 =

698 let stop = return res

699 let stop = return res

699 single a = return $ min_ a res

700 single a = return $ min_ a res

700− yield a r = go (min_ a res) r

701+ yieldk a r = go (min_ a res) r

701− in (K.runStream m1) Nothing stop single yield

702+ in (K.unStream m1) Nothing stop single yieldk

702

703

703 min_ a res = case res of

704 min_ a res = case res of

704 Nothing -> Just a

705 Nothing -> Just a

@@ -714,8 +715,8 @@ maximum m = go Nothing (toStream m)

714 go res m1 =

715 go res m1 =

715 let stop = return res

716 let stop = return res

716 single a = return $ max_ a res

717 single a = return $ max_ a res

717− yield a r = go (max_ a res) r

718+ yieldk a r = go (max_ a res) r

718− in (K.runStream m1) Nothing stop single yield

719+ in (K.unStream m1) Nothing stop single yieldk

719

720

720 max_ a res = case res of

721 max_ a res = case res of

721 Nothing -> Just a

722 Nothing -> Just a

@@ -745,8 +746,8 @@ mapM f m = go (toStream m)

745 where

746 where

746 go m1 = fromStream $ K.Stream $ \svr stp sng yld ->

747 go m1 = fromStream $ K.Stream $ \svr stp sng yld ->

747 let single a = f a >>= sng

748 let single a = f a >>= sng

748− yield a r = K.runStream (toStream (f a |: (go r))) svr stp sng yld

749+ yieldk a r = K.unStream (toStream (f a |: (go r))) svr stp sng yld

749− in (K.runStream m1) Nothing stp single yield

750+ in (K.unStream m1) Nothing stp single yieldk

750

751

751-- | Map a 'Maybe' returning function to a stream, filter out the 'Nothing'

752-- | Map a 'Maybe' returning function to a stream, filter out the 'Nothing'

752-- elements, and return a stream of values extracted from 'Just'.

753-- elements, and return a stream of values extracted from 'Just'.

@@ -760,10 +761,10 @@ mapMaybe f m = go (toStream m)

760 let single a = case f a of

761 let single a = case f a of

761 Just b -> sng b

762 Just b -> sng b

762 Nothing -> stp

763 Nothing -> stp

763− yield a r = case f a of

764+ yieldk a r = case f a of

764 Just b -> yld b (toStream $ go r)

765 Just b -> yld b (toStream $ go r)

765− Nothing -> (K.runStream r) Nothing stp single yield

766+ Nothing -> (K.unStream r) Nothing stp single yieldk

766− in (K.runStream m1) Nothing stp single yield

767+ in (K.unStream m1) Nothing stp single yieldk

767

768

768-- | Like 'mapMaybe' but maps a monadic function.

769-- | Like 'mapMaybe' but maps a monadic function.

769--

770--

@@ -786,8 +787,8 @@ mapM_ f m = go (toStream m)

786 go m1 =

787 go m1 =

787 let stop = return ()

788 let stop = return ()

788 single a = void (f a)

789 single a = void (f a)

789− yield a r = f a >> go r

790+ yieldk a r = f a >> go r

790− in (K.runStream m1) Nothing stop single yield

791+ in (K.unStream m1) Nothing stop single yieldk

791

792

792-- | Reduce a stream of monadic actions to a stream of the output of those

793-- | Reduce a stream of monadic actions to a stream of the output of those

793-- actions.

794-- actions.

@@ -808,5 +809,5 @@ sequence m = go (toStream m)

808 where

809 where

809 go m1 = fromStream $ K.Stream $ \svr stp sng yld ->

810 go m1 = fromStream $ K.Stream $ \svr stp sng yld ->

810 let single ma = ma >>= sng

811 let single ma = ma >>= sng

811− yield ma r = K.runStream (toStream $ ma |: go r) svr stp sng yld

812+ yieldk ma r = K.unStream (toStream $ ma |: go r) svr stp sng yld

812− in (K.runStream m1) Nothing stp single yield

813+ in (K.unStream m1) Nothing stp single yieldk

src/Streamly/Streams/Ahead.hs+27−27

@@ -141,9 +141,9 @@ workLoopAhead sv q heap = runHeap

141 yieldOutput seqNo a r = do

142 continue <- liftIO $ send sv (ChildYield a)

143 if continue

144− then (runStream r) (Just sv) (runQueueToken seqNo)

144+ then unStream r (Just sv) (runQueueToken seqNo)

145− (singleOutput seqNo)

145+ (singleOutput seqNo)

146− (yieldOutput seqNo)

146+ (yieldOutput seqNo)

147 else liftIO $ do

148 atomicModifyIORefCAS_ heap $ \(h, _) ->

149 (H.insert (Entry seqNo (AheadEntryStream r)) h, seqNo)

@@ -160,15 +160,15 @@ workLoopAhead sv q heap = runHeap

160 Just (m, seqNo) -> do

161 if seqNo == prevSeqNo + 1

162 then

163− (runStream m) (Just sv) (runQueueToken seqNo)

163+ unStream m (Just sv) (runQueueToken seqNo)

164− (singleOutput seqNo)

164+ (singleOutput seqNo)

165− (yieldOutput seqNo)

165+ (yieldOutput seqNo)

166 else do

167 liftIO $ atomicModifyIORefCAS_ heap $ \(h, _) ->

168 (h, prevSeqNo + 1)

169− (runStream m) (Just sv) runHeap

169+ unStream m (Just sv) runHeap

170− (singleToHeap seqNo)

170+ (singleToHeap seqNo)

171− (yieldToHeap seqNo)

171+ (yieldToHeap seqNo)

172 runQueueNoToken = do

173 work <- dequeueAhead q

174 case work of

@@ -176,13 +176,13 @@ workLoopAhead sv q heap = runHeap

176 Just (m, seqNo) -> do

177 if seqNo == 0

178 then

179− (runStream m) (Just sv) (runQueueToken seqNo)

179+ unStream m (Just sv) (runQueueToken seqNo)

180− (singleOutput seqNo)

180+ (singleOutput seqNo)

181− (yieldOutput seqNo)

181+ (yieldOutput seqNo)

182 else

183− (runStream m) (Just sv) runHeap

183+ unStream m (Just sv) runHeap

184− (singleToHeap seqNo)

184+ (singleToHeap seqNo)

185− (yieldToHeap seqNo)

185+ (yieldToHeap seqNo)

186

187 {-# NOINLINE runHeap #-}

188 runHeap = do

@@ -203,9 +203,9 @@ workLoopAhead sv q heap = runHeap

203 case hent of

204 AheadEntryPure a -> singleOutput seqNo a

205 AheadEntryStream r ->

206− (runStream r) (Just sv) (runQueueToken seqNo)

206+ unStream r (Just sv) (runQueueToken seqNo)

207− (singleOutput seqNo)

207+ (singleOutput seqNo)

208− (yieldOutput seqNo)

208+ (yieldOutput seqNo)

209

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

211-- WAhead

@@ -220,11 +220,11 @@ workLoopAhead sv q heap = runHeap

220forkSVarAhead :: MonadAsync m => Stream m a -> Stream m a -> Stream m a

221forkSVarAhead m1 m2 = Stream $ \_ stp sng yld -> do

222 sv <- newAheadVar (concurrently m1 m2) workLoopAhead

223− (runStream (fromSVar sv)) Nothing stp sng yld

223+ unStream (fromSVar sv) Nothing stp sng yld

224 where

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

226 liftIO $ enqueue (fromJust svr) mb

227− (runStream ma) Nothing stp sng yld

227+ unStream ma Nothing stp sng yld

228

229{-# INLINE aheadS #-}

230aheadS :: MonadAsync m => Stream m a -> Stream m a -> Stream m a

@@ -235,14 +235,14 @@ aheadS m1 m2 = Stream $ \svr stp sng yld -> do

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

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

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

238− (runStream m1) Nothing stp sng yld

238+ unStream m1 Nothing stp sng yld

239− _ -> runStream (forkSVarAhead m1 m2) Nothing stp sng yld

239+ _ -> unStream (forkSVarAhead m1 m2) Nothing stp sng yld

240

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

242-- of combining streams using ahead.

243{-# INLINE consMAhead #-}

244consMAhead :: MonadAsync m => m a -> Stream m a -> Stream m a

245−consMAhead m r = once m `aheadS` r

245+consMAhead m r = yieldM m `aheadS` r

246

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

248-- AheadT

@@ -356,10 +356,10 @@ aheadbind m f = go m

356 where

357 go (Stream g) =

358 Stream $ \ctx stp sng yld ->

359− let runIt x = (runStream x) ctx stp sng yld

359+ let run x = unStream x ctx stp sng yld

360− single a = runIt $ f a

360+ single a = run $ f a

361− yield a r = runIt $ f a `aheadS` go r

361+ yieldk a r = run $ f a `aheadS` go r

362− in g Nothing stp single yield

362+ in g Nothing stp single yieldk

363

364instance MonadAsync m => Monad (AheadT m) where

365 return = pure

src/Streamly/Streams/Async.hs+12−12

@@ -64,15 +64,15 @@ import Streamly.SVar

64{-# INLINE runStreamLIFO #-}

65runStreamLIFO :: MonadIO m

66 => SVar Stream m a -> IORef [Stream m a] -> Stream m a -> m () -> m ()

67−runStreamLIFO sv q m stop = runStream m (Just sv) stop single yield

67+runStreamLIFO sv q m stop = unStream m (Just sv) stop single yieldk

68 where

69 single a = do

70 res <- liftIO $ send sv (ChildYield a)

71 if res then stop else liftIO $ sendStop sv

72− yield a r = do

72+ yieldk a r = do

73 res <- liftIO $ send sv (ChildYield a)

74 if res

75− then (runStream r) (Just sv) stop single yield

75+ then (unStream r) (Just sv) stop single yieldk

76 else liftIO $ enqueueLIFO sv q r >> sendStop sv

78-------------------------------------------------------------------------------

@@ -82,12 +82,12 @@ runStreamLIFO sv q m stop = runStream m (Just sv) stop single yield

82{-# INLINE runStreamFIFO #-}

83runStreamFIFO :: MonadIO m

84 => SVar Stream m a -> LinkedQueue (Stream m a) -> Stream m a -> m () -> m ()

85−runStreamFIFO sv q m stop = runStream m (Just sv) stop single yield

85+runStreamFIFO sv q m stop = unStream m (Just sv) stop single yieldk

86 where

87 single a = do

88 res <- liftIO $ send sv (ChildYield a)

89 if res then stop else liftIO $ sendStop sv

90− yield a r = do

90+ yieldk a r = do

91 res <- liftIO $ send sv (ChildYield a)

92 liftIO (enqueueFIFO sv q r)

93 if res then stop else liftIO $ sendStop sv

@@ -278,11 +278,11 @@ forkSVarAsync style m1 m2 = Stream $ \_ stp sng yld -> do

278 AsyncVar -> newAsyncVar (concurrently m1 m2)

279 WAsyncVar -> newWAsyncVar (concurrently m1 m2)

280 _ -> error "illegal svar type"

281− runStream (fromSVar sv) Nothing stp sng yld

281+ unStream (fromSVar sv) Nothing stp sng yld

282 where

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

284 liftIO $ enqueue (fromJust svr) mb

285− (runStream ma) svr stp sng yld

285+ unStream ma svr stp sng yld

286

287{-# INLINE joinStreamVarAsync #-}

288joinStreamVarAsync :: MonadAsync m

@@ -290,8 +290,8 @@ joinStreamVarAsync :: MonadAsync m

290joinStreamVarAsync style m1 m2 = Stream $ \svr stp sng yld ->

291 case svr of

292 Just sv | svarStyle sv == style ->

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

293+ liftIO (enqueue sv m2) >> unStream m1 svr stp sng yld

294− _ -> runStream (forkSVarAsync style m1 m2) Nothing stp sng yld

294+ _ -> unStream (forkSVarAsync style m1 m2) Nothing stp sng yld

295

296------------------------------------------------------------------------------

297-- Semigroup and Monoid style compositions for parallel actions

@@ -323,7 +323,7 @@ async m1 m2 = fromStream $

323-- of combining streams using async.

324{-# INLINE consMAsync #-}

325consMAsync :: MonadAsync m => m a -> Stream m a -> Stream m a

326−consMAsync m r = once m `asyncS` r

326+consMAsync m r = yieldM m `asyncS` r

327

328------------------------------------------------------------------------------

329-- AsyncT

@@ -447,7 +447,7 @@ wAsyncS = joinStreamVarAsync WAsyncVar

447-- of combining streams using wAsync.

448{-# INLINE consMWAsync #-}

449consMWAsync :: MonadAsync m => m a -> Stream m a -> Stream m a

450−consMWAsync m r = once m `wAsyncS` r

450+consMWAsync m r = yieldM m `wAsyncS` r

451

452-- | Polymorphic version of the 'Semigroup' operation '<>' of 'WAsyncT'.

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

@@ -567,4 +567,4 @@ MONAD_COMMON_INSTANCES(WAsyncT, MONADPARALLEL)

567-- @since 0.1.0

568{-# DEPRECATED runAsyncT "Please use 'runStream . asyncly' instead." #-}

569runAsyncT :: Monad m => AsyncT m a -> m ()

570−runAsyncT = run

570+runAsyncT = runStream

src/Streamly/Streams/Instances.hs+1−1

@@ -9,7 +9,7 @@

10#define MONAD_APPLICATIVE_INSTANCE(STREAM,CONSTRAINT) \

11instance (Monad m CONSTRAINT) => Applicative (STREAM m) where { \

12− pure = STREAM . singleton; \

12+ pure = STREAM . yield; \

13 (<*>) = ap }

15#define MONAD_COMMON_INSTANCES(STREAM,CONSTRAINT) \

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

@@ -58,7 +58,7 @@ import Streamly.SVar

59{-# NOINLINE runOne #-}

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

61−runOne sv m = (runStream m) (Just sv) stop single yield

61+runOne sv m = (unStream m) (Just sv) stop single yieldk

63 where

@@ -69,7 +69,7 @@ runOne sv m = (runStream m) (Just sv) stop single yield

69 -- queue and queue it back on that and exit the thread when the outputQueue

70 -- overflows. Parallel is dangerous because it can accumulate unbounded

71 -- output in the buffer.

72− yield a r = void (sendit a) >> runOne sv r

72+ yieldk a r = void (sendit a) >> runOne sv r

74{-# NOINLINE forkSVarPar #-}

75forkSVarPar :: MonadAsync m => Stream m a -> Stream m a -> Stream m a

@@ -77,7 +77,7 @@ forkSVarPar m r = Stream $ \_ stp sng yld -> do

77 sv <- newParallelVar

78 pushWorkerPar sv (runOne sv m)

79 pushWorkerPar sv (runOne sv r)

80− (runStream (fromSVar sv)) Nothing stp sng yld

80+ (unStream (fromSVar sv)) Nothing stp sng yld

82{-# INLINE joinStreamVarPar #-}

83joinStreamVarPar :: MonadAsync m

@@ -85,8 +85,8 @@ joinStreamVarPar :: MonadAsync m

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

86 case svr of

87 Just sv | svarStyle sv == style -> do

88− pushWorkerPar sv (runOne sv m1) >> (runStream m2) svr stp sng yld

88+ pushWorkerPar sv (runOne sv m1) >> (unStream m2) svr stp sng yld

89− _ -> runStream (forkSVarPar m1 m2) Nothing stp sng yld

89+ _ -> unStream (forkSVarPar m1 m2) Nothing stp sng yld

91{-# INLINE parallelStream #-}

92parallelStream :: MonadAsync m => Stream m a -> Stream m a -> Stream m a

@@ -96,7 +96,7 @@ parallelStream = joinStreamVarPar ParallelVar

96-- of combining streams using parallel.

97{-# INLINE consMParallel #-}

98consMParallel :: MonadAsync m => m a -> Stream m a -> Stream m a

99−consMParallel m r = once m `parallelStream` r

99+consMParallel m r = yieldM m `parallelStream` r

100

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

102-- Merges two streams concurrently.

@@ -125,7 +125,7 @@ applyWith :: (IsStream t, MonadAsync m) => (t m a -> t m b) -> t m a -> t m b

125applyWith f m = fromStream $ Stream $ \svr stp sng yld -> do

126 sv <- newParallelVar

127 pushWorkerPar sv (runOne sv (toStream m))

128− runStream (toStream $ f $ fromSVar sv) svr stp sng yld

128+ unStream (toStream $ f $ fromSVar sv) svr stp sng yld

129

130------------------------------------------------------------------------------

131-- Stream runner concurrent function application

@@ -369,4 +369,4 @@ MONAD_COMMON_INSTANCES(ParallelT, MONADPARALLEL)

369-- @since 0.1.0

370{-# DEPRECATED runParallelT "Please use 'runStream . parallely' instead." #-}

371runParallelT :: Monad m => ParallelT m a -> m ()

372−runParallelT = run

372+runParallelT = runStream

src/Streamly/Streams/Prelude.hs+20−66

@@ -21,13 +21,8 @@

21--

22module Streamly.Streams.Prelude

23 (

24− -- * Construction

25− streamBuild

26− , fromCallback

27−

28 -- * Elimination

24 -- * Elimination

29− , streamFold

30− , runStream

25+ runStream

31 , runStreaming -- deprecated

26 , runStreaming -- deprecated

33 -- * Fold Utilities

28 -- * Fold Utilities

@@ -38,11 +33,28 @@ module Streamly.Streams.Prelude

38where

33where

40import Streamly.Streams.Serial (SerialT)

35import Streamly.Streams.Serial (SerialT)

41−import Streamly.SVar (SVar)

42−

43import Streamly.Streams.StreamK hiding (runStream)

36import Streamly.Streams.StreamK hiding (runStream)

44import qualified Streamly.Streams.StreamK as K

37import qualified Streamly.Streams.StreamK as K

39+------------------------------------------------------------------------------

40+-- Eliminating a stream

41+------------------------------------------------------------------------------

42+

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

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

45+-- combinators for example @runStream . 'asyncly'@.

46+--

47+-- @since 0.2.0

48+runStream :: Monad m => SerialT m a -> m ()

49+runStream = K.runStream

50+

51+-- | Same as 'runStream'

52+--

53+-- @since 0.1.0

54+{-# DEPRECATED runStreaming "Please use runStream instead." #-}

55+runStreaming :: (Monad m, IsStream t) => t m a -> m ()

56+runStreaming = runStream . adapt

57+

46------------------------------------------------------------------------------

58------------------------------------------------------------------------------

47-- Fold Utilities

59-- Fold Utilities

48------------------------------------------------------------------------------

60------------------------------------------------------------------------------

@@ -79,30 +91,6 @@ forEachWith :: (IsStream t, Foldable f)

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

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

80forEachWith f xs g = foldr (f . g) nil xs

92forEachWith f xs g = foldr (f . g) nil xs

82−------------------------------------------------------------------------------

83−-- Generation

84−------------------------------------------------------------------------------

85−

86−-- | Build a stream from its church encoding. The function passed maps

87−-- directly to the underlying representation of the stream type. The second

88−-- parameter to the function is the "yield" function yielding a value and the

89−-- remaining stream if any otherwise 'Nothing'. The third parameter is to

90−-- represent an "empty" stream.

91−streamBuild :: IsStream t

92− => (forall r. Maybe (SVar Stream m a)

93− -> (a -> t m a -> m r)

94− -> (a -> m r)

95− -> m r

96− -> m r)

97− -> t m a

98−streamBuild k = fromStream $ Stream $ \svr stp sng yld ->

99− let yield a r = yld a (toStream r)

100− in k svr yield sng stp

101−

102−-- | Build a singleton stream from a callback function.

103−fromCallback :: IsStream t => (forall r. (a -> m r) -> m r) -> t m a

104−fromCallback k = fromStream $ Stream $ \_ _ sng _ -> k sng

105−

106-------------------------------------------------------------------------------

94-------------------------------------------------------------------------------

107-- Generation by unfold

95-- Generation by unfold

108-------------------------------------------------------------------------------

96-------------------------------------------------------------------------------

@@ -128,37 +116,3 @@ unfoldrM step = go

128 Just (a, b) ->

116 Just (a, b) ->

129 K.runStream (a `cons` go b) svr stp sng yld

117 K.runStream (a `cons` go b) svr stp sng yld

130-}

118-}

131−

132−------------------------------------------------------------------------------

133−-- Destroying a stream

134−------------------------------------------------------------------------------

135−

136−-- | Fold a stream using its church encoding. The second argument is the "step"

137−-- function consuming an element and the remaining stream, if any. The third

138−-- argument is for consuming an "empty" stream that yields nothing.

139−streamFold

140− :: IsStream t

141− => Maybe (SVar Stream m a)

142− -> (a -> t m a -> m r)

143− -> (a -> m r)

144− -> m r

145− -> t m a

146− -> m r

147−streamFold svr step single blank m =

148− let yield a x = step a (fromStream x)

149− in (K.runStream (toStream m)) svr blank single yield

150−

151−-- | Run a streaming composition, discard the results. By default it interprets

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

153−-- combinators for example @runStream . 'asyncly'@.

154−--

155−-- @since 0.2.0

156−runStream :: Monad m => SerialT m a -> m ()

157−runStream = run

158−

159−-- | Same as 'runStream'

160−--

161−-- @since 0.1.0

162−{-# DEPRECATED runStreaming "Please use runStream instead." #-}

163−runStreaming :: (Monad m, IsStream t) => t m a -> m ()

164−runStreaming = runStream . adapt

src/Streamly/Streams/SVar.hs+3−3

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

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

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

46 -- streams.

47− runStream (processEvents $ reverse list) Nothing stp sng yld

47+ unStream (processEvents $ reverse list) Nothing stp sng yld

49 where

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

62 done <- postProcess sv

63 if done

64 then allDone stp

65− else runStream (fromStreamVar sv) Nothing stp sng yld

65+ else unStream (fromStreamVar sv) Nothing stp sng yld

67 processEvents (ev : es) = Stream $ \_ stp sng yld -> do

68 let rest = processEvents es

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

71 ChildStop tid e -> do

72 accountThread sv tid

73 case e of

74− Nothing -> runStream rest Nothing stp sng yld

74+ Nothing -> unStream rest Nothing stp sng yld

75 Just ex -> throwM ex

77{-# INLINE fromSVar #-}

src/Streamly/Streams/Serial.hs+14−14

@@ -171,10 +171,10 @@ serial m1 m2 = fromStream $ C.serial (toStream m1) (toStream m2)

171instance Monad m => Monad (SerialT m) where

172 return = pure

173 (SerialT (Stream m)) >>= f = SerialT $ Stream $ \_ stp sng yld ->

174− let runIt x = (runStream x) Nothing stp sng yld

174+ let run x = (unStream x) Nothing stp sng yld

175− single a = runIt $ toStream (f a)

175+ single a = run $ toStream (f a)

176− yield a r = runIt $ toStream $ f a <> (fromStream r >>= f)

176+ yieldk a r = run $ toStream $ f a <> (fromStream r >>= f)

177− in m Nothing stp single yield

177+ in m Nothing stp single yieldk

178

179------------------------------------------------------------------------------

180-- Other instances

@@ -272,10 +272,10 @@ instance IsStream WSerialT where

272{-# INLINE interleave #-}

273interleave :: Stream m a -> Stream m a -> Stream m a

274interleave m1 m2 = Stream $ \_ stp sng yld -> do

275− let stop = (runStream m2) Nothing stp sng yld

275+ let stop = (unStream m2) Nothing stp sng yld

276− single a = yld a m2

276+ single a = yld a m2

277− yield a r = yld a (interleave m2 r)

277+ yieldk a r = yld a (interleave m2 r)

278− (runStream m1) Nothing stop single yield

278+ (unStream m1) Nothing stop single yieldk

279

280-- | Polymorphic version of the 'Semigroup' operation '<>' of 'WSerialT'.

281-- Interleaves two streams, yielding one element from each stream alternately.

@@ -313,10 +313,10 @@ instance Monoid (WSerialT m a) where

313instance Monad m => Monad (WSerialT m) where

314 return = pure

315 (WSerialT (Stream m)) >>= f = WSerialT $ Stream $ \_ stp sng yld ->

316− let runIt x = (runStream x) Nothing stp sng yld

316+ let run x = (unStream x) Nothing stp sng yld

317− single a = runIt $ toStream (f a)

317+ single a = run $ toStream (f a)

318− yield a r = runIt $ toStream $ f a <> (fromStream r >>= f)

318+ yieldk a r = run $ toStream $ f a <> (fromStream r >>= f)

319− in m Nothing stp single yield

319+ in m Nothing stp single yieldk

320

321------------------------------------------------------------------------------

322-- Other instances

@@ -334,11 +334,11 @@ MONAD_COMMON_INSTANCES(WSerialT,)

334-- @since 0.1.0

335{-# DEPRECATED runStreamT "Please use runStream instead." #-}

336runStreamT :: Monad m => SerialT m a -> m ()

337−runStreamT = run

337+runStreamT = runStream

338

339-- | Same as @runStream . wSerially@.

340--

341-- @since 0.1.0

342{-# DEPRECATED runInterleavedT "Please use 'runStream . interleaving' instead." #-}

343runInterleavedT :: Monad m => InterleavedT m a -> m ()

344−runInterleavedT = run

344+runInterleavedT = runStream

src/Streamly/Streams/StreamK.hs+124−47

@@ -20,32 +20,47 @@

20-- Portability : GHC

21--

22--

23+-- Continuation passing style stream implementation.

24+--

25+-- import qualified Streamly.Streams.StreamK as K

26+--

23module Streamly.Streams.StreamK

27module Streamly.Streams.StreamK

24 (

28 (

25 -- * Streams

29 -- * Streams

26 IsStream (..)

30 IsStream (..)

27 , adapt

31 , adapt

28− , Streaming -- deprecated

32+

29 , Stream (..)

33 , Stream (..)

30− , run

34+ , mkStream

35+ , foldStream

36+ , runStream

32 -- * Construction

38 -- * Construction

33 , nil

39 , nil

40+ , yield

41+ , yieldM

34 , cons

42 , cons

35 , (.:)

43 , (.:)

36− , consMSerial

44+

45+ -- * Asynchronous construction

46+ , nilK

47+ , yieldK

48+ , consK

38 -- * Generation

50 -- * Generation

39− , singleton

40− , once

41 , repeat

51 , repeat

43 -- * Semigroup Style Composition

53 -- * Semigroup Style Composition

44 , serial

54 , serial

46 -- * Utilities

56 -- * Utilities

57+ , consMSerial

47 , bindWith

58 , bindWith

48 , withLocal

59 , withLocal

60+

61+ -- * Deprecated

62+ , Streaming -- deprecated

63+ , once -- deprecated

49 )

64 )

50where

65where

@@ -77,7 +92,7 @@ import Streamly.SVar

77--

92--

78newtype Stream m a =

93newtype Stream m a =

79 Stream {

94 Stream {

80− runStream :: forall r.

95+ unStream :: forall r.

81 Maybe (SVar Stream m a) -- local state

96 Maybe (SVar Stream m a) -- local state

82 -> m r -- stop

97 -> m r -- stop

83 -> (a -> m r) -- singleton

98 -> (a -> m r) -- singleton

@@ -152,6 +167,23 @@ type Streaming = IsStream

152adapt :: (IsStream t1, IsStream t2) => t1 m a -> t2 m a

167adapt :: (IsStream t1, IsStream t2) => t1 m a -> t2 m a

153adapt = fromStream . toStream

168adapt = fromStream . toStream

154

169

170+------------------------------------------------------------------------------

171+-- Building a stream

172+------------------------------------------------------------------------------

173+

174+-- | Build a stream from an 'SVar', a stop continuation, a singleton stream

175+-- continuation and a yield continuation.

176+mkStream:: IsStream t

177+ => (forall r. Maybe (SVar Stream m a)

178+ -> m r

179+ -> (a -> m r)

180+ -> (a -> t m a -> m r)

181+ -> m r)

182+ -> t m a

183+mkStream k = fromStream $ Stream $ \svr stp sng yld ->

184+ let yieldk a r = yld a (toStream r)

185+ in k svr stp sng yieldk

186+

155------------------------------------------------------------------------------

187------------------------------------------------------------------------------

156-- Construction

188-- Construction

157------------------------------------------------------------------------------

189------------------------------------------------------------------------------

@@ -167,11 +199,41 @@ adapt = fromStream . toStream

167nil :: IsStream t => t m a

199nil :: IsStream t => t m a

168nil = fromStream $ Stream $ \_ stp _ _ -> stp

200nil = fromStream $ Stream $ \_ stp _ _ -> stp

169

201

202+-- faster than yieldM because there is no bind.

203+-- | Create a singleton stream from a pure value. Same as @yieldM (return a)@

204+-- but more efficient.

205+yield :: IsStream t => a -> t m a

206+yield a = fromStream $ Stream $ \_ _ single _ -> single a

207+

208+-- | Create a singleton stream from a monadic action. Same as @m \`consM` nil@

209+-- but more efficient.

210+--

211+-- @

212+-- > toList $ yieldM getLine

213+-- hello

214+-- ["hello"]

215+-- @

216+--

217+-- @since 0.4.0

218+{-# INLINE yieldM #-}

219+yieldM :: (Monad m, IsStream t) => m a -> t m a

220+yieldM m = fromStream $ Stream $ \_ _ single _ -> m >>= single

221+

222+-- | Same as yieldM

223+--

224+-- @since 0.2.0

225+{-# DEPRECATED once "Please use yieldM instead." #-}

226+{-# INLINE once #-}

227+once :: (Monad m, IsStream t) => m a -> t m a

228+once = yieldM

229+

170infixr 5 `cons`

230infixr 5 `cons`

171

231

172-- faster than consM because there is no bind.

232-- faster than consM because there is no bind.

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

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

174−-- stream. For pure values it can be faster than 'consM'. For example:

234+-- stream. For serial streams this is the same as @(return a) \`consM` r@ but

235+-- more efficient. For concurrent streams this is not concurrent whereas

236+-- 'consM' is concurrent. For example:

175--

237--

176-- @

238-- @

177-- > toList $ 1 \`cons` 2 \`cons` 3 \`cons` nil

239-- > toList $ 1 \`cons` 2 \`cons` 3 \`cons` nil

@@ -210,6 +272,25 @@ infixr 5 .:

210consMSerial :: (Monad m) => m a -> Stream m a -> Stream m a

272consMSerial :: (Monad m) => m a -> Stream m a -> Stream m a

211consMSerial m r = Stream $ \_ _ _ yld -> m >>= \a -> yld a r

273consMSerial m r = Stream $ \_ _ _ yld -> m >>= \a -> yld a r

212

274

275+------------------------------------------------------------------------------

276+-- Asynchronous construction

277+------------------------------------------------------------------------------

278+

279+-- | Make an empty stream from a callback function.

280+nilK :: IsStream t => (forall r. m r -> m r) -> t m a

281+nilK k = fromStream $ Stream $ \_ stp _ _ -> k stp

282+

283+-- | Make a singleton stream from a one shot callback function.

284+yieldK :: IsStream t => (forall r. (a -> m r) -> m r) -> t m a

285+yieldK k = fromStream $ Stream $ \_ _ sng _ -> k sng

286+

287+-- | Construct a stream from a callback function.

288+consK :: IsStream t => (forall r. (a -> m r) -> m r) -> t m a -> t m a

289+consK k r = fromStream $ Stream $ \_ _ _ yld -> k (\x -> yld x (toStream r))

290+

291+-- XXX consK with concurrent callbacks

292+-- XXX Build a stream from a repeating callback function.

293+

213-------------------------------------------------------------------------------

294-------------------------------------------------------------------------------

214-- IsStream Stream

295-- IsStream Stream

215-------------------------------------------------------------------------------

296-------------------------------------------------------------------------------

@@ -228,37 +309,33 @@ instance IsStream Stream where

228 (|:) :: Monad m => m a -> Stream m a -> Stream m a

309 (|:) :: Monad m => m a -> Stream m a -> Stream m a

229 (|:) = consMSerial

310 (|:) = consMSerial

230

311

312+-- | Fold a stream by providing an SVar, a stop continuation, a singleton

313+-- continuation and a yield continuation.

314+foldStream

315+ :: IsStream t

316+ => Maybe (SVar Stream m a)

317+ -> m r

318+ -> (a -> m r)

319+ -> (a -> t m a -> m r)

320+ -> t m a

321+ -> m r

322+foldStream svr blank single step m =

323+ let yieldk a x = step a (fromStream x)

324+ in (unStream (toStream m)) svr blank single yieldk

325+

231−run :: (Monad m, IsStream t) => t m a -> m ()

326+runStream :: (Monad m, IsStream t) => t m a -> m ()

232−run m = go (toStream m)

327+runStream m = go (toStream m)

233 where

328 where

234 go m1 =

329 go m1 =

235 let stop = return ()

330 let stop = return ()

236 single _ = return ()

331 single _ = return ()

237− yield _ r = go (toStream r)

332+ yieldk _ r = go (toStream r)

238− in (runStream m1) Nothing stop single yield

333+ in (unStream m1) Nothing stop single yieldk

239

334

240-------------------------------------------------------------------------------

335-------------------------------------------------------------------------------

241-- Special generation

336-- Special generation

242-------------------------------------------------------------------------------

337-------------------------------------------------------------------------------

243

338

244−-- | Same as @once . return@ but may be faster because there is no bind

245−singleton :: IsStream t => a -> t m a

246−singleton a = fromStream $ Stream $ \_ _ single _ -> single a

247−

248−-- | Create a singleton stream by executing a monadic action once. Same as

249−-- @m \`consM` nil@ but more efficient.

250−--

251−-- @

252−-- > toList $ once getLine

253−-- hello

254−-- ["hello"]

255−-- @

256−--

257−-- @since 0.2.0

258−{-# INLINE once #-}

259−once :: (Monad m, IsStream t) => m a -> t m a

260−once m = fromStream $ Stream $ \_ _ single _ -> m >>= single

261−

262repeat :: IsStream t => a -> t m a

339repeat :: IsStream t => a -> t m a

263repeat a = let x = cons a x in x

340repeat a = let x = cons a x in x

264

341

@@ -273,10 +350,10 @@ serial :: Stream m a -> Stream m a -> Stream m a

273serial m1 m2 = go m1

350serial m1 m2 = go m1

274 where

351 where

275 go (Stream m) = Stream $ \_ stp sng yld ->

352 go (Stream m) = Stream $ \_ stp sng yld ->

276− let stop = (runStream m2) Nothing stp sng yld

353+ let stop = (unStream m2) Nothing stp sng yld

277− single a = yld a m2

354+ single a = yld a m2

278− yield a r = yld a (go r)

355+ yieldk a r = yld a (go r)

279− in m Nothing stop single yield

356+ in m Nothing stop single yieldk

280

357

281instance Semigroup (Stream m a) where

358instance Semigroup (Stream m a) where

282 (<>) = serial

359 (<>) = serial

@@ -295,9 +372,9 @@ instance Monoid (Stream m a) where

295

372

296instance Monad m => Functor (Stream m) where

373instance Monad m => Functor (Stream m) where

297 fmap f m = Stream $ \_ stp sng yld ->

374 fmap f m = Stream $ \_ stp sng yld ->

298− let single = sng . f

375+ let single = sng . f

299− yield a r = yld (f a) (fmap f r)

376+ yieldk a r = yld (f a) (fmap f r)

300− in (runStream m) Nothing stp single yield

377+ in (unStream m) Nothing stp single yieldk

301

378

302-------------------------------------------------------------------------------

379-------------------------------------------------------------------------------

303-- Bind utility

380-- Bind utility

@@ -313,10 +390,10 @@ bindWith par m f = go m

313 where

390 where

314 go (Stream g) =

391 go (Stream g) =

315 Stream $ \ctx stp sng yld ->

392 Stream $ \ctx stp sng yld ->

316− let runIt x = (runStream x) ctx stp sng yld

393+ let run x = (unStream x) ctx stp sng yld

317− single a = runIt $ f a

394+ single a = run $ f a

318− yield a r = runIt $ f a `par` go r

395+ yieldk a r = run $ f a `par` go r

319− in g Nothing stp single yield

396+ in g Nothing stp single yieldk

320

397

321------------------------------------------------------------------------------

398------------------------------------------------------------------------------

322-- Alternative & MonadPlus

399-- Alternative & MonadPlus

@@ -324,8 +401,8 @@ bindWith par m f = go m

324

401

325_alt :: Stream m a -> Stream m a -> Stream m a

402_alt :: Stream m a -> Stream m a -> Stream m a

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

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

327− let stop = runStream m2 Nothing stp sng yld

404+ let stop = unStream m2 Nothing stp sng yld

328− in runStream m1 Nothing stop sng yld

405+ in unStream m1 Nothing stop sng yld

329

406

330------------------------------------------------------------------------------

407------------------------------------------------------------------------------

331-- MonadReader

408-- MonadReader

@@ -335,8 +412,8 @@ withLocal :: MonadReader r m => (r -> r) -> Stream m a -> Stream m a

335withLocal f m =

412withLocal f m =

336 Stream $ \_ stp sng yld ->

413 Stream $ \_ stp sng yld ->

337 let single = local f . sng

414 let single = local f . sng

338− yield a r = local f $ yld a (withLocal f r)

415+ yieldk a r = local f $ yld a (withLocal f r)

339− in (runStream m) Nothing (local f stp) single yield

416+ in (unStream m) Nothing (local f stp) single yieldk

340

417

341------------------------------------------------------------------------------

418------------------------------------------------------------------------------

342-- MonadError

419-- MonadError

@@ -349,9 +426,9 @@ withCatchError

349 => Stream m a -> (e -> Stream m a) -> Stream m a

426 => Stream m a -> (e -> Stream m a) -> Stream m a

350withCatchError m h =

427withCatchError m h =

351 Stream $ \_ stp sng yld ->

428 Stream $ \_ stp sng yld ->

352− let run x = runStream x Nothing stp sng yield

429+ let run x = unStream x Nothing stp sng yieldk

353 handle r = r `catchError` \e -> run $ h e

430 handle r = r `catchError` \e -> run $ h e

354− yield a r = yld a (withCatchError r h)

431+ yieldk a r = yld a (withCatchError r h)

355 in handle $ run m

432 in handle $ run m

356-}

433-}

357

434

@@ -360,4 +437,4 @@ withCatchError m h =

360-------------------------------------------------------------------------------

437-------------------------------------------------------------------------------

361

438

362instance MonadTrans Stream where

439instance MonadTrans Stream where

363− lift = once

440+ lift = yieldM

src/Streamly/Streams/Zip.hs+9−9

@@ -61,10 +61,10 @@ zipWithS f m1 m2 = go m1 m2

61 let merge a ra =

62 let single2 b = sng (f a b)

63 yield2 b rb = yld (f a b) (go ra rb)

64− in (runStream my) Nothing stp single2 yield2

64+ in unStream my Nothing stp single2 yield2

65 let single1 a = merge a nil

66 yield1 a ra = merge a ra

67− (runStream mx) Nothing stp single1 yield1

67+ unStream mx Nothing stp single1 yield1

69-- | Zip two streams serially using a pure zipping function.

70--

@@ -81,13 +81,13 @@ zipWithM f m1 m2 = fromStream $ go (toStream m1) (toStream m2)

81 where

82 go mx my = Stream $ \_ stp sng yld -> do

83 let merge a ra =

84− let runIt x = runStream x Nothing stp sng yld

84+ let runIt x = unStream x Nothing stp sng yld

85 single2 b = runIt $ toStream (f a b)

86 yield2 b rb = runIt $ toStream (f a b) <> go ra rb

87− in (runStream my) Nothing stp single2 yield2

87+ in unStream my Nothing stp single2 yield2

88 let single1 a = merge a nil

89 yield1 a ra = merge a ra

90− (runStream mx) Nothing stp single1 yield1

90+ unStream mx Nothing stp single1 yield1

92------------------------------------------------------------------------------

93-- Serially Zipping Streams

@@ -168,7 +168,7 @@ zipAsyncWith :: (IsStream t, MonadAsync m)

168zipAsyncWith f m1 m2 = fromStream $ Stream $ \_ stp sng yld -> do

169 ma <- mkAsync m1

170 mb <- mkAsync m2

171− (runStream (toStream (zipWith f ma mb))) Nothing stp sng yld

171+ unStream (toStream (zipWith f ma mb)) Nothing stp sng yld

172

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

174-- concurrently) using a monadic zipping function.

@@ -179,7 +179,7 @@ zipAsyncWithM :: (IsStream t, MonadAsync m)

179zipAsyncWithM f m1 m2 = fromStream $ Stream $ \_ stp sng yld -> do

180 ma <- mkAsync m1

181 mb <- mkAsync m2

182− (runStream (toStream (zipWithM f ma mb))) Nothing stp sng yld

182+ unStream (toStream (zipWithM f ma mb)) Nothing stp sng yld

183

184------------------------------------------------------------------------------

185-- Parallely Zipping Streams

@@ -245,11 +245,11 @@ instance MonadAsync m => Applicative (ZipAsyncM m) where

245-- @since 0.1.0

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

247runZipStream :: Monad m => ZipSerialM m a -> m ()

248−runZipStream = run

248+runZipStream = runStream

249

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

251--

252-- @since 0.1.0

253{-# DEPRECATED runZipAsync "Please use 'runStream . zipAsyncly instead." #-}

254runZipAsync :: Monad m => ZipAsyncM m a -> m ()

255−runZipAsync = run

255+runZipAsync = runStream

test/Main.hs+19−19

@@ -40,7 +40,7 @@ main = hspec $ do

40 it "simple serially" $

41 (runStream . serially) (return (0 :: Int)) `shouldReturn` ()

42 it "simple serially with IO" $

43− (runStream . serially) (A.once $ putStrLn "hello") `shouldReturn` ()

43+ (runStream . serially) (A.yieldM $ putStrLn "hello") `shouldReturn` ()

45 describe "Empty" $ do

46 it "Monoid - mempty" $

@@ -381,12 +381,12 @@ main = hspec $ do

381

382 it "asyncly crosses thread limit (2000 threads)" $

383 runStream (asyncly $ fold $

384− replicate 2000 $ A.once $ threadDelay 1000000)

384+ replicate 2000 $ A.yieldM $ threadDelay 1000000)

385 `shouldReturn` ()

386

387 it "aheadly crosses thread limit (4000 threads)" $

388 runStream (aheadly $ fold $

389− replicate 4000 $ A.once $ threadDelay 1000000)

389+ replicate 4000 $ A.yieldM $ threadDelay 1000000)

390 `shouldReturn` ()

391

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

@@ -578,7 +578,7 @@ nestTwoParallelApp =

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

579

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

581−timed x = A.once (threadDelay (x * 100000)) >> return x

581+timed x = A.yieldM (threadDelay (x * 100000)) >> return x

582

583interleaveCheck :: IsStream t

584 => (t IO Int -> SerialT IO Int)

@@ -602,7 +602,7 @@ parallelCheck t f = do

602 (A.toList . t) (event 4 `f` (event 3 `f` (event 2 `f` event 1)))

603 `shouldReturn` ([1..4])

604

605− where event n = (A.once $ threadDelay (n * 100000)) >> (return n)

605+ where event n = (A.yieldM $ threadDelay (n * 100000)) >> (return n)

606

607compose :: (IsStream t, Semigroup (t IO Int))

608 => (t IO Int -> SerialT IO Int) -> t IO Int -> ([Int] -> [Int]) -> Spec

@@ -682,12 +682,12 @@ loops t tsrt hsrt = do

682 where

683 loopHead x = do

684 -- this print line is important for the test (causes a bind)

685− A.once $ putStrLn "LoopHead..."

685+ A.yieldM $ putStrLn "LoopHead..."

686 t $ (if x < 3 then loopHead (x + 1) else nil) <> return x

687

688 loopTail x = do

689 -- this print line is important for the test (causes a bind)

690− A.once $ putStrLn "LoopTail..."

690+ A.yieldM $ putStrLn "LoopTail..."

691 t $ return x <> (if x < 3 then loopTail (x + 1) else nil)

692

693bindAndComposeSimple

@@ -755,21 +755,21 @@ mixedOps = do

755

756 composeMixed :: SerialT IO Int

757 composeMixed = do

758− A.once $ return ()

758+ A.yieldM $ return ()

759− A.once $ putStr ""

759+ A.yieldM $ putStr ""

760 x <- return 1

761 y <- return 2

762 z <- do

763 x1 <- wAsyncly $ return 1 <> return 2

764− A.once $ return ()

764+ A.yieldM $ return ()

765− A.once $ putStr ""

765+ A.yieldM $ putStr ""

766 y1 <- asyncly $ return 1 <> return 2

767 z1 <- do

768 x11 <- return 1 <> return 2

769 y11 <- asyncly $ return 1 <> return 2

770 z11 <- wSerially $ return 1 <> return 2

771− A.once $ return ()

771+ A.yieldM $ return ()

772− A.once $ putStr ""

772+ A.yieldM $ putStr ""

773 return (x11 + y11 + z11)

774 return (x1 + y1 + z1)

775 return (x + y + z)

@@ -785,21 +785,21 @@ mixedOpsAheadly = do

785

786 composeMixed :: SerialT IO Int

787 composeMixed = do

788− A.once $ return ()

788+ A.yieldM $ return ()

789− A.once $ putStr ""

789+ A.yieldM $ putStr ""

790 x <- return 1

791 y <- return 2

792 z <- do

793 x1 <- wAsyncly $ return 1 <> return 2

794− A.once $ return ()

794+ A.yieldM $ return ()

795− A.once $ putStr ""

795+ A.yieldM $ putStr ""

796 y1 <- aheadly $ return 1 <> return 2

797 z1 <- do

798 x11 <- return 1 <> return 2

799 y11 <- aheadly $ return 1 <> return 2

800 z11 <- parallely $ return 1 <> return 2

801− A.once $ return ()

801+ A.yieldM $ return ()

802− A.once $ putStr ""

802+ A.yieldM $ putStr ""

803 return (x11 + y11 + z11)

804 return (x1 + y1 + z1)

805 return (x + y + z)

test/Prop.hs+3−3

@@ -159,10 +159,10 @@ concurrentUnfoldrM eq op n =

159 -- results may not be yielded in order, in case of

160 -- Async/WAsync/Parallel. So we use an increasing count

161 -- instead.

162− i <- A.once $ readIORef cnt

162+ i <- A.yieldM $ readIORef cnt

163− A.once $ modifyIORef cnt (+1)

163+ A.yieldM $ modifyIORef cnt (+1)

164 let msg = show i ++ "/" ++ show n

165− A.once $ do

165+ A.yieldM $ do

166 if even i

167 then do

168 dbgMVar ("first take concurrentUnfoldrM " ++ msg)

test/loops.hs+11−11

@@ -1,6 +1,6 @@

1import Streamly

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

3−import Streamly.Prelude (nil, once)

3+import Streamly.Prelude (nil, yieldM)

5main = do

6 hSetBuffering stdout LineBuffering

@@ -8,32 +8,32 @@ main = do

8 putStrLn $ "\nloopTail:\n"

9 runStream $ do

10 x <- loopTail 0

11− once $ print (x :: Int)

11+ yieldM $ print (x :: Int)

13 putStrLn $ "\nloopHead:\n"

14 runStream $ do

15 x <- loopHead 0

16− once $ print (x :: Int)

16+ yieldM $ print (x :: Int)

18 putStrLn $ "\nloopTailA:\n"

19 runStream $ do

20 x <- loopTailA 0

21− once $ print (x :: Int)

21+ yieldM $ print (x :: Int)

23 putStrLn $ "\nloopHeadA:\n"

24 runStream $ do

25 x <- loopHeadA 0

26− once $ print (x :: Int)

26+ yieldM $ print (x :: Int)

28 putStrLn $ "\nwSerial:\n"

29 runStream $ do

30 x <- (return 0 <> return 1) `wSerial` (return 100 <> return 101)

31− once $ print (x :: Int)

31+ yieldM $ print (x :: Int)

33 putStrLn $ "\nParallel interleave:\n"

34 runStream $ do

35 x <- (return 0 <> return 1) `wAsync` (return 100 <> return 101)

36− once $ print (x :: Int)

36+ yieldM $ print (x :: Int)

38 where

@@ -45,7 +45,7 @@ main = do

45 -- stream. Interleaves the generator and the consumer.

46 loopTail :: Int -> Serial Int

47 loopTail x = do

48− once $ putStrLn "LoopTail..."

48+ yieldM $ putStrLn "LoopTail..."

49 return x <> (if x < 3 then loopTail (x + 1) else nil)

51 -- Loops and then generates a value. The consumer can run only after the

@@ -53,7 +53,7 @@ main = do

53 -- at all.

54 loopHead :: Int -> Serial Int

55 loopHead x = do

56− once $ putStrLn "LoopHead..."

56+ yieldM $ putStrLn "LoopHead..."

57 (if x < 3 then loopHead (x + 1) else nil) <> return x

59-------------------------------------------------------------------------------

@@ -62,12 +62,12 @@ main = do

63 loopTailA :: Int -> Serial Int

64 loopTailA x = do

65− once $ putStrLn "LoopTailA..."

65+ yieldM $ putStrLn "LoopTailA..."

66 return x `async` (if x < 3 then loopTailA (x + 1) else nil)

68 loopHeadA :: Int -> Serial Int

69 loopHeadA x = do

70− once $ putStrLn "LoopHeadA..."

70+ yieldM $ putStrLn "LoopHeadA..."

71 (if x < 3 then loopHeadA (x + 1) else nil) `async` return x

73-------------------------------------------------------------------------------

test/nested-loops.hs+5−5

@@ -2,23 +2,23 @@ import Control.Concurrent (myThreadId)

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

3import System.Random (randomIO)

4import Streamly

5−import Streamly.Prelude (nil, once)

5+import Streamly.Prelude (nil, yieldM)

7main = runStream $ do

8− once $ hSetBuffering stdout LineBuffering

8+ yieldM $ hSetBuffering stdout LineBuffering

9 x <- loop "A " 2

10 y <- loop "B " 2

11− once $ myThreadId >>= putStr . show

11+ yieldM $ myThreadId >>= putStr . show

12 >> putStr " "

13 >> print (x, y)

15 where

17 -- we can just use

18− -- parallely $ mconcat $ replicate n $ once (...)

18+ -- parallely $ mconcat $ replicate n $ yieldM (...)

19 loop :: String -> Int -> SerialT IO String

20 loop name n = do

21− rnd <- once (randomIO :: IO Int)

21+ rnd <- yieldM (randomIO :: IO Int)

22 let result = (name ++ show rnd)

23 repeat = if n > 1 then loop name (n - 1) else nil

24 in (return result) `wAsync` repeat

test/parallel-loops.hs+5−5

@@ -8,19 +8,19 @@ main = do

8 hSetBuffering stdout LineBuffering

9 runStream $ do

10 x <- S.take 10 $ loop "A" `parallel` loop "B"

11− S.once $ myThreadId >>= putStr . show

11+ S.yieldM $ myThreadId >>= putStr . show

12 >> putStr " got "

13 >> print x

15 where

17 -- we can just use

18− -- parallely $ cycle1 $ once (...)

18+ -- parallely $ cycle1 $ yieldM (...)

19 loop :: String -> Serial (String, Int)

20 loop name = do

21− S.once $ threadDelay 1000000

21+ S.yieldM $ threadDelay 1000000

22− rnd <- S.once (randomIO :: IO Int)

22+ rnd <- S.yieldM (randomIO :: IO Int)

23− S.once $ myThreadId >>= putStr . show

23+ S.yieldM $ myThreadId >>= putStr . show

24 >> putStr " yielding "

25 >> print rnd

26 return (name, rnd) `parallel` loop name