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

git.iko.soy/streamly

on 5d096df help sign in

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

456 lines 14 kB history raw

1{-# LANGUAGE TypeFamilies #-}

2-- Must come after TypeFamilies, otherwise it is re-enabled.

3-- MonoLocalBinds enabled by TypeFamilies causes perf regressions in general.

4{-# LANGUAGE NoMonoLocalBinds #-}

5{-# LANGUAGE UndecidableInstances #-}

7{-# OPTIONS_GHC -Wno-deprecations #-}

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

10-- |

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

13--

14-- License : BSD3

15-- Maintainer : streamly@composewell.com

16-- Stability : experimental

17-- Portability : GHC

18--

19-- To run examples in this module:

20--

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

22--

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

24 (

25 -- * Serial appending stream

26 SerialT(..)

27 , toStreamK

28 , fromStreamK

29 , Serial

30 , serial

32 -- * Serial interleaving stream

33 , WSerialT(..)

34 , WSerial

35 , wSerial

36 , wSerialFst

37 , wSerialMin

38 , consMWSerial

40 -- * Construction

41 , cons

42 , consM

43 , repeat

44 , unfoldrM

45 , fromList

47 -- * Elimination

48 , toList

50 -- * Transformation

51 , map

52 , mapM

53 )

54where

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

57import Control.Applicative (liftA2)

58#endif

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

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

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

62#endif

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

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

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

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

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

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

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

70import Data.Maybe (fromMaybe)

71import Data.Semigroup (Endo(..))

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

73import Text.Read

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

75 , readListPrecDefault)

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

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

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

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

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

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

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

86#include "Instances.hs"

87#include "inline.hs"

89-- $setup

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

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

93{-# INLINABLE withLocal #-}

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

95withLocal f m =

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

97 let single = local f . sng

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

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

100

101------------------------------------------------------------------------------

102-- SerialT

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

104

105-- | For 'SerialT' streams:

106--

107-- @

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

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

110-- @

111--

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

113--

114-- >>> :{

115-- IsStream.toList $ do

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

117-- return x

118-- :}

119-- [1,2]

120--

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

122--

123-- >>> :{

124-- IsStream.toList $ do

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

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

127-- return (x, y)

128-- :}

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

130--

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

132--

133-- @since 0.8.0

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

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

136 deriving (Semigroup, Monoid)

137

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

139-- for more details.

140--

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

142--

143-- @since 0.8.0

144type Serial = SerialT IO

145

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

147toStreamK = getSerialT

148

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

150fromStreamK = SerialT

151

152------------------------------------------------------------------------------

153-- Generation

154------------------------------------------------------------------------------

155

156infixr 5 `cons`

157

158{-# INLINE cons #-}

159cons :: a -> SerialT m a -> SerialT m a

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

161

162infixr 5 `consM`

163

164{-# INLINE consM #-}

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

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

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

168

169-- |

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

171--

172{-# INLINE_NORMAL repeat #-}

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

174repeat = SerialT . D.toStreamK . D.repeat

175

176------------------------------------------------------------------------------

177-- Combining

178------------------------------------------------------------------------------

179

180{-# INLINE serial #-}

181serial :: SerialT m a -> SerialT m a -> SerialT m a

182serial = (<>)

183

184------------------------------------------------------------------------------

185-- Monad

186------------------------------------------------------------------------------

187

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

189 return = pure

190

191 -- Benchmarks better with StreamD bind and pure:

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

193 --

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

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

196

197 -- Benchmarks better with CPS bind and pure:

198 -- Prime sieve (25x)

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

200 --

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

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

203

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

205 (>>) = (*>)

206

207instance MonadTrans SerialT where

208 {-# INLINE lift #-}

209 lift = SerialT . K.fromEffect

210

211------------------------------------------------------------------------------

212-- Other instances

213------------------------------------------------------------------------------

214

215{-# INLINE mapM #-}

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

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

218

219-- |

220-- @

221-- map = fmap

222-- @

223--

224-- Same as 'fmap'.

225--

226-- @

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

228-- [2,3,4]

229-- @

230--

231-- @since 0.4.0

232{-# INLINE map #-}

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

234map f = mapM (return . f)

235

236{-# INLINE apSerial #-}

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

238apSerial (SerialT m1) (SerialT m2) =

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

240

241{-# INLINE apSequence #-}

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

243apSequence (SerialT m1) (SerialT m2) =

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

245

246{-# INLINE apDiscardSnd #-}

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

248apDiscardSnd (SerialT m1) (SerialT m2) =

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

250

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

252-- INLINE them.

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

254 {-# INLINE pure #-}

255 pure = SerialT . K.fromPure

256

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

258 (<*>) = apSerial

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

260

261 {-# INLINE liftA2 #-}

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

263

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

265 (*>) = apSequence

266 -- (*>) = K.apSerialDiscardFst

267

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

269 (<*) = apDiscardSnd

270 -- (<*) = K.apSerialDiscardSnd

271

272MONAD_COMMON_INSTANCES(SerialT,)

273LIST_INSTANCES(SerialT)

274NFDATA1_INSTANCE(SerialT)

275FOLDABLE_INSTANCE(SerialT)

276TRAVERSABLE_INSTANCE(SerialT)

277

278------------------------------------------------------------------------------

279-- WSerialT

280------------------------------------------------------------------------------

281

282-- | For 'WSerialT' streams:

283--

284-- @

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

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

287-- @

288--

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

290-- in the resulting stream.

291--

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

293--

294-- >>> :{

295-- IsStream.toList $ IsStream.fromWSerial $ do

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

297-- return x

298-- :}

299-- [1,2]

300--

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

302--

303-- >>> :{

304-- IsStream.toList $ IsStream.fromWSerial $ do

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

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

307-- return (x, y)

308-- :}

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

310--

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

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

313-- @2@:

314--

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

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

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

318--

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

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

321--

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

323--

324-- @since 0.8.0

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

326

327instance MonadTrans WSerialT where

328 {-# INLINE lift #-}

329 lift = WSerialT . K.fromEffect

330

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

332-- documentation for more details.

333--

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

335--

336-- @since 0.8.0

337type WSerial = WSerialT IO

338

339{-# INLINE consMWSerial #-}

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

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

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

343

344------------------------------------------------------------------------------

345-- Semigroup

346------------------------------------------------------------------------------

347

348infixr 6 `wSerial`

349

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

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

352-- the output stream.

353--

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

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

356--

357-- /Not fused/

358

359-- NOTE:

360--

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

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

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

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

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

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

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

368-- each subexpression on the right.

369--

370{-# INLINE wSerial #-}

371wSerial :: WSerialT m a -> WSerialT m a -> WSerialT m a

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

373

374{-# INLINE wSerialFst #-}

375wSerialFst :: WSerialT m a -> WSerialT m a -> WSerialT m a

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

377

378{-# INLINE wSerialMin #-}

379wSerialMin :: WSerialT m a -> WSerialT m a -> WSerialT m a

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

381

382instance Semigroup (WSerialT m a) where

383 (<>) = wSerial

384

385------------------------------------------------------------------------------

386-- Monoid

387------------------------------------------------------------------------------

388

389instance Monoid (WSerialT m a) where

390 mempty = WSerialT K.nil

391 mappend = (<>)

392

393{-# INLINE apWSerial #-}

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

395apWSerial (WSerialT m1) (WSerialT m2) =

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

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

398

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

400 {-# INLINE pure #-}

401 pure = WSerialT . K.fromPure

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

403 (<*>) = apWSerial

404

405------------------------------------------------------------------------------

406-- Monad

407------------------------------------------------------------------------------

408

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

410 return = pure

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

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

413

414------------------------------------------------------------------------------

415-- Other instances

416------------------------------------------------------------------------------

417

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

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

420 liftBase = liftBaseDefault

421#endif

422

423MONAD_COMMON_INSTANCES(WSerialT,)

424LIST_INSTANCES(WSerialT)

425NFDATA1_INSTANCE(WSerialT)

426FOLDABLE_INSTANCE(WSerialT)

427TRAVERSABLE_INSTANCE(WSerialT)

428

429------------------------------------------------------------------------------

430-- Construction

431------------------------------------------------------------------------------

432

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

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

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

436-- example,

437--

438-- @

439-- let f b =

440-- if b > 3

441-- then return Nothing

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

443-- in drain $ unfoldrM f 0

444-- @

445-- @

446-- 0

447-- 1

448-- 2

449-- 3

450-- @

451--

452-- /Pre-release/

453--

454{-# INLINE unfoldrM #-}

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

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