src/Streamly/Internal/Data/Stream/Serial.hs · streamly

git.iko.soy/streamly

on 8aee462 help sign in

streamly/src/Streamly/Internal/Data/Stream/Serial.hs

451 lines 14 kB history raw

1{-# LANGUAGE UndecidableInstances #-}

2{-# OPTIONS_GHC -Wno-deprecations #-}

3{-# OPTIONS_GHC -Wno-redundant-constraints #-}

5-- |

6-- Module : Streamly.Internal.Data.Stream.Serial

8--

9-- License : BSD3

10-- Maintainer : streamly@composewell.com

11-- Stability : experimental

12-- Portability : GHC

13--

14-- To run examples in this module:

15--

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

17--

18module Streamly.Internal.Data.Stream.Serial {-# DEPRECATED "Please use \"Streamly.Internal.Data.Stream\" from streamly-core package instead." #-}

19 (

20 -- * Serial appending stream

21 SerialT(..)

22 , toStreamK

23 , fromStreamK

24 , Serial

25 , serial

27 -- * Serial interleaving stream

28 , WSerialT(..)

29 , WSerial

30 , wSerial

31 , wSerialFst

32 , wSerialMin

33 , consMWSerial

35 -- * Construction

36 , cons

37 , consM

38 , repeat

39 , unfoldrM

40 , fromList

42 -- * Elimination

43 , toList

45 -- * Transformation

46 , map

47 , mapM

48 )

49where

51#if !(MIN_VERSION_base(4,18,0))

52import Control.Applicative (liftA2)

53#endif

54import Control.DeepSeq (NFData(..), NFData1(..))

55#if !(MIN_VERSION_transformers(0,6,0))

56import Control.Monad.Base (MonadBase(..), liftBaseDefault)

57#endif

58import Control.Monad.Catch (MonadThrow, throwM)

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

60import Control.Monad.Reader.Class (MonadReader(..))

61import Control.Monad.State.Class (MonadState(..))

62import Control.Monad.Trans.Class (MonadTrans(lift))

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

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

65import Data.Maybe (fromMaybe)

66import Data.Semigroup (Endo(..))

67import GHC.Exts (IsList(..), IsString(..), oneShot)

68import Text.Read

69 ( Lexeme(Ident), lexP, parens, prec, readPrec, readListPrec

70 , readListPrecDefault)

71import Streamly.Internal.BaseCompat ((#.))

72import Streamly.Internal.Data.Maybe.Strict (Maybe'(..), toMaybe)

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

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

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

77import qualified Streamly.Internal.Data.StreamK as K

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

81#include "Instances.hs"

82#include "inline.hs"

84-- $setup

85-- >>> :set -fno-warn-deprecations

86-- >>> import qualified Streamly.Prelude as IsStream

88{-# INLINABLE withLocal #-}

89withLocal :: MonadReader r m => (r -> r) -> K.Stream m a -> K.Stream m a

90withLocal f m =

91 K.mkStream $ \st yld sng stp ->

92 let single = local f . sng

93 yieldk a r = local f $ yld a (withLocal f r)

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

96------------------------------------------------------------------------------

97-- SerialT

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

100-- | For 'SerialT' streams:

101--

102-- @

103-- (<>) = 'Streamly.Prelude.serial' -- 'Semigroup'

104-- (>>=) = flip . 'Streamly.Prelude.concatMapWith' 'Streamly.Prelude.serial' -- 'Monad'

105-- @

106--

107-- A single 'Monad' bind behaves like a @for@ loop:

108--

109-- >>> :{

110-- IsStream.toList $ do

111-- x <- IsStream.fromList [1,2] -- foreach x in stream

112-- return x

113-- :}

114-- [1,2]

115--

116-- Nested monad binds behave like nested @for@ loops:

117--

118-- >>> :{

119-- IsStream.toList $ do

120-- x <- IsStream.fromList [1,2] -- foreach x in stream

121-- y <- IsStream.fromList [3,4] -- foreach y in stream

122-- return (x, y)

123-- :}

124-- [(1,3),(1,4),(2,3),(2,4)]

125--

126-- /Since: 0.2.0 ("Streamly")/

127--

128-- @since 0.8.0

129newtype SerialT m a = SerialT {getSerialT :: Stream m a}

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

131 deriving (Semigroup, Monoid)

132

133-- | A serial IO stream of elements of type @a@. See 'SerialT' documentation

134-- for more details.

135--

136-- /Since: 0.2.0 ("Streamly")/

137--

138-- @since 0.8.0

139type Serial = SerialT IO

140

141toStreamK :: SerialT m a -> Stream m a

142toStreamK = getSerialT

143

144fromStreamK :: Stream m a -> SerialT m a

145fromStreamK = SerialT

146

147------------------------------------------------------------------------------

148-- Generation

149------------------------------------------------------------------------------

150

151infixr 5 `cons`

152

153{-# INLINE cons #-}

154cons :: a -> SerialT m a -> SerialT m a

155cons x (SerialT ms) = SerialT $ K.cons x ms

156

157infixr 5 `consM`

158

159{-# INLINE consM #-}

160{-# SPECIALIZE consM :: IO a -> SerialT IO a -> SerialT IO a #-}

161consM :: Monad m => m a -> SerialT m a -> SerialT m a

162consM m (SerialT ms) = SerialT $ K.consM m ms

163

164-- |

165-- Generate an infinite stream by repeating a pure value.

166--

167{-# INLINE_NORMAL repeat #-}

168repeat :: Monad m => a -> SerialT m a

169repeat = SerialT . D.toStreamK . D.repeat

170

171------------------------------------------------------------------------------

172-- Combining

173------------------------------------------------------------------------------

174

175{-# INLINE serial #-}

176serial :: SerialT m a -> SerialT m a -> SerialT m a

177serial = (<>)

178

179------------------------------------------------------------------------------

180-- Monad

181------------------------------------------------------------------------------

182

183instance Monad m => Monad (SerialT m) where

184 return = pure

185

186 -- Benchmarks better with StreamD bind and pure:

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

188 --

189 -- pure = SerialT . D.fromStreamD . D.fromPure

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

191

192 -- Benchmarks better with CPS bind and pure:

193 -- Prime sieve (25x)

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

195 --

196 {-# INLINE (>>=) #-}

197 (>>=) (SerialT m) f = SerialT $ K.bindWith K.append m (getSerialT . f)

198

199 {-# INLINE (>>) #-}

200 (>>) = (*>)

201

202instance MonadTrans SerialT where

203 {-# INLINE lift #-}

204 lift = SerialT . K.fromEffect

205

206------------------------------------------------------------------------------

207-- Other instances

208------------------------------------------------------------------------------

209

210{-# INLINE mapM #-}

211mapM :: Monad m => (a -> m b) -> SerialT m a -> SerialT m b

212mapM f (SerialT m) = SerialT $ D.toStreamK $ D.mapM f $ D.fromStreamK m

213

214-- |

215-- @

216-- map = fmap

217-- @

218--

219-- Same as 'fmap'.

220--

221-- @

222-- > S.toList $ S.map (+1) $ S.fromList [1,2,3]

223-- [2,3,4]

224-- @

225--

226-- @since 0.4.0

227{-# INLINE map #-}

228map :: Monad m => (a -> b) -> SerialT m a -> SerialT m b

229map f = mapM (return . f)

230

231{-# INLINE apSerial #-}

232apSerial :: Monad m => SerialT m (a -> b) -> SerialT m a -> SerialT m b

233apSerial (SerialT m1) (SerialT m2) =

234 SerialT $ D.toStreamK $ D.fromStreamK m1 `D.crossApply` D.fromStreamK m2

235

236{-# INLINE apSequence #-}

237apSequence :: Monad m => SerialT m a -> SerialT m b -> SerialT m b

238apSequence (SerialT m1) (SerialT m2) =

239 SerialT $ D.toStreamK $ D.fromStreamK m1 `D.crossApplySnd` D.fromStreamK m2

240

241{-# INLINE apDiscardSnd #-}

242apDiscardSnd :: Monad m => SerialT m a -> SerialT m b -> SerialT m a

243apDiscardSnd (SerialT m1) (SerialT m2) =

244 SerialT $ D.toStreamK $ D.fromStreamK m1 `D.crossApplyFst` D.fromStreamK m2

245

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

247-- INLINE them.

248instance Monad m => Applicative (SerialT m) where

249 {-# INLINE pure #-}

250 pure = SerialT . K.fromPure

251

252 {-# INLINE (<*>) #-}

253 (<*>) = apSerial

254 -- (<*>) = K.apSerial

255

256 {-# INLINE liftA2 #-}

257 liftA2 f x = (<*>) (fmap f x)

258

259 {-# INLINE (*>) #-}

260 (*>) = apSequence

261 -- (*>) = K.apSerialDiscardFst

262

263 {-# INLINE (<*) #-}

264 (<*) = apDiscardSnd

265 -- (<*) = K.apSerialDiscardSnd

266

267MONAD_COMMON_INSTANCES(SerialT,)

268LIST_INSTANCES(SerialT)

269NFDATA1_INSTANCE(SerialT)

270FOLDABLE_INSTANCE(SerialT)

271TRAVERSABLE_INSTANCE(SerialT)

272

273------------------------------------------------------------------------------

274-- WSerialT

275------------------------------------------------------------------------------

276

277-- | For 'WSerialT' streams:

278--

279-- @

280-- (<>) = 'Streamly.Prelude.wSerial' -- 'Semigroup'

281-- (>>=) = flip . 'Streamly.Prelude.concatMapWith' 'Streamly.Prelude.wSerial' -- 'Monad'

282-- @

283--

284-- Note that '<>' is associative only if we disregard the ordering of elements

285-- in the resulting stream.

286--

287-- A single 'Monad' bind behaves like a @for@ loop:

288--

289-- >>> :{

290-- IsStream.toList $ IsStream.fromWSerial $ do

291-- x <- IsStream.fromList [1,2] -- foreach x in stream

292-- return x

293-- :}

294-- [1,2]

295--

296-- Nested monad binds behave like interleaved nested @for@ loops:

297--

298-- >>> :{

299-- IsStream.toList $ IsStream.fromWSerial $ do

300-- x <- IsStream.fromList [1,2] -- foreach x in stream

301-- y <- IsStream.fromList [3,4] -- foreach y in stream

302-- return (x, y)

303-- :}

304-- [(1,3),(2,3),(1,4),(2,4)]

305--

306-- It is a result of interleaving all the nested iterations corresponding to

307-- element @1@ in the first stream with all the nested iterations of element

308-- @2@:

309--

310-- >>> import Streamly.Prelude (wSerial)

311-- >>> IsStream.toList $ IsStream.fromList [(1,3),(1,4)] `IsStream.wSerial` IsStream.fromList [(2,3),(2,4)]

312-- [(1,3),(2,3),(1,4),(2,4)]

313--

314-- The @W@ in the name stands for @wide@ or breadth wise scheduling in

315-- contrast to the depth wise scheduling behavior of 'SerialT'.

316--

317-- /Since: 0.2.0 ("Streamly")/

318--

319-- @since 0.8.0

320newtype WSerialT m a = WSerialT {getWSerialT :: Stream m a}

321

322instance MonadTrans WSerialT where

323 {-# INLINE lift #-}

324 lift = WSerialT . K.fromEffect

325

326-- | An interleaving serial IO stream of elements of type @a@. See 'WSerialT'

327-- documentation for more details.

328--

329-- /Since: 0.2.0 ("Streamly")/

330--

331-- @since 0.8.0

332type WSerial = WSerialT IO

333

334{-# INLINE consMWSerial #-}

335{-# SPECIALIZE consMWSerial :: IO a -> WSerialT IO a -> WSerialT IO a #-}

336consMWSerial :: Monad m => m a -> WSerialT m a -> WSerialT m a

337consMWSerial m (WSerialT ms) = WSerialT $ K.consM m ms

338

339------------------------------------------------------------------------------

340-- Semigroup

341------------------------------------------------------------------------------

342

343infixr 6 `wSerial`

344

345-- | Interleaves two streams, yielding one element from each stream

346-- alternately. When one stream stops the rest of the other stream is used in

347-- the output stream.

348--

349-- This gives exponential priority to earlier streams than the ones joining

350-- later. Because of exponential weighting it can be used with 'concatMapWith'.

351--

352-- /Not fused/

353

354-- NOTE:

355--

356-- Note that evaluation of @a \`wSerial` b \`wSerial` c@ does not interleave

357-- @a@, @b@ and @c@ with equal priority. This expression is equivalent to @a

358-- \`wSerial` (b \`wSerial` c)@, therefore, it fairly interleaves @a@ with the

359-- result of @b \`wSerial` c@. For example, @Stream.fromList [1,2] \`wSerial`

360-- Stream.fromList [3,4] \`wSerial` Stream.fromList [5,6]@ would result in

361-- [1,3,2,5,4,6]. In other words, the leftmost stream gets the same scheduling

362-- priority as the rest of the streams taken together. The same is true for

363-- each subexpression on the right.

364--

365{-# INLINE wSerial #-}

366wSerial :: WSerialT m a -> WSerialT m a -> WSerialT m a

367wSerial (WSerialT m1) (WSerialT m2) = WSerialT $ K.interleave m1 m2

368

369{-# INLINE wSerialFst #-}

370wSerialFst :: WSerialT m a -> WSerialT m a -> WSerialT m a

371wSerialFst (WSerialT m1) (WSerialT m2) = WSerialT $ K.interleaveFst m1 m2

372

373{-# INLINE wSerialMin #-}

374wSerialMin :: WSerialT m a -> WSerialT m a -> WSerialT m a

375wSerialMin (WSerialT m1) (WSerialT m2) = WSerialT $ K.interleaveMin m1 m2

376

377instance Semigroup (WSerialT m a) where

378 (<>) = wSerial

379

380------------------------------------------------------------------------------

381-- Monoid

382------------------------------------------------------------------------------

383

384instance Monoid (WSerialT m a) where

385 mempty = WSerialT K.nil

386 mappend = (<>)

387

388{-# INLINE apWSerial #-}

389apWSerial :: WSerialT m (a -> b) -> WSerialT m a -> WSerialT m b

390apWSerial (WSerialT m1) (WSerialT m2) =

391 let f x1 = K.concatMapWith K.interleave (K.fromPure . x1) m2

392 in WSerialT $ K.concatMapWith K.interleave f m1

393

394instance Monad m => Applicative (WSerialT m) where

395 {-# INLINE pure #-}

396 pure = WSerialT . K.fromPure

397 {-# INLINE (<*>) #-}

398 (<*>) = apWSerial

399

400------------------------------------------------------------------------------

401-- Monad

402------------------------------------------------------------------------------

403

404instance Monad m => Monad (WSerialT m) where

405 return = pure

406 {-# INLINE (>>=) #-}

407 (>>=) (WSerialT m) f = WSerialT $ K.bindWith K.interleave m (getWSerialT . f)

408

409------------------------------------------------------------------------------

410-- Other instances

411------------------------------------------------------------------------------

412

413#if !(MIN_VERSION_transformers(0,6,0))

414instance (MonadBase b m, Monad m) => MonadBase b (SerialT m) where

415 liftBase = liftBaseDefault

416#endif

417

418MONAD_COMMON_INSTANCES(WSerialT,)

419LIST_INSTANCES(WSerialT)

420NFDATA1_INSTANCE(WSerialT)

421FOLDABLE_INSTANCE(WSerialT)

422TRAVERSABLE_INSTANCE(WSerialT)

423

424------------------------------------------------------------------------------

425-- Construction

426------------------------------------------------------------------------------

427

428-- | Build a stream by unfolding a /monadic/ step function starting from a

429-- seed. The step function returns the next element in the stream and the next

430-- seed value. When it is done it returns 'Nothing' and the stream ends. For

431-- example,

432--

433-- @

434-- let f b =

435-- if b > 3

436-- then return Nothing

437-- else print b >> return (Just (b, b + 1))

438-- in drain $ unfoldrM f 0

439-- @

440-- @

441-- 0

442-- 1

443-- 2

444-- 3

445-- @

446--

447-- /Pre-release/

448--

449{-# INLINE unfoldrM #-}

450unfoldrM :: Monad m => (b -> m (Maybe (a, b))) -> b -> SerialT m a

451unfoldrM step seed = SerialT $ D.toStreamK (D.unfoldrM step seed)