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

git.iko.soy/streamly

on 0bd3d4e help sign in

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

443 lines 14 kB history raw

1{-# LANGUAGE UndecidableInstances #-}

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

4-- |

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

7--

8-- License : BSD3

9-- Maintainer : streamly@composewell.com

10-- Stability : experimental

11-- Portability : GHC

12--

13-- To run examples in this module:

14--

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

16--

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

18 (

19 -- * Serial appending stream

20 SerialT(..)

21 , toStreamK

22 , fromStreamK

23 , Serial

24 , serial

26 -- * Serial interleaving stream

27 , WSerialT(..)

28 , WSerial

29 , wSerial

30 , wSerialFst

31 , wSerialMin

32 , consMWSerial

34 -- * Construction

35 , cons

36 , consM

37 , repeat

38 , unfoldrM

39 , fromList

41 -- * Elimination

42 , toList

44 -- * Transformation

45 , map

46 , mapM

47 )

48where

50import Control.Applicative (liftA2)

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

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

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

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

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

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

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

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

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

60import Data.Maybe (fromMaybe)

61import Data.Semigroup (Endo(..))

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

63import Text.Read

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

65 , readListPrecDefault)

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

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

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

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

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

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

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

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

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

78#include "Instances.hs"

79#include "inline.hs"

81-- $setup

82-- >>> :set -fno-warn-deprecations

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

85{-# INLINABLE withLocal #-}

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

87withLocal f m =

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

89 let single = local f . sng

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

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

93------------------------------------------------------------------------------

94-- SerialT

95------------------------------------------------------------------------------

97-- | For 'SerialT' streams:

98--

99-- @

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

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

102-- @

103--

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

105--

106-- >>> :{

107-- IsStream.toList $ do

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

109-- return x

110-- :}

111-- [1,2]

112--

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

114--

115-- >>> :{

116-- IsStream.toList $ do

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

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

119-- return (x, y)

120-- :}

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

122--

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

124--

125-- @since 0.8.0

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

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

128 deriving (Semigroup, Monoid)

129

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

131-- for more details.

132--

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

134--

135-- @since 0.8.0

136type Serial = SerialT IO

137

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

139toStreamK = getSerialT

140

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

142fromStreamK = SerialT

143

144------------------------------------------------------------------------------

145-- Generation

146------------------------------------------------------------------------------

147

148infixr 5 `cons`

149

150{-# INLINE cons #-}

151cons :: a -> SerialT m a -> SerialT m a

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

153

154infixr 5 `consM`

155

156{-# INLINE consM #-}

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

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

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

160

161-- |

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

163--

164{-# INLINE_NORMAL repeat #-}

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

166repeat = SerialT . D.toStreamK . D.repeat

167

168------------------------------------------------------------------------------

169-- Combining

170------------------------------------------------------------------------------

171

172{-# INLINE serial #-}

173serial :: SerialT m a -> SerialT m a -> SerialT m a

174serial = (<>)

175

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

177-- Monad

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

179

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

181 return = pure

182

183 -- Benchmarks better with StreamD bind and pure:

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

185 --

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

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

188

189 -- Benchmarks better with CPS bind and pure:

190 -- Prime sieve (25x)

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

192 --

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

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

195

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

197 (>>) = (*>)

198

199instance MonadTrans SerialT where

200 {-# INLINE lift #-}

201 lift = SerialT . K.fromEffect

202

203------------------------------------------------------------------------------

204-- Other instances

205------------------------------------------------------------------------------

206

207{-# INLINE mapM #-}

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

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

210

211-- |

212-- @

213-- map = fmap

214-- @

215--

216-- Same as 'fmap'.

217--

218-- @

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

220-- [2,3,4]

221-- @

222--

223-- @since 0.4.0

224{-# INLINE map #-}

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

226map f = mapM (return . f)

227

228{-# INLINE apSerial #-}

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

230apSerial (SerialT m1) (SerialT m2) =

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

232

233{-# INLINE apSequence #-}

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

235apSequence (SerialT m1) (SerialT m2) =

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

237

238{-# INLINE apDiscardSnd #-}

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

240apDiscardSnd (SerialT m1) (SerialT m2) =

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

242

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

244-- INLINE them.

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

246 {-# INLINE pure #-}

247 pure = SerialT . K.fromPure

248

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

250 (<*>) = apSerial

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

252

253 {-# INLINE liftA2 #-}

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

255

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

257 (*>) = apSequence

258 -- (*>) = K.apSerialDiscardFst

259

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

261 (<*) = apDiscardSnd

262 -- (<*) = K.apSerialDiscardSnd

263

264MONAD_COMMON_INSTANCES(SerialT,)

265LIST_INSTANCES(SerialT)

266NFDATA1_INSTANCE(SerialT)

267FOLDABLE_INSTANCE(SerialT)

268TRAVERSABLE_INSTANCE(SerialT)

269

270------------------------------------------------------------------------------

271-- WSerialT

272------------------------------------------------------------------------------

273

274-- | For 'WSerialT' streams:

275--

276-- @

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

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

279-- @

280--

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

282-- in the resulting stream.

283--

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

285--

286-- >>> :{

287-- IsStream.toList $ IsStream.fromWSerial $ do

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

289-- return x

290-- :}

291-- [1,2]

292--

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

294--

295-- >>> :{

296-- IsStream.toList $ IsStream.fromWSerial $ do

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

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

299-- return (x, y)

300-- :}

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

302--

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

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

305-- @2@:

306--

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

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

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

310--

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

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

313--

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

315--

316-- @since 0.8.0

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

318

319instance MonadTrans WSerialT where

320 {-# INLINE lift #-}

321 lift = WSerialT . K.fromEffect

322

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

324-- documentation for more details.

325--

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

327--

328-- @since 0.8.0

329type WSerial = WSerialT IO

330

331{-# INLINE consMWSerial #-}

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

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

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

335

336------------------------------------------------------------------------------

337-- Semigroup

338------------------------------------------------------------------------------

339

340infixr 6 `wSerial`

341

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

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

344-- the output stream.

345--

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

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

348--

349-- /Not fused/

350

351-- NOTE:

352--

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

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

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

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

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

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

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

360-- each subexpression on the right.

361--

362{-# INLINE wSerial #-}

363wSerial :: WSerialT m a -> WSerialT m a -> WSerialT m a

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

365

366{-# INLINE wSerialFst #-}

367wSerialFst :: WSerialT m a -> WSerialT m a -> WSerialT m a

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

369

370{-# INLINE wSerialMin #-}

371wSerialMin :: WSerialT m a -> WSerialT m a -> WSerialT m a

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

373

374instance Semigroup (WSerialT m a) where

375 (<>) = wSerial

376

377------------------------------------------------------------------------------

378-- Monoid

379------------------------------------------------------------------------------

380

381instance Monoid (WSerialT m a) where

382 mempty = WSerialT K.nil

383 mappend = (<>)

384

385{-# INLINE apWSerial #-}

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

387apWSerial (WSerialT m1) (WSerialT m2) =

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

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

390

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

392 {-# INLINE pure #-}

393 pure = WSerialT . K.fromPure

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

395 (<*>) = apWSerial

396

397------------------------------------------------------------------------------

398-- Monad

399------------------------------------------------------------------------------

400

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

402 return = pure

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

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

405

406------------------------------------------------------------------------------

407-- Other instances

408------------------------------------------------------------------------------

409

410MONAD_COMMON_INSTANCES(WSerialT,)

411LIST_INSTANCES(WSerialT)

412NFDATA1_INSTANCE(WSerialT)

413FOLDABLE_INSTANCE(WSerialT)

414TRAVERSABLE_INSTANCE(WSerialT)

415

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

417-- Construction

418------------------------------------------------------------------------------

419

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

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

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

423-- example,

424--

425-- @

426-- let f b =

427-- if b > 3

428-- then return Nothing

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

430-- in drain $ unfoldrM f 0

431-- @

432-- @

433-- 0

434-- 1

435-- 2

436-- 3

437-- @

438--

439-- /Pre-release/

440--

441{-# INLINE unfoldrM #-}

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

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