3bb7ff1 · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Use only runStream to run streams with optional combinators

3bb7ff1Harendra Kumar9 years ago

examples/loops.hs+6−6

compare with master

unifiedsplit

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

5 liftIO $ hSetBuffering stdout LineBuffering

7 putStrLn $ "\nloopTail:\n"

8− runSerialT $ do

8+ runStream $ do

9 x <- loopTail 0

10 liftIO $ print (x :: Int)

12 putStrLn $ "\nloopHead:\n"

13− runSerialT $ do

13+ runStream $ do

14 x <- loopHead 0

15 liftIO $ print (x :: Int)

17 putStrLn $ "\nloopTailA:\n"

18− runSerialT $ do

18+ runStream $ do

19 x <- loopTailA 0

20 liftIO $ print (x :: Int)

22 putStrLn $ "\nloopHeadA:\n"

23− runSerialT $ do

23+ runStream $ do

24 x <- loopHeadA 0

25 liftIO $ print (x :: Int)

27 putStrLn $ "\ninterleave:\n"

28− runSerialT $ do

28+ runStream $ do

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

30 liftIO $ print (x :: Int)

32 putStrLn $ "\nParallel interleave:\n"

33− runSerialT $ do

33+ runStream $ do

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

35 liftIO $ print (x :: Int)

examples/nested-loops.hs+1−1

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

5import Streamly

6import Streamly.Prelude (nil)

8−main = runSerialT $ do

8+main = runStream $ do

9 liftIO $ hSetBuffering stdout LineBuffering

10 x <- loop "A " 2

11 y <- loop "B " 2

examples/parallel-loops.hs+1−1

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

4import System.Random (randomIO)

5import Streamly

7−main = runSerialT $ do

7+main = runStream $ do

8 liftIO $ hSetBuffering stdout LineBuffering

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

10 liftIO $ myThreadId >>= putStr . show

src/Streamly.hs+20−48

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

30 , ZipSerial

31 , ZipAsync

33− -- * Type Independent Sum Operations

33+ -- * Polymorphic Sum Operations

34 -- $sum

35 , append

36 , interleave

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

53 -- * Running Streams

54 , runStream

55− , runSerialT

56− , runInterleavedT

57− , runAParallelT

58− , runParallelT

59− , runZipSerial

60− , runZipAsync

62 -- * Fold Utilities

56 -- * Fold Utilities

63 -- $foldutils

57 -- $foldutils

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

77 , Streaming

71 , Streaming

78 , runStreaming

72 , runStreaming

79 , runStreamT

73 , runStreamT

74+ , runInterleavedT

75+ , runParallelT

76+ , runZipAsync

80 , runAsyncT

77 , runAsyncT

81 , runZipStream

78 , runZipStream

82 , StreamT

79 , StreamT

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

182-- not monads.

179-- not monads.

183

180

184-- $sum

181-- $sum

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

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

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

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

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

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

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

185+-- of the type.

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

190

186

191-- $adapters

187-- $adapters

192--

188--

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

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

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

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

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

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

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

192+--

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

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

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

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

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

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

200−-- representation.

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

201−--

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

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

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

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

200+-- unnecessarily on polymorphic code.

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.

229--

201--

230

202

231-- $foldutils

203-- $foldutils

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

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

42acidRainGame = do

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

44 \ type \"potion\" or \"quit\""

45− _ <- runStateT (runSerialT game) 60

45+ _ <- runStateT (runStream game) 60

46 return ()

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

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

87circlingSquare = do

88 sdlInit

89 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

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

4import Path.IO (listDir, getCurrentDir)

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

6−import Streamly (runAParallelT)

6+import Streamly (runStream, aparallely)

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

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

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

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

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

11listDirRecursive :: IO ()

12listDirRecursive = do

13 hSetBuffering stdout LineBuffering

14− runAParallelT $ getCurrentDir >>= readdir

14+ runStream . aparallely $ getCurrentDir >>= readdir

15 where readdir d = do

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

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

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

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

7searchEngineQuery :: IO ()

8searchEngineQuery = do

9 putStrLn "Using parallel alternative"

10− runParallelT $ google <> bing <> duckduckgo

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

12 putStrLn "\nUsing parallel applicative zip"

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

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

15 where

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

src/Streamly/Streams.hs+37−47

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

77 , adapt

79 -- * Running Streams

80− , runSerialT

81 , runStreamT -- deprecated

80 , runStreamT -- deprecated

82− , runInterleavedT

81+ , runInterleavedT -- deprecated

83− , runAParallelT

84 , runAsyncT -- deprecated

82 , runAsyncT -- deprecated

85− , runParallelT

83+ , runParallelT -- deprecated

86− , runZipSerial

87 , runZipStream -- deprecated

84 , runZipStream -- deprecated

88− , runZipAsync

85+ , runZipAsync -- deprecated

90 -- * Zipping

87 -- * Zipping

91 , zipWith

88 , zipWith

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

274-- element of the stream, serially.

271-- element of the stream, serially.

275--

272--

276-- @

273-- @

277−-- main = 'runSerialT' $ do

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

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

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

279-- liftIO $ print x

276-- liftIO $ print x

280-- @

277-- @

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

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

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

287--

284--

288-- @

285-- @

289−-- main = 'runSerialT' $ do

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

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

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

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

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

292-- liftIO $ print (x, y)

289-- liftIO $ print (x, y)

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

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

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

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

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

306--

303--

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

305+-- 'SerialT'.

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

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

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

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

309--

308--

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

334-- Semigroup

333-- Semigroup

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

334------------------------------------------------------------------------------

336

335

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

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

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

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

339−-- elements from the second stream.

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

340{-# INLINE append #-}

339{-# INLINE append #-}

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

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

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

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

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

435--

434--

436--

435--

437-- @

436-- @

438−-- main = 'runInterleavedT' $ do

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

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

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

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

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

441-- liftIO $ print (x, y)

440-- liftIO $ print (x, y)

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

468-- Semigroup

467-- Semigroup

469------------------------------------------------------------------------------

468------------------------------------------------------------------------------

470

469

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

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

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

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

473{-# INLINE interleave #-}

472{-# INLINE interleave #-}

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

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

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

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

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

586-- import "Streamly"

585-- import "Streamly"

587-- import Control.Concurrent

586-- import Control.Concurrent

588--

587--

589−-- main = 'runAParallelT' $ do

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

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

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

591-- liftIO $ do

590-- liftIO $ do

592-- threadDelay (n * 1000000)

591-- threadDelay (n * 1000000)

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

627-- Semigroup

626-- Semigroup

628------------------------------------------------------------------------------

627------------------------------------------------------------------------------

629

628

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

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

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

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

632−-- left one when available.

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

633{-# INLINE aparallel #-}

632{-# INLINE aparallel #-}

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

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

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

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

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

753-- import "Streamly"

752-- import "Streamly"

754-- import Control.Concurrent

753-- import Control.Concurrent

755--

754--

756−-- main = 'runParallelT' $ do

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

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

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

758-- liftIO $ do

757-- liftIO $ do

759-- threadDelay (n * 1000000)

758-- threadDelay (n * 1000000)

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

792-- Semigroup

791-- Semigroup

793------------------------------------------------------------------------------

792------------------------------------------------------------------------------

794

793

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

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

796−-- concurrently choosing elements from both fairly.

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

797{-# INLINE parallel #-}

796{-# INLINE parallel #-}

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

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

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

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

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

1066-- Type adapting combinators

1065-- Type adapting combinators

1067-------------------------------------------------------------------------------

1066-------------------------------------------------------------------------------

1068

1067

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

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

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

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

1071adapt = fromStream . toStream

1070adapt = fromStream . toStream

1072

1071

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

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

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

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

1075serially = adapt

1074serially = adapt

1076

1075

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

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

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

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

1079interleaving = adapt

1078interleaving = adapt

1080

1079

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

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

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

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

1083aparallely = adapt

1082aparallely = adapt

1084

1083

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

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

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

1088asyncly = aparallely

1087asyncly = aparallely

1089

1088

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

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

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

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

1092parallely = adapt

1091parallely = adapt

1093

1092

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

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

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

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

1096zipping = adapt

1095zipping = adapt

1097

1096

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

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

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

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

1100zippingAsync = adapt

1099zippingAsync = adapt

1101

1100

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

1103-- Running Streams, convenience functions specialized to types

1102-- Running Streams, convenience functions specialized to types

1104-------------------------------------------------------------------------------

1103-------------------------------------------------------------------------------

1105

1104

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

1105+-- | Same as @runStream@.

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

1108−runSerialT = runStream

1109−

1110−-- | Same as @runSerialT@.

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

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

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

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

1113−runStreamT = runSerialT

1108+runStreamT = runStream

1114

1109

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

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

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

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

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

1117runInterleavedT = runStream . interleaving

1113runInterleavedT = runStream . interleaving

1118

1114

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

1115-- | 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." #-}

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

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

1126−runAsyncT = runAParallelT

1118+runAsyncT = runStream . aparallely

1127

1119

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

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

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

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

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

1130runParallelT = runStream . parallely

1123runParallelT = runStream . parallely

1131

1124

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

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

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

1134−runZipSerial = runStream . zipping

1135−

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

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

1137−-- | Same as ZipSerial.

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

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

1139−runZipStream = runZipSerial

1128+runZipStream = runStream . zipping

1140

1129

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

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

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

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

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

1143runZipAsync = runStream . zippingAsync

1133runZipAsync = runStream . zippingAsync

1144

1134