7febfe9 · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Add consM, once and |:, fixes #51

7febfe9Harendra Kumar8 years ago

examples/loops.hs+11−11

compare with master

unifiedsplit

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

1import Streamly

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

3−import Streamly.Prelude (nil)

3+import Streamly.Prelude (nil, once)

5main = do

6 hSetBuffering stdout LineBuffering

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

8 putStrLn $ "\nloopTail:\n"

9 runStream $ do

10 x <- loopTail 0

11− fromIO $ print (x :: Int)

11+ once $ print (x :: Int)

13 putStrLn $ "\nloopHead:\n"

14 runStream $ do

15 x <- loopHead 0

16− fromIO $ print (x :: Int)

16+ once $ print (x :: Int)

18 putStrLn $ "\nloopTailA:\n"

19 runStream $ do

20 x <- loopTailA 0

21− fromIO $ print (x :: Int)

21+ once $ print (x :: Int)

23 putStrLn $ "\nloopHeadA:\n"

24 runStream $ do

25 x <- loopHeadA 0

26− fromIO $ print (x :: Int)

26+ once $ print (x :: Int)

28 putStrLn $ "\ncosplice:\n"

29 runStream $ do

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

31− fromIO $ print (x :: Int)

31+ once $ print (x :: Int)

33 putStrLn $ "\nParallel interleave:\n"

34 runStream $ do

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

36− fromIO $ print (x :: Int)

36+ once $ print (x :: Int)

38 where

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

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

50 loopTail :: Int -> Stream Int

51 loopTail x = do

52− fromIO $ putStrLn "LoopTail..."

52+ once $ putStrLn "LoopTail..."

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

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

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

57 -- at all.

58 loopHead :: Int -> Stream Int

59 loopHead x = do

60− fromIO $ putStrLn "LoopHead..."

60+ once $ putStrLn "LoopHead..."

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

63-------------------------------------------------------------------------------

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

72 loopTailA :: Int -> Stream Int

73 loopTailA x = do

74− fromIO $ putStrLn "LoopTailA..."

74+ once $ putStrLn "LoopTailA..."

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

77 loopHeadA :: Int -> Stream Int

78 loopHeadA x = do

79− fromIO $ putStrLn "LoopHeadA..."

79+ once $ putStrLn "LoopHeadA..."

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

82-------------------------------------------------------------------------------

examples/nested-loops.hs+4−4

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

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

3import System.Random (randomIO)

4import Streamly

5−import Streamly.Prelude (nil)

5+import Streamly.Prelude (nil, once)

7main = runStream $ do

8− fromIO $ hSetBuffering stdout LineBuffering

8+ once $ hSetBuffering stdout LineBuffering

9 x <- loop "A " 2

10 y <- loop "B " 2

11− fromIO $ myThreadId >>= putStr . show

11+ once $ myThreadId >>= putStr . show

12 >> putStr " "

13 >> print (x, y)

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

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

18 loop name n = do

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

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

20 let result = (name ++ show rnd)

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

22 in (return result) `parallel` repeat

examples/parallel-loops.hs+5−4

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

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

3import System.Random (randomIO)

4import Streamly

5+import Streamly.Prelude (once)

6main = runStream $ do

7main = runStream $ do

7− fromIO $ hSetBuffering stdout LineBuffering

8+ once $ hSetBuffering stdout LineBuffering

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

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

9− fromIO $ myThreadId >>= putStr . show

10+ once $ myThreadId >>= putStr . show

10 >> putStr " "

11 >> putStr " "

11 >> print x

12 >> print x

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

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

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

16 loop name = do

17 loop name = do

17− fromIO $ threadDelay 1000000

18+ once $ threadDelay 1000000

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

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

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

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

src/Streamly.hs−2

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

24 , Parallel

25 , ZipStream

26 , ZipParallel

27− , fromIO

28− , fromM

30 -- * Stream transformers

28 -- * Stream transformers

31 , IsStream

29 , IsStream

src/Streamly/Core.hs+11−1

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

27 -- * Construction

28 , singleton

29+ , once

29 , cons

30 , cons

31+ , consM

30 , repeat

32 , repeat

31 , nil

33 , nil

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

175nil :: Stream m a

177nil :: Stream m a

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

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

177

179

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

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

182+

178{-# INLINE singleton #-}

183{-# INLINE singleton #-}

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

179singleton :: a -> Stream m a

185singleton :: a -> Stream m a

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

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

181

187

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@

182cons :: a -> Stream m a -> Stream m a

192cons :: a -> Stream m a -> Stream m a

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

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

184

194

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

597-------------------------------------------------------------------------------

607-------------------------------------------------------------------------------

598

608

599instance MonadTrans Stream where

609instance MonadTrans Stream where

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

610+ lift = once

601

611

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

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

603withLocal f m =

613withLocal f m =

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

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

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

16userAction :: MonadIO m => StreamT m Event

17−userAction = cycle1 $ fromIO askUser

17+userAction = cycle1 $ liftIO askUser

18 where

19 askUser = do

20 command <- getLine

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

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

26acidRain :: MonadIO m => StreamT m Event

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

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

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

30game = do

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

37 h <- get

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

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

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

41acidRainGame :: IO ()

42acidRainGame = do

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

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

10import Data.IORef

11import Graphics.UI.SDL as SDL

12import Streamly

13+import Streamly.Prelude (once)

13import Streamly.Time

14import Streamly.Time

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

16------------------------------------------------------------------------------

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

87circlingSquare = do

88circlingSquare = do

88 sdlInit

89 sdlInit

89 cref <- newIORef (0,0)

90 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

6import System.Random (getStdGen, randoms)

7import Data.List (sort)

8import Streamly

9+import Streamly.Prelude (once)

9import qualified Streamly.Prelude as A

10import qualified Streamly.Prelude as A

11getSorted :: Stream Word16

12getSorted :: Stream Word16

12getSorted = do

13getSorted = do

13− g <- fromIO getStdGen

14+ g <- once getStdGen

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

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

15 foldMap return (sort ls)

16 foldMap return (sort ls)

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

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

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

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

19mergeAsync a b = do

20mergeAsync a b = do

20− x <- fromIO $ async a

21+ x <- once $ async a

21− y <- fromIO $ async b

22+ y <- once $ async b

22 merge x y

23 merge x y

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

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

25merge a b = do

26merge a b = do

26− a1 <- fromIO $ A.uncons a

27+ a1 <- once $ A.uncons a

27 case a1 of

28 case a1 of

28 Nothing -> b

29 Nothing -> b

29 Just (x, ma) -> do

30 Just (x, ma) -> do

30− b1 <- fromIO $ A.uncons b

31+ b1 <- once $ A.uncons b

31 case b1 of

32 case b1 of

32 Nothing -> return x <> ma

33 Nothing -> return x <> ma

33 Just (y, mb) ->

34 Just (y, mb) ->

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

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

1module Streamly.Examples.SearchEngineQuery where

3import Streamly

4+import Streamly.Prelude (once)

4import Network.HTTP.Simple

5import Network.HTTP.Simple

6-- Runs three search engine queries in parallel.

7-- Runs three search engine queries in parallel.

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

15 where

16 where

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

17 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))

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

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

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

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

src/Streamly/Prelude.hs+3

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

20 -- * Construction and Deconstruction

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

22 nil

23+ , once

24+ , consM

25+ , (|:)

23 , cons

26 , cons

24 , (.:)

27 , (.:)

25 , uncons

28 , uncons

src/Streamly/Streams.hs+55−42

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

35 -- * Construction

36 , nil

37+ , once

37 , cons

38 , cons

38 , (.:)

39 , (.:)

40+ , consM

41+ , (|:)

39 , streamBuild

42 , streamBuild

40 , fromCallback

43 , fromCallback

41 , fromSVar

44 , fromSVar

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

64 , Parallel

67 , Parallel

65 , ZipStream

68 , ZipStream

66 , ZipParallel

69 , ZipParallel

67− , fromIO

68− , fromM

70 -- * Stream Transformers

71 -- * Stream Transformers

71 , StreamT

72 , StreamT

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

122import Data.Semigroup (Semigroup(..))

123import Data.Semigroup (Semigroup(..))

123import Prelude hiding (zipWith)

124import Prelude hiding (zipWith)

124import Streamly.Core ( MonadParallel

125import Streamly.Core ( MonadParallel

125− , singleton

126 , SVar, SVarStyle(..)

127 , SVarTag(..), SVarSched(..))

128import qualified Streamly.Core as S

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

145-- Constructing a stream

146------------------------------------------------------------------------------

147

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

148+-- | An empty stream.

149+--

150+-- @

151+-- > toList nil

152+-- []

153+-- @

149nil :: IsStream t => t m a

154nil :: IsStream t => t m a

150nil = fromStream S.nil

155nil = fromStream S.nil

151

156

157+-- | Create a single element (singleton) stream from a monadic action.

158+--

159+-- @

160+-- > toList $ once getLine

161+-- hello

162+-- ["hello"]

163+-- @

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

165+once = fromStream . S.once

166+

167+infixr 5 `consM`

168+

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

170+-- stream. For example:

171+--

172+-- @

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

174+-- hello

175+-- world

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

177+-- @

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

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

180+

181+infixr 5 |:

182+

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

184+--

185+-- @

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

187+-- hello

188+-- world

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

190+-- @

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

192+(|:) = consM

193+

152infixr 5 `cons`

194infixr 5 `cons`

153

195

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

196+-- | Same as @consM . return@. Constructs a stream by adding a pure value at

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

197+-- the head of an existing stream. For example:

156--

198--

157-- @

199-- @

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

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

159−-- > (toList . streamly) stream

160-- [1,2,3]

201-- [1,2,3]

161-- @

202-- @

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

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

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

164

205

165infixr 5 .:

206infixr 5 .:

166

207

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

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

168−-- values more succinctly like this:

169--

209--

170-- @

210-- @

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

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

172−-- > (toList . streamly) stream

173-- [1,2,3]

212-- [1,2,3]

174-- @

213-- @

175--

214--

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−--

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

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

194(.:) = cons

216(.:) = cons

195

217

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

285

307

286#define MONAD_APPLICATIVE_INSTANCE(STREAM,CONSTRAINT) \

308#define MONAD_APPLICATIVE_INSTANCE(STREAM,CONSTRAINT) \

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

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

288− pure = STREAM . singleton; \

310+ pure = STREAM . S.singleton; \

289 (<*>) = ap }

311 (<*>) = ap }

290

312

291#define MONAD_COMMON_INSTANCES(STREAM,CONSTRAINT) \

313#define MONAD_COMMON_INSTANCES(STREAM,CONSTRAINT) \

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

413-- yielding one element from each stream alternately.

435-- yielding one element from each stream alternately.

414--

436--

415-- @

437-- @

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

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

417-- @

439-- @

418-- @

440-- @

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

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

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

424--

446--

425--

447--

426-- @

448-- @

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

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

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

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

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

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

430-- liftIO $ print (x, y)

452-- liftIO $ print (x, y)

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

436-- (2,4)

458-- (2,4)

437-- @

459-- @

438--

460--

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

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

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

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

441--

463--

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

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

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

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

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

900type ZipParallel a = ZipParallelM IO a

922type ZipParallel a = ZipParallelM IO a

901

923

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−

911------------------------------------------------------------------------------

924------------------------------------------------------------------------------

912-- Fold Utilities

925-- Fold Utilities

913------------------------------------------------------------------------------

926------------------------------------------------------------------------------