66efe33 · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Include singleton as a continuation in stream type

66efe33Harendra Kumar8 years ago

src/Streamly/Core.hs+72−113

compare with master

unifiedsplit

@@ -25,6 +25,7 @@ module Streamly.Core

25 , Stream (..)

27 -- * Construction

28+ , singleton

28 , cons

29 , cons

29 , repeat

30 , repeat

30 , nil

31 , nil

@@ -145,91 +146,43 @@ data SVar m a =

145-- The stream type

146-- The stream type

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

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

147

148

148−-- TBD use a functor instead of the bare type a?

149−

150-- | The type 'Stream m a' represents a monadic stream of values of type 'a'

149-- | The type 'Stream m a' represents a monadic stream of values of type 'a'

151−-- constructed using actions in monad 'm'. It uses a stop continuation and a

150+-- constructed using actions in monad 'm'. It uses stop, singleton and yield

152−-- yield continuation. You can consider it a rough equivalent of direct style

151+-- continuations equivalent to the following direct style type:

153−-- type:

154−--

155−-- data Stream m a = Stop | Yield a (Maybe (Stream m a))

156−--

157−-- Our goal is to be able to represent finite as well as infinite streams and

158−-- being able to compose a large number of small streams efficiently. In

159−-- addition we want to compose streams in parallel, to facilitate that we

160−-- maintain a local state in an SVar that is shared across and is used for

161−-- synchronization of the streams being composed.

162−--

163−-- Using this type, there are two ways to indicate the end of a stream, one is

164−-- by calling the stop continuation and the other one is by yielding the last

165−-- value along with 'Nothing' as the rest of the stream.

166−--

167−-- Why do we have this redundancy? Why can't we use (a -> Stream m a -> m r) as

168−-- the type of the yield continuation and always use the stop continuation to

169−-- indicate the end of the stream? The reason is that when we compose a large

170−-- number of short or singleton streams then using the stop continuation

171−-- becomes expensive, just to know that there is no next element we have to

172−-- call the continuation, introducing an indirection, it seems when using CPS

173−-- GHC is not able to optimize this out as efficiently as it can be in direct

174−-- style because of the function call involved. In direct style it will just be

175−-- a constructor check and a memory access instead of a function call. So we

176−-- could use:

177−--

178−-- data Stream m a = Stop | Yield a (Stream m a)

179--

152--

180−-- In CPS style, when we use the 'Maybe' argument of yield to indicate the end

153+-- data Stream m a = Stop | Singleton a | Yield a (Stream m a)

181−-- then just like direct style we can figure out that there is no next element

182−-- without a function call.

183--

154--

184−-- Then why not get rid of the stop continuation and use only yield to indicate

155+-- To facilitate parallel composition we maintain a local state in an SVar that

185−-- the end of stream? The answer is, in that case to indicate the end of the

156+-- is shared across and is used for synchronization of the streams being

186−-- stream we would have to yield at least one element so there is no way to

157+-- composed.

187−-- represent an empty stream.

188--

158--

189−-- Whenever we make a singleton stream or in general when we build a stream

159+-- The singleton case can be expressed in terms of stop and yield but we have

190−-- strictly i.e. when we know all the elements of the stream in advance we can

160+-- it as a separate case to optimize composition operations for streams with

191−-- use the last yield to indicate the end of the stream, because we know in

192−-- advance at the time of the last yield that the stream is ending. We build

193−-- singleton streams in the implementation of 'pure' for Applicative and Monad,

161+-- single element. We build singleton streams in the implementation of 'pure'

194−-- and in 'lift' for MonadTrans, in these places we use yield with 'Nothing' to

162+-- for Applicative and Monad, and in 'lift' for MonadTrans.

195−-- indicate the end of the stream. Note that, the only advantage of Maybe is

196−-- when we have to build a large number of singleton or short streams. For

197−-- larger streams anyway the overhead of a separate stop continuation is not

198−-- significant. This could be significant when we breakdown a large stream into

199−-- its elements, process them in some way and then recompose it from the

200−-- pieces. Zipping streams is one such example. Zipping with streamly is the

201−-- fastest among all streaming libraries.

202−--

203−-- However in a lazy computation we cannot know in advance that the stream is

204−-- ending therefore we cannot use 'Maybe', we use the stop continuation in that

205−-- case. For example when building a stream from a lazy container using a right

206−-- fold.

207--

163--

208newtype Stream m a =

164newtype Stream m a =

209 Stream {

165 Stream {

210 runStream :: forall r.

166 runStream :: forall r.

211− Maybe (SVar m a) -- local state

167+ Maybe (SVar m a) -- local state

212− -> m r -- stop

168+ -> m r -- stop

169+ -> (a -> m r) -- singleton

213− -> (a -> Maybe (Stream m a) -> m r) -- yield

170+ -> (a -> Stream m a -> m r) -- yield

214 -> m r

171 -> m r

215 }

172 }

216

173

217−-- | A monad that can perform concurrent or parallel IO operations. Streams

174+nil :: Stream m a

218−-- that can be composed concurrently require the underlying monad to be

175+nil = Stream $ \_ stp _ _ -> stp

219−-- 'MonadParallel'.

220−type MonadParallel m = (MonadIO m, MonadBaseControl IO m, MonadThrow m)

221

176

222−{-# DEPRECATED MonadAsync "Please use MonadParallel instead." #-}

177+{-# INLINE singleton #-}

223−type MonadAsync m = MonadParallel m

178+singleton :: a -> Stream m a

179+singleton a = Stream $ \_ _ single _ -> single a

224

180

225−cons :: a -> Maybe (Stream m a) -> Stream m a

181+cons :: a -> Stream m a -> Stream m a

226−cons a r = Stream $ \_ _ yld -> yld a r

182+cons a r = Stream $ \_ _ _ yld -> yld a r

227

183

228repeat :: a -> Stream m a

184repeat :: a -> Stream m a

229−repeat a = let x = cons a (Just x) in x

185+repeat a = let x = cons a x in x

230−

231−nil :: Stream m a

232−nil = Stream $ \_ stp _ -> stp

233

186

234------------------------------------------------------------------------------

187------------------------------------------------------------------------------

235-- Composing streams

188-- Composing streams

@@ -249,11 +202,11 @@ nil = Stream $ \_ stp _ -> stp

249serial :: Stream m a -> Stream m a -> Stream m a

202serial :: Stream m a -> Stream m a -> Stream m a

250serial m1 m2 = go m1

203serial m1 m2 = go m1

251 where

204 where

252− go (Stream m) = Stream $ \_ stp yld ->

205+ go (Stream m) = Stream $ \_ stp sng yld ->

253− let stop = (runStream m2) Nothing stp yld

206+ let stop = (runStream m2) Nothing stp sng yld

254− yield a Nothing = yld a (Just m2)

207+ single a = yld a m2

255− yield a (Just r) = yld a (Just (go r))

208+ yield a r = yld a (go r)

256− in m Nothing stop yield

209+ in m Nothing stop single yield

257

210

258instance Semigroup (Stream m a) where

211instance Semigroup (Stream m a) where

259 (<>) = serial

212 (<>) = serial

@@ -271,16 +224,24 @@ instance Monoid (Stream m a) where

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

224------------------------------------------------------------------------------

272

225

273coserial :: Stream m a -> Stream m a -> Stream m a

226coserial :: Stream m a -> Stream m a -> Stream m a

274−coserial m1 m2 = Stream $ \_ stp yld -> do

227+coserial m1 m2 = Stream $ \_ stp sng yld -> do

275− let stop = (runStream m2) Nothing stp yld

228+ let stop = (runStream m2) Nothing stp sng yld

276− yield a Nothing = yld a (Just m2)

229+ single a = yld a m2

277− yield a (Just r) = yld a (Just (coserial m2 r))

230+ yield a r = yld a (coserial m2 r)

278− (runStream m1) Nothing stop yield

231+ (runStream m1) Nothing stop single yield

279

232

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

233------------------------------------------------------------------------------

281-- Spawning threads and collecting result in streamed fashion

234-- Spawning threads and collecting result in streamed fashion

282------------------------------------------------------------------------------

235------------------------------------------------------------------------------

283

236

237+-- | A monad that can perform concurrent or parallel IO operations. Streams

238+-- that can be composed concurrently require the underlying monad to be

239+-- 'MonadParallel'.

240+type MonadParallel m = (MonadIO m, MonadBaseControl IO m, MonadThrow m)

241+

242+{-# DEPRECATED MonadAsync "Please use MonadParallel instead." #-}

243+type MonadAsync m = MonadParallel m

244+

284{-# INLINE doFork #-}

245{-# INLINE doFork #-}

285doFork :: MonadBaseControl IO m

246doFork :: MonadBaseControl IO m

286 => m ()

247 => m ()

@@ -322,11 +283,11 @@ runqueueLIFO sv q = run

322 work <- dequeue

283 work <- dequeue

323 case work of

284 case work of

324 Nothing -> sendStop sv

285 Nothing -> sendStop sv

325− Just m -> (runStream m) (Just sv) run yield

286+ Just m -> (runStream m) (Just sv) run single yield

326

287

327 sendit a = send sv (ChildYield a)

288 sendit a = send sv (ChildYield a)

328− yield a Nothing = sendit a >> run

289+ single a = sendit a >> run

329− yield a (Just r) = sendit a >> (runStream r) (Just sv) run yield

290+ yield a r = sendit a >> (runStream r) (Just sv) run single yield

330

291

331 dequeue = liftIO $ atomicModifyIORefCAS q $ \case

292 dequeue = liftIO $ atomicModifyIORefCAS q $ \case

332 [] -> ([], Nothing)

293 [] -> ([], Nothing)

@@ -345,12 +306,12 @@ runqueueFIFO sv q = run

345 work <- dequeue

306 work <- dequeue

346 case work of

307 case work of

347 Nothing -> sendStop sv

308 Nothing -> sendStop sv

348− Just m -> (runStream m) (Just sv) run yield

309+ Just m -> (runStream m) (Just sv) run single yield

349

310

350 dequeue = liftIO $ tryPopR q

311 dequeue = liftIO $ tryPopR q

351 sendit a = send sv (ChildYield a)

312 sendit a = send sv (ChildYield a)

352− yield a Nothing = sendit a >> run

313+ single a = sendit a >> run

353− yield a (Just r) = sendit a >> liftIO (enqueueFIFO q r) >> run

314+ yield a r = sendit a >> liftIO (enqueueFIFO q r) >> run

354

315

355-- Thread tracking is needed for two reasons:

316-- Thread tracking is needed for two reasons:

356--

317--

@@ -406,7 +367,7 @@ sendWorkerWait sv = do

406-- | Pull a stream from an SVar.

367-- | Pull a stream from an SVar.

407{-# NOINLINE fromStreamVar #-}

368{-# NOINLINE fromStreamVar #-}

408fromStreamVar :: MonadParallel m => SVar m a -> Stream m a

369fromStreamVar :: MonadParallel m => SVar m a -> Stream m a

409−fromStreamVar sv = Stream $ \_ stp yld -> do

370+fromStreamVar sv = Stream $ \_ stp sng yld -> do

410 -- XXX if reading the IORef is costly we can use a flag in the SVar to

371 -- XXX if reading the IORef is costly we can use a flag in the SVar to

411 -- indicate we are done.

372 -- indicate we are done.

412 done <- allThreadsDone sv

373 done <- allThreadsDone sv

@@ -419,7 +380,7 @@ fromStreamVar sv = Stream $ \_ stp yld -> do

419 -- To avoid lock overhead we read all events at once instead of reading

380 -- To avoid lock overhead we read all events at once instead of reading

420 -- one at a time. We just reverse the list to process the events in the

381 -- one at a time. We just reverse the list to process the events in the

421 -- order they arrived. Maybe we can use a queue instead?

382 -- order they arrived. Maybe we can use a queue instead?

422− (runStream $ processEvents (reverse list)) Nothing stp yld

383+ (runStream $ processEvents (reverse list)) Nothing stp sng yld

423

384

424 where

385 where

425

386

@@ -430,15 +391,15 @@ fromStreamVar sv = Stream $ \_ stp yld -> do

430 throwM e

391 throwM e

431

392

432 {-# INLINE processEvents #-}

393 {-# INLINE processEvents #-}

433− processEvents [] = Stream $ \_ stp yld -> do

394+ processEvents [] = Stream $ \_ stp sng yld -> do

434 done <- allThreadsDone sv

395 done <- allThreadsDone sv

435 if not done

396 if not done

436− then (runStream (fromStreamVar sv)) Nothing stp yld

397+ then (runStream (fromStreamVar sv)) Nothing stp sng yld

437 else stp

398 else stp

438

399

439− processEvents (ev : es) = Stream $ \_ stp yld -> do

400+ processEvents (ev : es) = Stream $ \_ stp sng yld -> do

440− let continue = (runStream (processEvents es)) Nothing stp yld

401+ let continue = (runStream (processEvents es)) Nothing stp sng yld

441− yield a = yld a (Just (processEvents es))

402+ yield a = yld a (processEvents es)

442

403

443 case ev of

404 case ev of

444 ChildYield a -> yield a

405 ChildYield a -> yield a

@@ -567,9 +528,9 @@ toStreamVar sv m = do

567{-# NOINLINE withNewSVar2 #-}

528{-# NOINLINE withNewSVar2 #-}

568withNewSVar2 :: MonadParallel m

529withNewSVar2 :: MonadParallel m

569 => SVarStyle -> Stream m a -> Stream m a -> Stream m a

530 => SVarStyle -> Stream m a -> Stream m a -> Stream m a

570−withNewSVar2 style m1 m2 = Stream $ \_ stp yld -> do

531+withNewSVar2 style m1 m2 = Stream $ \_ stp sng yld -> do

571 sv <- newStreamVar2 style m1 m2

532 sv <- newStreamVar2 style m1 m2

572− (runStream (fromStreamVar sv)) Nothing stp yld

533+ (runStream (fromStreamVar sv)) Nothing stp sng yld

573

534

574-- | Join two computations on the currently running 'SVar' queue for concurrent

535-- | Join two computations on the currently running 'SVar' queue for concurrent

575-- execution. The 'SVarStyle' required by the current composition context is

536-- execution. The 'SVarStyle' required by the current composition context is

@@ -602,11 +563,11 @@ withNewSVar2 style m1 m2 = Stream $ \_ stp yld -> do

602{-# INLINE joinStreamVar2 #-}

563{-# INLINE joinStreamVar2 #-}

603joinStreamVar2 :: MonadParallel m

564joinStreamVar2 :: MonadParallel m

604 => SVarStyle -> Stream m a -> Stream m a -> Stream m a

565 => SVarStyle -> Stream m a -> Stream m a -> Stream m a

605−joinStreamVar2 style m1 m2 = Stream $ \st stp yld ->

566+joinStreamVar2 style m1 m2 = Stream $ \svr stp sng yld ->

606− case st of

567+ case svr of

607 Just sv | svarStyle sv == style ->

568 Just sv | svarStyle sv == style ->

608− liftIO ((enqueue sv) m2) >> (runStream m1) st stp yld

569+ liftIO ((enqueue sv) m2) >> (runStream m1) svr stp sng yld

609− _ -> (runStream (withNewSVar2 style m1 m2)) Nothing stp yld

570+ _ -> (runStream (withNewSVar2 style m1 m2)) Nothing stp sng yld

610

571

611------------------------------------------------------------------------------

572------------------------------------------------------------------------------

612-- Semigroup and Monoid style compositions for parallel actions

573-- Semigroup and Monoid style compositions for parallel actions

@@ -646,10 +607,9 @@ instance Monad m => Monad (Stream m) where

646------------------------------------------------------------------------------

607------------------------------------------------------------------------------

647

608

648alt :: Stream m a -> Stream m a -> Stream m a

609alt :: Stream m a -> Stream m a -> Stream m a

649−alt m1 m2 = Stream $ \_ stp yld ->

610+alt m1 m2 = Stream $ \_ stp sng yld ->

650− let stop = runStream m2 Nothing stp yld

611+ let stop = runStream m2 Nothing stp sng yld

651− yield = yld

652− in runStream m1 Nothing stop yield

612+ in runStream m1 Nothing stop sng yld

653

613

654instance MonadParallel m => Alternative (Stream m) where

614instance MonadParallel m => Alternative (Stream m) where

655 empty = nil

615 empty = nil

@@ -664,7 +624,7 @@ instance MonadParallel m => MonadPlus (Stream m) where

664-------------------------------------------------------------------------------

624-------------------------------------------------------------------------------

665

625

666instance MonadTrans Stream where

626instance MonadTrans Stream where

667− lift mx = Stream $ \_ _ yld -> mx >>= (\a -> (yld a Nothing))

627+ lift mx = Stream $ \_ _ single _ -> mx >>= single

668

628

669instance (MonadBase b m, Monad m) => MonadBase b (Stream m) where

629instance (MonadBase b m, Monad m) => MonadBase b (Stream m) where

670 liftBase = liftBaseDefault

630 liftBase = liftBaseDefault

@@ -682,18 +642,17 @@ instance MonadThrow m => MonadThrow (Stream m) where

682-- XXX handle and test cross thread state transfer

642-- XXX handle and test cross thread state transfer

683instance MonadError e m => MonadError e (Stream m) where

643instance MonadError e m => MonadError e (Stream m) where

684 throwError = lift . throwError

644 throwError = lift . throwError

685− catchError m h = Stream $ \st stp yld ->

645+ catchError m h = Stream $ \svr stp sng yld ->

686− let handle r = r `catchError` \e -> (runStream (h e)) st stp yld

646+ let handle r = r `catchError` \e -> (runStream (h e)) svr stp sng yld

687− yield a Nothing = yld a Nothing

688− yield a (Just r) = yld a (Just (catchError r h))

647+ yield a r = yld a (catchError r h)

689− in handle $ (runStream m) st stp yield

648+ in handle $ (runStream m) svr stp sng yield

690

649

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

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

692 ask = lift ask

651 ask = lift ask

693− local f m = Stream $ \st stp yld ->

652+ local f m = Stream $ \svr stp sng yld ->

694− let yield a Nothing = local f $ yld a Nothing

653+ let single = local f . sng

695− yield a (Just r) = local f $ yld a (Just (local f r))

654+ yield a r = local f $ yld a (local f r)

696− in (runStream m) st (local f stp) yield

655+ in (runStream m) svr (local f stp) single yield

697

656

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

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

699 get = lift get

658 get = lift get

src/Streamly/Prelude.hs+131−137

@@ -114,20 +114,20 @@ import Streamly.Streams

114unfoldr :: IsStream t => (b -> Maybe (a, b)) -> b -> t m a

115unfoldr step = fromStream . go

116 where

117− go s = Stream $ \_ stp yld ->

117+ go s = Stream $ \_ stp _ yld ->

118 case step s of

119 Nothing -> stp

120− Just (a, b) -> yld a (Just (go b))

120+ Just (a, b) -> yld a (go b)

121

122-- | Build a Stream by unfolding monadic steps starting from a seed.

123unfoldrM :: (IsStream t, Monad m) => (b -> m (Maybe (a, b))) -> b -> t m a

124unfoldrM step = fromStream . go

125 where

126− go s = Stream $ \_ stp yld -> do

126+ go s = Stream $ \_ stp _ yld -> do

127 mayb <- step s

128 case mayb of

129 Nothing -> stp

130− Just (a, b) -> yld a (Just (go b))

130+ Just (a, b) -> yld a (go b)

131

132-- | Construct a stream from a 'Foldable' container.

133{-# INLINE fromFoldable #-}

@@ -144,27 +144,27 @@ each = fromFoldable

144iterate :: IsStream t => (a -> a) -> a -> t m a

145iterate step = fromStream . go

146 where

147− go s = S.cons s (Just (go (step s)))

147+ go s = S.cons s (go (step s))

148

149-- | Iterate a monadic function from a seed value, streaming the results forever

150iterateM :: (IsStream t, Monad m) => (a -> m a) -> a -> t m a

151iterateM step = fromStream . go

152 where

153− go s = Stream $ \_ _ yld -> do

153+ go s = Stream $ \_ _ _ yld -> do

154 a <- step s

155− yld s (Just (go a))

155+ yld s (go a)

156

157-- | Read lines from an IO Handle into a stream of Strings.

158fromHandle :: (IsStream t, MonadIO m) => IO.Handle -> t m String

159fromHandle h = fromStream go

160 where

161− go = Stream $ \_ stp yld -> do

161+ go = Stream $ \_ stp _ yld -> do

162 eof <- liftIO $ IO.hIsEOF h

163 if eof

164 then stp

165 else do

166 str <- liftIO $ IO.hGetLine h

167− yld str (Just go)

167+ yld str go

168

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

170-- Elimination

@@ -181,9 +181,9 @@ foldr step acc m = go (toStream m)

181 where

182 go m1 =

183 let stop = return acc

184− yield a Nothing = return (step a acc)

184+ single a = return (step a acc)

185− yield a (Just x) = go x >>= \b -> return (step a b)

185+ yield a r = go r >>= \b -> return (step a b)

186− in (S.runStream m1) Nothing stop yield

186+ in (S.runStream m1) Nothing stop single yield

187

188-- | Lazy right fold with a monadic step function. For example, to fold a

189-- stream into a list:

@@ -198,9 +198,9 @@ foldrM step acc m = go (toStream m)

198 where

199 go m1 =

200 let stop = return acc

201− yield a Nothing = step a acc

201+ single a = step a acc

202− yield a (Just x) = (go x) >>= (step a)

202+ yield a r = go r >>= step a

203− in (S.runStream m1) Nothing stop yield

203+ in (S.runStream m1) Nothing stop single yield

204

205-- | Strict left scan with an extraction function. Like 'scanl'', but applies a

206-- user supplied extraction function (the third argument) at each step. This is

@@ -210,13 +210,12 @@ foldrM step acc m = go (toStream m)

210scanx :: IsStream t => (x -> a -> x) -> x -> (x -> b) -> t m a -> t m b

211scanx step begin done m = cons (done begin) $ fromStream $ go (toStream m) begin

212 where

213− go m1 !acc = Stream $ \_ stp yld ->

213+ go m1 !acc = Stream $ \_ stp sng yld ->

214− let stop = stp

215− yield a Nothing = yld (done $ step acc a) Nothing

214+ let single a = sng (done $ step acc a)

216− yield a (Just x) =

215+ yield a r =

217 let s = step acc a

216 let s = step acc a

218− in yld (done s) (Just (go x s))

217+ in yld (done s) (go r s)

219− in S.runStream m1 Nothing stop yield

218+ in S.runStream m1 Nothing stp single yield

220

219

221{-# DEPRECATED scan "Please use scanx instead." #-}

220{-# DEPRECATED scan "Please use scanx instead." #-}

222scan :: IsStream t => (x -> a -> x) -> x -> (x -> b) -> t m a -> t m b

221scan :: IsStream t => (x -> a -> x) -> x -> (x -> b) -> t m a -> t m b

@@ -240,22 +239,20 @@ foldx step begin done m = get $ go (toStream m) begin

240 where

239 where

241 {-# NOINLINE get #-}

240 {-# NOINLINE get #-}

242 get m1 =

241 get m1 =

243− let yield a Nothing = return $ done a

242+ let single = return . done

244− yield _ _ = undefined

245− in (S.runStream m1) Nothing undefined yield

243+ in (S.runStream m1) Nothing undefined single undefined

246

244

247 -- Note, this can be implemented by making a recursive call to "go",

245 -- Note, this can be implemented by making a recursive call to "go",

248 -- however that is more expensive because of unnecessary recursion

246 -- however that is more expensive because of unnecessary recursion

249 -- that cannot be tail call optimized. Unfolding recursion explicitly via

247 -- that cannot be tail call optimized. Unfolding recursion explicitly via

250 -- continuations is much more efficient.

248 -- continuations is much more efficient.

251− go m1 !acc = Stream $ \_ _ yld ->

249+ go m1 !acc = Stream $ \_ _ sng yld ->

252− let stop = yld acc Nothing

250+ let stop = sng acc

251+ single a = sng $ step acc a

253 yield a r =

252 yield a r =

254− let s = step acc a

253+ let stream = go r (step acc a)

255− in case r of

256− Nothing -> yld s Nothing

257− Just x -> (S.runStream (go x s)) Nothing undefined yld

254+ in (S.runStream stream) Nothing undefined sng yld

258− in (S.runStream m1) Nothing stop yield

255+ in (S.runStream m1) Nothing stop single yield

259

256

260{-# DEPRECATED foldl "Please use foldx instead." #-}

257{-# DEPRECATED foldl "Please use foldx instead." #-}

261foldl :: Monad m => (x -> a -> x) -> x -> (x -> b) -> SerialT m a -> m b

258foldl :: Monad m => (x -> a -> x) -> x -> (x -> b) -> SerialT m a -> m b

@@ -273,9 +270,9 @@ foldxM step begin done m = go begin (toStream m)

273 where

270 where

274 go !acc m1 =

271 go !acc m1 =

275 let stop = acc >>= done

272 let stop = acc >>= done

276− yield a Nothing = acc >>= \b -> step b a >>= done

273+ single a = acc >>= \b -> step b a >>= done

277− yield a (Just x) = acc >>= \b -> go (step b a) x

274+ yield a r = acc >>= \b -> go (step b a) r

278− in (S.runStream m1) Nothing stop yield

275+ in (S.runStream m1) Nothing stop single yield

279

276

280{-# DEPRECATED foldlM "Please use foldxM instead." #-}

277{-# DEPRECATED foldlM "Please use foldxM instead." #-}

281foldlM :: Monad m => (x -> a -> m x) -> m x -> (x -> m b) -> SerialT m a -> m b

278foldlM :: Monad m => (x -> a -> m x) -> m x -> (x -> m b) -> SerialT m a -> m b

@@ -291,9 +288,9 @@ foldlM' step begin m = foldxM step (return begin) return m

291uncons :: (IsStream t, Monad m) => SerialT m a -> m (Maybe (a, t m a))

288uncons :: (IsStream t, Monad m) => SerialT m a -> m (Maybe (a, t m a))

292uncons m =

289uncons m =

293 let stop = return Nothing

290 let stop = return Nothing

294− yield a Nothing = return (Just (a, nil))

291+ single a = return (Just (a, nil))

295− yield a (Just x) = return (Just (a, fromStream x))

292+ yield a r = return (Just (a, fromStream r))

296− in (S.runStream (toStream m)) Nothing stop yield

293+ in (S.runStream (toStream m)) Nothing stop single yield

297

294

298-- | Write a stream of Strings to an IO Handle.

295-- | Write a stream of Strings to an IO Handle.

299toHandle :: MonadIO m => IO.Handle -> SerialT m String -> m ()

296toHandle :: MonadIO m => IO.Handle -> SerialT m String -> m ()

@@ -301,9 +298,9 @@ toHandle h m = go (toStream m)

301 where

298 where

302 go m1 =

299 go m1 =

303 let stop = return ()

300 let stop = return ()

304− yield a Nothing = liftIO (IO.hPutStrLn h a)

301+ single a = liftIO (IO.hPutStrLn h a)

305− yield a (Just x) = liftIO (IO.hPutStrLn h a) >> go x

302+ yield a r = liftIO (IO.hPutStrLn h a) >> go r

306− in (S.runStream m1) Nothing stop yield

303+ in (S.runStream m1) Nothing stop single yield

307

304

308------------------------------------------------------------------------------

305------------------------------------------------------------------------------

309-- Special folds

306-- Special folds

@@ -319,46 +316,45 @@ toList = foldrM (\a xs -> return (a : xs)) []

319take :: IsStream t => Int -> t m a -> t m a

316take :: IsStream t => Int -> t m a -> t m a

320take n m = fromStream $ go n (toStream m)

317take n m = fromStream $ go n (toStream m)

321 where

318 where

322− go n1 m1 = Stream $ \ctx stp yld ->

319+ go n1 m1 = Stream $ \ctx stp sng yld ->

323− let yield a Nothing = yld a Nothing

324− yield a (Just x) = yld a (Just (go (n1 - 1) x))

320+ let yield a r = yld a (go (n1 - 1) r)

325− in if n1 <= 0 then stp else (S.runStream m1) ctx stp yield

321+ in if n1 <= 0 then stp else (S.runStream m1) ctx stp sng yield

326

322

327-- | Include only those elements that pass a predicate.

323-- | Include only those elements that pass a predicate.

328{-# INLINE filter #-}

324{-# INLINE filter #-}

329filter :: IsStream t => (a -> Bool) -> t m a -> t m a

325filter :: IsStream t => (a -> Bool) -> t m a -> t m a

330filter p m = fromStream $ go (toStream m)

326filter p m = fromStream $ go (toStream m)

331 where

327 where

332− go m1 = Stream $ \ctx stp yld ->

328+ go m1 = Stream $ \ctx stp sng yld ->

333− let yield a Nothing | p a = yld a Nothing

329+ let single a | p a = sng a

334− | otherwise = stp

330+ | otherwise = stp

335− yield a (Just x) | p a = yld a (Just (go x))

331+ yield a r | p a = yld a (go r)

336− | otherwise = (S.runStream x) ctx stp yield

332+ | otherwise = (S.runStream r) ctx stp single yield

337− in (S.runStream m1) ctx stp yield

333+ in (S.runStream m1) ctx stp single yield

338

334

339-- | End the stream as soon as the predicate fails on an element.

335-- | End the stream as soon as the predicate fails on an element.

340{-# INLINE takeWhile #-}

336{-# INLINE takeWhile #-}

341takeWhile :: IsStream t => (a -> Bool) -> t m a -> t m a

337takeWhile :: IsStream t => (a -> Bool) -> t m a -> t m a

342takeWhile p m = fromStream $ go (toStream m)

338takeWhile p m = fromStream $ go (toStream m)

343 where

339 where

344− go m1 = Stream $ \ctx stp yld ->

340+ go m1 = Stream $ \ctx stp sng yld ->

345− let yield a Nothing | p a = yld a Nothing

341+ let single a | p a = sng a

346− | otherwise = stp

342+ | otherwise = stp

347− yield a (Just x) | p a = yld a (Just (go x))

343+ yield a r | p a = yld a (go r)

348− | otherwise = stp

344+ | otherwise = stp

349− in (S.runStream m1) ctx stp yield

345+ in (S.runStream m1) ctx stp single yield

350

346

351-- | Discard first 'n' elements from the stream and take the rest.

347-- | Discard first 'n' elements from the stream and take the rest.

352drop :: IsStream t => Int -> t m a -> t m a

348drop :: IsStream t => Int -> t m a -> t m a

353drop n m = fromStream $ go n (toStream m)

349drop n m = fromStream $ go n (toStream m)

354 where

350 where

355− go n1 m1 = Stream $ \ctx stp yld ->

351+ go n1 m1 = Stream $ \ctx stp sng yld ->

356− let yield _ Nothing = stp

352+ let single _ = stp

357− yield _ (Just x) = (S.runStream $ go (n1 - 1) x) ctx stp yld

353+ yield _ r = (S.runStream $ go (n1 - 1) r) ctx stp sng yld

358 -- Somehow "<=" check performs better than a ">"

354 -- Somehow "<=" check performs better than a ">"

359 in if n1 <= 0

355 in if n1 <= 0

360− then (S.runStream m1) ctx stp yld

356+ then (S.runStream m1) ctx stp sng yld

361− else (S.runStream m1) ctx stp yield

357+ else (S.runStream m1) ctx stp single yield

362

358

363-- | Drop elements in the stream as long as the predicate succeeds and then

359-- | Drop elements in the stream as long as the predicate succeeds and then

364-- take the rest of the stream.

360-- take the rest of the stream.

@@ -366,34 +362,34 @@ drop n m = fromStream $ go n (toStream m)

366dropWhile :: IsStream t => (a -> Bool) -> t m a -> t m a

362dropWhile :: IsStream t => (a -> Bool) -> t m a -> t m a

367dropWhile p m = fromStream $ go (toStream m)

363dropWhile p m = fromStream $ go (toStream m)

368 where

364 where

369− go m1 = Stream $ \ctx stp yld ->

365+ go m1 = Stream $ \ctx stp sng yld ->

370− let yield a Nothing | p a = stp

366+ let single a | p a = stp

371− | otherwise = yld a Nothing

367+ | otherwise = sng a

372− yield a (Just x) | p a = (S.runStream x) ctx stp yield

368+ yield a r | p a = (S.runStream r) ctx stp single yield

373− | otherwise = yld a (Just x)

369+ | otherwise = yld a r

374− in (S.runStream m1) ctx stp yield

370+ in (S.runStream m1) ctx stp single yield

375

371

376-- | Determine whether all elements of a stream satisfy a predicate.

372-- | Determine whether all elements of a stream satisfy a predicate.

377all :: Monad m => (a -> Bool) -> SerialT m a -> m Bool

373all :: Monad m => (a -> Bool) -> SerialT m a -> m Bool

378all p m = go (toStream m)

374all p m = go (toStream m)

379 where

375 where

380 go m1 =

376 go m1 =

381− let yield a Nothing | p a = return True

377+ let single a | p a = return True

382− | otherwise = return False

378+ | otherwise = return False

383− yield a (Just x) | p a = go x

379+ yield a r | p a = go r

384− | otherwise = return False

380+ | otherwise = return False

385− in (S.runStream m1) Nothing (return True) yield

381+ in (S.runStream m1) Nothing (return True) single yield

386

382

387-- | Determine whether any of the elements of a stream satisfy a predicate.

383-- | Determine whether any of the elements of a stream satisfy a predicate.

388any :: Monad m => (a -> Bool) -> SerialT m a -> m Bool

384any :: Monad m => (a -> Bool) -> SerialT m a -> m Bool

389any p m = go (toStream m)

385any p m = go (toStream m)

390 where

386 where

391 go m1 =

387 go m1 =

392− let yield a Nothing | p a = return True

388+ let single a | p a = return True

393− | otherwise = return False

389+ | otherwise = return False

394− yield a (Just x) | p a = return True

390+ yield a r | p a = return True

395− | otherwise = go x

391+ | otherwise = go r

396− in (S.runStream m1) Nothing (return False) yield

392+ in (S.runStream m1) Nothing (return False) single yield

397

393

398-- | Determine the sum of all elements of a stream of numbers

394-- | Determine the sum of all elements of a stream of numbers

399sum :: (Monad m, Num a) => SerialT m a -> m a

395sum :: (Monad m, Num a) => SerialT m a -> m a

@@ -407,16 +403,17 @@ product = foldl (*) 1 id

407head :: Monad m => SerialT m a -> m (Maybe a)

403head :: Monad m => SerialT m a -> m (Maybe a)

408head m =

404head m =

409 let stop = return Nothing

405 let stop = return Nothing

406+ single a = return (Just a)

410 yield a _ = return (Just a)

407 yield a _ = return (Just a)

411− in (S.runStream (toStream m)) Nothing stop yield

408+ in (S.runStream (toStream m)) Nothing stop single yield

412

409

413-- | Extract all but the first element of the stream, if any.

410-- | Extract all but the first element of the stream, if any.

414tail :: (IsStream t, Monad m) => SerialT m a -> m (Maybe (t m a))

411tail :: (IsStream t, Monad m) => SerialT m a -> m (Maybe (t m a))

415tail m =

412tail m =

416− let stop = return Nothing

413+ let stop = return Nothing

417− yield _ Nothing = return $ Just nil

414+ single _ = return $ Just nil

418− yield _ (Just t) = return $ Just $ fromStream t

415+ yield _ r = return $ Just $ fromStream r

419− in (S.runStream (toStream m)) Nothing stop yield

416+ in (S.runStream (toStream m)) Nothing stop single yield

420

417

421-- | Extract the last element of the stream, if any.

418-- | Extract the last element of the stream, if any.

422{-# INLINE last #-}

419{-# INLINE last #-}

@@ -427,28 +424,29 @@ last = foldl (\_ y -> Just y) Nothing id

427null :: Monad m => SerialT m a -> m Bool

424null :: Monad m => SerialT m a -> m Bool

428null m =

425null m =

429 let stop = return True

426 let stop = return True

427+ single _ = return False

430 yield _ _ = return False

428 yield _ _ = return False

431− in (S.runStream (toStream m)) Nothing stop yield

429+ in (S.runStream (toStream m)) Nothing stop single yield

432

430

433-- | Determine whether an element is present in the stream.

431-- | Determine whether an element is present in the stream.

434elem :: (Monad m, Eq a) => a -> SerialT m a -> m Bool

432elem :: (Monad m, Eq a) => a -> SerialT m a -> m Bool

435elem e m = go (toStream m)

433elem e m = go (toStream m)

436 where

434 where

437 go m1 =

435 go m1 =

438− let stop = return False

436+ let stop = return False

439− yield a Nothing = return (a == e)

437+ single a = return (a == e)

440− yield a (Just x) = if a == e then return True else go x

438+ yield a r = if a == e then return True else go r

441− in (S.runStream m1) Nothing stop yield

439+ in (S.runStream m1) Nothing stop single yield

442

440

443-- | Determine whether an element is not present in the stream.

441-- | Determine whether an element is not present in the stream.

444notElem :: (Monad m, Eq a) => a -> SerialT m a -> m Bool

442notElem :: (Monad m, Eq a) => a -> SerialT m a -> m Bool

445notElem e m = go (toStream m)

443notElem e m = go (toStream m)

446 where

444 where

447 go m1 =

445 go m1 =

448− let stop = return True

446+ let stop = return True

449− yield a Nothing = return (a /= e)

447+ single a = return (a /= e)

450− yield a (Just x) = if a == e then return False else go x

448+ yield a r = if a == e then return False else go r

451− in (S.runStream m1) Nothing stop yield

449+ in (S.runStream m1) Nothing stop single yield

452

450

453-- | Determine the length of the stream.

451-- | Determine the length of the stream.

454length :: Monad m => SerialT m a -> m Int

452length :: Monad m => SerialT m a -> m Int

@@ -457,28 +455,27 @@ length = foldl (\n _ -> n + 1) 0 id

457-- | Returns the elements of the stream in reverse order.

455-- | Returns the elements of the stream in reverse order.

458-- The stream must be finite.

456-- The stream must be finite.

459reverse :: (IsStream t) => t m a -> t m a

457reverse :: (IsStream t) => t m a -> t m a

460−reverse m = fromStream $ go Nothing (toStream m)

458+reverse m = fromStream $ go S.nil (toStream m)

461 where

459 where

462− go rev rest = Stream $ \svr stp yld ->

460+ go rev rest = Stream $ \svr stp sng yld ->

463− let stop = case rev of

464− Nothing -> stp

465− Just str -> S.runStream str svr stp yld

461+ let run x = S.runStream x svr stp sng yld

466− yield a Nothing = S.runStream (a `S.cons` rev) svr stp yld

462+ stop = run rev

463+ single a = run $ a `S.cons` rev

467− yield a (Just x) = S.runStream (go (Just $ a `S.cons` rev) x) svr stp yld

464+ yield a r = run $ go (a `S.cons` rev) r

468− in S.runStream rest svr stop yield

465+ in S.runStream rest svr stop single yield

469

466

470-- XXX replace the recursive "go" with continuation

467-- XXX replace the recursive "go" with continuation

471-- | Determine the minimum element in a stream.

468-- | Determine the minimum element in a stream.

472minimum :: (Monad m, Ord a) => SerialT m a -> m (Maybe a)

469minimum :: (Monad m, Ord a) => SerialT m a -> m (Maybe a)

473minimum m = go Nothing (toStream m)

470minimum m = go Nothing (toStream m)

474 where

471 where

475− go r m1 =

472+ go res m1 =

476− let stop = return r

473+ let stop = return res

477− yield a Nothing = return $ min_ a r

474+ single a = return $ min_ a res

478− yield a (Just x) = go (min_ a r) x

475+ yield a r = go (min_ a res) r

479− in (S.runStream m1) Nothing stop yield

476+ in (S.runStream m1) Nothing stop single yield

480

477

481− min_ a r = case r of

478+ min_ a res = case res of

482 Nothing -> Just a

479 Nothing -> Just a

483 Just e -> Just $ min a e

480 Just e -> Just $ min a e

484

481

@@ -487,13 +484,13 @@ minimum m = go Nothing (toStream m)

487maximum :: (Monad m, Ord a) => SerialT m a -> m (Maybe a)

484maximum :: (Monad m, Ord a) => SerialT m a -> m (Maybe a)

488maximum m = go Nothing (toStream m)

485maximum m = go Nothing (toStream m)

489 where

486 where

490− go r m1 =

487+ go res m1 =

491− let stop = return r

488+ let stop = return res

492− yield a Nothing = return $ max_ a r

489+ single a = return $ max_ a res

493− yield a (Just x) = go (max_ a r) x

490+ yield a r = go (max_ a res) r

494− in (S.runStream m1) Nothing stop yield

491+ in (S.runStream m1) Nothing stop single yield

495

492

496− max_ a r = case r of

493+ max_ a res = case res of

497 Nothing -> Just a

494 Nothing -> Just a

498 Just e -> Just $ max a e

495 Just e -> Just $ max a e

499

496

@@ -509,11 +506,10 @@ maximum m = go Nothing (toStream m)

509mapM :: (IsStream t, Monad m) => (a -> m b) -> t m a -> t m b

506mapM :: (IsStream t, Monad m) => (a -> m b) -> t m a -> t m b

510mapM f m = fromStream $ go (toStream m)

507mapM f m = fromStream $ go (toStream m)

511 where

508 where

512− go m1 = Stream $ \_ stp yld ->

509+ go m1 = Stream $ \_ stp sng yld ->

513− let stop = stp

510+ let single a = f a >>= sng

514− yield a Nothing = f a >>= \b -> yld b Nothing

515− yield a (Just x) = f a >>= \b -> yld b (Just (go x))

511+ yield a r = f a >>= \b -> yld b (go r)

516− in (S.runStream m1) Nothing stop yield

512+ in (S.runStream m1) Nothing stp single yield

517

513

518-- | Apply a monadic action to each element of the stream and discard the

514-- | Apply a monadic action to each element of the stream and discard the

519-- output of the action.

515-- output of the action.

@@ -522,29 +518,28 @@ mapM_ f m = go (toStream m)

522 where

518 where

523 go m1 =

519 go m1 =

524 let stop = return ()

520 let stop = return ()

525− yield a Nothing = void (f a)

521+ single a = void (f a)

526− yield a (Just x) = f a >> go x

522+ yield a r = f a >> go r

527− in (S.runStream m1) Nothing stop yield

523+ in (S.runStream m1) Nothing stop single yield

528

524

529-- | Reduce a stream of monadic actions to a stream of the output of those

525-- | Reduce a stream of monadic actions to a stream of the output of those

530-- actions.

526-- actions.

531sequence :: (IsStream t, Monad m) => t m (m a) -> t m a

527sequence :: (IsStream t, Monad m) => t m (m a) -> t m a

532sequence m = fromStream $ go (toStream m)

528sequence m = fromStream $ go (toStream m)

533 where

529 where

534− go m1 = Stream $ \_ stp yld ->

530+ go m1 = Stream $ \_ stp sng yld ->

535− let stop = stp

531+ let single ma = ma >>= sng

536− yield a Nothing = a >>= \b -> yld b Nothing

537− yield a (Just x) = a >>= \b -> yld b (Just (go x))

532+ yield ma r = ma >>= \b -> yld b (go r)

538− in (S.runStream m1) Nothing stop yield

533+ in (S.runStream m1) Nothing stp single yield

539

534

540-- | Generate a stream by performing an action @n@ times.

535-- | Generate a stream by performing an action @n@ times.

541replicateM :: (IsStream t, Monad m) => Int -> m a -> t m a

536replicateM :: (IsStream t, Monad m) => Int -> m a -> t m a

542replicateM n m = fromStream $ go n

537replicateM n m = fromStream $ go n

543 where

538 where

544− go cnt = Stream $ \_ stp yld ->

539+ go cnt = Stream $ \_ stp _ yld ->

545 if cnt <= 0

540 if cnt <= 0

546 then stp

541 then stp

547− else m >>= \a -> yld a (Just $ go (cnt - 1))

542+ else m >>= \a -> yld a (go (cnt - 1))

548

543

549------------------------------------------------------------------------------

544------------------------------------------------------------------------------

550-- Serially Zipping Streams

545-- Serially Zipping Streams

@@ -554,16 +549,15 @@ replicateM n m = fromStream $ go n

554zipWithM :: IsStream t => (a -> b -> t m c) -> t m a -> t m b -> t m c

549zipWithM :: IsStream t => (a -> b -> t m c) -> t m a -> t m b -> t m c

555zipWithM f m1 m2 = fromStream $ go (toStream m1) (toStream m2)

550zipWithM f m1 m2 = fromStream $ go (toStream m1) (toStream m2)

556 where

551 where

557− go mx my = Stream $ \_ stp yld -> do

552+ go mx my = Stream $ \_ stp sng yld -> do

558 let merge a ra =

553 let merge a ra =

559− let yield2 b Nothing = (S.runStream (g a b)) Nothing stp yld

554+ let run x = S.runStream x Nothing stp sng yld

560− yield2 b (Just rb) =

555+ single2 b = run $ toStream (f a b)

561− (S.runStream (g a b <> go ra rb)) Nothing stp yld

556+ yield2 b rb = run $ toStream (f a b) <> go ra rb

562− in (S.runStream my) Nothing stp yield2

557+ in (S.runStream my) Nothing stp single2 yield2

563− let yield1 a Nothing = merge a S.nil

558+ let single1 a = merge a S.nil

564− yield1 a (Just ra) = merge a ra

559+ yield1 a ra = merge a ra

565− (S.runStream mx) Nothing stp yield1

560+ (S.runStream mx) Nothing stp single1 yield1

566− g a b = toStream $ f a b

567

561

568------------------------------------------------------------------------------

562------------------------------------------------------------------------------

569-- Parallely Zipping Streams

563-- Parallely Zipping Streams

@@ -573,10 +567,10 @@ zipWithM f m1 m2 = fromStream $ go (toStream m1) (toStream m2)

573-- concurrently) using a monadic zipping function.

567-- concurrently) using a monadic zipping function.

574zipParallelWithM :: (IsStream t, MonadParallel m)

568zipParallelWithM :: (IsStream t, MonadParallel m)

575 => (a -> b -> t m c) -> t m a -> t m b -> t m c

569 => (a -> b -> t m c) -> t m a -> t m b -> t m c

576−zipParallelWithM f m1 m2 = fromStream $ Stream $ \_ stp yld -> do

570+zipParallelWithM f m1 m2 = fromStream $ Stream $ \_ stp sng yld -> do

577 ma <- async m1

571 ma <- async m1

578 mb <- async m2

572 mb <- async m2

579− (S.runStream (toStream (zipWithM f ma mb))) Nothing stp yld

573+ (S.runStream (toStream (zipWithM f ma mb))) Nothing stp sng yld

580

574

581{-# DEPRECATED zipAsyncWithM "Please use zipParallelWithM instead." #-}

575{-# DEPRECATED zipAsyncWithM "Please use zipParallelWithM instead." #-}

582zipAsyncWithM :: (IsStream t, MonadParallel m)

576zipAsyncWithM :: (IsStream t, MonadParallel m)

src/Streamly/Streams.hs+78−79

@@ -114,6 +114,7 @@ import Control.Monad.Trans.Class (MonadTrans)

114import Data.Semigroup (Semigroup(..))

115import Prelude hiding (zipWith)

116import Streamly.Core ( MonadParallel, Stream(Stream)

117+ , singleton

117 , SVar, SVarStyle(..)

118 , SVar, SVarStyle(..)

118 , SVarTag(..), SVarSched(..))

119 , SVarTag(..), SVarSched(..))

119import qualified Streamly.Core as S

120import qualified Streamly.Core as S

@@ -151,7 +152,7 @@ infixr 5 `cons`

151-- [1,2,3]

152-- [1,2,3]

152-- @

153-- @

153cons :: IsStream t => a -> t m a -> t m a

154cons :: IsStream t => a -> t m a -> t m a

154−cons a r = fromStream $ S.cons a (Just (toStream r))

155+cons a r = fromStream $ S.cons a (toStream r)

155

156

156infixr 5 .:

157infixr 5 .:

157

158

@@ -191,18 +192,18 @@ infixr 5 .:

191-- represent an "empty" stream.

192-- represent an "empty" stream.

192streamBuild :: IsStream t

193streamBuild :: IsStream t

193 => (forall r. Maybe (SVar m a)

194 => (forall r. Maybe (SVar m a)

194− -> (a -> Maybe (t m a) -> m r)

195+ -> (a -> t m a -> m r)

196+ -> (a -> m r)

195 -> m r

197 -> m r

196 -> m r)

198 -> m r)

197 -> t m a

199 -> t m a

198−streamBuild k = fromStream $ Stream $ \sv stp yld ->

200+streamBuild k = fromStream $ Stream $ \svr stp sng yld ->

199− let yield a Nothing = yld a Nothing

200− yield a (Just r) = yld a (Just (toStream r))

201+ let yield a r = yld a (toStream r)

201− in k sv yield stp

202+ in k svr yield sng stp

202

203

203-- | Build a singleton stream from a callback function.

204-- | Build a singleton stream from a callback function.

204fromCallback :: IsStream t => (forall r. (a -> m r) -> m r) -> t m a

205fromCallback :: IsStream t => (forall r. (a -> m r) -> m r) -> t m a

205−fromCallback k = fromStream $ Stream $ \_ _ yld -> k (\a -> yld a Nothing)

206+fromCallback k = fromStream $ Stream $ \_ _ sng _ -> k sng

206

207

207-- | Read an SVar to get a stream.

208-- | Read an SVar to get a stream.

208fromSVar :: (MonadParallel m, IsStream t) => SVar m a -> t m a

209fromSVar :: (MonadParallel m, IsStream t) => SVar m a -> t m a

@@ -215,12 +216,17 @@ fromSVar sv = fromStream $ S.fromStreamVar sv

215-- | Fold a stream using its church encoding. The second argument is the "step"

216-- | Fold a stream using its church encoding. The second argument is the "step"

216-- function consuming an element and the remaining stream, if any. The third

217-- function consuming an element and the remaining stream, if any. The third

217-- argument is for consuming an "empty" stream that yields nothing.

218-- argument is for consuming an "empty" stream that yields nothing.

218−streamFold :: IsStream t

219+streamFold

220+ :: IsStream t

219− => Maybe (SVar m a) -> (a -> Maybe (t m a) -> m r) -> m r -> t m a -> m r

221+ => Maybe (SVar m a)

222+ -> (a -> t m a -> m r)

223+ -> (a -> m r)

224+ -> m r

225+ -> t m a

226+ -> m r

220−streamFold sv step blank m =

227+streamFold svr step single blank m =

221− let yield a Nothing = step a Nothing

222− yield a (Just x) = step a (Just (fromStream x))

228+ let yield a x = step a (fromStream x)

223− in (S.runStream (toStream m)) sv blank yield

229+ in (S.runStream (toStream m)) svr blank single yield

224

230

225-- | Run a streaming composition, discard the results. By default it interprets

231-- | Run a streaming composition, discard the results. By default it interprets

226-- the stream as 'SerialT', to run other types of streams use the type adapting

232-- the stream as 'SerialT', to run other types of streams use the type adapting

@@ -230,9 +236,9 @@ runStream m = go (toStream m)

230 where

236 where

231 go m1 =

237 go m1 =

232 let stop = return ()

238 let stop = return ()

233− yield _ Nothing = stop

239+ single _ = return ()

234− yield _ (Just x) = go x

240+ yield _ r = go r

235− in (S.runStream m1) Nothing stop yield

241+ in (S.runStream m1) Nothing stop single yield

236

242

237-- | Same as 'runStream'

243-- | Same as 'runStream'

238{-# Deprecated runStreaming "Please use runStream instead." #-}

244{-# Deprecated runStreaming "Please use runStream instead." #-}

@@ -354,18 +360,18 @@ serial m1 m2 = fromStream $ S.serial (toStream m1) (toStream m2)

354

360

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

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

356 return = pure

362 return = pure

357− (SerialT (Stream m)) >>= f = SerialT $ Stream $ \_ stp yld ->

363+ (SerialT (Stream m)) >>= f = SerialT $ Stream $ \_ stp sng yld ->

358− let run x = (S.runStream x) Nothing stp yld

364+ let run x = (S.runStream x) Nothing stp sng yld

359− yield a Nothing = run $ toStream (f a)

365+ single a = run $ toStream (f a)

360− yield a (Just r) = run $ toStream $ f a <> (fromStream r >>= f)

366+ yield a r = run $ toStream $ f a <> (fromStream r >>= f)

361− in m Nothing stp yield

367+ in m Nothing stp single yield

362

368

363------------------------------------------------------------------------------

369------------------------------------------------------------------------------

364-- Applicative

370-- Applicative

365------------------------------------------------------------------------------

371------------------------------------------------------------------------------

366

372

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

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

368− pure a = SerialT $ S.cons a Nothing

374+ pure = SerialT . singleton

369 (<*>) = ap

375 (<*>) = ap

370

376

371------------------------------------------------------------------------------

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

@@ -373,11 +379,10 @@ instance Monad m => Applicative (SerialT m) where

373------------------------------------------------------------------------------

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

374

380

375instance Monad m => Functor (SerialT m) where

381instance Monad m => Functor (SerialT m) where

376− fmap f (SerialT (Stream m)) = SerialT $ Stream $ \_ stp yld ->

382+ fmap f m = fromStream $ Stream $ \_ stp sng yld ->

377− let yield a Nothing = yld (f a) Nothing

383+ let single = sng . f

378− yield a (Just r) = yld (f a)

384+ yield a r = yld (f a) (toStream $ fmap f (SerialT r))

379− (Just (getSerialT (fmap f (SerialT r))))

380− in m Nothing stp yield

385+ in (S.runStream (toStream m)) Nothing stp single yield

381

386

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

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

383-- Num

388-- Num

@@ -501,18 +506,18 @@ infixr 5 <=>

501

506

502instance Monad m => Monad (CoserialT m) where

507instance Monad m => Monad (CoserialT m) where

503 return = pure

508 return = pure

504− (CoserialT (Stream m)) >>= f = CoserialT $ Stream $ \_ stp yld ->

509+ (CoserialT (Stream m)) >>= f = CoserialT $ Stream $ \_ stp sng yld ->

505− let run x = (S.runStream x) Nothing stp yld

510+ let run x = (S.runStream x) Nothing stp sng yld

506− yield a Nothing = run $ toStream (f a)

511+ single a = run $ toStream (f a)

507− yield a (Just r) = run $ toStream $ f a <> (fromStream r >>= f)

512+ yield a r = run $ toStream $ f a <> (fromStream r >>= f)

508− in m Nothing stp yield

513+ in m Nothing stp single yield

509

514

510------------------------------------------------------------------------------

515------------------------------------------------------------------------------

511-- Applicative

516-- Applicative

512------------------------------------------------------------------------------

517------------------------------------------------------------------------------

513

518

514instance Monad m => Applicative (CoserialT m) where

519instance Monad m => Applicative (CoserialT m) where

515− pure a = CoserialT $ S.cons a Nothing

520+ pure = CoserialT . singleton

516 (<*>) = ap

521 (<*>) = ap

517

522

518------------------------------------------------------------------------------

523------------------------------------------------------------------------------

@@ -520,11 +525,10 @@ instance Monad m => Applicative (CoserialT m) where

520------------------------------------------------------------------------------

525------------------------------------------------------------------------------

521

526

522instance Monad m => Functor (CoserialT m) where

527instance Monad m => Functor (CoserialT m) where

523− fmap f (CoserialT (Stream m)) = CoserialT $ Stream $ \_ stp yld ->

528+ fmap f m = fromStream $ Stream $ \_ stp sng yld ->

524− let yield a Nothing = yld (f a) Nothing

529+ let single = sng . f

525− yield a (Just r) =

526− yld (f a) (Just (getCoserialT (fmap f (CoserialT r))))

530+ yield a r = yld (f a) (toStream $ fmap f (CoserialT r))

527− in m Nothing stp yield

531+ in (S.runStream (toStream m)) Nothing stp single yield

528

532

529------------------------------------------------------------------------------

533------------------------------------------------------------------------------

530-- Num

534-- Num

@@ -666,24 +670,23 @@ parbind

666parbind par m f = go m

670parbind par m f = go m

667 where

671 where

668 go (Stream g) =

672 go (Stream g) =

669− Stream $ \ctx stp yld ->

673+ Stream $ \ctx stp sng yld ->

670− let run x = (S.runStream x) ctx stp yld

674+ let run x = (S.runStream x) ctx stp sng yld

671− yield a Nothing = run $ f a

675+ single a = run $ f a

672− yield a (Just r) = run $ f a `par` go r

676+ yield a r = run $ f a `par` go r

673− in g Nothing stp yield

677+ in g Nothing stp single yield

674

678

675instance MonadParallel m => Monad (CoparallelT m) where

679instance MonadParallel m => Monad (CoparallelT m) where

676 return = pure

680 return = pure

677− (CoparallelT m) >>= f = CoparallelT $ parbind par m g

681+ (CoparallelT m) >>= f = CoparallelT $ parbind par m (getCoparallelT . f)

678− where g x = getCoparallelT (f x)

679− par = S.joinStreamVar2 (SVarStyle Conjunction LIFO)

682+ where par = S.joinStreamVar2 (SVarStyle Conjunction LIFO)

680

683

681------------------------------------------------------------------------------

684------------------------------------------------------------------------------

682-- Applicative

685-- Applicative

683------------------------------------------------------------------------------

686------------------------------------------------------------------------------

684

687

685instance MonadParallel m => Applicative (CoparallelT m) where

688instance MonadParallel m => Applicative (CoparallelT m) where

686− pure a = CoparallelT $ S.cons a Nothing

689+ pure = CoparallelT . singleton

687 (<*>) = ap

690 (<*>) = ap

688

691

689------------------------------------------------------------------------------

692------------------------------------------------------------------------------

@@ -691,10 +694,10 @@ instance MonadParallel m => Applicative (CoparallelT m) where

691------------------------------------------------------------------------------

694------------------------------------------------------------------------------

692

695

693instance Monad m => Functor (CoparallelT m) where

696instance Monad m => Functor (CoparallelT m) where

694− fmap f (CoparallelT (Stream m)) = CoparallelT $ Stream $ \_ stp yld ->

697+ fmap f m = fromStream $ Stream $ \_ stp sng yld ->

695− let yield a Nothing = yld (f a) Nothing

698+ let single = sng . f

696− yield a (Just r) = yld (f a) (Just (getCoparallelT (fmap f (CoparallelT r))))

699+ yield a r = yld (f a) (toStream $ fmap f (CoparallelT r))

697− in m Nothing stp yield

700+ in (S.runStream (toStream m)) Nothing stp single yield

698

701

699------------------------------------------------------------------------------

702------------------------------------------------------------------------------

700-- Num

703-- Num

@@ -817,16 +820,15 @@ instance MonadParallel m => Semigroup (ParallelT m a) where

817

820

818instance MonadParallel m => Monad (ParallelT m) where

821instance MonadParallel m => Monad (ParallelT m) where

819 return = pure

822 return = pure

820− (ParallelT m) >>= f = ParallelT $ parbind par m g

823+ (ParallelT m) >>= f = ParallelT $ parbind par m (getParallelT . f)

821− where g x = getParallelT (f x)

822− par = S.joinStreamVar2 (SVarStyle Conjunction FIFO)

824+ where par = S.joinStreamVar2 (SVarStyle Conjunction FIFO)

823

825

824------------------------------------------------------------------------------

826------------------------------------------------------------------------------

825-- Applicative

827-- Applicative

826------------------------------------------------------------------------------

828------------------------------------------------------------------------------

827

829

828instance MonadParallel m => Applicative (ParallelT m) where

830instance MonadParallel m => Applicative (ParallelT m) where

829− pure a = ParallelT $ S.cons a Nothing

831+ pure = ParallelT . singleton

830 (<*>) = ap

832 (<*>) = ap

831

833

832------------------------------------------------------------------------------

834------------------------------------------------------------------------------

@@ -834,11 +836,10 @@ instance MonadParallel m => Applicative (ParallelT m) where

834------------------------------------------------------------------------------

836------------------------------------------------------------------------------

835

837

836instance Monad m => Functor (ParallelT m) where

838instance Monad m => Functor (ParallelT m) where

837− fmap f (ParallelT (Stream m)) = ParallelT $ Stream $ \_ stp yld ->

839+ fmap f m = fromStream $ Stream $ \_ stp sng yld ->

838− let yield a Nothing = yld (f a) Nothing

840+ let single = sng . f

839− yield a (Just r) = yld (f a)

841+ yield a r = yld (f a) (toStream $ fmap f (ParallelT r))

840− (Just (getParallelT (fmap f (ParallelT r))))

841− in m Nothing stp yield

842+ in (S.runStream (toStream m)) Nothing stp single yield

842

843

843------------------------------------------------------------------------------

844------------------------------------------------------------------------------

844-- Num

845-- Num

@@ -892,14 +893,14 @@ instance (MonadParallel m, Floating a) => Floating (ParallelT m a) where

892zipWith :: IsStream t => (a -> b -> c) -> t m a -> t m b -> t m c

893zipWith :: IsStream t => (a -> b -> c) -> t m a -> t m b -> t m c

893zipWith f m1 m2 = fromStream $ go (toStream m1) (toStream m2)

894zipWith f m1 m2 = fromStream $ go (toStream m1) (toStream m2)

894 where

895 where

895− go mx my = Stream $ \_ stp yld -> do

896+ go mx my = Stream $ \_ stp sng yld -> do

896 let merge a ra =

897 let merge a ra =

897− let yield2 b Nothing = yld (f a b) Nothing

898+ let single2 b = sng (f a b)

898− yield2 b (Just rb) = yld (f a b) (Just (go ra rb))

899+ yield2 b rb = yld (f a b) (go ra rb)

899− in (S.runStream my) Nothing stp yield2

900+ in (S.runStream my) Nothing stp single2 yield2

900− let yield1 a Nothing = merge a S.nil

901+ let single1 a = merge a S.nil

901− yield1 a (Just ra) = merge a ra

902+ yield1 a ra = merge a ra

902− (S.runStream mx) Nothing stp yield1

903+ (S.runStream mx) Nothing stp single1 yield1

903

904

904-- | The applicative instance of 'ZipSerial' zips a number of streams serially

905-- | The applicative instance of 'ZipSerial' zips a number of streams serially

905-- i.e. it produces one element from each stream serially and then zips all

906-- i.e. it produces one element from each stream serially and then zips all

@@ -928,11 +929,10 @@ type ZipStream = ZipSerial

928deriving instance MonadParallel m => Alternative (ZipSerial m)

929deriving instance MonadParallel m => Alternative (ZipSerial m)

929

930

930instance Monad m => Functor (ZipSerial m) where

931instance Monad m => Functor (ZipSerial m) where

931− fmap f (ZipSerial (Stream m)) = ZipSerial $ Stream $ \_ stp yld ->

932+ fmap f m = fromStream $ Stream $ \_ stp sng yld ->

932− let yield a Nothing = yld (f a) Nothing

933+ let single = sng . f

933− yield a (Just r) = yld (f a)

934+ yield a r = yld (f a) (toStream $ fmap f (ZipSerial r))

934− (Just (getZipSerial (fmap f (ZipSerial r))))

935− in m Nothing stp yield

935+ in (S.runStream (toStream m)) Nothing stp single yield

936

937instance Monad m => Applicative (ZipSerial m) where

938 pure = ZipSerial . S.repeat

@@ -990,10 +990,10 @@ instance (Monad m, Floating a) => Floating (ZipSerial m a) where

990-- generated concurrently) using a pure zipping function.

991zipParallelWith :: (IsStream t, MonadParallel m)

992 => (a -> b -> c) -> t m a -> t m b -> t m c

993−zipParallelWith f m1 m2 = fromStream $ Stream $ \_ stp yld -> do

993+zipParallelWith f m1 m2 = fromStream $ Stream $ \_ stp sng yld -> do

994 ma <- async m1

995 mb <- async m2

996− (S.runStream (toStream (zipWith f ma mb))) Nothing stp yld

996+ (S.runStream (toStream (zipWith f ma mb))) Nothing stp sng yld

997

998{-# DEPRECATED zipAsyncWith "Please use zipParallelWith instead." #-}

999zipAsyncWith :: (IsStream t, MonadParallel m)

@@ -1025,11 +1025,10 @@ type ZipAsync = ZipParallel

1025deriving instance MonadParallel m => Alternative (ZipParallel m)

1026

1027instance Monad m => Functor (ZipParallel m) where

1028− fmap f (ZipParallel (Stream m)) = ZipParallel $ Stream $ \_ stp yld ->

1028+ fmap f m = fromStream $ Stream $ \_ stp sng yld ->

1029− let yield a Nothing = yld (f a) Nothing

1029+ let single = sng . f

1030− yield a (Just r) = yld (f a)

1031− (Just (getZipParallel (fmap f (ZipParallel r))))

1030+ yield a r = yld (f a) (toStream $ fmap f (ZipParallel r))

1032− in m Nothing stp yield

1031+ in (S.runStream (toStream m)) Nothing stp single yield

1033

1032

1034instance MonadParallel m => Applicative (ZipParallel m) where

1033instance MonadParallel m => Applicative (ZipParallel m) where

1035 pure = ZipParallel . S.repeat

1034 pure = ZipParallel . S.repeat