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

git.iko.soy/streamly

on bcac95d help sign in

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

444 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

51import Control.Applicative (liftA2)

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

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

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

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

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

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

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

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

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

61import Data.Maybe (fromMaybe)

62import Data.Semigroup (Endo(..))

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

64import Text.Read

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

66 , readListPrecDefault)

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

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

69import Streamly.Internal.Data.Stream.StreamK.Type (Stream)

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

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

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

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

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

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

79#include "Instances.hs"

80#include "inline.hs"

82-- $setup

83-- >>> :set -fno-warn-deprecations

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

86{-# INLINABLE withLocal #-}

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

88withLocal f m =

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

90 let single = local f . sng

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

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

94------------------------------------------------------------------------------

95-- SerialT

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

98-- | For 'SerialT' streams:

99--

100-- @

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

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

103-- @

104--

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

106--

107-- >>> :{

108-- IsStream.toList $ do

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

110-- return x

111-- :}

112-- [1,2]

113--

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

115--

116-- >>> :{

117-- IsStream.toList $ do

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

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

120-- return (x, y)

121-- :}

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

123--

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

125--

126-- @since 0.8.0

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

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

129 deriving (Semigroup, Monoid)

130

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

132-- for more details.

133--

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

135--

136-- @since 0.8.0

137type Serial = SerialT IO

138

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

140toStreamK = getSerialT

141

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

143fromStreamK = SerialT

144

145------------------------------------------------------------------------------

146-- Generation

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

148

149infixr 5 `cons`

150

151{-# INLINE cons #-}

152cons :: a -> SerialT m a -> SerialT m a

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

154

155infixr 5 `consM`

156

157{-# INLINE consM #-}

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

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

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

161

162-- |

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

164--

165{-# INLINE_NORMAL repeat #-}

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

167repeat = SerialT . D.toStreamK . D.repeat

168

169------------------------------------------------------------------------------

170-- Combining

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

172

173{-# INLINE serial #-}

174serial :: SerialT m a -> SerialT m a -> SerialT m a

175serial = (<>)

176

177------------------------------------------------------------------------------

178-- Monad

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

180

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

182 return = pure

183

184 -- Benchmarks better with StreamD bind and pure:

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

186 --

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

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

189

190 -- Benchmarks better with CPS bind and pure:

191 -- Prime sieve (25x)

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

193 --

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

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

196

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

198 (>>) = (*>)

199

200instance MonadTrans SerialT where

201 {-# INLINE lift #-}

202 lift = SerialT . K.fromEffect

203

204------------------------------------------------------------------------------

205-- Other instances

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

207

208{-# INLINE mapM #-}

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

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

211

212-- |

213-- @

214-- map = fmap

215-- @

216--

217-- Same as 'fmap'.

218--

219-- @

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

221-- [2,3,4]

222-- @

223--

224-- @since 0.4.0

225{-# INLINE map #-}

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

227map f = mapM (return . f)

228

229{-# INLINE apSerial #-}

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

231apSerial (SerialT m1) (SerialT m2) =

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

233

234{-# INLINE apSequence #-}

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

236apSequence (SerialT m1) (SerialT m2) =

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

238

239{-# INLINE apDiscardSnd #-}

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

241apDiscardSnd (SerialT m1) (SerialT m2) =

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

243

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

245-- INLINE them.

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

247 {-# INLINE pure #-}

248 pure = SerialT . K.fromPure

249

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

251 (<*>) = apSerial

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

253

254 {-# INLINE liftA2 #-}

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

256

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

258 (*>) = apSequence

259 -- (*>) = K.apSerialDiscardFst

260

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

262 (<*) = apDiscardSnd

263 -- (<*) = K.apSerialDiscardSnd

264

265MONAD_COMMON_INSTANCES(SerialT,)

266LIST_INSTANCES(SerialT)

267NFDATA1_INSTANCE(SerialT)

268FOLDABLE_INSTANCE(SerialT)

269TRAVERSABLE_INSTANCE(SerialT)

270

271------------------------------------------------------------------------------

272-- WSerialT

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

274

275-- | For 'WSerialT' streams:

276--

277-- @

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

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

280-- @

281--

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

283-- in the resulting stream.

284--

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

286--

287-- >>> :{

288-- IsStream.toList $ IsStream.fromWSerial $ do

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

290-- return x

291-- :}

292-- [1,2]

293--

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

295--

296-- >>> :{

297-- IsStream.toList $ IsStream.fromWSerial $ do

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

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

300-- return (x, y)

301-- :}

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

303--

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

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

306-- @2@:

307--

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

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

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

311--

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

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

314--

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

316--

317-- @since 0.8.0

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

319

320instance MonadTrans WSerialT where

321 {-# INLINE lift #-}

322 lift = WSerialT . K.fromEffect

323

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

325-- documentation for more details.

326--

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

328--

329-- @since 0.8.0

330type WSerial = WSerialT IO

331

332{-# INLINE consMWSerial #-}

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

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

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

336

337------------------------------------------------------------------------------

338-- Semigroup

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

340

341infixr 6 `wSerial`

342

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

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

345-- the output stream.

346--

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

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

349--

350-- /Not fused/

351

352-- NOTE:

353--

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

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

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

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

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

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

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

361-- each subexpression on the right.

362--

363{-# INLINE wSerial #-}

364wSerial :: WSerialT m a -> WSerialT m a -> WSerialT m a

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

366

367{-# INLINE wSerialFst #-}

368wSerialFst :: WSerialT m a -> WSerialT m a -> WSerialT m a

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

370

371{-# INLINE wSerialMin #-}

372wSerialMin :: WSerialT m a -> WSerialT m a -> WSerialT m a

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

374

375instance Semigroup (WSerialT m a) where

376 (<>) = wSerial

377

378------------------------------------------------------------------------------

379-- Monoid

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

381

382instance Monoid (WSerialT m a) where

383 mempty = WSerialT K.nil

384 mappend = (<>)

385

386{-# INLINE apWSerial #-}

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

388apWSerial (WSerialT m1) (WSerialT m2) =

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

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

391

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

393 {-# INLINE pure #-}

394 pure = WSerialT . K.fromPure

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

396 (<*>) = apWSerial

397

398------------------------------------------------------------------------------

399-- Monad

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

401

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

403 return = pure

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

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

406

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

408-- Other instances

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

410

411MONAD_COMMON_INSTANCES(WSerialT,)

412LIST_INSTANCES(WSerialT)

413NFDATA1_INSTANCE(WSerialT)

414FOLDABLE_INSTANCE(WSerialT)

415TRAVERSABLE_INSTANCE(WSerialT)

416

417------------------------------------------------------------------------------

418-- Construction

419------------------------------------------------------------------------------

420

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

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

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

424-- example,

425--

426-- @

427-- let f b =

428-- if b > 3

429-- then return Nothing

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

431-- in drain $ unfoldrM f 0

432-- @

433-- @

434-- 0

435-- 1

436-- 2

437-- 3

438-- @

439--

440-- /Pre-release/

441--

442{-# INLINE unfoldrM #-}

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

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