8c8b9c3 · streamly · git.iko.soy

benchmark/Streamly/Benchmark/Data/Stream/Common.hs+44−43

compare with master

unifiedsplit

@@ -71,6 +71,7 @@ import Control.DeepSeq (NFData)

71import Control.Exception (try)

72import GHC.Exception (ErrorCall)

73import Streamly.Internal.Data.Stream (Stream)

74+import Streamly.Internal.Data.Stream.Cross (CrossStream(..))

74import System.Random (randomRIO)

75import System.Random (randomRIO)

75

76

76import qualified Streamly.Internal.Data.Fold as Fold

77import qualified Streamly.Internal.Data.Fold as Fold

@@ -227,9 +228,9 @@ sourceConcatMapId value n =

227{-# INLINE apDiscardFst #-}

228{-# INLINE apDiscardFst #-}

228apDiscardFst :: MonadAsync m =>

229apDiscardFst :: MonadAsync m =>

229 Int -> Int -> m ()

230 Int -> Int -> m ()

230−apDiscardFst linearCount start = drain $

231+apDiscardFst linearCount start = drain $ getCrossStream $

231− sourceUnfoldrM nestedCount2 start

232+ CrossStream (sourceUnfoldrM nestedCount2 start)

232− *> sourceUnfoldrM nestedCount2 start

233+ *> CrossStream (sourceUnfoldrM nestedCount2 start)

233

234

234 where

235 where

235

236

@@ -237,9 +238,9 @@ apDiscardFst linearCount start = drain $

237

238

238{-# INLINE apDiscardSnd #-}

239{-# INLINE apDiscardSnd #-}

239apDiscardSnd :: MonadAsync m => Int -> Int -> m ()

240apDiscardSnd :: MonadAsync m => Int -> Int -> m ()

240−apDiscardSnd linearCount start = drain $

241+apDiscardSnd linearCount start = drain $ getCrossStream $

241− sourceUnfoldrM nestedCount2 start

242+ CrossStream (sourceUnfoldrM nestedCount2 start)

242− <* sourceUnfoldrM nestedCount2 start

243+ <* CrossStream (sourceUnfoldrM nestedCount2 start)

243

244

244 where

245 where

245

246

@@ -247,9 +248,9 @@ apDiscardSnd linearCount start = drain $

247

248

248{-# INLINE apLiftA2 #-}

249{-# INLINE apLiftA2 #-}

249apLiftA2 :: MonadAsync m => Int -> Int -> m ()

250apLiftA2 :: MonadAsync m => Int -> Int -> m ()

250−apLiftA2 linearCount start = drain $

251+apLiftA2 linearCount start = drain $ getCrossStream $

251− liftA2 (+) (sourceUnfoldrM nestedCount2 start)

252+ liftA2 (+) (CrossStream (sourceUnfoldrM nestedCount2 start))

252− (sourceUnfoldrM nestedCount2 start)

253+ (CrossStream (sourceUnfoldrM nestedCount2 start))

253

254

254 where

255 where

255

256

@@ -257,9 +258,9 @@ apLiftA2 linearCount start = drain $

257

258

258{-# INLINE toNullAp #-}

259{-# INLINE toNullAp #-}

259toNullAp :: MonadAsync m => Int -> Int -> m ()

260toNullAp :: MonadAsync m => Int -> Int -> m ()

260−toNullAp linearCount start = drain $

261+toNullAp linearCount start = drain $ getCrossStream $

261− (+) <$> sourceUnfoldrM nestedCount2 start

262+ (+) <$> CrossStream (sourceUnfoldrM nestedCount2 start)

262− <*> sourceUnfoldrM nestedCount2 start

263+ <*> CrossStream (sourceUnfoldrM nestedCount2 start)

263

264

264 where

265 where

265

266

@@ -267,9 +268,9 @@ toNullAp linearCount start = drain $

267

268

268{-# INLINE monadThen #-}

269{-# INLINE monadThen #-}

269monadThen :: MonadAsync m => Int -> Int -> m ()

270monadThen :: MonadAsync m => Int -> Int -> m ()

270−monadThen linearCount start = drain $ do

271+monadThen linearCount start = drain $ getCrossStream $ do

271− sourceUnfoldrM nestedCount2 start >>

272+ CrossStream (sourceUnfoldrM nestedCount2 start) >>

272− sourceUnfoldrM nestedCount2 start

273+ CrossStream (sourceUnfoldrM nestedCount2 start)

273

274

274 where

275 where

275

276

@@ -277,9 +278,9 @@ monadThen linearCount start = drain $ do

277

278

278{-# INLINE toNullM #-}

279{-# INLINE toNullM #-}

279toNullM :: MonadAsync m => Int -> Int -> m ()

280toNullM :: MonadAsync m => Int -> Int -> m ()

280−toNullM linearCount start = drain $ do

281+toNullM linearCount start = drain $ getCrossStream $ do

281− x <- sourceUnfoldrM nestedCount2 start

282+ x <- CrossStream (sourceUnfoldrM nestedCount2 start)

282− y <- sourceUnfoldrM nestedCount2 start

283+ y <- CrossStream (sourceUnfoldrM nestedCount2 start)

283 return $ x + y

284 return $ x + y

284

285

285 where

286 where

@@ -288,56 +289,56 @@ toNullM linearCount start = drain $ do

288

289

289{-# INLINE toNullM3 #-}

290{-# INLINE toNullM3 #-}

290toNullM3 :: MonadAsync m => Int -> Int -> m ()

291toNullM3 :: MonadAsync m => Int -> Int -> m ()

291−toNullM3 linearCount start = drain $ do

292+toNullM3 linearCount start = drain $ getCrossStream $ do

292− x <- sourceUnfoldrM nestedCount3 start

293+ x <- CrossStream (sourceUnfoldrM nestedCount3 start)

293− y <- sourceUnfoldrM nestedCount3 start

294+ y <- CrossStream (sourceUnfoldrM nestedCount3 start)

294− z <- sourceUnfoldrM nestedCount3 start

295+ z <- CrossStream (sourceUnfoldrM nestedCount3 start)

295 return $ x + y + z

296 return $ x + y + z

296 where

297 where

297 nestedCount3 = round (fromIntegral linearCount**(1/3::Double))

298 nestedCount3 = round (fromIntegral linearCount**(1/3::Double))

298

299

299{-# INLINE filterAllOutM #-}

300{-# INLINE filterAllOutM #-}

300filterAllOutM :: MonadAsync m => Int -> Int -> m ()

301filterAllOutM :: MonadAsync m => Int -> Int -> m ()

301−filterAllOutM linearCount start = drain $ do

302+filterAllOutM linearCount start = drain $ getCrossStream $ do

302− x <- sourceUnfoldrM nestedCount2 start

303+ x <- CrossStream (sourceUnfoldrM nestedCount2 start)

303− y <- sourceUnfoldrM nestedCount2 start

304+ y <- CrossStream (sourceUnfoldrM nestedCount2 start)

304 let s = x + y

305 let s = x + y

305 if s < 0

306 if s < 0

306 then return s

307 then return s

307− else Stream.nil

308+ else CrossStream Stream.nil

308 where

309 where

309 nestedCount2 = round (fromIntegral linearCount**(1/2::Double))

310 nestedCount2 = round (fromIntegral linearCount**(1/2::Double))

310

311

311{-# INLINE filterAllInM #-}

312{-# INLINE filterAllInM #-}

312filterAllInM :: MonadAsync m => Int -> Int -> m ()

313filterAllInM :: MonadAsync m => Int -> Int -> m ()

313−filterAllInM linearCount start = drain $ do

314+filterAllInM linearCount start = drain $ getCrossStream $ do

314− x <- sourceUnfoldrM nestedCount2 start

315+ x <- CrossStream (sourceUnfoldrM nestedCount2 start)

315− y <- sourceUnfoldrM nestedCount2 start

316+ y <- CrossStream (sourceUnfoldrM nestedCount2 start)

316 let s = x + y

317 let s = x + y

317 if s > 0

318 if s > 0

318 then return s

319 then return s

319− else Stream.nil

320+ else CrossStream Stream.nil

320 where

321 where

321 nestedCount2 = round (fromIntegral linearCount**(1/2::Double))

322 nestedCount2 = round (fromIntegral linearCount**(1/2::Double))

322

323

323{-# INLINE filterSome #-}

324{-# INLINE filterSome #-}

324filterSome :: MonadAsync m => Int -> Int -> m ()

325filterSome :: MonadAsync m => Int -> Int -> m ()

325−filterSome linearCount start = drain $ do

326+filterSome linearCount start = drain $ getCrossStream $ do

326− x <- sourceUnfoldrM nestedCount2 start

327+ x <- CrossStream (sourceUnfoldrM nestedCount2 start)

327− y <- sourceUnfoldrM nestedCount2 start

328+ y <- CrossStream (sourceUnfoldrM nestedCount2 start)

328 let s = x + y

329 let s = x + y

329 if s > 1100000

330 if s > 1100000

330 then return s

331 then return s

331− else Stream.nil

332+ else CrossStream Stream.nil

332 where

333 where

333 nestedCount2 = round (fromIntegral linearCount**(1/2::Double))

334 nestedCount2 = round (fromIntegral linearCount**(1/2::Double))

334

335

335{-# INLINE breakAfterSome #-}

336{-# INLINE breakAfterSome #-}

336breakAfterSome :: Int -> Int -> IO ()

337breakAfterSome :: Int -> Int -> IO ()

337breakAfterSome linearCount start = do

338breakAfterSome linearCount start = do

338− (_ :: Either ErrorCall ()) <- try $ drain $ do

339+ (_ :: Either ErrorCall ()) <- try $ drain $ getCrossStream $ do

339− x <- sourceUnfoldrM nestedCount2 start

340+ x <- CrossStream (sourceUnfoldrM nestedCount2 start)

340− y <- sourceUnfoldrM nestedCount2 start

341+ y <- CrossStream (sourceUnfoldrM nestedCount2 start)

341 let s = x + y

342 let s = x + y

342 if s > 1100000

343 if s > 1100000

343 then error "break"

344 then error "break"

@@ -348,9 +349,9 @@ breakAfterSome linearCount start = do

348

349

349{-# INLINE toListM #-}

350{-# INLINE toListM #-}

350toListM :: MonadAsync m => Int -> Int -> m [Int]

351toListM :: MonadAsync m => Int -> Int -> m [Int]

351−toListM linearCount start = Stream.fold Fold.toList $ do

352+toListM linearCount start = Stream.fold Fold.toList $ getCrossStream $ do

352− x <- sourceUnfoldrM nestedCount2 start

353+ x <- CrossStream (sourceUnfoldrM nestedCount2 start)

353− y <- sourceUnfoldrM nestedCount2 start

354+ y <- CrossStream (sourceUnfoldrM nestedCount2 start)

354 return $ x + y

355 return $ x + y

355 where

356 where

356 nestedCount2 = round (fromIntegral linearCount**(1/2::Double))

357 nestedCount2 = round (fromIntegral linearCount**(1/2::Double))

@@ -360,9 +361,9 @@ toListM linearCount start = Stream.fold Fold.toList $ do

360{-# INLINE toListSome #-}

361{-# INLINE toListSome #-}

361toListSome :: MonadAsync m => Int -> Int -> m [Int]

362toListSome :: MonadAsync m => Int -> Int -> m [Int]

362toListSome linearCount start =

363toListSome linearCount start =

363− Stream.fold Fold.toList $ Stream.take 10000 $ do

364+ Stream.fold Fold.toList $ Stream.take 10000 $ getCrossStream $ do

364− x <- sourceUnfoldrM nestedCount2 start

365+ x <- CrossStream (sourceUnfoldrM nestedCount2 start)

365− y <- sourceUnfoldrM nestedCount2 start

366+ y <- CrossStream (sourceUnfoldrM nestedCount2 start)

366 return $ x + y

367 return $ x + y

367 where

368 where

368 nestedCount2 = round (fromIntegral linearCount**(1/2::Double))

369 nestedCount2 = round (fromIntegral linearCount**(1/2::Double))

benchmark/Streamly/Benchmark/Data/Stream/Expand.hs+9−8

@@ -25,8 +25,8 @@

25

26module Stream.Expand (benchmarks) where

27

28−import Control.Monad.Trans.Class (lift)

29import Streamly.Internal.Data.Stream (Stream)

28import Streamly.Internal.Data.Stream (Stream)

29+import Streamly.Internal.Data.Stream.Cross (CrossStream(..))

30

31import qualified Control.Applicative as AP

32

@@ -70,12 +70,13 @@ iterateN g initial count = f count initial

70

71-- Iterate a transformation over a singleton stream

72{-# INLINE iterateSingleton #-}

73−iterateSingleton :: Monad m

73+iterateSingleton ::

74− => (Int -> Stream m Int -> Stream m Int)

74+ (Int -> CrossStream m Int -> CrossStream m Int)

75 -> Int

76 -> Int

77 -> Stream m Int

78−iterateSingleton g count n = iterateN g (return n) count

78+iterateSingleton g count n =

79+ getCrossStream $ iterateN g (CrossStream (S.fromPure n)) count

79

80

80{-

81{-

81-- XXX need to check why this is slower than the explicit recursion above, even

82-- XXX need to check why this is slower than the explicit recursion above, even

@@ -409,12 +410,12 @@ o_1_space_monad value =

409-- new prime we keep appending a division filter for all the future numbers.

410-- new prime we keep appending a division filter for all the future numbers.

410{-# INLINE sieve #-}

411{-# INLINE sieve #-}

411sieve :: Monad m => Stream m Int -> Stream m Int

412sieve :: Monad m => Stream m Int -> Stream m Int

412−sieve s = do

413+sieve s = S.concatM $ do

413− r <- lift $ S.uncons s

414+ r <- S.uncons s

414 case r of

415 case r of

415 Just (prime, rest) ->

416 Just (prime, rest) ->

416− prime `S.cons` sieve (S.filter (\n -> n `mod` prime /= 0) rest)

417+ pure $ prime `S.cons` sieve (S.filter (\n -> n `mod` prime /= 0) rest)

417− Nothing -> S.nil

418+ Nothing -> pure S.nil

418

419

419o_n_space_monad :: Int -> [Benchmark]

420o_n_space_monad :: Int -> [Benchmark]

420o_n_space_monad value =

421o_n_space_monad value =

benchmark/Streamly/Benchmark/Data/Stream/Generate.hs+4−3

@@ -35,6 +35,7 @@ import qualified Prelude

35import Gauge

36import Streamly.Benchmark.Common

37import Streamly.Internal.Data.Stream (Stream)

38+import Streamly.Internal.Data.Stream.Cross (CrossStream(..))

38#ifdef USE_PRELUDE

39#ifdef USE_PRELUDE

39import Streamly.Prelude (MonadAsync)

40import Streamly.Prelude (MonadAsync)

40import Stream.Common hiding (MonadAsync)

41import Stream.Common hiding (MonadAsync)

@@ -178,10 +179,10 @@ fromIndicesM value n = S.take value $ S.fromIndicesM (return <$> (+ n))

178mfixUnfold :: Int -> Int -> Stream IO (Int, Int)

179mfixUnfold :: Int -> Int -> Stream IO (Int, Int)

179mfixUnfold count start = Stream.mfix f

180mfixUnfold count start = Stream.mfix f

180 where

181 where

181− f action = do

182+ f action = getCrossStream $ do

182 let incr n act = fmap ((+n) . snd) $ unsafeInterleaveIO act

183 let incr n act = fmap ((+n) . snd) $ unsafeInterleaveIO act

183− x <- Common.fromListM [incr 1 action, incr 2 action]

184+ x <- CrossStream (Common.fromListM [incr 1 action, incr 2 action])

184− y <- Common.sourceUnfoldr count start

185+ y <- CrossStream (Common.sourceUnfoldr count start)

185 return (x, y)

186 return (x, y)

186

187

187o_1_space_generation :: Int -> [Benchmark]

188o_1_space_generation :: Int -> [Benchmark]

benchmark/Streamly/Benchmark/Data/Stream/Lift.hs+10−11

@@ -16,7 +16,6 @@

16module Stream.Lift (benchmarks) where

17

18import Control.DeepSeq (NFData(..))

19−import Control.Monad.Trans.Class (lift)

20import Control.Monad.State.Strict (StateT, get, put)

19import Control.Monad.State.Strict (StateT, get, put)

21import Data.Functor.Identity (Identity)

20import Data.Functor.Identity (Identity)

22import Stream.Common

21import Stream.Common

@@ -97,26 +96,26 @@ iterateStateIO n = do

97

96

98{-# INLINE iterateStateT #-}

97{-# INLINE iterateStateT #-}

99iterateStateT :: Int -> Stream (StateT Int IO) Int

98iterateStateT :: Int -> Stream (StateT Int IO) Int

100−iterateStateT n = do

99+iterateStateT n = Stream.concatM $ do

101− x <- lift get

100+ x <- get

102 if x > n

101 if x > n

103 then do

102 then do

104− lift $ put (x - 1)

103+ put (x - 1)

105− iterateStateT n

104+ return $ iterateStateT n

106− else return x

105+ else return $ Stream.fromPure x

107

106

108{-# INLINE iterateState #-}

107{-# INLINE iterateState #-}

109{-# SPECIALIZE iterateState :: Int -> Stream (StateT Int IO) Int #-}

108{-# SPECIALIZE iterateState :: Int -> Stream (StateT Int IO) Int #-}

110iterateState :: Monad m =>

109iterateState :: Monad m =>

111 Int

110 Int

112 -> Stream (StateT Int m) Int

111 -> Stream (StateT Int m) Int

113−iterateState n = do

112+iterateState n = Stream.concatM $ do

114− x <- Stream.fromEffect get

113+ x <- get

115 if x > n

114 if x > n

116 then do

115 then do

117− Stream.fromEffect $ put (x - 1)

116+ put (x - 1)

118− iterateState n

117+ return $ iterateState n

119− else return x

118+ else return $ Stream.fromPure x

120

119

121o_n_heap_transformer :: Int -> [Benchmark]

120o_n_heap_transformer :: Int -> [Benchmark]

122o_n_heap_transformer value =

121o_n_heap_transformer value =

benchmark/Streamly/Benchmark/Data/Stream/Reduce.hs+3−3

@@ -62,12 +62,12 @@ iterateN g initial count = f count initial

62

63-- Iterate a transformation over a singleton stream

64{-# INLINE iterateSingleton #-}

65−iterateSingleton :: Monad m

65+iterateSingleton ::

66− => (Int -> Stream m Int -> Stream m Int)

66+ (Int -> Stream m Int -> Stream m Int)

67 -> Int

68 -> Int

69 -> Stream m Int

70−iterateSingleton g count n = iterateN g (return n) count

70+iterateSingleton g count n = iterateN g (S.fromPure n) count

71

72{-

73-- XXX need to check why this is slower than the explicit recursion above, even

benchmark/Streamly/Benchmark/Data/Stream/Transform.hs+7−3

@@ -174,13 +174,17 @@ foldrS n = composeN n $ Stream.foldrS Stream.cons Stream.nil

174foldrSMap :: MonadIO m => Int -> Stream m Int -> m ()

175foldrSMap n = composeN n $ Stream.foldrS (\x xs -> x + 1 `Stream.cons` xs) Stream.nil

176

177+{-

177{-# INLINE foldrT #-}

178{-# INLINE foldrT #-}

178foldrT :: MonadIO m => Int -> Stream m Int -> m ()

179foldrT :: MonadIO m => Int -> Stream m Int -> m ()

179−foldrT n = composeN n $ Stream.foldrT Stream.cons Stream.nil

180+foldrT n = composeN n (getCrossStream . Stream.foldrT cns (CrossStream Stream.nil))

181+

182+ where cns x (CrossStream xs) = CrossStream (Stream.cons x xs)

180

183

181{-# INLINE foldrTMap #-}

184{-# INLINE foldrTMap #-}

182foldrTMap :: MonadIO m => Int -> Stream m Int -> m ()

185foldrTMap :: MonadIO m => Int -> Stream m Int -> m ()

183foldrTMap n = composeN n $ Stream.foldrT (\x xs -> x + 1 `Stream.cons` xs) Stream.nil

186foldrTMap n = composeN n $ Stream.foldrT (\x xs -> x + 1 `Stream.cons` xs) Stream.nil

187+-}

184

188

185{-# INLINE trace #-}

189{-# INLINE trace #-}

186trace :: MonadAsync m => Int -> Stream m Int -> m ()

190trace :: MonadAsync m => Int -> Stream m Int -> m ()

@@ -194,8 +198,8 @@ o_1_space_mapping value =

194 -- Right folds

198 -- Right folds

195 benchIOSink value "foldrS" (foldrS 1)

199 benchIOSink value "foldrS" (foldrS 1)

196 , benchIOSink value "foldrSMap" (foldrSMap 1)

200 , benchIOSink value "foldrSMap" (foldrSMap 1)

197− , benchIOSink value "foldrT" (foldrT 1)

201+ -- , benchIOSink value "foldrT" (foldrT 1)

198− , benchIOSink value "foldrTMap" (foldrTMap 1)

202+ -- , benchIOSink value "foldrTMap" (foldrTMap 1)

199

203

200 -- Mapping

204 -- Mapping

201 , benchIOSink value "map" (mapN 1)

205 , benchIOSink value "map" (mapN 1)

benchmark/Streamly/Benchmark/Prelude/Adaptive.hs+8−5

@@ -36,11 +36,14 @@ source range = S.replicateM value $ do

36

37{-# INLINE run #-}

38run :: IsStream t => (Int, Int) -> (Int, Int) -> (t IO Int -> SerialT IO Int) -> IO ()

39−run srange crange t = S.drain $ do

39+run srange crange t = S.drain $ S.mapM action (t $ source srange)

40− n <- t $ source srange

40+

41+ where

42+

43+ action x = liftIO $ do

41− d <- liftIO (randomRIO crange)

44+ d <- randomRIO crange

42− when (d /= 0) $ liftIO $ threadDelay d

45+ when (d /= 0) $ threadDelay d

43− return n

46+ return x

44

47

45low, medium, high :: Int

48low, medium, high :: Int

46low = 10

49low = 10

core/src/Streamly/Internal/Data/List.hs+31−26

@@ -51,7 +51,10 @@

51--

52module Streamly.Internal.Data.List

53 (

54− List (.., Nil, Cons)

54+ List (Nil, Cons)

55+

56+ , toStream

57+ , fromStream

55

58

56 -- XXX we may want to use rebindable syntax for variants instead of using

59 -- XXX we may want to use rebindable syntax for variants instead of using

57 -- different types (applicative do and apWith).

60 -- different types (applicative do and apWith).

@@ -64,37 +67,41 @@ where

64import Control.Arrow (second)

67import Control.Arrow (second)

65import Data.Functor.Identity (Identity, runIdentity)

68import Data.Functor.Identity (Identity, runIdentity)

66import GHC.Exts (IsList(..), IsString(..))

69import GHC.Exts (IsList(..), IsString(..))

70+import Streamly.Internal.Data.Stream.Cross (CrossStream(..))

67import Streamly.Internal.Data.Stream.Type (Stream)

71import Streamly.Internal.Data.Stream.Type (Stream)

68import Streamly.Internal.Data.Stream.Zip (ZipStream(..))

72import Streamly.Internal.Data.Stream.Zip (ZipStream(..))

73+import Text.Read (readPrec)

69

74

70import qualified Streamly.Internal.Data.Stream.StreamK.Type as K

75import qualified Streamly.Internal.Data.Stream.StreamK.Type as K

71import qualified Streamly.Internal.Data.Stream.Type as Stream

76import qualified Streamly.Internal.Data.Stream.Type as Stream

72

77

73−-- We implement list as a newtype instead of a type synonym to make type

78+-- XXX Rename to PureStream.

74−-- inference easier when using -XOverloadedLists and -XOverloadedStrings. When

79+

75−-- using a stream type the programmer needs to specify the Monad otherwise the

76−-- type remains ambiguous.

77−--

78−-- XXX once we separate consM from IsStream or remove the MonadIO and

79−-- MonadBaseControlIO dependency from it, then we can make this an instance of

80−-- IsStream and use the regular polymorphic functions on Lists as well. Once

81−-- that happens we can change the Show and Read instances as well to use "1 >:

82−-- 2 >: nil" etc. or should we use a separate constructor indicating the "List"

83−-- type ":>" for better inference?

84−--

85-- | @List a@ is a replacement for @[a]@.

80-- | @List a@ is a replacement for @[a]@.

86--

81--

87-- /Pre-release/

82-- /Pre-release/

88−newtype List a = List { toStream :: Stream Identity a }

83+newtype List a = List { toCrossStream :: CrossStream Identity a }

89 deriving

84 deriving

90− ( Show, Read, Eq, Ord

85+ ( Eq, Ord

91 , Semigroup, Monoid, Functor, Foldable

86 , Semigroup, Monoid, Functor, Foldable

92 , Applicative, Traversable, Monad, IsList)

87 , Applicative, Traversable, Monad, IsList)

93

88

89+toStream :: List a -> Stream Identity a

90+toStream = getCrossStream . toCrossStream

91+

92+fromStream :: Stream Identity a -> List a

93+fromStream xs = List (CrossStream xs)

94+

94instance (a ~ Char) => IsString (List a) where

95instance (a ~ Char) => IsString (List a) where

95 {-# INLINE fromString #-}

96 {-# INLINE fromString #-}

96 fromString = List . fromList

97 fromString = List . fromList

97

98

99+instance Show a => Show (List a) where

100+ show (List x) = show $ getCrossStream x

101+

102+instance Read a => Read (List a) where

103+ readPrec = fromStream <$> readPrec

104+

98------------------------------------------------------------------------------

105------------------------------------------------------------------------------

99-- Patterns

106-- Patterns

100------------------------------------------------------------------------------

107------------------------------------------------------------------------------

@@ -104,31 +111,32 @@ instance (a ~ Char) => IsString (List a) where

104-- perform the pattern match, it should not be too bad as it works lazily in

111-- perform the pattern match, it should not be too bad as it works lazily in

105-- the Identity monad. We need these patterns only when not using that

112-- the Identity monad. We need these patterns only when not using that

106-- extension.

113-- extension.

107−--

114+

108-- | An empty list constructor and pattern that matches an empty 'List'.

115-- | An empty list constructor and pattern that matches an empty 'List'.

109-- Corresponds to '[]' for Haskell lists.

116-- Corresponds to '[]' for Haskell lists.

110--

117--

111−-- @since 0.6.0

112pattern Nil :: List a

118pattern Nil :: List a

113−pattern Nil <- (runIdentity . K.null . Stream.toStreamK . toStream -> True) where

119+pattern Nil <- (runIdentity . K.null . Stream.toStreamK . toStream -> True)

120+

121+ where

122+

114− Nil = List $ Stream.fromStreamK K.nil

123+ Nil = List $ CrossStream (Stream.fromStreamK K.nil)

115

124

116infixr 5 `Cons`

125infixr 5 `Cons`

117

126

118-- | A list constructor and pattern that deconstructs a 'List' into its head

127-- | A list constructor and pattern that deconstructs a 'List' into its head

119-- and tail. Corresponds to ':' for Haskell lists.

128-- and tail. Corresponds to ':' for Haskell lists.

120--

129--

121−-- @since 0.6.0

122pattern Cons :: a -> List a -> List a

130pattern Cons :: a -> List a -> List a

123pattern Cons x xs <-

131pattern Cons x xs <-

124− (fmap (second (List . Stream.fromStreamK))

132+ (fmap (second (List . CrossStream . Stream.fromStreamK))

125 . runIdentity . K.uncons . Stream.toStreamK . toStream

133 . runIdentity . K.uncons . Stream.toStreamK . toStream

126 -> Just (x, xs)

134 -> Just (x, xs)

127 )

135 )

128

136

129 where

137 where

130

138

131− Cons x xs = List $ Stream.cons x (toStream xs)

139+ Cons x xs = List $ CrossStream $ Stream.cons x (toStream xs)

132

140

133{-# COMPLETE Nil, Cons #-}

141{-# COMPLETE Nil, Cons #-}

134

142

@@ -139,7 +147,6 @@ pattern Cons x xs <-

139-- | Just like 'List' except that it has a zipping 'Applicative' instance

147-- | Just like 'List' except that it has a zipping 'Applicative' instance

140-- and no 'Monad' instance.

148-- and no 'Monad' instance.

141--

149--

142−-- @since 0.6.0

143newtype ZipList a = ZipList { toZipStream :: ZipStream Identity a }

150newtype ZipList a = ZipList { toZipStream :: ZipStream Identity a }

144 deriving

151 deriving

145 ( Show, Read, Eq, Ord

152 ( Show, Read, Eq, Ord

@@ -153,12 +160,10 @@ instance (a ~ Char) => IsString (ZipList a) where

153

160

154-- | Convert a 'ZipList' to a regular 'List'

161-- | Convert a 'ZipList' to a regular 'List'

155--

162--

156−-- @since 0.6.0

157fromZipList :: ZipList a -> List a

163fromZipList :: ZipList a -> List a

158−fromZipList (ZipList zs) = List $ getZipStream zs

164+fromZipList (ZipList zs) = List $ CrossStream (getZipStream zs)

159

165

160-- | Convert a regular 'List' to a 'ZipList'

166-- | Convert a regular 'List' to a 'ZipList'

161--

167--

162−-- @since 0.6.0

163toZipList :: List a -> ZipList a

168toZipList :: List a -> ZipList a

164toZipList = ZipList . ZipStream . toStream

169toZipList = ZipList . ZipStream . toStream

core/src/Streamly/Internal/Data/Stream/Cross.hsnew file+163

@@ -0,0 +1,163 @@

1+{-# LANGUAGE StandaloneDeriving #-}

2+{-# LANGUAGE UndecidableInstances #-}

3+

4+-- |

5+-- Module : Streamly.Internal.Data.Stream.Cross

7+--

8+-- License : BSD3

9+-- Maintainer : streamly@composewell.com

10+-- Stability : experimental

11+-- Portability : GHC

12+--

13+module Streamly.Internal.Data.Stream.Cross

14+ (

15+ CrossStream (..)

16+ )

17+where

18+

19+import Control.Applicative (liftA2)

20+import Control.Monad.Catch (MonadThrow, throwM)

21+import Control.Monad.IO.Class (MonadIO(..))

22+import Data.Functor.Identity (Identity(..))

23+import GHC.Exts (IsList(..), IsString(..))

24+import Streamly.Internal.Data.Stream.Type (Stream)

25+

26+import qualified Streamly.Internal.Data.Stream.Type as Stream

27+import qualified Streamly.Internal.Data.Stream.StreamK.Type as K

28+

29+-- $setup

30+-- >>> import Streamly.Internal.Data.Stream.Cross (CrossStream(..))

31+-- >>> import qualified Streamly.Data.Fold as Fold

32+-- >>> import qualified Streamly.Data.Stream as Stream

33+

34+------------------------------------------------------------------------------

35+-- Stream with a cross product style monad instance

36+------------------------------------------------------------------------------

37+

38+-- | A newtype wrapper for the 'Stream' type with a cross product style monad

39+-- instance.

40+--

41+-- Semigroup instance appends two streams.

42+--

43+-- A 'Monad' bind behaves like a @for@ loop:

44+--

45+-- >>> :{

46+-- Stream.fold Fold.toList $ getCrossStream $ do

47+-- x <- CrossStream (Stream.fromList [1,2]) -- foreach x in stream

48+-- return x

49+-- :}

50+-- [1,2]

51+--

52+-- Nested monad binds behave like nested @for@ loops:

53+--

54+-- >>> :{

55+-- Stream.fold Fold.toList $ getCrossStream $ do

56+-- x <- CrossStream (Stream.fromList [1,2]) -- foreach x in stream

57+-- y <- CrossStream (Stream.fromList [3,4]) -- foreach y in stream

58+-- return (x, y)

59+-- :}

60+-- [(1,3),(1,4),(2,3),(2,4)]

61+--

62+newtype CrossStream m a = CrossStream {getCrossStream :: Stream m a}

63+ deriving (Functor, Semigroup, Monoid, Foldable)

64+

65+-- deriving instance (Foldable m, Monad m) => Foldable (CrossStream m)

66+deriving instance Traversable (CrossStream Identity)

67+deriving instance IsList (CrossStream Identity a)

68+deriving instance (a ~ Char) => IsString (CrossStream Identity a)

69+deriving instance Eq a => Eq (CrossStream Identity a)

70+deriving instance Ord a => Ord (CrossStream Identity a)

71+deriving instance Show a => Show (CrossStream Identity a)

72+deriving instance Read a => Read (CrossStream Identity a)

73+

74+{-

75+instance Show a => Show (CrossStream Identity a) where

76+ showsPrec p dl = showParen (p > 10) $

77+ showString "fromList " . shows (toList dl)

78+

79+instance Read a => Read (ZipStream Identity a) where

80+ readPrec = parens $ prec 10 $ do

81+ Ident "fromList" <- lexP

82+ fromList <$> readPrec

83+ readListPrec = readListPrecDefault

84+-}

85+

86+------------------------------------------------------------------------------

87+-- Applicative

88+------------------------------------------------------------------------------

89+

90+{-# INLINE apSerial #-}

91+apSerial :: Monad m => Stream m (a -> b) -> Stream m a -> Stream m b

92+apSerial m1 m2 =

93+ Stream.fromStreamD $ Stream.toStreamD m1 <*> Stream.toStreamD m2

94+

95+{-# INLINE apSequence #-}

96+apSequence :: Monad m => Stream m a -> Stream m b -> Stream m b

97+apSequence m1 m2 =

98+ Stream.fromStreamD $ Stream.toStreamD m1 *> Stream.toStreamD m2

99+

100+{-# INLINE apDiscardSnd #-}

101+apDiscardSnd :: Monad m => Stream m a -> Stream m b -> Stream m a

102+apDiscardSnd m1 m2 =

103+ Stream.fromStreamD $ Stream.toStreamD m1 <* Stream.toStreamD m2

104+

105+-- Note: we need to define all the typeclass operations because we want to

106+-- INLINE them.

107+instance Monad m => Applicative (CrossStream m) where

108+ {-# INLINE pure #-}

109+ pure x = CrossStream (Stream.fromPure x)

110+

111+ {-# INLINE (<*>) #-}

112+ (CrossStream s1) <*> (CrossStream s2) = CrossStream (apSerial s1 s2)

113+ -- (<*>) = K.apSerial

114+

115+ {-# INLINE liftA2 #-}

116+ liftA2 f x = (<*>) (fmap f x)

117+

118+ {-# INLINE (*>) #-}

119+ (CrossStream s1) *> (CrossStream s2) = CrossStream (apSequence s1 s2)

120+ -- (*>) = K.apSerialDiscardFst

121+

122+ {-# INLINE (<*) #-}

123+ (CrossStream s1) <* (CrossStream s2) = CrossStream (apDiscardSnd s1 s2)

124+ -- (<*) = K.apSerialDiscardSnd

125+

126+------------------------------------------------------------------------------

127+-- Monad

128+------------------------------------------------------------------------------

129+

130+instance Monad m => Monad (CrossStream m) where

131+ return = pure

132+

133+ -- Benchmarks better with StreamD bind and pure:

134+ -- toList, filterAllout, *>, *<, >> (~2x)

135+ --

136+ -- pure = Stream . D.fromStreamD . D.fromPure

137+ -- m >>= f = D.fromStreamD $ D.concatMap (D.toStreamD . f) (D.toStreamD m)

138+

139+ -- Benchmarks better with CPS bind and pure:

140+ -- Prime sieve (25x)

141+ -- n binds, breakAfterSome, filterAllIn, state transformer (~2x)

142+ --

143+ {-# INLINE (>>=) #-}

144+ (>>=) (CrossStream m) f =

145+ CrossStream

146+ (Stream.fromStreamK

147+ $ K.bindWith

148+ K.serial

149+ (Stream.toStreamK m)

150+ (Stream.toStreamK . getCrossStream . f))

151+

152+ {-# INLINE (>>) #-}

153+ (>>) = (*>)

154+

155+------------------------------------------------------------------------------

156+-- Transformers

157+------------------------------------------------------------------------------

158+

159+instance (MonadIO m) => MonadIO (CrossStream m) where

160+ liftIO x = CrossStream (Stream.fromEffect $ liftIO x)

161+

162+instance (MonadThrow m) => MonadThrow (CrossStream m) where

163+ throwM x = CrossStream (Stream.fromEffect $ throwM x)

core/src/Streamly/Internal/Data/Stream/Generate.hs+4−3

@@ -96,6 +96,7 @@ import Prelude hiding (iterate, replicate, repeat, take)

96-- >>> import Control.Concurrent (threadDelay)

97-- >>> import Data.Function (fix, (&))

98-- >>> import Data.Semigroup (cycle1)

99+-- >>> import Streamly.Internal.Data.Stream.Cross (CrossStream(..))

99-- >>> import qualified Streamly.Data.Fold as Fold

100-- >>> import qualified Streamly.Data.Fold as Fold

100-- >>> import qualified Streamly.Data.Unfold as Unfold

101-- >>> import qualified Streamly.Data.Unfold as Unfold

101-- >>> import qualified Streamly.Internal.Data.Stream as Stream

102-- >>> import qualified Streamly.Internal.Data.Stream as Stream

@@ -372,10 +373,10 @@ iterateM step = fromStreamD . D.iterateM step

372-- >>> :{

373-- >>> :{

373-- main = Stream.fold (Fold.drainMapM print) $ Stream.mfix f

374-- main = Stream.fold (Fold.drainMapM print) $ Stream.mfix f

374-- where

375-- where

375−-- f action = do

376+-- f action = getCrossStream $ do

376-- let incr n act = fmap ((+n) . snd) $ unsafeInterleaveIO act

377-- let incr n act = fmap ((+n) . snd) $ unsafeInterleaveIO act

377−-- x <- Stream.sequence $ Stream.fromList [incr 1 action, incr 2 action]

378+-- x <- CrossStream (Stream.sequence $ Stream.fromList [incr 1 action, incr 2 action])

378−-- y <- Stream.fromList [4,5]

379+-- y <- CrossStream (Stream.fromList [4,5])

379-- return (x, y)

380-- return (x, y)

380-- :}

381-- :}

381--

382--

core/src/Streamly/Internal/Data/Stream/Top.hs+27−24

@@ -44,12 +44,12 @@ where

44

45import Control.Monad.Catch (MonadThrow)

46import Control.Monad.IO.Class (MonadIO(..))

47−import Control.Monad.Trans.Class (lift)

48import Control.Monad.Trans.State.Strict (get, put)

47import Control.Monad.Trans.State.Strict (get, put)

49import Data.Function ((&))

48import Data.Function ((&))

50import Data.IORef (newIORef, readIORef, modifyIORef')

49import Data.IORef (newIORef, readIORef, modifyIORef')

51import Data.Maybe (isJust)

50import Data.Maybe (isJust)

52import Streamly.Internal.Data.Stream.Common ()

51import Streamly.Internal.Data.Stream.Common ()

52+import Streamly.Internal.Data.Stream.Cross (CrossStream (..))

53import Streamly.Internal.Data.Stream.Type (Stream, fromStreamD, toStreamD)

54

55import qualified Data.List as List

@@ -163,10 +163,10 @@ sortBy cmp =

163-- /Pre-release/

164{-# INLINE crossJoin #-}

165crossJoin :: Monad m => Stream m a -> Stream m b -> Stream m (a, b)

166−crossJoin s1 s2 = do

166+crossJoin s1 s2 = getCrossStream $ do

167 -- XXX use concatMap instead?

168− a <- s1

168+ a <- CrossStream s1

169− b <- s2

169+ b <- CrossStream s2

170 return (a, b)

171

172-- XXX We can do this concurrently.

@@ -246,25 +246,25 @@ joinInnerMerge = undefined

246{-# INLINE joinLeft #-}

247joinLeft :: Monad m =>

248 (a -> b -> Bool) -> Stream m a -> Stream m b -> Stream m (a, Maybe b)

249−joinLeft eq s1 s2 = Stream.evalStateT (return False) $ do

249+joinLeft eq s1 s2 = Stream.evalStateT (return False) $ getCrossStream $ do

250− a <- Stream.liftInner s1

250+ a <- CrossStream (Stream.liftInner s1)

251 -- XXX should we use StreamD monad here?

252 -- XXX Is there a better way to perform some action at the end of a loop

253 -- iteration?

254− lift $ put False

254+ CrossStream (Stream.fromEffect $ put False)

255− let final = do

255+ let final = Stream.concatM $ do

256− r <- lift get

256+ r <- get

257 if r

258− then Stream.nil

258+ then pure Stream.nil

259− else Stream.fromPure Nothing

259+ else pure (Stream.fromPure Nothing)

260− b <- fmap Just (Stream.liftInner s2) <> final

260+ b <- CrossStream (fmap Just (Stream.liftInner s2) <> final)

261 case b of

262 Just b1 ->

263 if a `eq` b1

264 then do

265− lift $ put True

265+ CrossStream (Stream.fromEffect $ put True)

266 return (a, Just b1)

267− else Stream.nil

267+ else CrossStream Stream.nil

268 Nothing -> return (a, Nothing)

269

270-- | Like 'joinLeft' but works only on sorted streams.

@@ -323,29 +323,32 @@ joinOuter eq s1 s =

323 ) stream1 stream2

324 ) & Stream.catMaybes

325

326− go inputArr foundArr = Stream.evalStateT (return False) $ do

326+ evalState = Stream.evalStateT (return False) . getCrossStream

327+

328+ go inputArr foundArr = evalState $ do

327− a <- Stream.liftInner s1

329+ a <- CrossStream (Stream.liftInner s1)

328 -- XXX should we use StreamD monad here?

330 -- XXX should we use StreamD monad here?

329 -- XXX Is there a better way to perform some action at the end of a loop

331 -- XXX Is there a better way to perform some action at the end of a loop

330 -- iteration?

332 -- iteration?

331− lift $ put False

333+ CrossStream (Stream.fromEffect $ put False)

332− let final = do

334+ let final = Stream.concatM $ do

333− r <- lift get

335+ r <- get

334 if r

336 if r

335− then Stream.nil

337+ then pure Stream.nil

336− else Stream.fromPure Nothing

338+ else pure (Stream.fromPure Nothing)

337 (i, b) <-

339 (i, b) <-

338 let stream = Array.read inputArr

340 let stream = Array.read inputArr

341+ in CrossStream

339− in Stream.indexed $ fmap Just (Stream.liftInner stream) <> final

342+ (Stream.indexed $ fmap Just (Stream.liftInner stream) <> final)

340

343

341 case b of

344 case b of

342 Just b1 ->

345 Just b1 ->

343 if a `eq` b1

346 if a `eq` b1

344 then do

347 then do

345− lift $ put True

348+ CrossStream (Stream.fromEffect $ put True)

346 MA.putIndex i True foundArr

349 MA.putIndex i True foundArr

347 return (Just a, Just b1)

350 return (Just a, Just b1)

348− else Stream.nil

351+ else CrossStream Stream.nil

349 Nothing -> return (Just a, Nothing)

352 Nothing -> return (Just a, Nothing)

350

353

351-- | Like 'joinOuter' but works only on sorted streams.

354-- | Like 'joinOuter' but works only on sorted streams.

core/src/Streamly/Internal/Data/Stream/Type.hs+1−101

@@ -41,9 +41,6 @@ where

41#include "inline.hs"

42

43import Control.Applicative (liftA2)

44−import Control.Monad.Catch (MonadThrow, throwM)

45−import Control.Monad.IO.Class (MonadIO(..))

46−import Control.Monad.Trans.Class (MonadTrans(lift))

47import Data.Foldable (Foldable(foldl'), fold)

44import Data.Foldable (Foldable(foldl'), fold)

48import Data.Functor.Identity (Identity(..), runIdentity)

45import Data.Functor.Identity (Identity(..), runIdentity)

49import Data.Maybe (fromMaybe)

46import Data.Maybe (fromMaybe)

@@ -72,33 +69,9 @@ import qualified Streamly.Internal.Data.Stream.StreamK.Type as K

72--

69--

73-- >>> (<>) = Stream.append

70-- >>> (<>) = Stream.append

74--

71--

75−-- Monad bind maps a stream generator function on the stream and flattens the

76−-- resulting stream:

77−--

78−-- >>> (>>=) = flip (Stream.concatMapWith Stream.append)

79−--

80−-- A 'Monad' bind behaves like a @for@ loop:

81−--

82−-- >>> :{

83−-- Stream.fold Fold.toList $ do

84−-- x <- Stream.fromList [1,2] -- foreach x in stream

85−-- return x

86−-- :}

87−-- [1,2]

88−--

89−-- Nested monad binds behave like nested @for@ loops:

90−--

91−-- >>> :{

92−-- Stream.fold Fold.toList $ do

93−-- x <- Stream.fromList [1,2] -- foreach x in stream

94−-- y <- Stream.fromList [3,4] -- foreach y in stream

95−-- return (x, y)

96−-- :}

97−-- [(1,3),(1,4),(2,3),(2,4)]

98−--

99newtype Stream m a = Stream (K.Stream m a)

72newtype Stream m a = Stream (K.Stream m a)

100 -- XXX when deriving do we inherit an INLINE?

73 -- XXX when deriving do we inherit an INLINE?

101− deriving (Semigroup, Monoid, MonadTrans)

74+ deriving (Semigroup, Monoid)

102

75

103------------------------------------------------------------------------------

76------------------------------------------------------------------------------

104-- Conversions

77-- Conversions

@@ -165,79 +138,6 @@ instance Monad m => Functor (Stream m) where

165 {-# INLINE (<$) #-}

138 {-# INLINE (<$) #-}

166 (<$) = fmap . const

139 (<$) = fmap . const

167

140

168−------------------------------------------------------------------------------

169−-- Applicative

170−------------------------------------------------------------------------------

171−

172−{-# INLINE apSerial #-}

173−apSerial :: Monad m => Stream m (a -> b) -> Stream m a -> Stream m b

174−apSerial (Stream m1) (Stream m2) =

175− fromStreamK $ D.toStreamK $ D.fromStreamK m1 <*> D.fromStreamK m2

176−

177−{-# INLINE apSequence #-}

178−apSequence :: Monad m => Stream m a -> Stream m b -> Stream m b

179−apSequence (Stream m1) (Stream m2) =

180− fromStreamK $ D.toStreamK $ D.fromStreamK m1 *> D.fromStreamK m2

181−

182−{-# INLINE apDiscardSnd #-}

183−apDiscardSnd :: Monad m => Stream m a -> Stream m b -> Stream m a

184−apDiscardSnd (Stream m1) (Stream m2) =

185− fromStreamK $ D.toStreamK $ D.fromStreamK m1 <* D.fromStreamK m2

186−

187−-- Note: we need to define all the typeclass operations because we want to

188−-- INLINE them.

189−instance Monad m => Applicative (Stream m) where

190− {-# INLINE pure #-}

191− pure = fromStreamK . K.fromPure

192−

193− {-# INLINE (<*>) #-}

194− (<*>) = apSerial

195− -- (<*>) = K.apSerial

196−

197− {-# INLINE liftA2 #-}

198− liftA2 f x = (<*>) (fmap f x)

199−

200− {-# INLINE (*>) #-}

201− (*>) = apSequence

202− -- (*>) = K.apSerialDiscardFst

203−

204− {-# INLINE (<*) #-}

205− (<*) = apDiscardSnd

206− -- (<*) = K.apSerialDiscardSnd

207−

208−------------------------------------------------------------------------------

209−-- Monad

210−------------------------------------------------------------------------------

211−

212−instance Monad m => Monad (Stream m) where

213− return = pure

214−

215− -- Benchmarks better with StreamD bind and pure:

216− -- toList, filterAllout, *>, *<, >> (~2x)

217− --

218− -- pure = Stream . D.fromStreamD . D.fromPure

219− -- m >>= f = D.fromStreamD $ D.concatMap (D.toStreamD . f) (D.toStreamD m)

220−

221− -- Benchmarks better with CPS bind and pure:

222− -- Prime sieve (25x)

223− -- n binds, breakAfterSome, filterAllIn, state transformer (~2x)

224− --

225− {-# INLINE (>>=) #-}

226− (>>=) m f = fromStreamK $ K.bindWith K.serial (toStreamK m) (toStreamK . f)

227−

228− {-# INLINE (>>) #-}

229− (>>) = (*>)

230−

231−------------------------------------------------------------------------------

232−-- Transformers

233−------------------------------------------------------------------------------

234−

235−instance (MonadIO m) => MonadIO (Stream m) where

236− liftIO = lift . liftIO

237−

238−instance (MonadThrow m) => MonadThrow (Stream m) where

239− throwM = lift . throwM

240−

241------------------------------------------------------------------------------

141------------------------------------------------------------------------------

242-- Lists

142-- Lists

243------------------------------------------------------------------------------

143------------------------------------------------------------------------------

core/streamly-core.cabal+1

@@ -319,6 +319,7 @@ library

319 , Streamly.Internal.Data.Stream

320 , Streamly.Internal.Data.Stream.Chunked

321 , Streamly.Internal.Data.Stream.Zip

322+ , Streamly.Internal.Data.Stream.Cross

322 , Streamly.Internal.Data.Stream.TypeGen

323 , Streamly.Internal.Data.Stream.TypeGen

323 , Streamly.Internal.Data.List

324 , Streamly.Internal.Data.List

324

325

docs/User/Tutorials/ConcurrentStreams.hs+13−28

@@ -12,6 +12,9 @@

12

13module User.Tutorials.ConcurrentStreams

14 (

15+ -- * Imports

16+ -- $setup

17+

15 -- * Concurrent Streams

18 -- * Concurrent Streams

16 -- $concurrentStreams

19 -- $concurrentStreams

17

20

@@ -99,33 +102,14 @@ import Control.Monad.Trans.Class (MonadTrans (lift))

99-- XXX This tutorial has to be rewritten.

102-- XXX This tutorial has to be rewritten.

100

103

101-- $setup

104-- $setup

102−-- >>> :m

103−-- >>> :set -fno-warn-deprecations

104−-- >>> import Data.Function ((&))

105−-- >>> import Streamly.Prelude ((|:), (|&))

106−-- >>> import qualified Streamly.Prelude as Stream

107−-- >>> import qualified Streamly.Data.Fold as Fold

108−--

109−-- >>> import Control.Concurrent (threadDelay, myThreadId)

110−-- >>> :{

111−-- delay n = Stream.fromEffect $ do

112−-- threadDelay (n * 1000000)

113−-- tid <- myThreadId

114−-- putStrLn (show tid ++ ": Delay " ++ show n)

115−-- :}

116−--

117−-- >>> import System.IO (stdout, hSetBuffering, BufferMode(LineBuffering))

118−-- >>> hSetBuffering stdout LineBuffering

119−--

120−

121−-- $imports

122−--

123-- In most of example snippets we do not repeat the imports. Where imports are

105-- In most of example snippets we do not repeat the imports. Where imports are

124-- not explicitly specified use the imports shown below.

106-- not explicitly specified use the imports shown below.

125--

107--

126-- >>> :m

108-- >>> :m

109+-- >>> :set -fno-warn-deprecations

127-- >>> import Data.Function ((&))

110-- >>> import Data.Function ((&))

128-- >>> import Streamly.Prelude ((|:), (|&))

111-- >>> import Streamly.Prelude ((|:), (|&))

112+-- >>> import Streamly.Internal.Data.Stream.Cross (CrossStream(..))

129-- >>> import qualified Streamly.Prelude as Stream

113-- >>> import qualified Streamly.Prelude as Stream

130-- >>> import qualified Streamly.Data.Fold as Fold

114-- >>> import qualified Streamly.Data.Fold as Fold

131--

115--

@@ -726,7 +710,7 @@ import Control.Monad.Trans.Class (MonadTrans (lift))

726-- 'fromZipAsync' type combinator can be used to switch to parallel applicative

710-- 'fromZipAsync' type combinator can be used to switch to parallel applicative

727-- zip composition:

711-- zip composition:

728--

712--

729−-- >>> d n = delay n >> return n

713+-- >>> d n = getCrossStream (CrossStream (delay n) >> return n)

730-- >>> s1 = Stream.fromSerial $ d 2 <> d 4

714-- >>> s1 = Stream.fromSerial $ d 2 <> d 4

731-- >>> s2 = Stream.fromSerial $ d 3 <> d 1

715-- >>> s2 = Stream.fromSerial $ d 3 <> d 1

732-- >>> (Stream.toList $ Stream.fromZipAsync $ (,) <$> s1 <*> s2) >>= print

716-- >>> (Stream.toList $ Stream.fromZipAsync $ (,) <$> s1 <*> s2) >>= print

@@ -760,13 +744,14 @@ import Control.Monad.Trans.Class (MonadTrans (lift))

760-- >>> import Data.List (sum)

744-- >>> import Data.List (sum)

761-- >>> :{

745-- >>> :{

762-- main = do

746-- main = do

763−-- z <- Stream.toList

747+-- z <- Stream.toList

764-- $ Stream.fromSerial -- Serial monadic processing (sqrt below)

748-- $ Stream.fromSerial -- Serial monadic processing (sqrt below)

765−-- $ do

749+-- $ getCrossStream $ do

766−-- x2 <- Stream.concatForFoldableWith Stream.async [1..100] $ -- Concurrent @"for"@ loop

750+-- -- Concurrent @"for"@ loop

751+-- x2 <- CrossStream (Stream.concatForFoldableWith Stream.async [1..100] $

767−-- \x -> return $ x * x -- body of the loop

752+-- \x -> Stream.fromPure $ x * x) -- body of the loop

768−-- y2 <- Stream.concatForFoldableWith Stream.async [1..100] $

753+-- y2 <- CrossStream (Stream.concatForFoldableWith Stream.async [1..100] $

769−-- \y -> return $ y * y

754+-- \y -> Stream.fromPure $ y * y)

770-- return $ sqrt (x2 + y2)

755-- return $ sqrt (x2 + y2)

771-- print $ sum z

756-- print $ sum z

772-- :}

757-- :}

docs/User/Tutorials/Tutorial.hs+41−67

@@ -13,6 +13,9 @@

13

14module User.Tutorials.Tutorial

15 (

16+ -- * Imports

17+ -- $setup

18+

16 -- * Stream Types

19 -- * Stream Types

17 -- $streams

20 -- $streams

18

21

@@ -86,22 +89,16 @@ import Control.Monad.Trans.Class (MonadTrans (lift))

86-- XXX This tutorial has to be re-written.

89-- XXX This tutorial has to be re-written.

87

90

88-- $setup

91-- $setup

89−-- >>> :m

90−-- >>> :set -fno-warn-deprecations

91−-- >>> import Data.Function ((&))

92−-- >>> import Streamly.Prelude ((|:), (|&))

93−-- >>> import qualified Streamly.Prelude as Stream

94−-- >>> import qualified Streamly.Data.Fold as Fold

95−--

96−

97−-- $imports

98--

92--

99−-- In most of example snippets we do not repeat the imports. Where imports are

93+-- In most of example snippets we do not repeat the imports. Where

100−-- not explicitly specified use the imports shown below.

94+-- imports are not explicitly specified use the imports shown below.

101--

95--

102-- >>> :m

96-- >>> :m

97+-- >>> :set -fno-warn-deprecations

103-- >>> import Data.Function ((&))

98-- >>> import Data.Function ((&))

104-- >>> import Streamly.Prelude ((|:), (|&))

99-- >>> import Streamly.Prelude ((|:), (|&))

100+-- >>> import Streamly.Data.Stream (Stream)

101+-- >>> import Streamly.Internal.Data.Stream.Cross (CrossStream(..))

105-- >>> import qualified Streamly.Prelude as Stream

102-- >>> import qualified Streamly.Prelude as Stream

106-- >>> import qualified Streamly.Data.Fold as Fold

103-- >>> import qualified Streamly.Data.Fold as Fold

107--

104--

@@ -134,13 +131,8 @@ import Control.Monad.Trans.Class (MonadTrans (lift))

134-- world

131-- world

135-- ["hello","world"]

132-- ["hello","world"]

136--

133--

137−-- To create a singleton stream from a pure value use 'fromPure' or 'pure' and to

134+-- To create a singleton stream from a pure value use 'fromPure' and to

138−-- create a singleton stream from a monadic action use 'fromEffect'. Note that in

135+-- create a singleton stream from a monadic action use 'fromEffect'.

139−-- case of Zip applicative streams "pure" repeats the value to generate an

140−-- infinite stream.

141−--

142−-- >>> Stream.toList $ pure 1

143−-- [1]

144--

136--

145-- >>> Stream.toList $ Stream.fromPure 1

137-- >>> Stream.toList $ Stream.fromPure 1

146-- [1]

138-- [1]

@@ -354,11 +346,8 @@ import Control.Monad.Trans.Class (MonadTrans (lift))

354

346

355-- $regularSerial

347-- $regularSerial

356--

348--

357−-- The 'Monad' composition of the 'Serial' type behaves like a standard list

349+-- The 'Monad' composition of the 'CrossStream' type behaves like a standard list

358−-- transformer. This is the default when we do not use an explicit type

350+-- transformer.

359−-- combinator. However, the 'fromSerial' type combinator can be used to switch to

360−-- this style of composition. We will see how this style of composition works

361−-- in the following examples.

362--

351--

363-- Let's start with an example with a simple @for@ loop without any nesting.

352-- Let's start with an example with a simple @for@ loop without any nesting.

364-- For simplicity of illustration we are using streams of pure values in all

353-- For simplicity of illustration we are using streams of pure values in all

@@ -366,9 +355,9 @@ import Control.Monad.Trans.Class (MonadTrans (lift))

366-- instead.

355-- instead.

367--

356--

368-- >>> :{

357-- >>> :{

369−-- Stream.drain $ do

358+-- Stream.drain $ getCrossStream $ do

370−-- x <- Stream.fromFoldable [3,2,1]

359+-- x <- CrossStream (Stream.fromFoldable [3,2,1])

371−-- Stream.fromEffect $ print x

360+-- CrossStream (Stream.fromEffect $ print x)

372-- :}

361-- :}

373-- 3

362-- 3

374-- 2

363-- 2

@@ -384,19 +373,19 @@ import Control.Monad.Trans.Class (MonadTrans (lift))

384--

373--

385-- >>> :{

374-- >>> :{

386-- main =

375-- main =

387−-- Stream.drain $ do

376+-- Stream.drain $ getCrossStream $ do

388−-- x <- Stream.repeatM getLine

377+-- x <- CrossStream (Stream.repeatM getLine)

389−-- Stream.fromEffect $ putStrLn x

378+-- CrossStream (Stream.fromEffect $ putStrLn x)

390-- :}

379-- :}

391--

380--

392-- When multiple streams are composed using this style they nest in a DFS

381-- When multiple streams are composed using this style they nest in a DFS

393-- manner:

382-- manner:

394--

383--

395-- >>> :{

384-- >>> :{

396−-- Stream.drain $ do

385+-- Stream.drain $ getCrossStream $ do

397−-- x <- Stream.fromFoldable [1,2]

386+-- x <- CrossStream (Stream.fromFoldable [1,2])

398−-- y <- Stream.fromFoldable [3,4]

387+-- y <- CrossStream (Stream.fromFoldable [3,4])

399−-- Stream.fromEffect $ print (x, y)

388+-- CrossStream (Stream.fromEffect $ print (x, y))

400-- :}

389-- :}

401-- (1,3)

390-- (1,3)

402-- (1,4)

391-- (1,4)

@@ -412,14 +401,16 @@ import Control.Monad.Trans.Class (MonadTrans (lift))

412

401

413-- $interleavedNesting

402-- $interleavedNesting

414--

403--

404+-- TBD: generate the WSerial type using the TH utils.

405+--

415-- The 'Monad' composition of 'WSerial' type interleaves the iterations of

406-- The 'Monad' composition of 'WSerial' type interleaves the iterations of

416-- outer and inner loops in a nested loop composition.

407-- outer and inner loops in a nested loop composition.

417--

408--

418−-- >>> :{

409+-- >> :{

419−-- Stream.drain $ Stream.fromWSerial $ do

410+-- Stream.drain $ getWSerial $ do

420−-- x <- Stream.fromFoldable [1,2]

411+-- x <- WSerialStream (Stream.fromFoldable [1,2])

421−-- y <- Stream.fromFoldable [3,4]

412+-- y <- WSerialStream (Stream.fromFoldable [3,4])

422−-- Stream.fromEffect $ print (x, y)

413+-- WSerialStream (Stream.fromEffect $ print (x, y))

423-- :}

414-- :}

424-- (1,3)

415-- (1,3)

425-- (2,3)

416-- (2,3)

@@ -436,36 +427,19 @@ import Control.Monad.Trans.Class (MonadTrans (lift))

436

427

437-- $monadExercise

428-- $monadExercise

438--

429--

439−-- Streamly code is usually written in a way that is agnostic of the

440−-- specific monadic composition type. We use a polymorphic type with a

441−-- 'IsStream' type class constraint. When running the stream we can choose the

442−-- specific mode of composition. For example take a look at the following code.

443−--

444-- >>> :{

430-- >>> :{

445−-- composed :: (Stream.IsStream t, Monad (t IO)) => t IO ()

431+-- composed :: Stream IO ()

446−-- composed = do

432+-- composed = getCrossStream $ do

447−-- sz <- sizes

433+-- sz <- CrossStream sizes

448−-- cl <- colors

434+-- cl <- CrossStream colors

449−-- sh <- shapes

435+-- sh <- CrossStream shapes

450−-- Stream.fromEffect $ print (sz, cl, sh)

436+-- CrossStream (Stream.fromEffect $ print (sz, cl, sh))

451-- where

437-- where

452-- sizes = Stream.fromFoldable [1, 2, 3]

438-- sizes = Stream.fromFoldable [1, 2, 3]

453-- colors = Stream.fromFoldable ["red", "green", "blue"]

439-- colors = Stream.fromFoldable ["red", "green", "blue"]

454-- shapes = Stream.fromFoldable ["triangle", "square", "circle"]

440-- shapes = Stream.fromFoldable ["triangle", "square", "circle"]

455-- :}

441-- :}

456--

442--

457−-- Now we can interpret this in whatever way we want:

458−--

459−-- @

460−-- main = Stream.'drain' $ Stream.'fromSerial' $ composed

461−-- main = Stream.'drain' $ Stream.'fromWSerial' $ composed

462−-- main = Stream.'drain' $ Stream.'fromAsync' $ composed

463−-- main = Stream.'drain' $ Stream.'fromWAsync' $ composed

464−-- main = Stream.'drain' $ Stream.'fromParallel' $ composed

465−-- @

466−--

467−-- As an exercise try to figure out the output of this code for each mode of

468−-- composition.

469

443

470-- $functor

444-- $functor

471--

445--

@@ -482,17 +456,17 @@ import Control.Monad.Trans.Class (MonadTrans (lift))

482-- $applicative

456-- $applicative

483--

457--

484-- Applicative is precisely the same as the 'ap' operation of 'Monad'. For

458-- Applicative is precisely the same as the 'ap' operation of 'Monad'. For

485−-- zipping applicatives separate types 'ZipSerial' and 'ZipAsync' are

459+-- zipping applicatives separate types 'ZipStream' and 'ZipAsync' are

486-- provided.

460-- provided.

487--

461--

488−-- The following is an example of 'Serial' applicative, it runs all iterations

462+-- The following is an example of 'CrossStream' applicative, it runs all iterations

489-- serially:

463-- serially:

490--

464--

491−-- >>> p n = Stream.fromEffect (print n) >> return n

465+-- >>> p n = Stream.fromEffect (print n >> pure n)

492-- >>> s1 = p 1 <> p 2

466-- >>> s1 = p 1 <> p 2

493-- >>> s2 = p 3 <> p 4

467-- >>> s2 = p 3 <> p 4

494--

468--

495−-- >>> (Stream.toList $ Stream.fromSerial $ (,) <$> s1 <*> s2) >>= print

469+-- >>> (Stream.toList $ getCrossStream $ (,) <$> CrossStream s1 <*> CrossStream s2) >>= print

496-- 1

470-- 1

497-- 3

471-- 3

498-- 4

472-- 4

@@ -504,7 +478,7 @@ import Control.Monad.Trans.Class (MonadTrans (lift))

504-- Similarly, 'WSerial' applicative runs the iterations in an interleaved

478-- Similarly, 'WSerial' applicative runs the iterations in an interleaved

505-- order but being serial it too takes a total of 17 seconds:

479-- order but being serial it too takes a total of 17 seconds:

506--

480--

507−-- >>> (Stream.toList $ Stream.fromWSerial $ (,) <$> s1 <*> s2) >>= print

481+-- >> (Stream.toList $ Stream.WSerialStream $ (,) <$> WSerialStream s1 <*> WSerialStream s2) >>= print

508-- 1

482-- 1

509-- 3

483-- 3

510-- 2

484-- 2

@@ -528,7 +502,7 @@ import Control.Monad.Trans.Class (MonadTrans (lift))

528-- 'fromZipSerial' type combinator can be used to switch to serial applicative

502-- 'fromZipSerial' type combinator can be used to switch to serial applicative

529-- zip composition:

503-- zip composition:

530--

504--

531−-- >>> p n = Stream.fromEffect (print n) >> return n

505+-- >>> p n = Stream.fromEffect (print n >> pure n)

532-- >>> s1 = Stream.fromSerial $ p 1 <> p 2

506-- >>> s1 = Stream.fromSerial $ p 1 <> p 2

533-- >>> s2 = Stream.fromSerial $ p 3 <> p 4

507-- >>> s2 = Stream.fromSerial $ p 3 <> p 4

534-- >>> (Stream.toList $ Stream.fromZipSerial $ (,) <$> s1 <*> s2) >>= print

508-- >>> (Stream.toList $ Stream.fromZipSerial $ (,) <$> s1 <*> s2) >>= print

src/Streamly/Internal/Data/Fold/Extra.hs+1

@@ -81,6 +81,7 @@ import Streamly.Internal.Data.Fold

81

82-- $setup

83-- >>> :m

84+-- >>> :set -XFlexibleContexts

84-- >>> import qualified Streamly.Data.Fold as Fold

85-- >>> import qualified Streamly.Data.Fold as Fold

85-- >>> import qualified Streamly.Data.Stream as Stream

86-- >>> import qualified Streamly.Data.Stream as Stream

86-- >>> import qualified Streamly.Internal.Data.Fold.Extra as Fold

87-- >>> import qualified Streamly.Internal.Data.Fold.Extra as Fold

src/Streamly/Internal/Data/Stream/Concurrent.hs+1−1

@@ -606,7 +606,7 @@ parApply :: MonadAsync m =>

606parApply modifier stream1 stream2 =

607 parConcatMap

608 modifier

609− (\g -> parConcatMap modifier (pure . g) stream2)

609+ (\g -> parConcatMap modifier (Stream.fromPure . g) stream2)

610 stream1

611

612{-# INLINE apply #-}

src/Streamly/Internal/Data/Stream/IsStream/Generate.hs−24

@@ -538,30 +538,6 @@ iterateMSerial step = fromStreamD . D.iterateM step

538-- returned by the stream monad. Thus, we can use the argument to construct

539-- itself.

540--

541−-- In the following example, the argument @action@ of the function @f@

542−-- represents the tuple @(x,y)@ returned by it in a given iteration. We define

543−-- the first element of the tuple in terms of the second.

544−--

545−-- >>> import Streamly.Internal.Data.Stream.IsStream as Stream

546−-- >>> import System.IO.Unsafe (unsafeInterleaveIO)

547−--

548−-- >>> :{

549−-- main = Stream.mapM_ print $ Stream.mfix f

550−-- where

551−-- f action = do

552−-- let incr n act = fmap ((+n) . snd) $ unsafeInterleaveIO act

553−-- x <- Stream.fromListM [incr 1 action, incr 2 action]

554−-- y <- Stream.fromList [4,5]

555−-- return (x, y)

556−-- :}

557−--

558−-- Note: you cannot achieve this by just changing the order of the monad

559−-- statements because that would change the order in which the stream elements

560−-- are generated.

561−--

562−-- Note that the function @f@ must be lazy in its argument, that's why we use

563−-- 'unsafeInterleaveIO' on @action@ because IO monad is strict.

564−--

565-- /Pre-release/

541-- /Pre-release/

566{-# INLINE mfix #-}

542{-# INLINE mfix #-}

567mfix :: (IsStream t, Monad m) => (m a -> t m a) -> t m a

543mfix :: (IsStream t, Monad m) => (m a -> t m a) -> t m a

src/Streamly/Internal/Data/Stream/IsStream/Reduce.hs+1−1

@@ -1499,7 +1499,7 @@ classifySessionsByGeneric _ tick reset ejectPred tmout

1499-- $ Stream.classifySessionsBy 1 False (const (return False)) 3 (Fold.take 3 Fold.toList)

1500-- $ Stream.timestamped

1501-- $ Stream.delay 0.1

1502−-- $ (,) <$> Stream.fromList [1,2,3] <*> Stream.fromList ['a','b','c']

1502+-- $ Stream.fromList ((,) <$> [1,2,3] <*> ['a','b','c'])

1503-- :}

1504-- (1,"abc")

1505-- (2,"abc")

src/Streamly/Internal/Data/Stream/IsStream/Top.hs+14−10

@@ -41,10 +41,10 @@ module Streamly.Internal.Data.Stream.IsStream.Top {-# DEPRECATED "Please use \"S

41 , joinInner

42 , joinInnerMap

43 , joinInnerMerge

44− , joinLeft

44+ -- , joinLeft

45 , mergeLeftJoin

46 , joinLeftMap

47− , joinOuter

47+ -- , joinOuter

48 , mergeOuterJoin

49 , joinOuterMap

50 )

@@ -54,12 +54,12 @@ where

54

55import Control.Monad.Catch (MonadCatch)

56import Control.Monad.IO.Class (MonadIO(..))

57−import Control.Monad.Trans.Class (lift)

57+-- import Control.Monad.Trans.Class (lift)

58−import Control.Monad.Trans.State.Strict (get, put)

58+-- import Control.Monad.Trans.State.Strict (get, put)

59−import Data.Function ((&))

59+-- import Data.Function ((&))

60import Data.IORef (newIORef, readIORef, modifyIORef')

61import Data.Kind (Type)

62−import Data.Maybe (isJust)

62+-- import Data.Maybe (isJust)

63import Streamly.Internal.Control.Concurrent (MonadAsync)

64import Streamly.Internal.Data.Stream.IsStream.Common (concatM)

65import Streamly.Internal.Data.Stream.IsStream.Type

@@ -69,14 +69,14 @@ import Streamly.Internal.Data.Time.Units (NanoSecond64(..), toRelTime64)

69

70import qualified Data.List as List

71import qualified Data.Map.Strict as Map

72−import qualified Streamly.Internal.Data.Array.Generic as Array

72+-- import qualified Streamly.Internal.Data.Array.Generic as Array

73− (fromStream, length, read)

73+-- (fromStream, length, read)

74−import qualified Streamly.Data.Array.Mut as MA

74+-- import qualified Streamly.Data.Array.Mut as MA

75import qualified Streamly.Internal.Data.Fold as Fold

76 (one, last, toStream, toStreamRev)

77import qualified Streamly.Internal.Data.Parser as Parser

78 (groupByRollingEither)

79−import qualified Streamly.Internal.Data.Stream.IsStream.Lift as Stream

79+-- import qualified Streamly.Internal.Data.Stream.IsStream.Lift as Stream

80import qualified Streamly.Internal.Data.Stream.IsStream.Eliminate as Stream

81import qualified Streamly.Internal.Data.Stream.IsStream.Generate as Stream

82import qualified Streamly.Internal.Data.Stream.IsStream.Expand as Stream

@@ -386,6 +386,7 @@ joinInnerMap s1 s2 =

386joinInnerMerge :: (a -> b -> Ordering) -> t m a -> t m b -> t m (a, b)

387joinInnerMerge = undefined

388

389+{-

389-- XXX We can do this concurrently.

390-- XXX We can do this concurrently.

390-- XXX If the second stream is sorted and passed as an Array or a seek capable

391-- XXX If the second stream is sorted and passed as an Array or a seek capable

391-- stream then we could use binary search if we have an Ord instance or

392-- stream then we could use binary search if we have an Ord instance or

@@ -433,6 +434,7 @@ joinLeft eq s1 s2 = Stream.evalStateT (return False) $ do

433 return (a, Just b1)

434 return (a, Just b1)

434 else Stream.nil

435 else Stream.nil

435 Nothing -> return (a, Nothing)

436 Nothing -> return (a, Nothing)

437+-}

436

438

437-- | Like 'joinLeft' but uses a hashmap for efficiency.

439-- | Like 'joinLeft' but uses a hashmap for efficiency.

438--

440--

@@ -468,6 +470,7 @@ mergeLeftJoin :: -- Monad m =>

468 (a -> b -> Ordering) -> t m a -> t m b -> t m (a, Maybe b)

470 (a -> b -> Ordering) -> t m a -> t m b -> t m (a, Maybe b)

469mergeLeftJoin _eq _s1 _s2 = undefined

471mergeLeftJoin _eq _s1 _s2 = undefined

470

472

473+{-

471-- XXX We can do this concurrently.

474-- XXX We can do this concurrently.

472--

475--

473-- | For all elements in @t m a@, for all elements in @t m b@ if @a@ and @b@

476-- | For all elements in @t m a@, for all elements in @t m b@ if @a@ and @b@

@@ -536,6 +539,7 @@ joinOuter eq s1 s =

536 return (Just a, Just b1)

539 return (Just a, Just b1)

537 else Stream.nil

540 else Stream.nil

538 Nothing -> return (Just a, Nothing)

541 Nothing -> return (Just a, Nothing)

542+-}

539

543

540-- Put the b's that have been paired, in another hash or mutate the hash to set

544-- Put the b's that have been paired, in another hash or mutate the hash to set

541-- a flag. At the end go through @t m b@ and find those that are not in that

545-- a flag. At the end go through @t m b@ and find those that are not in that

src/Streamly/Internal/Data/Stream/IsStream/Transform.hs+1−1

@@ -318,7 +318,7 @@ transform pipe xs = fromStreamD $ D.transform pipe (toStreamD xs)

318--

319-- Find if any element in the stream is 'True':

320--

321−-- >>> Stream.toList $ Stream.foldrS (\x xs -> if odd x then return True else xs) (return False) $ (Stream.fromList (2:4:5:undefined) :: Stream.SerialT IO Int)

321+-- >>> Stream.toList $ Stream.foldrS (\x xs -> if odd x then (Stream.fromPure True) else xs) (Stream.fromPure False) $ (Stream.fromList (2:4:5:undefined) :: Stream.SerialT IO Int)

322-- [True]

323--

324-- Map (+2) on odd elements and filter out the even elements:

src/Streamly/Internal/Data/Stream/Serial.hs+79−3

@@ -68,13 +68,14 @@ import Streamly.Data.Stream (Stream)

68

69import qualified Streamly.Data.Fold as Fold

70import qualified Streamly.Data.Stream as Stream

71−import qualified Streamly.Internal.Data.Stream as Stream (toStreamK, fromStreamK)

71+import qualified Streamly.Internal.Data.Stream as Stream

72+ (toStreamD, fromStreamD, toStreamK, fromStreamK)

72import qualified Streamly.Internal.Data.Stream.Common as P

73import qualified Streamly.Internal.Data.Stream.Common as P

73import qualified Streamly.Internal.Data.Stream.StreamD as D

74import qualified Streamly.Internal.Data.Stream.StreamD as D

74 (fromStreamK, toStreamK, mapM)

75 (fromStreamK, toStreamK, mapM)

75import qualified Streamly.Internal.Data.Stream.StreamK.Type as K

76import qualified Streamly.Internal.Data.Stream.StreamK.Type as K

76 (Stream, mkStream, foldStream, cons, consM, nil, concatMapWith, fromPure

77 (Stream, mkStream, foldStream, cons, consM, nil, concatMapWith, fromPure

77− , bindWith, interleave, interleaveFst, interleaveMin)

78+ , bindWith, interleave, interleaveFst, interleaveMin, serial)

78

79

79import Prelude hiding (map, mapM, repeat, filter)

80import Prelude hiding (map, mapM, repeat, filter)

80

81

@@ -95,9 +96,84 @@ withLocal f m =

95 in K.foldStream st yieldk single (local f stp) m

96 in K.foldStream st yieldk single (local f stp) m

96

97

97------------------------------------------------------------------------------

98------------------------------------------------------------------------------

98−-- mtl orphan instances

99+-- Applicative - orphan instances

99------------------------------------------------------------------------------

100------------------------------------------------------------------------------

100

101

102+{-# INLINE apSerial #-}

103+apSerial :: Monad m => Stream m (a -> b) -> Stream m a -> Stream m b

104+apSerial m1 m2 =

105+ Stream.fromStreamD $ Stream.toStreamD m1 <*> Stream.toStreamD m2

106+

107+{-# INLINE apSequence #-}

108+apSequence :: Monad m => Stream m a -> Stream m b -> Stream m b

109+apSequence m1 m2 =

110+ Stream.fromStreamD $ Stream.toStreamD m1 *> Stream.toStreamD m2

111+

112+{-# INLINE apDiscardSnd #-}

113+apDiscardSnd :: Monad m => Stream m a -> Stream m b -> Stream m a

114+apDiscardSnd m1 m2 =

115+ Stream.fromStreamD $ Stream.toStreamD m1 <* Stream.toStreamD m2

116+

117+-- Note: we need to define all the typeclass operations because we want to

118+-- INLINE them.

119+instance Monad m => Applicative (Stream m) where

120+ {-# INLINE pure #-}

121+ pure = Stream.fromStreamK . K.fromPure

122+

123+ {-# INLINE (<*>) #-}

124+ (<*>) = apSerial

125+ -- (<*>) = K.apSerial

126+

127+ {-# INLINE liftA2 #-}

128+ liftA2 f x = (<*>) (fmap f x)

129+

130+ {-# INLINE (*>) #-}

131+ (*>) = apSequence

132+ -- (*>) = K.apSerialDiscardFst

133+

134+ {-# INLINE (<*) #-}

135+ (<*) = apDiscardSnd

136+ -- (<*) = K.apSerialDiscardSnd

137+

138+------------------------------------------------------------------------------

139+-- Monad - orphan instances

140+------------------------------------------------------------------------------

141+

142+instance Monad m => Monad (Stream m) where

143+ return = pure

144+

145+ -- Benchmarks better with StreamD bind and pure:

146+ -- toList, filterAllout, *>, *<, >> (~2x)

147+ --

148+ -- pure = Stream . D.fromStreamD . D.fromPure

149+ -- m >>= f = D.fromStreamD $ D.concatMap (D.toStreamD . f) (D.toStreamD m)

150+

151+ -- Benchmarks better with CPS bind and pure:

152+ -- Prime sieve (25x)

153+ -- n binds, breakAfterSome, filterAllIn, state transformer (~2x)

154+ --

155+ {-# INLINE (>>=) #-}

156+ (>>=) m f =

157+ Stream.fromStreamK

158+ $ K.bindWith K.serial (Stream.toStreamK m) (Stream.toStreamK . f)

159+

160+ {-# INLINE (>>) #-}

161+ (>>) = (*>)

162+

163+------------------------------------------------------------------------------

164+-- Transformers - orphan instances

165+------------------------------------------------------------------------------

166+

167+instance (MonadIO m) => MonadIO (Stream m) where

168+ liftIO = lift . liftIO

169+

170+instance (MonadThrow m) => MonadThrow (Stream m) where

171+ throwM = lift . throwM

172+

173+instance MonadTrans Stream where

174+ {-# INLINE lift #-}

175+ lift = Stream.fromEffect

176+

101instance (MonadReader r m) => MonadReader r (Stream m) where

177instance (MonadReader r m) => MonadReader r (Stream m) where

102 ask = lift ask

178 ask = lift ask

103

179

src/Streamly/Internal/Data/Stream/Time.hs+1−1

@@ -714,7 +714,7 @@ classifySessionsByGeneric _ tick reset ejectPred tmout

714-- $ Stream.classifySessionsBy 1 False (const (return False)) 3 (Fold.take 3 Fold.toList)

715-- $ Stream.timestamped

716-- $ Stream.delay 0.1

717−-- $ (,) <$> Stream.fromList [1,2,3] <*> Stream.fromList ['a','b','c']

717+-- $ Stream.fromList ((,) <$> [1,2,3] <*> ['a','b','c'])

718-- :}

719-- (1,"abc")

720-- (2,"abc")

test/Streamly/Test/Data/List.hs+3−3

@@ -58,7 +58,7 @@ main = hspec $

58

59 it "Monad comprehension" $ do

60 [(x,y) | x <- [1..2], y <- [1..2]] `shouldBe`

61− ([(1,1), (1,2), (2,1), (2,2)] :: Stream Identity (Int, Int))

61+ ([(1,1), (1,2), (2,1), (2,2)] :: List (Int, Int))

62

63 it "Foldable (sum)" $ sum ([1..3] :: Stream Identity Int)

64 `shouldBe` 6

@@ -110,7 +110,7 @@ main = hspec $

110 it "Show instance" $ do

111 show ([1..3] :: List Int) `shouldBe`

112#ifdef USE_STREAMLY_LIST

113− "List {toStream = fromList [1,2,3]}"

113+ "fromList [1,2,3]"

114#else

115 "[1,2,3]"

116#endif

@@ -118,7 +118,7 @@ main = hspec $

118 it "Read instance" $ do

119 (read

120#ifdef USE_STREAMLY_LIST

121− "List {toStream = fromList [1,2,3]}"

121+ "fromList [1,2,3]"

122#else

123 "[1,2,3]"

124#endif

test/Streamly/Test/Data/Stream/Concurrent.hs+9−9

@@ -122,17 +122,17 @@ exceptionPropagation ::

122 (Stream IO Int -> Stream IO Int -> Stream IO Int) -> Spec

123exceptionPropagation f = do

124 it "append throwM, nil" $

125− try (tl (throwM (ExampleException "E") `f` Stream.nil))

125+ try (tl (Stream.fromEffect (throwM (ExampleException "E")) `f` Stream.nil))

126 `shouldReturn`

127 (Left (ExampleException "E") :: Either ExampleException [Int])

128 it "append nil, throwM" $

129− try (tl (Stream.nil `f` throwM (ExampleException "E")))

129+ try (tl (Stream.nil `f` Stream.fromEffect (throwM (ExampleException "E"))))

130 `shouldReturn`

131 (Left (ExampleException "E") :: Either ExampleException [Int])

132 it "append nested throwM" $ do

133 let nested =

134 Stream.fromFoldable [1..10]

135− `f` throwM (ExampleException "E")

135+ `f` Stream.fromEffect (throwM (ExampleException "E"))

136 `f` Stream.fromFoldable [1..10]

137 try (tl (Stream.nil `f` nested `f` Stream.fromFoldable [1..10]))

138 `shouldReturn`

@@ -152,8 +152,8 @@ exceptionPropagation f = do

152 in bind s1 $ \x ->

153 bind s2 $ \y ->

154 if x + y > 10

155− then throwM (ExampleException "E")

155+ then Stream.fromEffect (throwM (ExampleException "E"))

156− else return (x + y)

156+ else Stream.fromPure (x + y)

157 )

158 `shouldReturn`

159 (Left (ExampleException "E") :: Either ExampleException [Int])

@@ -289,14 +289,14 @@ main = hspec

289 in cmp stream (==) [1,2,3,4,5,6,7,8,9,10]

290

291 prop1 "apply (async arg1)"

292− $ let s1 = Async.apply (pure (,)) (pure 1 `Async.append2` pure 2)

292+ $ let s1 = Async.apply (Stream.fromPure (,)) (Stream.fromPure 1 `Async.append2` Stream.fromPure 2)

293− s2 = Async.apply s1 (pure 3) :: Stream IO (Int, Int)

293+ s2 = Async.apply s1 (Stream.fromPure 3) :: Stream IO (Int, Int)

294 xs = Stream.fold Fold.toList s2

295 in sort <$> xs `shouldReturn` [(1, 3), (2, 3)]

296

297 prop1 "apply (async arg2)"

298− $ let s1 = pure (1,)

298+ $ let s1 = Stream.fromPure (1,)

299− s2 = Async.apply s1 (pure 2 `Async.append2` pure 3)

299+ s2 = Async.apply s1 (Stream.fromPure 2 `Async.append2` Stream.fromPure 3)

300 xs = Stream.fold Fold.toList s2 :: IO [(Int, Int)]

301 in sort <$> xs `shouldReturn` [(1, 2), (1, 3)]

302

test/Streamly/Test/Network/Inet/TCP.hs+12−8

@@ -81,14 +81,18 @@ remoteAddr :: (Word8,Word8,Word8,Word8)

81remoteAddr = (127, 0, 0, 1)

82

83sender :: PortNumber -> MVar () -> Stream IO Char

84−sender port sem = do

84+sender port sem = Stream.before action stream

85− _ <- liftIO $ takeMVar sem

85+

86+ where

87+

86− liftIO $ threadDelay 1000000 -- wait for server

88+ action = liftIO (takeMVar sem >> threadDelay 1000000) -- wait for server

89+

90+ stream =

87− Stream.replicate 1000 testData -- Stream IO String

91+ Stream.replicate 1000 testData -- Stream IO String

88− & Stream.concatMap Stream.fromList -- Stream IO Char

92+ & Stream.concatMap Stream.fromList -- Stream IO Char

89− & Unicode.encodeLatin1 -- Stream IO Word8

93+ & Unicode.encodeLatin1 -- Stream IO Word8

90− & TCP.pipeBytes remoteAddr port -- Stream IO Word8

94+ & TCP.pipeBytes remoteAddr port -- Stream IO Word8

91− & Unicode.decodeLatin1 -- Stream IO Char

95+ & Unicode.decodeLatin1 -- Stream IO Char

92

96

93execute

97execute

94 :: Unfold.Unfold IO PortNumber Socket

98 :: Unfold.Unfold IO PortNumber Socket

test/Streamly/Test/Network/Socket.hs+12−8

@@ -94,14 +94,18 @@ remoteAddr :: (Word8,Word8,Word8,Word8)

94remoteAddr = (127, 0, 0, 1)

95

96sender :: PortNumber -> MVar () -> Stream IO Char

97−sender port sem = do

97+sender port sem = Stream.before action stream

98− _ <- liftIO $ takeMVar sem

98+

99+ where

100+

99− liftIO $ threadDelay 1000000 -- wait for server

101+ action = liftIO (takeMVar sem >> threadDelay 1000000) -- wait for server

102+

103+ stream =

100− Stream.replicate 1000 testData -- Stream IO String

104+ Stream.replicate 1000 testData -- Stream IO String

101− & Stream.concatMap Stream.fromList -- Stream IO Char

105+ & Stream.concatMap Stream.fromList -- Stream IO Char

102− & Unicode.encodeLatin1 -- Stream IO Word8

106+ & Unicode.encodeLatin1 -- Stream IO Word8

103− & TCP.pipeBytes remoteAddr port -- Stream IO Word8

107+ & TCP.pipeBytes remoteAddr port -- Stream IO Word8

104− & Unicode.decodeLatin1 -- Stream IO Char

108+ & Unicode.decodeLatin1 -- Stream IO Char

105

109

106execute :: PortNumber -> Int -> (Socket -> IO ()) -> IO (Stream IO Char)

110execute :: PortNumber -> Int -> (Socket -> IO ()) -> IO (Stream IO Char)

107execute port size handler = do

111execute port size handler = do

test/Streamly/Test/Prelude/Top.hs+6−2

@@ -87,6 +87,7 @@ joinOuterList ls0 ls1 =

87 v4 = filter (\(_, a2, _) -> isNothing a2) v3

88 in v2 ++ v4

89

90+{-

90joinOuter :: Property

91joinOuter :: Property

91joinOuter =

92joinOuter =

92 forAll (listOf (chooseInt (min_value, max_value))) $ \ls0 ->

93 forAll (listOf (chooseInt (min_value, max_value))) $ \ls0 ->

@@ -103,6 +104,7 @@ joinOuter =

103 (map (\a -> (a, a)) ls1)

104 (map (\a -> (a, a)) ls1)

104 v3 = map (\(_, v10, v20) -> (v10, v20)) v2

105 v3 = map (\(_, v10, v20) -> (v10, v20)) v2

105 assert (sort v1 == sort v3)

106 assert (sort v1 == sort v3)

107+-}

106

108

107

109

108joinOuterMap :: Property

110joinOuterMap :: Property

@@ -133,6 +135,7 @@ joinLeftList ls0 ls1 =

133 else return (fst i, fst i, Nothing)

135 else return (fst i, fst i, Nothing)

134 in v

136 in v

135

137

138+{-

136joinLeft :: Property

139joinLeft :: Property

137joinLeft =

140joinLeft =

138 forAll (listOf (chooseInt (min_value, max_value))) $ \ls0 ->

141 forAll (listOf (chooseInt (min_value, max_value))) $ \ls0 ->

@@ -150,6 +153,7 @@ joinLeft =

150 let v2 = joinLeftList (map (\a -> (a,a)) ls0) (map (\a -> (a,a)) ls1)

153 let v2 = joinLeftList (map (\a -> (a,a)) ls0) (map (\a -> (a,a)) ls1)

151 v3 = map (\ (_, x1, x2) -> (x1, x2)) v2

154 v3 = map (\ (_, x1, x2) -> (x1, x2)) v2

152 assert (v1 == v3)

155 assert (v1 == v3)

156+-}

153

157

154joinLeftMap :: Property

158joinLeftMap :: Property

155joinLeftMap =

159joinLeftMap =

@@ -209,9 +213,9 @@ main = hspec $ do

209 -- Joins

213 -- Joins

210 prop "joinInner" Main.joinInner

214 prop "joinInner" Main.joinInner

211 prop "joinInnerMap" Main.joinInnerMap

215 prop "joinInnerMap" Main.joinInnerMap

212− prop "joinOuter" Main.joinOuter

216+ -- prop "joinOuter" Main.joinOuter

213 prop "joinOuterMap" Main.joinOuterMap

217 prop "joinOuterMap" Main.joinOuterMap

214− prop "joinLeft" Main.joinLeft

218+ -- prop "joinLeft" Main.joinLeft

215 prop "joinLeftMap" Main.joinLeftMap

219 prop "joinLeftMap" Main.joinLeftMap

216 -- intersect

220 -- intersect

217 prop "intersectBy"

221 prop "intersectBy"

test/Streamly/Test/Unicode/Stream.hs+3−3

@@ -21,6 +21,7 @@ import Test.QuickCheck.Monadic (run, monadicIO, assert)

21import qualified Streamly.Data.Array as A

22import qualified Streamly.Internal.Data.Stream.Chunked as AS

23import qualified Streamly.Prelude as S

24+import qualified Streamly.Internal.Data.Stream.IsStream as S

24import qualified Streamly.Unicode.Stream as SS

25import qualified Streamly.Unicode.Stream as SS

25import qualified Streamly.Internal.Unicode.Stream as IUS

26import qualified Streamly.Internal.Unicode.Stream as IUS

26import qualified Streamly.Internal.Unicode.Array.Char as IUA

27import qualified Streamly.Internal.Unicode.Array.Char as IUA

@@ -66,9 +67,8 @@ propDecodeEncodeIdArrays :: Property

66propDecodeEncodeIdArrays =

67propDecodeEncodeIdArrays =

67 forAll genUnicode $ \list ->

68 forAll genUnicode $ \list ->

68 monadicIO $ do

69 monadicIO $ do

69− let wrds = SS.encodeUtf8' $ S.fromList list

70+ let wrds = S.arraysOf 8 $ SS.encodeUtf8' $ S.fromList list

70− chrs <- S.toList $ IUS.decodeUtf8Arrays

71+ chrs <- S.toList $ IUS.decodeUtf8Arrays wrds

71− (S.fold A.write wrds)

72 assert (chrs == list)

73

74unicodeTestData :: [Char]