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

git.iko.soy/streamly

on efe0be3 help sign in

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

453 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.Stream.StreamK.Type (Stream)

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

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

77import qualified Streamly.Internal.Data.Stream.StreamD.Transform as D

78import qualified Streamly.Internal.Data.Stream.StreamD.Type as D

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

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

83#include "Instances.hs"

84#include "inline.hs"

86-- $setup

87-- >>> :set -fno-warn-deprecations

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

90{-# INLINABLE withLocal #-}

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

92withLocal f m =

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

94 let single = local f . sng

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

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

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

99-- SerialT

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

101

102-- | For 'SerialT' streams:

103--

104-- @

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

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

107-- @

108--

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

110--

111-- >>> :{

112-- IsStream.toList $ do

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

114-- return x

115-- :}

116-- [1,2]

117--

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

119--

120-- >>> :{

121-- IsStream.toList $ do

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

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

124-- return (x, y)

125-- :}

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

127--

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

129--

130-- @since 0.8.0

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

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

133 deriving (Semigroup, Monoid)

134

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

136-- for more details.

137--

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

139--

140-- @since 0.8.0

141type Serial = SerialT IO

142

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

144toStreamK = getSerialT

145

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

147fromStreamK = SerialT

148

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

150-- Generation

151------------------------------------------------------------------------------

152

153infixr 5 `cons`

154

155{-# INLINE cons #-}

156cons :: a -> SerialT m a -> SerialT m a

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

158

159infixr 5 `consM`

160

161{-# INLINE consM #-}

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

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

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

165

166-- |

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

168--

169{-# INLINE_NORMAL repeat #-}

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

171repeat = SerialT . D.toStreamK . D.repeat

172

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

174-- Combining

175------------------------------------------------------------------------------

176

177{-# INLINE serial #-}

178serial :: SerialT m a -> SerialT m a -> SerialT m a

179serial = (<>)

180

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

182-- Monad

183------------------------------------------------------------------------------

184

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

186 return = pure

187

188 -- Benchmarks better with StreamD bind and pure:

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

190 --

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

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

193

194 -- Benchmarks better with CPS bind and pure:

195 -- Prime sieve (25x)

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

197 --

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

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

200

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

202 (>>) = (*>)

203

204instance MonadTrans SerialT where

205 {-# INLINE lift #-}

206 lift = SerialT . K.fromEffect

207

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

209-- Other instances

210------------------------------------------------------------------------------

211

212{-# INLINE mapM #-}

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

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

215

216-- |

217-- @

218-- map = fmap

219-- @

220--

221-- Same as 'fmap'.

222--

223-- @

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

225-- [2,3,4]

226-- @

227--

228-- @since 0.4.0

229{-# INLINE map #-}

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

231map f = mapM (return . f)

232

233{-# INLINE apSerial #-}

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

235apSerial (SerialT m1) (SerialT m2) =

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

237

238{-# INLINE apSequence #-}

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

240apSequence (SerialT m1) (SerialT m2) =

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

242

243{-# INLINE apDiscardSnd #-}

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

245apDiscardSnd (SerialT m1) (SerialT m2) =

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

247

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

249-- INLINE them.

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

251 {-# INLINE pure #-}

252 pure = SerialT . K.fromPure

253

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

255 (<*>) = apSerial

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

257

258 {-# INLINE liftA2 #-}

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

260

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

262 (*>) = apSequence

263 -- (*>) = K.apSerialDiscardFst

264

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

266 (<*) = apDiscardSnd

267 -- (<*) = K.apSerialDiscardSnd

268

269MONAD_COMMON_INSTANCES(SerialT,)

270LIST_INSTANCES(SerialT)

271NFDATA1_INSTANCE(SerialT)

272FOLDABLE_INSTANCE(SerialT)

273TRAVERSABLE_INSTANCE(SerialT)

274

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

276-- WSerialT

277------------------------------------------------------------------------------

278

279-- | For 'WSerialT' streams:

280--

281-- @

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

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

284-- @

285--

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

287-- in the resulting stream.

288--

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

290--

291-- >>> :{

292-- IsStream.toList $ IsStream.fromWSerial $ do

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

294-- return x

295-- :}

296-- [1,2]

297--

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

299--

300-- >>> :{

301-- IsStream.toList $ IsStream.fromWSerial $ do

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

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

304-- return (x, y)

305-- :}

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

307--

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

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

310-- @2@:

311--

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

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

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

315--

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

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

318--

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

320--

321-- @since 0.8.0

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

323

324instance MonadTrans WSerialT where

325 {-# INLINE lift #-}

326 lift = WSerialT . K.fromEffect

327

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

329-- documentation for more details.

330--

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

332--

333-- @since 0.8.0

334type WSerial = WSerialT IO

335

336{-# INLINE consMWSerial #-}

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

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

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

340

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

342-- Semigroup

343------------------------------------------------------------------------------

344

345infixr 6 `wSerial`

346

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

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

349-- the output stream.

350--

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

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

353--

354-- /Not fused/

355

356-- NOTE:

357--

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

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

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

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

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

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

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

365-- each subexpression on the right.

366--

367{-# INLINE wSerial #-}

368wSerial :: WSerialT m a -> WSerialT m a -> WSerialT m a

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

370

371{-# INLINE wSerialFst #-}

372wSerialFst :: WSerialT m a -> WSerialT m a -> WSerialT m a

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

374

375{-# INLINE wSerialMin #-}

376wSerialMin :: WSerialT m a -> WSerialT m a -> WSerialT m a

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

378

379instance Semigroup (WSerialT m a) where

380 (<>) = wSerial

381

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

383-- Monoid

384------------------------------------------------------------------------------

385

386instance Monoid (WSerialT m a) where

387 mempty = WSerialT K.nil

388 mappend = (<>)

389

390{-# INLINE apWSerial #-}

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

392apWSerial (WSerialT m1) (WSerialT m2) =

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

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

395

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

397 {-# INLINE pure #-}

398 pure = WSerialT . K.fromPure

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

400 (<*>) = apWSerial

401

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

403-- Monad

404------------------------------------------------------------------------------

405

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

407 return = pure

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

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

410

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

412-- Other instances

413------------------------------------------------------------------------------

414

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

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

417 liftBase = liftBaseDefault

418#endif

419

420MONAD_COMMON_INSTANCES(WSerialT,)

421LIST_INSTANCES(WSerialT)

422NFDATA1_INSTANCE(WSerialT)

423FOLDABLE_INSTANCE(WSerialT)

424TRAVERSABLE_INSTANCE(WSerialT)

425

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

427-- Construction

428------------------------------------------------------------------------------

429

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

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

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

433-- example,

434--

435-- @

436-- let f b =

437-- if b > 3

438-- then return Nothing

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

440-- in drain $ unfoldrM f 0

441-- @

442-- @

443-- 0

444-- 1

445-- 2

446-- 3

447-- @

448--

449-- /Pre-release/

450--

451{-# INLINE unfoldrM #-}

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

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