1f0dc07 · streamly · git.iko.soy

git.iko.soy/streamly

Merge pull request #72 from composewell/consM

Add consM, fromM and |:, fixes #51

commit1f0dc07343e56ce79bc580202211ba71a079ecc8 authorHarendra Kumar <harendra.kumar@gmail.com> dateThu 3 May 2018, 03:22 (8 years ago) parent20b592a f0fa660

11 files changed+104−74

compare with master

unifiedsplit

examples/loops.hs+11−11 examples/nested-loops.hs+4−4 examples/parallel-loops.hs+5−4 src/Streamly.hs−2 src/Streamly/Core.hs+11−1 src/Streamly/Examples/AcidRainGame.hs+3−3 src/Streamly/Examples/CirclingSquare.hs+3−1 src/Streamly/Examples/MergeSortedStreams.hs+6−5 src/Streamly/Examples/SearchEngineQuery.hs+2−1 src/Streamly/Prelude.hs+3 src/Streamly/Streams.hs+56−42

examples/loops.hs+11−11

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

11import Streamly

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

3−import Streamly.Prelude (nil)

3+import Streamly.Prelude (nil, once)

55main = do

66 hSetBuffering stdout LineBuffering

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

88 putStrLn $ "\nloopTail:\n"

99 runStream $ do

1010 x <- loopTail 0

11− fromIO $ print (x :: Int)

11+ once $ print (x :: Int)

1212

1313 putStrLn $ "\nloopHead:\n"

1414 runStream $ do

1515 x <- loopHead 0

16− fromIO $ print (x :: Int)

16+ once $ print (x :: Int)

1717

1818 putStrLn $ "\nloopTailA:\n"

1919 runStream $ do

2020 x <- loopTailA 0

21− fromIO $ print (x :: Int)

21+ once $ print (x :: Int)

2222

2323 putStrLn $ "\nloopHeadA:\n"

2424 runStream $ do

2525 x <- loopHeadA 0

26− fromIO $ print (x :: Int)

26+ once $ print (x :: Int)

2727

2828 putStrLn $ "\ncosplice:\n"

2929 runStream $ do

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

31− fromIO $ print (x :: Int)

31+ once $ print (x :: Int)

3232

3333 putStrLn $ "\nParallel interleave:\n"

3434 runStream $ do

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

36− fromIO $ print (x :: Int)

36+ once $ print (x :: Int)

3737

3838 where

3939

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

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

5050 loopTail :: Int -> Stream Int

5151 loopTail x = do

52− fromIO $ putStrLn "LoopTail..."

52+ once $ 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

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

5757 -- at all.

5858 loopHead :: Int -> Stream Int

5959 loopHead x = do

60− fromIO $ putStrLn "LoopHead..."

60+ once $ putStrLn "LoopHead..."

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

6262

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

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

7171

7272 loopTailA :: Int -> Stream Int

7373 loopTailA x = do

74− fromIO $ putStrLn "LoopTailA..."

74+ once $ putStrLn "LoopTailA..."

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

7676

7777 loopHeadA :: Int -> Stream Int

7878 loopHeadA x = do

79− fromIO $ putStrLn "LoopHeadA..."

79+ once $ putStrLn "LoopHeadA..."

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

8181

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

examples/nested-loops.hs+4−4

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

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

33import System.Random (randomIO)

44import Streamly

5−import Streamly.Prelude (nil)

5+import Streamly.Prelude (nil, once)

77main = runStream $ do

8− fromIO $ hSetBuffering stdout LineBuffering

8+ once $ hSetBuffering stdout LineBuffering

99 x <- loop "A " 2

1010 y <- loop "B " 2

11− fromIO $ myThreadId >>= putStr . show

11+ once $ myThreadId >>= putStr . show

1212 >> putStr " "

1313 >> print (x, y)

1414

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

1616

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

1818 loop name n = do

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

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

2020 let result = (name ++ show rnd)

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

2222 in (return result) `parallel` repeat

examples/parallel-loops.hs+5−4

@@ -2,11 +2,12 @@ import Control.Concurrent (myThreadId, threadDelay)

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

33import System.Random (randomIO)

44import Streamly

5+import Streamly.Prelude (once)

67main = runStream $ do

7− fromIO $ hSetBuffering stdout LineBuffering

8+ once $ hSetBuffering stdout LineBuffering

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

9− fromIO $ myThreadId >>= putStr . show

10+ once $ myThreadId >>= putStr . show

1011 >> putStr " "

1112 >> print x

1213

@@ -14,6 +15,6 @@ main = runStream $ do

1415

1516 loop :: String -> Stream (String, Int)

1617 loop name = do

17− fromIO $ threadDelay 1000000

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

18+ once $ threadDelay 1000000

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

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

src/Streamly.hs−2

@@ -24,8 +24,6 @@ module Streamly

2424 , Parallel

2525 , ZipStream

2626 , ZipParallel

27− , fromIO

28− , fromM

2927

3028 -- * Stream transformers

3129 , IsStream

src/Streamly/Core.hs+11−1

@@ -26,7 +26,9 @@ module Streamly.Core

2626

2727 -- * Construction

2828 , singleton

29+ , once

2930 , cons

31+ , consM

3032 , repeat

3133 , nil

3234

@@ -175,10 +177,18 @@ newtype Stream m a =

175177nil :: Stream m a

176178nil = Stream $ \_ stp _ _ -> stp

177179

180+once :: Monad m => m a -> Stream m a

181+once m = Stream $ \_ _ single _ -> m >>= single

182+

178183{-# INLINE singleton #-}

184+-- | Same as @once . return@

179185singleton :: a -> Stream m a

180186singleton a = Stream $ \_ _ single _ -> single a

181187

188+consM :: Monad m => m a -> Stream m a -> Stream m a

189+consM m r = Stream $ \_ _ _ yld -> m >>= \a -> yld a r

190+

191+-- | Same as @consM . return@

182192cons :: a -> Stream m a -> Stream m a

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

184194

@@ -597,7 +607,7 @@ alt m1 m2 = Stream $ \_ stp sng yld ->

597607-------------------------------------------------------------------------------

598608

599609instance MonadTrans Stream where

600− lift mx = Stream $ \_ _ single _ -> mx >>= single

610+ lift = once

601611

602612withLocal :: MonadReader r m => (r -> r) -> Stream m a -> Stream m a

603613withLocal f m =

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 $ fromIO askUser

17+userAction = cycle1 $ liftIO askUser

1818 where

1919 askUser = do

2020 command <- getLine

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

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

2525

2626acidRain :: MonadIO m => StreamT m Event

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

27+acidRain = cycle1 $ liftIO (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− fromIO $ putStrLn $ "Health = " ++ show h

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

4040

4141acidRainGame :: IO ()

4242acidRainGame = do

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

@@ -10,6 +10,7 @@ module Streamly.Examples.CirclingSquare where

1010import Data.IORef

1111import Graphics.UI.SDL as SDL

1212import Streamly

13+import Streamly.Prelude (once)

1314import Streamly.Time

1415

1516------------------------------------------------------------------------------

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

8788circlingSquare = do

8889 sdlInit

8990 cref <- newIORef (0,0)

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

91+ runStream $ once (updateController cref)

92+ `parallel` once (updateDisplay cref)

src/Streamly/Examples/MergeSortedStreams.hs+6−5

@@ -6,28 +6,29 @@ import Data.Word

66import System.Random (getStdGen, randoms)

77import Data.List (sort)

88import Streamly

9+import Streamly.Prelude (once)

910import qualified Streamly.Prelude as A

1011

1112getSorted :: Stream Word16

1213getSorted = do

13− g <- fromIO getStdGen

14+ g <- once getStdGen

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

1516 foldMap return (sort ls)

1617

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

1819mergeAsync :: Ord a => Stream a -> Stream a -> Stream a

1920mergeAsync a b = do

20− x <- fromIO $ async a

21− y <- fromIO $ async b

21+ x <- once $ async a

22+ y <- once $ async b

2223 merge x y

2324

2425merge :: Ord a => Stream a -> Stream a -> Stream a

2526merge a b = do

26− a1 <- fromIO $ A.uncons a

27+ a1 <- once $ A.uncons a

2728 case a1 of

2829 Nothing -> b

2930 Just (x, ma) -> do

30− b1 <- fromIO $ A.uncons b

31+ b1 <- once $ A.uncons b

3132 case b1 of

3233 Nothing -> return x <> ma

3334 Just (y, mb) ->

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

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

11module Streamly.Examples.SearchEngineQuery where

33import Streamly

4+import Streamly.Prelude (once)

45import Network.HTTP.Simple

67-- Runs three search engine queries in parallel.

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

1415

1516 where

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

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

18+ get s = once (httpNoBody (parseRequest_ s) >> putStrLn (show s))

1819

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

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

src/Streamly/Prelude.hs+3

@@ -20,6 +20,9 @@ module Streamly.Prelude

2020 -- * Construction and Deconstruction

2121 -- | Primitives to construct or inspect a stream.

2222 nil

23+ , consM

24+ , (|:)

25+ , once

2326 , cons

2427 , (.:)

2528 , uncons

src/Streamly/Streams.hs+56−42

@@ -34,8 +34,11 @@ module Streamly.Streams

3434

3535 -- * Construction

3636 , nil

37+ , once

3738 , cons

3839 , (.:)

40+ , consM

41+ , (|:)

3942 , streamBuild

4043 , fromCallback

4144 , fromSVar

@@ -64,8 +67,6 @@ module Streamly.Streams

6467 , Parallel

6568 , ZipStream

6669 , ZipParallel

67− , fromIO

68− , fromM

6970

7071 -- * Stream Transformers

7172 , StreamT

@@ -122,7 +123,6 @@ import Control.Monad.Trans.Class (MonadTrans (lift))

122123import Data.Semigroup (Semigroup(..))

123124import Prelude hiding (zipWith)

124125import Streamly.Core ( MonadParallel

125− , singleton

126126 , SVar, SVarStyle(..)

127127 , SVarTag(..), SVarSched(..))

128128import qualified Streamly.Core as S

@@ -145,18 +145,60 @@ type Streaming = IsStream

145145-- Constructing a stream

146146------------------------------------------------------------------------------

147147

148−-- | Represesnts an empty stream just like @[]@ represents an empty list.

148+-- | An empty stream.

149+--

150+-- @

151+-- > toList nil

152+-- []

153+-- @

149154nil :: IsStream t => t m a

150155nil = fromStream S.nil

151156

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

158+-- but more efficient.

159+--

160+-- @

161+-- > toList $ once getLine

162+-- hello

163+-- ["hello"]

164+-- @

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

166+once = fromStream . S.once

167+

168+infixr 5 `consM`

169+

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

171+-- stream. For example:

172+--

173+-- @

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

175+-- hello

176+-- world

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

178+-- @

179+consM :: (IsStream t, Monad m) => m a -> t m a -> t m a

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

181+

182+infixr 5 |:

183+

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

185+--

186+-- @

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

188+-- hello

189+-- world

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

191+-- @

192+(|:) :: (IsStream t, Monad m) => m a -> t m a -> t m a

193+(|:) = consM

194+

152195infixr 5 `cons`

153196

154−-- | Constructs a stream by adding a pure value at the head of an existing

155−-- stream, just like ':' constructs lists. For example:

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

198+-- stream. Same as @consM . return@. For example:

156199--

157200-- @

158−-- > let stream = 1 \`cons` 2 \`cons` 3 \`cons` nil

159−-- > (toList . streamly) stream

201+-- > toList $ 1 \`cons` 2 \`cons` 3 \`cons` nil

160202-- [1,2,3]

161203-- @

162204cons :: IsStream t => a -> t m a -> t m a

@@ -164,32 +206,13 @@ cons a r = fromStream $ S.cons a (toStream r)

164206

165207infixr 5 .:

166208

167−-- | Operator equivalent of 'cons' so that you can construct a stream of pure

168−-- values more succinctly like this:

209+-- | Operator equivalent of 'cons'.

169210--

170211-- @

171−-- > let stream = 1 .: 2 .: 3 .: nil

172−-- > (toList . streamly) stream

212+-- > toList $ 1 .: 2 .: 3 .: nil

173213-- [1,2,3]

174214-- @

175215--

176−-- '.:' constructs a stream just like ':' constructs a list.

177−--

178−-- Also note that another equivalent way of building streams from pure values

179−-- is:

180−--

181−-- @

182−-- > let stream = pure 1 <> pure 2 <> pure 3

183−-- > (toList . streamly) stream

184−-- [1,2,3]

185−-- @

186−--

187−-- In the first method we construct a stream by adding one element at a time.

188−-- In the second method we first construct singleton streams using 'pure' and

189−-- then compose all those streams together using the 'Semigroup' style

190−-- composition of streams. The former method is a bit more efficient than the

191−-- latter.

192−--

193216(.:) :: IsStream t => a -> t m a -> t m a

194217(.:) = cons

195218

@@ -285,7 +308,7 @@ async m = do

285308

286309#define MONAD_APPLICATIVE_INSTANCE(STREAM,CONSTRAINT) \

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

288− pure = STREAM . singleton; \

311+ pure = STREAM . S.singleton; \

289312 (<*>) = ap }

290313

291314#define MONAD_COMMON_INSTANCES(STREAM,CONSTRAINT) \

@@ -413,7 +436,7 @@ MONAD_COMMON_INSTANCES(StreamT,)

413436-- yielding one element from each stream alternately.

414437--

415438-- @

416−-- main = ('toList' . 'interleaving' $ (fromFoldable [1,2]) \<\> (fromFoldable [3,4])) >>= print

439+-- main = ('toList' . 'costreamly' $ (fromFoldable [1,2]) \<\> (fromFoldable [3,4])) >>= print

417440-- @

418441-- @

419442-- [1,3,2,4]

@@ -424,7 +447,7 @@ MONAD_COMMON_INSTANCES(StreamT,)

424447--

425448--

426449-- @

427−-- main = 'runStream' . 'interleaving' $ do

450+-- main = 'runStream' . 'costreamly' $ do

428451-- x <- return 1 \<\> return 2

429452-- y <- return 3 \<\> return 4

430453-- liftIO $ print (x, y)

@@ -436,7 +459,7 @@ MONAD_COMMON_INSTANCES(StreamT,)

436459-- (2,4)

437460-- @

438461--

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

462+-- Note that costreamly composition can only combine a finite number of

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

441464--

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

@@ -899,15 +922,6 @@ type ZipStream a = ZipStreamM IO a

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

900923type ZipParallel a = ZipParallelM IO a

901924

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−

911925------------------------------------------------------------------------------

912926-- Fold Utilities

913927------------------------------------------------------------------------------