12b970c · streamly · git.iko.soy

git.iko.soy/streamly

Add stream types specialized to IO monad

commit12b970cdd6ddd745774affcb10ce2f21fb555641 authorHarendra Kumar <harendra.kumar@gmail.com> dateWed 25 Apr 2018, 02:13 (8 years ago) parentdcf4297

10 files changed+121−65

compare with master

unifiedsplit

Changelog.md+1 examples/loops.hs+15−15 examples/nested-loops.hs+3−4 examples/parallel-loops.hs+5−6 src/Streamly.hs+15−3 src/Streamly/Examples/AcidRainGame.hs+3−3 src/Streamly/Examples/CirclingSquare.hs+1−1 src/Streamly/Examples/MergeSortedStreams.hs+10−10 src/Streamly/Examples/SearchEngineQuery.hs+1−1 src/Streamly/Streams.hs+67−22

Changelog.md+1

@@ -69,6 +69,7 @@

6969 * `foldlM'` strict left fold with a monadic fold function

7070 * `splice` run two streams serially one after the other

7171 * `parallel` run two streams in parallel (replaces `<|>`)

72+* Add simpler stream types that are specialized to the IO monad

7273

7374## 0.1.2

7475

examples/loops.hs+15−15

@@ -3,37 +3,37 @@ import System.IO (stdout, hSetBuffering, BufferMode(LineBuffering))

33import Streamly.Prelude (nil)

55main = do

6− liftIO $ hSetBuffering stdout LineBuffering

6+ hSetBuffering stdout LineBuffering

88 putStrLn $ "\nloopTail:\n"

99 runStream $ do

1010 x <- loopTail 0

11− liftIO $ print (x :: Int)

11+ fromIO $ print (x :: Int)

1212

1313 putStrLn $ "\nloopHead:\n"

1414 runStream $ do

1515 x <- loopHead 0

16− liftIO $ print (x :: Int)

16+ fromIO $ print (x :: Int)

1717

1818 putStrLn $ "\nloopTailA:\n"

1919 runStream $ do

2020 x <- loopTailA 0

21− liftIO $ print (x :: Int)

21+ fromIO $ print (x :: Int)

2222

2323 putStrLn $ "\nloopHeadA:\n"

2424 runStream $ do

2525 x <- loopHeadA 0

26− liftIO $ print (x :: Int)

26+ fromIO $ print (x :: Int)

2727

2828 putStrLn $ "\ncosplice:\n"

2929 runStream $ do

3030 x <- (return 0 <> return 1) `cosplice` (return 100 <> return 101)

31− liftIO $ print (x :: Int)

31+ fromIO $ print (x :: Int)

3232

3333 putStrLn $ "\nParallel interleave:\n"

3434 runStream $ do

3535 x <- (return 0 <> return 1) `parallel` (return 100 <> return 101)

36− liftIO $ print (x :: Int)

36+ fromIO $ print (x :: Int)

3737

3838 where

3939

@@ -47,17 +47,17 @@ main = do

4747

4848 -- Generates a value and then loops. Can be used to generate an infinite

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

50− loopTail :: Int -> StreamT IO Int

50+ loopTail :: Int -> Stream Int

5151 loopTail x = do

52− liftIO $ putStrLn "LoopTail..."

52+ fromIO $ putStrLn "LoopTail..."

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

5454

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

5656 -- loop has finished. An infinite generator will not let the consumer run

5757 -- at all.

58− loopHead :: Int -> StreamT IO Int

58+ loopHead :: Int -> Stream Int

5959 loopHead x = do

60− liftIO $ putStrLn "LoopHead..."

60+ fromIO $ putStrLn "LoopHead..."

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

6262

6363-------------------------------------------------------------------------------

@@ -69,14 +69,14 @@ main = do

6969 -- then the action on the right is also spawned concurrently. In other

7070 -- words, both actions may run concurrently based on the need.

7171

72− loopTailA :: Int -> StreamT IO Int

72+ loopTailA :: Int -> Stream Int

7373 loopTailA x = do

74− liftIO $ putStrLn "LoopTailA..."

74+ fromIO $ putStrLn "LoopTailA..."

7575 return x `coparallel` (if x < 3 then loopTailA (x + 1) else nil)

7676

77− loopHeadA :: Int -> StreamT IO Int

77+ loopHeadA :: Int -> Stream Int

7878 loopHeadA x = do

79− liftIO $ putStrLn "LoopHeadA..."

79+ fromIO $ putStrLn "LoopHeadA..."

8080 (if x < 3 then loopHeadA (x + 1) else nil) `coparallel` return x

8181

8282-------------------------------------------------------------------------------

examples/nested-loops.hs+3−4

@@ -1,15 +1,14 @@

11import Control.Concurrent (myThreadId)

2−import Control.Monad.IO.Class (liftIO)

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

43import System.Random (randomIO)

54import Streamly

65import Streamly.Prelude (nil)

87main = runStream $ do

9− liftIO $ hSetBuffering stdout LineBuffering

8+ fromIO $ hSetBuffering stdout LineBuffering

109 x <- loop "A " 2

1110 y <- loop "B " 2

12− liftIO $ myThreadId >>= putStr . show

11+ fromIO $ myThreadId >>= putStr . show

1312 >> putStr " "

1413 >> print (x, y)

1514

@@ -17,7 +16,7 @@ main = runStream $ do

1716

1817 loop :: String -> Int -> StreamT IO String

1918 loop name n = do

20− rnd <- liftIO (randomIO :: IO Int)

19+ rnd <- fromIO (randomIO :: IO Int)

2120 let result = (name ++ show rnd)

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

2322 in (return result) `parallel` repeat

examples/parallel-loops.hs+5−6

@@ -1,20 +1,19 @@

11import Control.Concurrent (myThreadId, threadDelay)

2−import Control.Monad.IO.Class (liftIO)

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

43import System.Random (randomIO)

54import Streamly

76main = runStream $ do

8− liftIO $ hSetBuffering stdout LineBuffering

7+ fromIO $ hSetBuffering stdout LineBuffering

98 x <- loop "A" `parallel` loop "B"

10− liftIO $ myThreadId >>= putStr . show

9+ fromIO $ myThreadId >>= putStr . show

1110 >> putStr " "

1211 >> print x

1312

1413 where

1514

16− loop :: String -> StreamT IO (String, Int)

15+ loop :: String -> Stream (String, Int)

1716 loop name = do

18− liftIO $ threadDelay 1000000

19− rnd <- liftIO (randomIO :: IO Int)

17+ fromIO $ threadDelay 1000000

18+ rnd <- fromIO (randomIO :: IO Int)

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

src/Streamly.hs+15−3

@@ -16,19 +16,31 @@ module Streamly

1616 -- $overview

1717

1818 MonadParallel

19+

20+ -- * IO Streams

21+ , Stream

22+ , Costream

23+ , Coparallel

24+ , Parallel

25+ , ZipStream

26+ , ZipParallel

27+ , fromIO

28+ , fromM

29+

30+ -- * Stream transformers

1931 , IsStream

2032

21− -- * Serial Streams

33+ -- ** Serial Streams

2234 -- $serial

2335 , StreamT

2436 , CostreamT

2537

26− -- * Parallel Streams

38+ -- ** Parallel Streams

2739 -- $parallel

2840 , CoparallelT

2941 , ParallelT

3042

31− -- * Zipping Streams

43+ -- ** Zipping Streams

3244 -- $zipping

3345 , ZipStreamM

3446 , ZipParallelM

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

@@ -14,7 +14,7 @@ import Data.Semigroup (cycle1)

1414data Event = Harm Int | Heal Int | Quit deriving (Show)

1515

1616userAction :: MonadIO m => StreamT m Event

17−userAction = cycle1 $ liftIO askUser

17+userAction = cycle1 $ fromIO askUser

1818 where

1919 askUser = do

2020 command <- getLine

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

2424 _ -> putStrLn "What?" >> askUser

2525

2626acidRain :: MonadIO m => StreamT m Event

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

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

2828

2929game :: (MonadParallel m, MonadState Int m) => StreamT m ()

3030game = do

@@ -36,7 +36,7 @@ game = do

3636

3737 h <- get

3838 when (h <= 0) $ fail "You die!"

39− liftIO $ putStrLn $ "Health = " ++ show h

39+ fromIO $ putStrLn $ "Health = " ++ show h

4040

4141acidRainGame :: IO ()

4242acidRainGame = do

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

@@ -87,4 +87,4 @@ circlingSquare :: IO ()

8787circlingSquare = do

8888 sdlInit

8989 cref <- newIORef (0,0)

90− runStream $ liftIO (updateController cref) `parallel` liftIO (updateDisplay cref)

90+ runStream $ fromIO (updateController cref) `parallel` fromIO (updateDisplay cref)

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

@@ -8,26 +8,26 @@ import Data.List (sort)

88import Streamly

99import qualified Streamly.Prelude as A

1010

11−getSorted :: MonadIO m => StreamT m Word16

11+getSorted :: Stream Word16

1212getSorted = do

13− g <- liftIO getStdGen

13+ g <- fromIO getStdGen

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

15− foldMapWith (<>) return (sort ls)

15+ foldMap return (sort ls)

1616

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

18− => StreamT m a -> StreamT m a -> StreamT m a

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

18+mergeAsync :: Ord a => Stream a -> Stream a -> Stream a

1919mergeAsync a b = do

20− x <- lift $ async a

21− y <- lift $ async b

20+ x <- fromIO $ async a

21+ y <- fromIO $ async b

2222 merge x y

2323

24−merge :: (Ord a, MonadParallel m) => StreamT m a -> StreamT m a -> StreamT m a

24+merge :: Ord a => Stream a -> Stream a -> Stream a

2525merge a b = do

26− a1 <- lift $ A.uncons a

26+ a1 <- fromIO $ A.uncons a

2727 case a1 of

2828 Nothing -> b

2929 Just (x, ma) -> do

30− b1 <- lift $ A.uncons b

30+ b1 <- fromIO $ A.uncons b

3131 case b1 of

3232 Nothing -> return x <> ma

3333 Just (y, mb) ->

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

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

1414

1515 where

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

17− get s = streamly $ liftIO (httpNoBody (parseRequest_ s) >> putStrLn (show s))

17+ get s = fromIO (httpNoBody (parseRequest_ s) >> putStrLn (show s))

1818

1919 google, bing, duckduckgo :: IsStream t => t IO ()

2020 google = get "https://www.google.com/search?q=haskell"

src/Streamly/Streams.hs+67−22

@@ -57,7 +57,17 @@ module Streamly.Streams

5757 , (<=>) --deprecated

5858 , (<|) --deprecated

5959

60− -- * Stream Styles

60+ -- * IO Streams

61+ , Stream

62+ , Costream

63+ , Coparallel

64+ , Parallel

65+ , ZipStream

66+ , ZipParallel

67+ , fromIO

68+ , fromM

69+

70+ -- * Stream Transformers

6171 , StreamT

6272 , CostreamT

6373 , InterleavedT -- deprecated

@@ -111,7 +121,7 @@ import Control.Monad.State.Class (MonadState(..))

111121import Control.Monad.Trans.Class (MonadTrans (lift))

112122import Data.Semigroup (Semigroup(..))

113123import Prelude hiding (zipWith)

114−import Streamly.Core ( MonadParallel, Stream(Stream)

124+import Streamly.Core ( MonadParallel

115125 , singleton

116126 , SVar, SVarStyle(..)

117127 , SVarTag(..), SVarSched(..))

@@ -124,8 +134,8 @@ import qualified Streamly.Core as S

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

125135-- some monad 'm'.

126136class IsStream t where

127− toStream :: t m a -> Stream m a

128− fromStream :: Stream m a -> t m a

137+ toStream :: t m a -> S.Stream m a

138+ fromStream :: S.Stream m a -> t m a

129139

130140-- | Same as 'IsStream'.

131141{-# Deprecated Streaming "Please use IsStream instead." #-}

@@ -195,13 +205,13 @@ streamBuild :: IsStream t

195205 -> m r

196206 -> m r)

197207 -> t m a

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

208+streamBuild k = fromStream $ S.Stream $ \svr stp sng yld ->

199209 let yield a r = yld a (toStream r)

200210 in k svr yield sng stp

201211

202212-- | Build a singleton stream from a callback function.

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

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

214+fromCallback k = fromStream $ S.Stream $ \_ _ sng _ -> k sng

205215

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

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

@@ -358,7 +368,7 @@ instance (MonadState s m CONSTRAINT) => MonadState s (STREAM m) where { \

358368-- Note that serial composition can be used to combine an infinite number of

359369-- streams as it explores only one stream at a time.

360370--

361−newtype StreamT m a = StreamT {getStreamT :: Stream m a}

371+newtype StreamT m a = StreamT {getStreamT :: S.Stream m a}

362372 deriving (Semigroup, Monoid, Functor, MonadTrans)

363373

364374instance IsStream StreamT where

@@ -382,7 +392,7 @@ splice m1 m2 = fromStream $ S.splice (toStream m1) (toStream m2)

382392

383393instance Monad m => Monad (StreamT m) where

384394 return = pure

385− (StreamT (Stream m)) >>= f = StreamT $ Stream $ \_ stp sng yld ->

395+ (StreamT (S.Stream m)) >>= f = StreamT $ S.Stream $ \_ stp sng yld ->

386396 let run x = (S.runStream x) Nothing stp sng yld

387397 single a = run $ toStream (f a)

388398 yield a r = run $ toStream $ f a <> (fromStream r >>= f)

@@ -429,7 +439,7 @@ MONAD_COMMON_INSTANCES(StreamT,)

429439-- Note that interleaving composition can only combine a finite number of

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

431441--

432−newtype CostreamT m a = CostreamT {getCostreamT :: Stream m a}

442+newtype CostreamT m a = CostreamT {getCostreamT :: S.Stream m a}

433443 deriving (Functor, MonadTrans)

434444

435445{-# DEPRECATED InterleavedT "Please use 'CostreamT' instead." #-}

@@ -474,7 +484,7 @@ instance Monoid (CostreamT m a) where

474484

475485instance Monad m => Monad (CostreamT m) where

476486 return = pure

477− (CostreamT (Stream m)) >>= f = CostreamT $ Stream $ \_ stp sng yld ->

487+ (CostreamT (S.Stream m)) >>= f = CostreamT $ S.Stream $ \_ stp sng yld ->

478488 let run x = (S.runStream x) Nothing stp sng yld

479489 single a = run $ toStream (f a)

480490 yield a r = run $ toStream $ f a <> (fromStream r >>= f)

@@ -530,7 +540,7 @@ MONAD_COMMON_INSTANCES(CostreamT,)

530540-- Note that this composition can be used to combine infinite number of streams

531541-- as it explores only a bounded number of streams at a time.

532542--

533−newtype CoparallelT m a = CoparallelT {getCoparallelT :: Stream m a}

543+newtype CoparallelT m a = CoparallelT {getCoparallelT :: S.Stream m a}

534544 deriving (Functor, MonadTrans)

535545

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

@@ -574,14 +584,14 @@ instance MonadParallel m => Monoid (CoparallelT m a) where

574584

575585{-# INLINE parbind #-}

576586parbind

577− :: (forall c. Stream m c -> Stream m c -> Stream m c)

578− -> Stream m a

579− -> (a -> Stream m b)

580− -> Stream m b

587+ :: (forall c. S.Stream m c -> S.Stream m c -> S.Stream m c)

588+ -> S.Stream m a

589+ -> (a -> S.Stream m b)

590+ -> S.Stream m b

581591parbind par m f = go m

582592 where

583− go (Stream g) =

584− Stream $ \ctx stp sng yld ->

593+ go (S.Stream g) =

594+ S.Stream $ \ctx stp sng yld ->

585595 let run x = (S.runStream x) ctx stp sng yld

586596 single a = run $ f a

587597 yield a r = run $ f a `par` go r

@@ -640,7 +650,7 @@ MONAD_COMMON_INSTANCES(CoparallelT, MONADPARALLEL)

640650-- Note that round robin composition can only combine a finite number of

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

642652--

643−newtype ParallelT m a = ParallelT {getParallelT :: Stream m a}

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

644654 deriving (Functor, MonadTrans)

645655

646656instance IsStream ParallelT where

@@ -692,7 +702,7 @@ MONAD_COMMON_INSTANCES(ParallelT, MONADPARALLEL)

692702zipWith :: IsStream t => (a -> b -> c) -> t m a -> t m b -> t m c

693703zipWith f m1 m2 = fromStream $ go (toStream m1) (toStream m2)

694704 where

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

705+ go mx my = S.Stream $ \_ stp sng yld -> do

696706 let merge a ra =

697707 let single2 b = sng (f a b)

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

@@ -718,7 +728,7 @@ zipWith f m1 m2 = fromStream $ go (toStream m1) (toStream m2)

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

719729-- 'StreamT'.

720730--

721−newtype ZipStreamM m a = ZipStreamM {getZipStreamM :: Stream m a}

731+newtype ZipStreamM m a = ZipStreamM {getZipStreamM :: S.Stream m a}

722732 deriving (Functor, Semigroup, Monoid)

723733

724734instance Monad m => Applicative (ZipStreamM m) where

@@ -737,7 +747,7 @@ instance IsStream ZipStreamM where

737747-- generated concurrently) using a pure zipping function.

738748zipParallelWith :: (IsStream t, MonadParallel m)

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

740−zipParallelWith f m1 m2 = fromStream $ Stream $ \_ stp sng yld -> do

750+zipParallelWith f m1 m2 = fromStream $ S.Stream $ \_ stp sng yld -> do

741751 ma <- async m1

742752 mb <- async m2

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

@@ -763,7 +773,7 @@ zipAsyncWith = zipParallelWith

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

764774-- 'StreamT'.

765775--

766−newtype ZipParallelM m a = ZipParallelM {getZipParallelM :: Stream m a}

776+newtype ZipParallelM m a = ZipParallelM {getZipParallelM :: S.Stream m a}

767777 deriving (Functor, Semigroup, Monoid)

768778

769779{-# DEPRECATED ZipAsync "Please use ZipParallelM instead." #-}

@@ -863,6 +873,41 @@ runZipStream = runStream . zipStreamly

863873runZipAsync :: Monad m => ZipParallelM m a -> m ()

864874runZipAsync = runStream . zipParallely

865875

876+------------------------------------------------------------------------------

877+-- IO Streams

878+------------------------------------------------------------------------------

879+

880+-- | A serial IO stream of elements of type @a@. See 'StreamT' documentation

881+-- for more details.

882+type Stream a = StreamT IO a

883+

884+-- | An interleaving serial IO stream of elements of type @a@. See 'CostreamT'

885+-- documentation for more details.

886+type Costream a = CostreamT IO a

887+

888+-- | A demand driven left biased parallelly composing IO stream of elements of

889+-- type @a@. See 'CoparallelT' documentation for more details.

890+type Coparallel a = CoparallelT IO a

891+

892+-- | A round robin parallelly composing IO stream of elements of type @a@.

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

894+type Parallel a = ParallelT IO a

895+

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

897+type ZipStream a = ZipStreamM IO a

898+

899+-- | An IO stream whose applicative instance zips streams parallely.

900+type ZipParallel a = ZipParallelM IO a

901+

902+-- | Lift a value from IO monad to any of the stream types. This works for all

903+-- stream types including those that are not monads e.g. 'ZipStream' and

904+-- 'ZipParallel'.

905+fromIO :: (IsStream t, MonadIO m) => IO a -> t m a

906+fromIO = streamly . liftIO

907+

908+fromM :: (IsStream t, Monad m) => m a -> t m a

909+fromM = streamly . lift

910+

866911------------------------------------------------------------------------------

867912-- Fold Utilities

868913------------------------------------------------------------------------------