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

git.iko.soy/streamly

on ffcb268 help sign in

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

372 lines 12 kB history raw

1{-# LANGUAGE UndecidableInstances #-}

2{-# OPTIONS_GHC -Wno-orphans #-}

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.Data.Stream as Stream

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

17--

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

19 (

20 -- * Serial appending stream

21 SerialT

22 , Serial

23 , serial

25 -- * Serial interleaving stream

26 , WSerialT(..)

27 , WSerial

28 , wSerial

29 , wSerialFst

30 , wSerialMin

31 , consMWSerial

33 -- * Construction

34 , Stream.cons

35 , Stream.consM

36 , Stream.repeat

37 , Stream.unfoldrM

38 , fromList

40 -- * Elimination

41 , toList

43 -- * Transformation

44 , map

45 , Stream.mapM

46 )

47where

49import Control.Applicative (liftA2)

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

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

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

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

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

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

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

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

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

59import Data.Maybe (fromMaybe)

60import Data.Semigroup (Endo(..))

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

62import Text.Read

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

64 , readListPrecDefault)

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

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

67import Streamly.Data.Stream (Stream)

69import qualified Streamly.Data.Fold as Fold

70import qualified Streamly.Data.Stream as Stream

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

72 (toStreamK, fromStreamK, crossApply, crossApplySnd, crossApplyFst)

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

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

75 (fromStreamK, toStreamK, mapM)

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

77 (Stream, mkStream, foldStream, cons, consM, nil, concatMapWith, fromPure

78 , bindWith, interleave, interleaveFst, interleaveMin, serial, fromEffect)

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

82#include "Instances.hs"

83#include "inline.hs"

85-- $setup

86-- >>> :set -fno-warn-deprecations

87-- >>> import qualified Streamly.Data.Stream as Stream

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-- Applicative - orphan instances

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

101

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

103-- INLINE them.

104instance Monad m => Applicative (Stream m) where

105 {-# INLINE pure #-}

106 pure = Stream.fromStreamK . K.fromPure

107

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

109 (<*>) = Stream.crossApply

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

111

112 {-# INLINE liftA2 #-}

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

114

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

116 (*>) = Stream.crossApplySnd

117 -- (*>) = K.apSerialDiscardFst

118

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

120 (<*) = Stream.crossApplyFst

121 -- (<*) = K.apSerialDiscardSnd

122

123------------------------------------------------------------------------------

124-- Monad - orphan instances

125------------------------------------------------------------------------------

126

127instance Monad m => Monad (Stream m) where

128 return = pure

129

130 -- Benchmarks better with StreamD bind and pure:

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

132 --

133 -- pure = Stream . D.fromStreamD . D.fromPure

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

135

136 -- Benchmarks better with CPS bind and pure:

137 -- Prime sieve (25x)

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

139 --

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

141 (>>=) m f =

142 Stream.fromStreamK

143 $ K.bindWith K.serial (Stream.toStreamK m) (Stream.toStreamK . f)

144

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

146 (>>) = (*>)

147

148------------------------------------------------------------------------------

149-- Transformers - orphan instances

150------------------------------------------------------------------------------

151

152instance (MonadIO m) => MonadIO (Stream m) where

153 liftIO = lift . liftIO

154

155instance (MonadThrow m) => MonadThrow (Stream m) where

156 throwM = lift . throwM

157

158instance MonadTrans Stream where

159 {-# INLINE lift #-}

160 lift = Stream.fromEffect

161

162instance (MonadReader r m) => MonadReader r (Stream m) where

163 ask = lift ask

164

165 local f m = Stream.fromStreamK $ withLocal f (Stream.toStreamK m)

166

167instance (MonadState s m) => MonadState s (Stream m) where

168 {-# INLINE get #-}

169 get = lift get

170

171 {-# INLINE put #-}

172 put x = lift (put x)

173

174 {-# INLINE state #-}

175 state k = lift (state k)

176

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

178-- NFData - orphan instances

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

180

181instance NFData a => NFData (Stream Identity a) where

182 {-# INLINE rnf #-}

183 rnf xs = runIdentity $ Stream.fold (Fold.foldl' (\_ x -> rnf x) ()) xs

184

185instance NFData1 (Stream Identity) where

186 {-# INLINE liftRnf #-}

187 liftRnf f xs = runIdentity $ Stream.fold (Fold.foldl' (\_ x -> f x) ()) xs

188

189------------------------------------------------------------------------------

190-- SerialT

191------------------------------------------------------------------------------

192

193type SerialT = Stream

194

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

196-- for more details.

197--

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

199--

200-- @since 0.8.0

201type Serial = SerialT IO

202

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

204-- Combining

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

206

207{-# INLINE serial #-}

208serial :: SerialT m a -> SerialT m a -> SerialT m a

209serial = (<>)

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 = Stream.mapM (return . f)

227

228------------------------------------------------------------------------------

229-- WSerialT

230------------------------------------------------------------------------------

231

232-- | For 'WSerialT' streams:

233--

234-- @

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

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

237-- @

238--

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

240-- in the resulting stream.

241--

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

243--

244-- >>> :{

245-- IsStream.toList $ IsStream.fromWSerial $ do

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

247-- return x

248-- :}

249-- [1,2]

250--

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

252--

253-- >>> :{

254-- IsStream.toList $ IsStream.fromWSerial $ do

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

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

257-- return (x, y)

258-- :}

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

260--

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

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

263-- @2@:

264--

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

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

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

268--

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

270-- contrast to the depth wise scheduling behavior of 'Stream'.

271--

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

273--

274-- @since 0.8.0

275newtype WSerialT m a = WSerialT {getWSerialT :: K.Stream m a}

276

277instance MonadTrans WSerialT where

278 {-# INLINE lift #-}

279 lift = WSerialT . K.fromEffect

280

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

282-- documentation for more details.

283--

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

285--

286-- @since 0.8.0

287type WSerial = WSerialT IO

288

289{-# INLINE consMWSerial #-}

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

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

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

293

294------------------------------------------------------------------------------

295-- Semigroup

296------------------------------------------------------------------------------

297

298infixr 6 `wSerial`

299

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

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

302-- the output stream.

303--

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

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

306--

307-- /Not fused/

308

309-- NOTE:

310--

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

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

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

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

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

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

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

318-- each subexpression on the right.

319--

320{-# INLINE wSerial #-}

321wSerial :: WSerialT m a -> WSerialT m a -> WSerialT m a

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

323

324{-# INLINE wSerialFst #-}

325wSerialFst :: WSerialT m a -> WSerialT m a -> WSerialT m a

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

327

328{-# INLINE wSerialMin #-}

329wSerialMin :: WSerialT m a -> WSerialT m a -> WSerialT m a

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

331

332instance Semigroup (WSerialT m a) where

333 (<>) = wSerial

334

335------------------------------------------------------------------------------

336-- Monoid

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

338

339instance Monoid (WSerialT m a) where

340 mempty = WSerialT K.nil

341 mappend = (<>)

342

343{-# INLINE apWSerial #-}

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

345apWSerial (WSerialT m1) (WSerialT m2) =

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

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

348

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

350 {-# INLINE pure #-}

351 pure = WSerialT . K.fromPure

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

353 (<*>) = apWSerial

354

355------------------------------------------------------------------------------

356-- Monad

357------------------------------------------------------------------------------

358

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

360 return = pure

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

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

363

364------------------------------------------------------------------------------

365-- Other instances

366------------------------------------------------------------------------------

367

368MONAD_COMMON_INSTANCES(WSerialT,)

369LIST_INSTANCES(WSerialT)

370NFDATA1_INSTANCE(WSerialT)

371FOLDABLE_INSTANCE(WSerialT)

372TRAVERSABLE_INSTANCE(WSerialT)