cf87d11 · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

implement maxYields to put a limit on total yields of a stream

cf87d11Harendra Kumar8 years ago

src/Streamly/Prelude.hs+2−1

compare with master

unifiedsplit

@@ -165,6 +165,7 @@ import qualified Prelude

165import qualified System.IO as IO

166

167import Streamly.SVar (MonadAsync, defState, rstState)

168+import Streamly.Streams.SVar (maxYields)

168import Streamly.Streams.StreamK (IsStream(..))

169import Streamly.Streams.StreamK (IsStream(..))

169import Streamly.Streams.Serial (SerialT)

170import Streamly.Streams.Serial (SerialT)

170

171

@@ -688,7 +689,7 @@ filterM p m = fromStreamD $ D.filterM p $ toStreamD m

688-- @since 0.1.0

689-- @since 0.1.0

689{-# INLINE take #-}

690{-# INLINE take #-}

690take :: (IsStream t, Monad m) => Int -> t m a -> t m a

691take :: (IsStream t, Monad m) => Int -> t m a -> t m a

691−take n m = fromStreamS $ S.take n $ toStreamS m

692+take n m = fromStreamS $ S.take n $ toStreamS (maxYields (Just n) m)

692

693

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

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

694--

695--

src/Streamly/SVar.hs+37−10

@@ -37,7 +37,7 @@ module Streamly.SVar

37 , atomicModifyIORefCAS

38 , ChildEvent (..)

39 , AheadHeapEntry (..)

40− , send

40+ , sendYield

41 , sendStop

42 , enqueueLIFO

43 , workLoopLIFO

@@ -46,7 +46,6 @@ module Streamly.SVar

46 , enqueueAhead

47 , pushWorkerPar

49− , maxHeap

50 , queueEmptyAhead

49 , queueEmptyAhead

51 , dequeueAhead

50 , dequeueAhead

52 , dequeueFromHeap

51 , dequeueFromHeap

@@ -158,6 +157,7 @@ data SVar t m a =

158

157

159 -- Shared output queue (events, length)

158 -- Shared output queue (events, length)

160 , outputQueue :: IORef ([ChildEvent a], Int)

159 , outputQueue :: IORef ([ChildEvent a], Int)

160+ , maxYieldLimit :: Maybe (IORef Int)

161 , outputDoorBell :: MVar () -- signal the consumer about output

162 , readOutputQ :: m [ChildEvent a]

163 , postProcess :: m Bool

@@ -187,9 +187,10 @@ data SVar t m a =

187 }

188

189data State t m a = State

190− { streamVar :: Maybe (SVar t m a)

190+ { streamVar :: Maybe (SVar t m a)

191+ , yieldLimit :: Maybe Int

191 , threadsHigh :: Int

192 , threadsHigh :: Int

192− , bufferHigh :: Int

193+ , bufferHigh :: Int

193 }

194 }

194

195

195defaultMaxThreads, defaultMaxBuffer :: Int

196defaultMaxThreads, defaultMaxBuffer :: Int

@@ -199,15 +200,23 @@ defaultMaxBuffer = 1500

199defState :: State t m a

200defState :: State t m a

200defState = State

201defState = State

201 { streamVar = Nothing

202 { streamVar = Nothing

203+ , yieldLimit = Nothing

202 , threadsHigh = defaultMaxThreads

204 , threadsHigh = defaultMaxThreads

203 , bufferHigh = defaultMaxBuffer

205 , bufferHigh = defaultMaxBuffer

204 }

206 }

205

207

208+-- XXX if perf gets affected we can have all the Nothing params in a single

209+-- structure so that we reset is fast. We can also use rewrite rules such that

210+-- reset occurs only in concurrent streams to reduce the impact on serial

211+-- streams.

206-- We can optimize this so that we clear it only if it is a Just value, it

212-- We can optimize this so that we clear it only if it is a Just value, it

207-- results in slightly better perf for zip/zipM but the performance of scan

213-- results in slightly better perf for zip/zipM but the performance of scan

208-- worsens a lot, it does not fuse.

214-- worsens a lot, it does not fuse.

209rstState :: State t m a -> State t m b

215rstState :: State t m a -> State t m b

216+rstState st = st

210−rstState st = st {streamVar = Nothing}

217+ { streamVar = Nothing

218+ , yieldLimit = Nothing

219+ }

211

220

212#ifdef DIAGNOSTICS

221#ifdef DIAGNOSTICS

213{-# NOINLINE dumpSVar #-}

222{-# NOINLINE dumpSVar #-}

@@ -337,7 +346,6 @@ doFork action exHandler =

337

346

338-- | This function is used by the producer threads to queue output for the

347-- | This function is used by the producer threads to queue output for the

339-- consumer thread to consume. Returns whether the queue has more space.

348-- consumer thread to consume. Returns whether the queue has more space.

340−{-# NOINLINE send #-}

341send :: Int -> SVar t m a -> ChildEvent a -> IO Bool

349send :: Int -> SVar t m a -> ChildEvent a -> IO Bool

342send maxOutputQLen sv msg = do

350send maxOutputQLen sv msg = do

343 len <- atomicModifyIORefCAS (outputQueue sv) $ \(es, n) ->

351 len <- atomicModifyIORefCAS (outputQueue sv) $ \(es, n) ->

@@ -355,6 +363,15 @@ send maxOutputQLen sv msg = do

355 void $ tryPutMVar (outputDoorBell sv) ()

363 void $ tryPutMVar (outputDoorBell sv) ()

356 return (len < maxOutputQLen || maxOutputQLen < 0)

364 return (len < maxOutputQLen || maxOutputQLen < 0)

357

365

366+{-# NOINLINE sendYield #-}

367+sendYield :: Int -> SVar t m a -> ChildEvent a -> IO Bool

368+sendYield maxOutputQLen sv msg = do

369+ ylimit <- case maxYieldLimit sv of

370+ Nothing -> return True

371+ Just ref -> atomicModifyIORefCAS ref $ \x -> (x - 1, x > 1)

372+ r <- send maxOutputQLen sv msg

373+ return $ r && ylimit

374+

358{-# NOINLINE sendStop #-}

375{-# NOINLINE sendStop #-}

359sendStop :: SVar t m a -> IO ()

376sendStop :: SVar t m a -> IO ()

360sendStop sv = do

377sendStop sv = do

@@ -534,9 +551,6 @@ enqueueAhead sv q m = do

534--

551--

535-- XXX review for livelock

552-- XXX review for livelock

536--

553--

537−maxHeap :: Int

538−maxHeap = 1500

539−

540{-# INLINE queueEmptyAhead #-}

554{-# INLINE queueEmptyAhead #-}

541queueEmptyAhead :: MonadIO m => IORef ([t m a], Int) -> m Bool

555queueEmptyAhead :: MonadIO m => IORef ([t m a], Int) -> m Bool

542queueEmptyAhead q = liftIO $ do

556queueEmptyAhead q = liftIO $ do

@@ -675,7 +689,15 @@ dispatchWorker maxWorkerLimit sv = do

675 -- executing. In that case we should either configure the maxWorker

689 -- executing. In that case we should either configure the maxWorker

676 -- count to higher or use parallel style instead of ahead or async

690 -- count to higher or use parallel style instead of ahead or async

677 -- style.

691 -- style.

692+ limit <- case maxYieldLimit sv of

693+ Nothing -> return maxWorkerLimit

694+ Just x -> do

695+ lim <- liftIO $ readIORef x

696+ return $

697+ if maxWorkerLimit > 0

698+ then min maxWorkerLimit lim

699+ else lim

678− when (cnt < maxWorkerLimit || maxWorkerLimit < 0) $ pushWorker sv

700+ when (cnt < limit || limit < 0) $ pushWorker sv

679

701

680{-# NOINLINE sendWorkerWait #-}

702{-# NOINLINE sendWorkerWait #-}

681sendWorkerWait :: MonadAsync m => Int -> SVar t m a -> m ()

703sendWorkerWait :: MonadAsync m => Int -> SVar t m a -> m ()

@@ -816,6 +838,9 @@ getAheadSVar st f = do

816 wfw <- newIORef False

838 wfw <- newIORef False

817 running <- newIORef S.empty

839 running <- newIORef S.empty

818 q <- newIORef ([], -1)

840 q <- newIORef ([], -1)

841+ yl <- case yieldLimit st of

842+ Nothing -> return Nothing

843+ Just x -> Just <$> newIORef x

819

844

820#ifdef DIAGNOSTICS

845#ifdef DIAGNOSTICS

821 disp <- newIORef 0

846 disp <- newIORef 0

@@ -826,6 +851,7 @@ getAheadSVar st f = do

826#endif

851#endif

827 let sv =

852 let sv =

828 SVar { outputQueue = outQ

853 SVar { outputQueue = outQ

854+ , maxYieldLimit = yl

829 , outputDoorBell = outQMv

855 , outputDoorBell = outQMv

830 , readOutputQ = readOutputQBounded (threadsHigh st) sv

856 , readOutputQ = readOutputQBounded (threadsHigh st) sv

831 , postProcess = postProcessBounded sv

857 , postProcess = postProcessBounded sv

@@ -879,6 +905,7 @@ getParallelSVar = do

879#endif

905#endif

880 let sv =

906 let sv =

881 SVar { outputQueue = outQ

907 SVar { outputQueue = outQ

908+ , maxYieldLimit = Nothing

882 , outputDoorBell = outQMv

909 , outputDoorBell = outQMv

883 , readOutputQ = readOutputQPar sv

910 , readOutputQ = readOutputQPar sv

884 , postProcess = allThreadsDone sv

911 , postProcess = allThreadsDone sv

src/Streamly/Streams/Ahead.hs+13−6

@@ -37,7 +37,7 @@ import Control.Monad.State.Class (MonadState(..))

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

38import Data.Atomics (atomicModifyIORefCAS_)

39import Data.Heap (Heap, Entry(..))

40−import Data.IORef (IORef)

40+import Data.IORef (IORef, readIORef)

41import Data.Maybe (fromJust)

42import Data.Semigroup (Semigroup(..))

@@ -51,7 +51,7 @@ import qualified Streamly.Streams.StreamK as K

52#ifdef DIAGNOSTICS

53import Control.Monad (when)

54−import Data.IORef (writeIORef, readIORef)

54+import Data.IORef (writeIORef)

55#endif

56import Prelude hiding (map)

@@ -128,7 +128,14 @@ workLoopAhead st q heap = runHeap

128 toHeap seqNo ent = do

129 hp <- liftIO $ atomicModifyIORefCAS heap $ \(h, snum) ->

130 ((H.insert (Entry seqNo ent) h, snum), h)

131+ (_, len) <- liftIO $ readIORef (outputQueue sv)

132+ let maxHeap = maxBuf - len

133+ limit <- case maxYieldLimit sv of

134+ Nothing -> return maxHeap

135+ Just ref -> do

136+ r <- liftIO $ readIORef ref

137+ return $ if r >= 0 then r else maxHeap

131− if H.size hp <= maxHeap

138+ if H.size hp <= limit

132 then runHeap

139 then runHeap

133 else liftIO $ sendStop sv

140 else liftIO $ sendStop sv

134

141

@@ -136,7 +143,7 @@ workLoopAhead st q heap = runHeap

136 yieldToHeap seqNo a r = toHeap seqNo (AheadEntryStream (a `K.cons` r))

143 yieldToHeap seqNo a r = toHeap seqNo (AheadEntryStream (a `K.cons` r))

137

144

138 singleOutput seqNo a = do

145 singleOutput seqNo a = do

139− continue <- liftIO $ send maxBuf sv (ChildYield a)

146+ continue <- liftIO $ sendYield maxBuf sv (ChildYield a)

140 if continue

147 if continue

141 then runQueueToken seqNo

148 then runQueueToken seqNo

142 else liftIO $ do

149 else liftIO $ do

@@ -144,7 +151,7 @@ workLoopAhead st q heap = runHeap

144 sendStop sv

151 sendStop sv

145

152

146 yieldOutput seqNo a r = do

153 yieldOutput seqNo a r = do

147− continue <- liftIO $ send maxBuf sv (ChildYield a)

154+ continue <- liftIO $ sendYield maxBuf sv (ChildYield a)

148 if continue

155 if continue

149 then unStream r st (runQueueToken seqNo)

156 then unStream r st (runQueueToken seqNo)

150 (singleOutput seqNo)

157 (singleOutput seqNo)

@@ -241,7 +248,7 @@ aheadS m1 m2 = Stream $ \st stp sng yld -> do

241 -- sequencing results. This means the left side cannot further

248 -- sequencing results. This means the left side cannot further

242 -- split into more ahead computations on the same SVar.

249 -- split into more ahead computations on the same SVar.

243 unStream m1 (rstState st) stp sng yld

250 unStream m1 (rstState st) stp sng yld

244− _ -> unStream (forkSVarAhead m1 m2) (rstState st) stp sng yld

251+ _ -> unStream (forkSVarAhead m1 m2) st stp sng yld

245

252

246-- | XXX we can implement it more efficienty by directly implementing instead

253-- | XXX we can implement it more efficienty by directly implementing instead

247-- of combining streams using ahead.

254-- of combining streams using ahead.

src/Streamly/Streams/Async.hs+14−6

@@ -72,10 +72,10 @@ runStreamLIFO st q m stop = unStream m st stop single yieldk

72 sv = fromJust $ streamVar st

73 maxBuf = bufferHigh st

74 single a = do

75− res <- liftIO $ send maxBuf sv (ChildYield a)

75+ res <- liftIO $ sendYield maxBuf sv (ChildYield a)

76 if res then stop else liftIO $ sendStop sv

77 yieldk a r = do

78− res <- liftIO $ send maxBuf sv (ChildYield a)

78+ res <- liftIO $ sendYield maxBuf sv (ChildYield a)

79 if res

80 then (unStream r) st stop single yieldk

81 else liftIO $ enqueueLIFO sv q r >> sendStop sv

@@ -97,10 +97,10 @@ runStreamFIFO st q m stop = unStream m st stop single yieldk

97 sv = fromJust $ streamVar st

98 maxBuf = bufferHigh st

99 single a = do

100− res <- liftIO $ send maxBuf sv (ChildYield a)

100+ res <- liftIO $ sendYield maxBuf sv (ChildYield a)

101 if res then stop else liftIO $ sendStop sv

102 yieldk a r = do

103− res <- liftIO $ send maxBuf sv (ChildYield a)

103+ res <- liftIO $ sendYield maxBuf sv (ChildYield a)

104 liftIO (enqueueFIFO sv q r)

105 if res then stop else liftIO $ sendStop sv

106

@@ -121,6 +121,9 @@ getLifoSVar st = do

121 wfw <- newIORef False

122 running <- newIORef S.empty

123 q <- newIORef []

124+ yl <- case yieldLimit st of

125+ Nothing -> return Nothing

126+ Just x -> Just <$> newIORef x

124#ifdef DIAGNOSTICS

127#ifdef DIAGNOSTICS

125 disp <- newIORef 0

128 disp <- newIORef 0

126 maxWrk <- newIORef 0

129 maxWrk <- newIORef 0

@@ -131,6 +134,7 @@ getLifoSVar st = do

131 let checkEmpty = null <$> readIORef q

134 let checkEmpty = null <$> readIORef q

132 let sv =

135 let sv =

133 SVar { outputQueue = outQ

136 SVar { outputQueue = outQ

137+ , maxYieldLimit = yl

134 , outputDoorBell = outQMv

138 , outputDoorBell = outQMv

135 , readOutputQ = readOutputQBounded (threadsHigh st) sv

139 , readOutputQ = readOutputQBounded (threadsHigh st) sv

136 , postProcess = postProcessBounded sv

140 , postProcess = postProcessBounded sv

@@ -163,6 +167,9 @@ getFifoSVar st = do

163 wfw <- newIORef False

167 wfw <- newIORef False

164 running <- newIORef S.empty

168 running <- newIORef S.empty

165 q <- newQ

169 q <- newQ

170+ yl <- case yieldLimit st of

171+ Nothing -> return Nothing

172+ Just x -> Just <$> newIORef x

166#ifdef DIAGNOSTICS

173#ifdef DIAGNOSTICS

167 disp <- newIORef 0

174 disp <- newIORef 0

168 maxWrk <- newIORef 0

175 maxWrk <- newIORef 0

@@ -172,6 +179,7 @@ getFifoSVar st = do

172#endif

179#endif

173 let sv =

180 let sv =

174 SVar { outputQueue = outQ

181 SVar { outputQueue = outQ

182+ , maxYieldLimit = yl

175 , outputDoorBell = outQMv

183 , outputDoorBell = outQMv

176 , readOutputQ = readOutputQBounded (threadsHigh st) sv

184 , readOutputQ = readOutputQBounded (threadsHigh st) sv

177 , postProcess = postProcessBounded sv

185 , postProcess = postProcessBounded sv

@@ -307,11 +315,11 @@ forkSVarAsync style m1 m2 = Stream $ \st stp sng yld -> do

307{-# INLINE joinStreamVarAsync #-}

315{-# INLINE joinStreamVarAsync #-}

308joinStreamVarAsync :: MonadAsync m

316joinStreamVarAsync :: MonadAsync m

309 => SVarStyle -> Stream m a -> Stream m a -> Stream m a

317 => SVarStyle -> Stream m a -> Stream m a -> Stream m a

310−joinStreamVarAsync style m1 m2 = Stream $ \st stp sng yld ->

318+joinStreamVarAsync style m1 m2 = Stream $ \st stp sng yld -> do

311 case streamVar st of

319 case streamVar st of

312 Just sv | svarStyle sv == style ->

320 Just sv | svarStyle sv == style ->

313 liftIO (enqueue sv m2) >> unStream m1 st stp sng yld

321 liftIO (enqueue sv m2) >> unStream m1 st stp sng yld

314− _ -> unStream (forkSVarAsync style m1 m2) (rstState st) stp sng yld

322+ _ -> unStream (forkSVarAsync style m1 m2) st stp sng yld

315

323

316------------------------------------------------------------------------------

324------------------------------------------------------------------------------

317-- Semigroup and Monoid style compositions for parallel actions

325-- Semigroup and Monoid style compositions for parallel actions

src/Streamly/Streams/Parallel.hs+1−1

@@ -67,7 +67,7 @@ runOne st m = unStream m st stop single yieldk

68 sv = fromJust $ streamVar st

69 stop = liftIO $ sendStop sv

70− sendit a = liftIO $ send (-1) sv (ChildYield a)

70+ sendit a = liftIO $ sendYield (-1) sv (ChildYield a)

71 single a = sendit a >> stop

72 -- XXX there is no flow control in parallel case. We should perhaps use a

73 -- queue and queue it back on that and exit the thread when the outputQueue

src/Streamly/Streams/SVar.hs+31

@@ -9,6 +9,8 @@

9{-# LANGUAGE UnboxedTuples #-}

10{-# LANGUAGE UndecidableInstances #-} -- XXX

12+#include "inline.h"

13+

12-- |

14-- |

13-- Module : Streamly.Streams.SVar

15-- Module : Streamly.Streams.SVar

@@ -25,6 +27,7 @@ module Streamly.Streams.SVar

25 , toSVar

27 , toSVar

26 , maxThreads

28 , maxThreads

27 , maxBuffer

29 , maxBuffer

30+ , maxYields

28 )

31 )

29where

32where

@@ -32,6 +35,7 @@ import Control.Monad.Catch (throwM)

33import Streamly.SVar

36import Streamly.SVar

34import Streamly.Streams.StreamK

37import Streamly.Streams.StreamK

38+import Streamly.Streams.Serial (SerialT)

36-- MVar diagnostics has some overhead - around 5% on asyncly null benchmark, we

40-- MVar diagnostics has some overhead - around 5% on asyncly null benchmark, we

37-- can keep it on in production to debug problems quickly if and when they

41-- can keep it on in production to debug problems quickly if and when they

@@ -85,6 +89,10 @@ fromSVar sv = fromStream $ fromStreamVar sv

85toSVar :: (IsStream t, MonadAsync m) => SVar Stream m a -> t m a -> m ()

89toSVar :: (IsStream t, MonadAsync m) => SVar Stream m a -> t m a -> m ()

86toSVar sv m = toStreamVar sv (toStream m)

90toSVar sv m = toStreamVar sv (toStream m)

92+-------------------------------------------------------------------------------

93+-- Concurrency control

94+-------------------------------------------------------------------------------

95+--

88-- | Specify the maximum number of threads that can be spawned concurrently

96-- | Specify the maximum number of threads that can be spawned concurrently

89-- when using concurrent streams. This is not the grand total number of threads

97-- when using concurrent streams. This is not the grand total number of threads

90-- but maximum threads at each point of concurrency.

98-- but maximum threads at each point of concurrency.

@@ -92,11 +100,16 @@ toSVar sv m = toStreamVar sv (toStream m)

92-- there is no limit. The default value is 1500.

100-- there is no limit. The default value is 1500.

93--

101--

94-- @since 0.4.0

102-- @since 0.4.0

103+{-# INLINE_NORMAL maxThreads #-}

95maxThreads :: IsStream t => Int -> t m a -> t m a

104maxThreads :: IsStream t => Int -> t m a -> t m a

96maxThreads n m = fromStream $ Stream $ \st stp sng yld -> do

105maxThreads n m = fromStream $ Stream $ \st stp sng yld -> do

97 let n' = if n == 0 then defaultMaxThreads else n

106 let n' = if n == 0 then defaultMaxThreads else n

98 unStream (toStream m) (st {threadsHigh = n'}) stp sng yld

107 unStream (toStream m) (st {threadsHigh = n'}) stp sng yld

108

109+{-# RULES "maxThreadsSerial serial" maxThreads = maxThreadsSerial #-}

110+maxThreadsSerial :: Int -> SerialT m a -> SerialT m a

111+maxThreadsSerial _ = id

112+

100-- | Specify the maximum size of the buffer for storing the results from

113-- | Specify the maximum size of the buffer for storing the results from

101-- concurrent computations. If the buffer becomes full we stop spawning more

114-- concurrent computations. If the buffer becomes full we stop spawning more

102-- concurrent tasks until there is space in the buffer.

115-- concurrent tasks until there is space in the buffer.

@@ -104,7 +117,25 @@ maxThreads n m = fromStream $ Stream $ \st stp sng yld -> do

104-- there is no limit. The default value is 1500.

117-- there is no limit. The default value is 1500.

105--

118--

106-- @since 0.4.0

119-- @since 0.4.0

120+{-# INLINE_NORMAL maxBuffer #-}

107maxBuffer :: IsStream t => Int -> t m a -> t m a

121maxBuffer :: IsStream t => Int -> t m a -> t m a

108maxBuffer n m = fromStream $ Stream $ \st stp sng yld -> do

122maxBuffer n m = fromStream $ Stream $ \st stp sng yld -> do

109 let n' = if n == 0 then defaultMaxBuffer else n

123 let n' = if n == 0 then defaultMaxBuffer else n

110 unStream (toStream m) (st {bufferHigh = n'}) stp sng yld

124 unStream (toStream m) (st {bufferHigh = n'}) stp sng yld

125+

126+{-# RULES "maxBuffer serial" maxBuffer = maxBufferSerial #-}

127+maxBufferSerial :: Int -> SerialT m a -> SerialT m a

128+maxBufferSerial _ = id

129+

130+-- Stop concurrent dispatches after this limit. This is useful in API's like

131+-- "take" where we want to dispatch only upto the number of elements "take"

132+-- needs. This value applies only to the immediate next level and is not

133+-- inherited by everything in enclosed scope.

134+{-# INLINE_NORMAL maxYields #-}

135+maxYields :: IsStream t => Maybe Int -> t m a -> t m a

136+maxYields n m = fromStream $ Stream $ \st stp sng yld -> do

137+ unStream (toStream m) (st {yieldLimit = n}) stp sng yld

138+

139+{-# RULES "maxYields serial" maxYields = maxYieldsSerial #-}

140+maxYieldsSerial :: Maybe Int -> SerialT m a -> SerialT m a

141+maxYieldsSerial _ = id

test/Main.hs+11

@@ -380,6 +380,10 @@ main = hspec $ do

380 describe "Composed MonadThrow parallely" $ composeWithMonadThrow parallely

381 describe "Composed MonadThrow aheadly" $ composeWithMonadThrow aheadly

382

383+ describe "take on infinite concurrent stream" $ takeInfinite asyncly

384+ describe "take on infinite concurrent stream" $ takeInfinite wAsyncly

385+ describe "take on infinite concurrent stream" $ takeInfinite aheadly

386+

383 it "asyncly crosses thread limit (2000 threads)" $

387 it "asyncly crosses thread limit (2000 threads)" $

384 runStream (asyncly $ fold $

388 runStream (asyncly $ fold $

385 replicate 2000 $ S.yieldM $ threadDelay 1000000)

389 replicate 2000 $ S.yieldM $ threadDelay 1000000)

@@ -390,6 +394,13 @@ main = hspec $ do

390 replicate 4000 $ S.yieldM $ threadDelay 1000000)

394 replicate 4000 $ S.yieldM $ threadDelay 1000000)

391 `shouldReturn` ()

395 `shouldReturn` ()

392

396

397+takeInfinite :: IsStream t => (t IO Int -> SerialT IO Int) -> Spec

398+takeInfinite t = do

399+ it "take 1" $

400+ (runStream $ t $

401+ S.take 1 $ S.repeatM (print "hello" >> return (1::Int)))

402+ `shouldReturn` ()

403+

393-- XXX need to test that we have promptly cleaned up everything after the error

404-- XXX need to test that we have promptly cleaned up everything after the error

394-- XXX We can also check the output that we are expected to get before the

405-- XXX We can also check the output that we are expected to get before the

395-- error occurs.

406-- error occurs.