3bb7ff1 · streamly · git.iko.soy

git.iko.soy/streamly

Use only runStream to run streams with optional combinators

commit3bb7ff10c598cdf1cf23d982e076a59217732bc0 authorHarendra Kumar <harendra.kumar@gmail.com> dateWed 18 Apr 2018, 13:25 (9 years ago) parent793f2b8

9 files changed+73−111

compare with master

unifiedsplit

examples/loops.hs+6−6 examples/nested-loops.hs+1−1 examples/parallel-loops.hs+1−1 src/Streamly.hs+20−48 src/Streamly/Examples/AcidRainGame.hs+1−1 src/Streamly/Examples/CirclingSquare.hs+1−1 src/Streamly/Examples/ListDirRecursive.hs+4−4 src/Streamly/Examples/SearchEngineQuery.hs+2−2 src/Streamly/Streams.hs+37−47

examples/loops.hs+6−6

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

55 liftIO $ hSetBuffering stdout LineBuffering

77 putStrLn $ "\nloopTail:\n"

8− runSerialT $ do

8+ runStream $ do

99 x <- loopTail 0

1010 liftIO $ print (x :: Int)

1111

1212 putStrLn $ "\nloopHead:\n"

13− runSerialT $ do

13+ runStream $ do

1414 x <- loopHead 0

1515 liftIO $ print (x :: Int)

1616

1717 putStrLn $ "\nloopTailA:\n"

18− runSerialT $ do

18+ runStream $ do

1919 x <- loopTailA 0

2020 liftIO $ print (x :: Int)

2121

2222 putStrLn $ "\nloopHeadA:\n"

23− runSerialT $ do

23+ runStream $ do

2424 x <- loopHeadA 0

2525 liftIO $ print (x :: Int)

2626

2727 putStrLn $ "\ninterleave:\n"

28− runSerialT $ do

28+ runStream $ do

2929 x <- (return 0 <> return 1) `interleave` (return 100 <> return 101)

3030 liftIO $ print (x :: Int)

3131

3232 putStrLn $ "\nParallel interleave:\n"

33− runSerialT $ do

33+ runStream $ do

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

3535 liftIO $ print (x :: Int)

3636

examples/nested-loops.hs+1−1

@@ -5,7 +5,7 @@ import System.Random (randomIO)

55import Streamly

66import Streamly.Prelude (nil)

8−main = runSerialT $ do

8+main = runStream $ do

99 liftIO $ hSetBuffering stdout LineBuffering

1010 x <- loop "A " 2

1111 y <- loop "B " 2

examples/parallel-loops.hs+1−1

@@ -4,7 +4,7 @@ import System.IO (stdout, hSetBuffering, BufferMode(LineBuffering))

44import System.Random (randomIO)

55import Streamly

7−main = runSerialT $ do

7+main = runStream $ do

88 liftIO $ hSetBuffering stdout LineBuffering

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

1010 liftIO $ myThreadId >>= putStr . show

src/Streamly.hs+20−48

@@ -30,7 +30,7 @@ module Streamly

3030 , ZipSerial

3131 , ZipAsync

3232

33− -- * Type Independent Sum Operations

33+ -- * Polymorphic Sum Operations

3434 -- $sum

3535 , append

3636 , interleave

@@ -52,12 +52,6 @@ module Streamly

5252

5353 -- * Running Streams

5454 , runStream

55− , runSerialT

56− , runInterleavedT

57− , runAParallelT

58− , runParallelT

59− , runZipSerial

60− , runZipAsync

6155

6256 -- * Fold Utilities

6357 -- $foldutils

@@ -77,6 +71,9 @@ module Streamly

7771 , Streaming

7872 , runStreaming

7973 , runStreamT

74+ , runInterleavedT

75+ , runParallelT

76+ , runZipAsync

8077 , runAsyncT

8178 , runZipStream

8279 , StreamT

@@ -182,50 +179,25 @@ import Control.Monad.Trans.Class (MonadTrans (..))

182179-- not monads.

183180

184181-- $sum

185−-- Each stream style provides its own way of merging streams. The 'Semigroup'

186−-- '<>' operation can be used to merge streams in the style of the current

187−-- type. In case you want to merge streams in a particular style you can either

188−-- use a type adapter to force that type of composition or alternatively use

189−-- the type independent merge operations described in this section.

182+-- The 'Semigroup' operation '<>' of each stream type combines two streams in a

183+-- type specific manner. This section provides polymorphic versions of '<>'

184+-- which can be used to combine two streams in a predetermined way irrespective

185+-- of the type.

190186

191187-- $adapters

192188--

193−-- Code using streamly is usually written such that it is agnostic of any

194−-- specific streaming type. We use a type variable (polymorphic type) with the

195−-- 'IsStream' class constraint. Finally, when running the monad we can specify

196−-- the actual type that we want to use to interpret the code. However, in

197−-- certain cases we may want to use a specific type to force a certain type of

198−-- composition. These combinators can be used to convert the stream types from

199−-- one to another at no cost as all the types have the same underlying

200−-- representation.

201−--

202−-- If you see an @ambiguous type variable@ error then most likely it is because

203−-- you have not specified the stream type. You either need a type annotation or

204−-- one of the following combinators to specify what type of stream you mean.

205−--

206−-- This code:

207−--

208−-- @

209−-- main = ('toList' $ (return 1 <> return 2)) >>= print

210−-- @

211−--

212−-- will result in a type error like this:

213−--

214−-- @

215−-- Ambiguous type variable ‘t0’ arising from a use of ...

216−-- @

217−--

218−-- To fix the error just tell 'toList' what kind of stream are we feeding it:

219−--

220−-- @

221−-- main = ('toList' $ 'serially' $ (return 1 <> return 2)) >>= print

222−-- @

223−-- @

224−-- main = ('toList' $ (return 1 <> return 2 :: SerialT IO Int)) >>= print

225−-- @

226−--

227−-- Note that using the combinators is easier as you do not have to think about

228−-- the specific types, they are just inferred.

189+-- You may want to use different stream types at different points in your

190+-- program. Stream types can be converted or adapted from one type to another

191+-- to make them interwork with each other.

192+--

193+-- To adapt from one monomorphic type (e.g. 'ParallelT') to another monomorphic

194+-- type (e.g. 'SerialT') use the 'adapt' combinator. To give a polymorphic code

195+-- a specific interpretation or to adapt a specific type to a polymorphic type

196+-- use the type specific combinators e.g. 'parallely' or 'interleaving'. You

197+-- cannot adapt polymorphic code to polymorphic code, as it would not know

198+-- which specific type you are converting from or to. If you see a an

199+-- @ambiguous type variable@ error then most likely you are using 'adapt'

200+-- unnecessarily on polymorphic code.

229201--

230202

231203-- $foldutils

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

@@ -42,5 +42,5 @@ acidRainGame :: IO ()

4242acidRainGame = do

4343 putStrLn "Your health is deteriorating due to acid rain,\

4444 \ type \"potion\" or \"quit\""

45− _ <- runStateT (runSerialT game) 60

45+ _ <- runStateT (runStream game) 60

4646 return ()

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

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

8787circlingSquare = do

8888 sdlInit

8989 cref <- newIORef (0,0)

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

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

src/Streamly/Examples/ListDirRecursive.hs+4−4

@@ -3,15 +3,15 @@ module Streamly.Examples.ListDirRecursive where

33import Control.Monad.IO.Class (liftIO)

44import Path.IO (listDir, getCurrentDir)

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

6−import Streamly (runAParallelT)

6+import Streamly (runStream, aparallely)

88-- | This example demonstrates that there is little difference between regular

9−-- IO code and concurrent streamly code. You can just remove 'runAParallelT'

10−-- and this is your regular IO code.

9+-- IO code and concurrent streamly code. You can just remove

10+-- 'runStream . aparallely' and this becomes your regular IO code.

1111listDirRecursive :: IO ()

1212listDirRecursive = do

1313 hSetBuffering stdout LineBuffering

14− runAParallelT $ getCurrentDir >>= readdir

14+ runStream . aparallely $ getCurrentDir >>= readdir

1515 where readdir d = do

1616 (ds, fs) <- liftIO $ listDir d

1717 liftIO $ mapM_ putStrLn $ map show fs ++ map show ds

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

@@ -7,10 +7,10 @@ import Network.HTTP.Simple

77searchEngineQuery :: IO ()

88searchEngineQuery = do

99 putStrLn "Using parallel alternative"

10− runParallelT $ google <> bing <> duckduckgo

10+ runStream . parallely $ google <> bing <> duckduckgo

1111

1212 putStrLn "\nUsing parallel applicative zip"

13− runZipAsync $ (,,) <$> google <*> bing <*> duckduckgo

13+ runStream . zippingAsync $ (,,) <$> google <*> bing <*> duckduckgo

1414

1515 where

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

src/Streamly/Streams.hs+37−47

@@ -77,15 +77,12 @@ module Streamly.Streams

7777 , adapt

7878

7979 -- * Running Streams

80− , runSerialT

8180 , runStreamT -- deprecated

82− , runInterleavedT

83− , runAParallelT

81+ , runInterleavedT -- deprecated

8482 , runAsyncT -- deprecated

85− , runParallelT

86− , runZipSerial

83+ , runParallelT -- deprecated

8784 , runZipStream -- deprecated

88− , runZipAsync

85+ , runZipAsync -- deprecated

8986

9087 -- * Zipping

9188 , zipWith

@@ -274,7 +271,7 @@ async m = do

274271-- element of the stream, serially.

275272--

276273-- @

277−-- main = 'runSerialT' $ do

274+-- main = 'runStream' . 'serially' $ do

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

279276-- liftIO $ print x

280277-- @

@@ -286,7 +283,7 @@ async m = do

286283-- 'SerialT' nests streams serially in a depth first manner.

287284--

288285-- @

289−-- main = 'runSerialT' $ do

286+-- main = 'runStream' . 'serially' $ do

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

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

292289-- liftIO $ print (x, y)

@@ -304,6 +301,8 @@ async m = do

304301-- nested @for@ loops and the monadic continuation is the body of the loop. The

305302-- loop iterates for all elements of the stream.

306303--

304+-- The 'serially' combinator can be omitted as the default stream type is

305+-- 'SerialT'.

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

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

309308--

@@ -334,9 +333,9 @@ type StreamT = SerialT

334333-- Semigroup

335334------------------------------------------------------------------------------

336335

337−-- | Same as the 'Semigroup' instance of 'SerialT'. Appends two streams

338−-- sequentially, yielding all elements from the first stream, and then all

339−-- elements from the second stream.

336+-- | Polymorphic version of the 'Semigroup' operation '<>' of 'SerialT'.

337+-- Appends two streams sequentially, yielding all elements from the first

338+-- stream, and then all elements from the second stream.

340339{-# INLINE append #-}

341340append :: IsStream t => t m a -> t m a -> t m a

342341append m1 m2 = fromStream $ S.append (toStream m1) (toStream m2)

@@ -435,7 +434,7 @@ instance (Monad m, Floating a) => Floating (SerialT m a) where

435434--

436435--

437436-- @

438−-- main = 'runInterleavedT' $ do

437+-- main = 'runStream' . 'interleaving' $ do

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

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

441440-- liftIO $ print (x, y)

@@ -468,8 +467,8 @@ instance IsStream InterleavedT where

468467-- Semigroup

469468------------------------------------------------------------------------------

470469

471−-- | Same as the 'Semigroup' instance of 'InterleavedT'. Interleaves two

472−-- streams, yielding one element from each stream alternately.

470+-- | Polymorphic version of the 'Semigroup' operation '<>' of 'InterleavedT'.

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

473472{-# INLINE interleave #-}

474473interleave :: IsStream t => t m a -> t m a -> t m a

475474interleave m1 m2 = fromStream $ S.interleave (toStream m1) (toStream m2)

@@ -586,7 +585,7 @@ instance (Monad m, Floating a) => Floating (InterleavedT m a) where

586585-- import "Streamly"

587586-- import Control.Concurrent

588587--

589−-- main = 'runAParallelT' $ do

588+-- main = 'runStream' . 'aparallely' $ do

590589-- n <- return 3 \<\> return 2 \<\> return 1

591590-- liftIO $ do

592591-- threadDelay (n * 1000000)

@@ -627,9 +626,9 @@ instance IsStream AParallelT where

627626-- Semigroup

628627------------------------------------------------------------------------------

629628

630−-- | Same as the 'Semigroup' operation of 'AParallelT', but polymorphic.

631−-- Merges two streams possibly concurrently, preferring the elements from the

632−-- left one when available.

629+-- | Polymorphic version of the 'Semigroup' operation '<>' of 'AParallelT', but

630+-- polymorphic. Merges two streams possibly concurrently, preferring the

631+-- elements from the left one when available.

633632{-# INLINE aparallel #-}

634633aparallel :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a

635634aparallel m1 m2 = fromStream $ S.aparallel (toStream m1) (toStream m2)

@@ -753,7 +752,7 @@ instance (MonadAsync m, Floating a) => Floating (AParallelT m a) where

753752-- import "Streamly"

754753-- import Control.Concurrent

755754--

756−-- main = 'runParallelT' $ do

755+-- main = 'runStream' . 'parallely' $ do

757756-- n <- return 3 \<\> return 2 \<\> return 1

758757-- liftIO $ do

759758-- threadDelay (n * 1000000)

@@ -792,8 +791,8 @@ instance IsStream ParallelT where

792791-- Semigroup

793792------------------------------------------------------------------------------

794793

795−-- | Same as the 'Semigroup' instance of 'ParallelT'. Merges two streams

796−-- concurrently choosing elements from both fairly.

794+-- | Polymorphic version of the 'Semigroup' operation '<>' of 'ParallelT'.

795+-- Merges two streams concurrently choosing elements from both fairly.

797796{-# INLINE parallel #-}

798797parallel :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a

799798parallel m1 m2 = fromStream $ S.parallel (toStream m1) (toStream m2)

@@ -1066,19 +1065,19 @@ instance (MonadAsync m, Floating a) => Floating (ZipAsync m a) where

10661065-- Type adapting combinators

10671066-------------------------------------------------------------------------------

10681067

1069−-- | Adapt one streaming type to another.

1068+-- | Adapt any specific stream type to any other specific stream type.

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

10711070adapt = fromStream . toStream

10721071

1073−-- | Interpret an ambiguously typed stream as 'SerialT'.

1072+-- | Fix the type of a polymorphic stream as 'SerialT'.

10741073serially :: IsStream t => SerialT m a -> t m a

10751074serially = adapt

10761075

1077−-- | Interpret an ambiguously typed stream as 'InterleavedT'.

1076+-- | Fix the type of a polymorphic stream as 'InterleavedT'.

10781077interleaving :: IsStream t => InterleavedT m a -> t m a

10791078interleaving = adapt

10801079

1081−-- | Interpret an ambiguously typed stream as 'AParallelT'.

1080+-- | Fix the type of a polymorphic stream as 'AParallelT'.

10821081aparallely :: IsStream t => AParallelT m a -> t m a

10831082aparallely = adapt

10841083

@@ -1087,15 +1086,15 @@ aparallely = adapt

10871086asyncly :: IsStream t => AParallelT m a -> t m a

10881087asyncly = aparallely

10891088

1090−-- | Interpret an ambiguously typed stream as 'ParallelT'.

1089+-- | Fix the type of a polymorphic stream as 'ParallelT'.

10911090parallely :: IsStream t => ParallelT m a -> t m a

10921091parallely = adapt

10931092

1094−-- | Interpret an ambiguously typed stream as 'ZipSerial'.

1093+-- | Fix the type of a polymorphic stream as 'ZipSerial'.

10951094zipping :: IsStream t => ZipSerial m a -> t m a

10961095zipping = adapt

10971096

1098−-- | Interpret an ambiguously typed stream as 'ZipAsync'.

1097+-- | Fix the type of a polymorphic stream as 'ZipAsync'.

10991098zippingAsync :: IsStream t => ZipAsync m a -> t m a

11001099zippingAsync = adapt

11011100

@@ -1103,42 +1102,33 @@ zippingAsync = adapt

11031102-- Running Streams, convenience functions specialized to types

11041103-------------------------------------------------------------------------------

11051104

1106−-- | Same as @runStream . serially@.

1107−runSerialT :: Monad m => SerialT m a -> m ()

1108−runSerialT = runStream

1109−

1110−-- | Same as @runSerialT@.

1111−{-# Deprecated runStreamT "Please use runSerialT instead." #-}

1105+-- | Same as @runStream@.

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

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

1113−runStreamT = runSerialT

1108+runStreamT = runStream

11141109

11151110-- | Same as @runStream . interleaving@.

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

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

11171113runInterleavedT = runStream . interleaving

11181114

11191115-- | Same as @runStream . aparallely@.

1120−runAParallelT :: Monad m => AParallelT m a -> m ()

1121−runAParallelT = runStream . aparallely

1122−

1123−-- | Same as @runAParallelT@.

1124−{-# Deprecated runAsyncT "Please use runAParallelT instead." #-}

1116+{-# DEPRECATED runAsyncT "Please use 'runStream . aparallely' instead." #-}

11251117runAsyncT :: Monad m => AParallelT m a -> m ()

1126−runAsyncT = runAParallelT

1118+runAsyncT = runStream . aparallely

11271119

11281120-- | Same as @runStream . parallely@.

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

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

11301123runParallelT = runStream . parallely

11311124

11321125-- | Same as @runStream . zipping@.

1133−runZipSerial :: Monad m => ZipSerial m a -> m ()

1134−runZipSerial = runStream . zipping

1135−

1136−{-# Deprecated runZipStream "Please use runZipSerial instead." #-}

1137−-- | Same as ZipSerial.

1126+{-# DEPRECATED runZipStream "Please use 'runStream . zipping instead." #-}

11381127runZipStream :: Monad m => ZipSerial m a -> m ()

1139−runZipStream = runZipSerial

1128+runZipStream = runStream . zipping

11401129

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

1131+{-# DEPRECATED runZipAsync "Please use 'runStream . zippingAsync instead." #-}

11421132runZipAsync :: Monad m => ZipAsync m a -> m ()

11431133runZipAsync = runStream . zippingAsync

11441134