ed7f4ab · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Implement bounded buffering for ParallelT streams

ed7f4abHarendra Kumar7 years ago

Changelog.md+6

compare with master

unifiedsplit

@@ -5,6 +5,12 @@

5* Change the signature of `foldrM` to ensure that it is lazy

6* `scanx` would now require an additional `Monad m` constraint.

7* Change the signature of `iterateM` to ensure that it is lazy.

8+* Earlier ParallelT was unaffected by `maxBuffer` directive, now `maxBuffer`

9+ can limit the buffer of a ParallelT stream as well. When the buffer becomes

10+ full the producer threads block.

11+* ParallelT streams no longer have an unlimited buffer by default. Now the

12+ buffer for parallel streams is limited to 1500 by default, the same as other

13+ concurrent stream types.

9### Bug Fixes

15### Bug Fixes

src/Streamly/SVar.hs+131−28

@@ -88,8 +88,9 @@ module Streamly.SVar

88 , workerRateControl

89 , updateYieldCount

90 , decrementYieldLimit

91− , decrementYieldLimitPost

92 , incrementYieldLimit

91 , incrementYieldLimit

92+ , decrementBufferLimit

93+ , incrementBufferLimit

93 , postProcessBounded

94 , postProcessBounded

94 , postProcessPaced

95 , postProcessPaced

95 , readOutputQBounded

96 , readOutputQBounded

@@ -107,7 +108,7 @@ where

107import Control.Concurrent

108import Control.Concurrent

108 (ThreadId, myThreadId, threadDelay, throwTo)

109 (ThreadId, myThreadId, threadDelay, throwTo)

109import Control.Concurrent.MVar

110import Control.Concurrent.MVar

110− (MVar, newEmptyMVar, tryPutMVar, takeMVar, newMVar, tryReadMVar)

111+ (MVar, newEmptyMVar, tryPutMVar, takeMVar, tryTakeMVar, newMVar, tryReadMVar)

111import Control.Exception

112import Control.Exception

112 (SomeException(..), catch, mask, assert, Exception, catches,

113 (SomeException(..), catch, mask, assert, Exception, catches,

113 throwIO, Handler(..), BlockedIndefinitelyOnMVar(..),

114 throwIO, Handler(..), BlockedIndefinitelyOnMVar(..),

@@ -350,6 +351,26 @@ data SVarStopStyle =

350 | StopBy -- stop when a specific stream finishes

351 | StopBy -- stop when a specific stream finishes

351 deriving (Eq, Show)

352 deriving (Eq, Show)

352

353

354+-- | Buffering policy for persistent push workers (in ParallelT). In a pull

355+-- style SVar (in AsyncT, AheadT etc.), the consumer side dispatches workers on

356+-- demand, workers terminate if the buffer is full or if the consumer is not

357+-- cosuming fast enough. In a push style SVar, a worker is dispatched only

358+-- once, workers are persistent and keep pushing work to the consumer via a

359+-- bounded buffer. If the buffer becomes full the worker either blocks, or it

360+-- can drop an item from the buffer to make space.

361+--

362+-- Pull style SVars are useful in lazy stream evaluation whereas push style

363+-- SVars are useful in strict left Folds.

364+--

365+-- XXX Maybe we can separate the implementation in two different types instead

366+-- of using a common SVar type.

367+--

368+data PushBufferPolicy =

369+ PushBufferDropNew -- drop the latest element and continue

370+ | PushBufferDropOld -- drop the oldest element and continue

371+ | PushBufferBlock -- block the thread until space

372+ -- becomes available

373+

353-- IMPORTANT NOTE: we cannot update the SVar after generating it as we have

374-- IMPORTANT NOTE: we cannot update the SVar after generating it as we have

354-- references to the original SVar stored in several functions which will keep

375-- references to the original SVar stored in several functions which will keep

355-- pointing to the original data and the new updates won't reflect there.

376-- pointing to the original data and the new updates won't reflect there.

@@ -392,6 +413,13 @@ data SVar t m a = SVar

392 -- case exceeding the requested buffer size.

413 -- case exceeding the requested buffer size.

393 , maxWorkerLimit :: Limit

414 , maxWorkerLimit :: Limit

394 , maxBufferLimit :: Limit

415 , maxBufferLimit :: Limit

416+ -- These two are valid and used only when maxBufferLimit is Limited.

417+ , pushBufferSpace :: IORef Count

418+ , pushBufferPolicy :: PushBufferPolicy

419+ -- [LOCKING] The consumer puts this MVar after emptying the buffer, workers

420+ -- block on it when the buffer becomes full. No overhead unless the buffer

421+ -- becomes full.

422+ , pushBufferMVar :: MVar ()

395

423

396 -- [LOCKING] Read only access by consumer when dispatching a worker.

424 -- [LOCKING] Read only access by consumer when dispatching a worker.

397 -- Decremented by workers when picking work and undo decrement if the

425 -- Decremented by workers when picking work and undo decrement if the

@@ -920,17 +948,6 @@ decrementYieldLimit sv =

920 r <- atomicModifyIORefCAS ref $ \x -> (x - 1, x)

948 r <- atomicModifyIORefCAS ref $ \x -> (x - 1, x)

921 return $ r >= 1

949 return $ r >= 1

922

950

923−-- decrementYieldLimit returns False when the old limit is 0. This one returns

924−-- False when the old limit is 1.

925−{-# INLINE decrementYieldLimitPost #-}

926−decrementYieldLimitPost :: SVar t m a -> IO Bool

927−decrementYieldLimitPost sv =

928− case remainingWork sv of

929− Nothing -> return True

930− Just ref -> do

931− r <- atomicModifyIORefCAS ref $ \x -> (x - 1, x)

932− return $ r > 1

933−

934{-# INLINE incrementYieldLimit #-}

951{-# INLINE incrementYieldLimit #-}

935incrementYieldLimit :: SVar t m a -> IO ()

952incrementYieldLimit :: SVar t m a -> IO ()

936incrementYieldLimit sv =

953incrementYieldLimit sv =

@@ -946,15 +963,81 @@ incrementYieldLimit sv =

946-- XXX Only yields should be counted in the buffer limit and not the Stop

963-- XXX Only yields should be counted in the buffer limit and not the Stop

947-- events.

964-- events.

948

965

966+{-# INLINE decrementBufferLimit #-}

967+decrementBufferLimit :: SVar t m a -> IO ()

968+decrementBufferLimit sv =

969+ case maxBufferLimit sv of

970+ Unlimited -> return ()

971+ Limited _ -> do

972+ let ref = pushBufferSpace sv

973+ old <- atomicModifyIORefCAS ref $ \x ->

974+ (if x >= 1 then x - 1 else x, x)

975+ when (old <= 0) $

976+ case pushBufferPolicy sv of

977+ PushBufferBlock -> blockAndRetry

978+ PushBufferDropNew -> do

979+ -- We just drop one item and proceed. It is possible

980+ -- that by the time we drop the item the consumer

981+ -- thread might have run and created space in the

982+ -- buffer, but we do not care about that condition.

983+ -- This is not pedantically correct but it should be

984+ -- fine in practice.

985+ -- XXX we may want to drop only if n == maxBuf

986+ -- otherwise we must have space in the buffer and a

987+ -- decrement should be possible.

988+ block <- atomicModifyIORefCAS (outputQueue sv) $

989+ \(es, n) ->

990+ case es of

991+ [] -> (([],n), True)

992+ _ : xs -> ((xs, n - 1), False)

993+ when block blockAndRetry

994+ -- XXX need a dequeue or ring buffer for this

995+ PushBufferDropOld -> undefined

996+

997+ where

998+

999+ blockAndRetry = do

1000+ let ref = pushBufferSpace sv

1001+ liftIO $ takeMVar (pushBufferMVar sv)

1002+ old <- atomicModifyIORefCAS ref $ \x ->

1003+ (if x >= 1 then x - 1 else x, x)

1004+ -- When multiple threads sleep on takeMVar, the first thread would

1005+ -- wakeup due to a putMVar by the consumer, but the rest of the threads

1006+ -- would have to put back the MVar after taking it and decrementing the

1007+ -- buffer count, otherwise all other threads will remain asleep.

1008+ if old >= 1

1009+ then void $ liftIO $ tryPutMVar (pushBufferMVar sv) ()

1010+ -- We do not put the MVar back in this case, instead we

1011+ -- wait for the consumer to put it.

1012+ else blockAndRetry

1013+

1014+{-# INLINE incrementBufferLimit #-}

1015+incrementBufferLimit :: SVar t m a -> IO ()

1016+incrementBufferLimit sv =

1017+ case maxBufferLimit sv of

1018+ Unlimited -> return ()

1019+ Limited _ -> do

1020+ atomicModifyIORefCAS_ (pushBufferSpace sv) (+ 1)

1021+ writeBarrier

1022+ void $ liftIO $ tryPutMVar (pushBufferMVar sv) ()

1023+

1024+{-# INLINE resetBufferLimit #-}

1025+resetBufferLimit :: SVar t m a -> IO ()

1026+resetBufferLimit sv =

1027+ case maxBufferLimit sv of

1028+ Unlimited -> return ()

1029+ Limited n -> atomicModifyIORefCAS_ (pushBufferSpace sv)

1030+ (const (fromIntegral n))

1031+

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

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

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

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

951−send :: SVar t m a -> ChildEvent a -> IO Bool

1034+send :: SVar t m a -> ChildEvent a -> IO Int

952send sv msg = do

1035send sv msg = do

953 -- XXX can the access to outputQueue and maxBufferLimit be made faster

1036 -- XXX can the access to outputQueue and maxBufferLimit be made faster

954 -- somehow?

1037 -- somehow?

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

1038+ oldlen <- atomicModifyIORefCAS (outputQueue sv) $ \(es, n) ->

956 ((msg : es, n + 1), n)

1039 ((msg : es, n + 1), n)

957− when (len <= 0) $ do

1040+ when (oldlen <= 0) $ do

958 -- The wake up must happen only after the store has finished otherwise

1041 -- The wake up must happen only after the store has finished otherwise

959 -- we can have lost wakeup problems.

1042 -- we can have lost wakeup problems.

960 writeBarrier

1043 writeBarrier

@@ -965,13 +1048,7 @@ send sv msg = do

965 -- The important point is that the consumer is guaranteed to receive a

1048 -- The important point is that the consumer is guaranteed to receive a

966 -- doorbell if something was added to the queue after it empties it.

1049 -- doorbell if something was added to the queue after it empties it.

967 void $ tryPutMVar (outputDoorBell sv) ()

1050 void $ tryPutMVar (outputDoorBell sv) ()

968−

1051+ return oldlen

969− let limit = maxBufferLimit sv

970− case limit of

971− Unlimited -> return True

972− Limited lim -> do

973− active <- readIORef (workerCount sv)

974− return $ len < (fromIntegral lim - active)

975

1052

976workerCollectLatency :: WorkerInfo -> IO (Maybe (Count, NanoSecond64))

1053workerCollectLatency :: WorkerInfo -> IO (Maybe (Count, NanoSecond64))

977workerCollectLatency winfo = do

1054workerCollectLatency winfo = do

@@ -1062,10 +1139,18 @@ workerRateControl sv yinfo winfo = do

1062-- XXX we should do rate control here but not latency update in case of ahead

1139-- XXX we should do rate control here but not latency update in case of ahead

1063-- streams. latency update must be done when we yield directly to outputQueue

1140-- streams. latency update must be done when we yield directly to outputQueue

1064-- or when we yield to heap.

1141-- or when we yield to heap.

1142+--

1143+-- returns whether the worker should continue (True) or stop (False).

1065{-# INLINE sendYield #-}

1144{-# INLINE sendYield #-}

1066sendYield :: SVar t m a -> Maybe WorkerInfo -> ChildEvent a -> IO Bool

1145sendYield :: SVar t m a -> Maybe WorkerInfo -> ChildEvent a -> IO Bool

1067sendYield sv mwinfo msg = do

1146sendYield sv mwinfo msg = do

1068− r <- send sv msg

1147+ oldlen <- send sv msg

1148+ let limit = maxBufferLimit sv

1149+ bufferSpaceOk <- case limit of

1150+ Unlimited -> return True

1151+ Limited lim -> do

1152+ active <- readIORef (workerCount sv)

1153+ return $ (oldlen + 1) < (fromIntegral lim - active)

1069 rateLimitOk <-

1154 rateLimitOk <-

1070 case mwinfo of

1155 case mwinfo of

1071 Just winfo ->

1156 Just winfo ->

@@ -1073,7 +1158,7 @@ sendYield sv mwinfo msg = do

1073 Nothing -> return True

1158 Nothing -> return True

1074 Just yinfo -> workerRateControl sv yinfo winfo

1159 Just yinfo -> workerRateControl sv yinfo winfo

1075 Nothing -> return True

1160 Nothing -> return True

1076− return $ r && rateLimitOk

1161+ return $ bufferSpaceOk && rateLimitOk

1077

1162

1078{-# INLINE workerStopUpdate #-}

1163{-# INLINE workerStopUpdate #-}

1079workerStopUpdate :: WorkerInfo -> YieldRateInfo -> IO ()

1164workerStopUpdate :: WorkerInfo -> YieldRateInfo -> IO ()

@@ -2102,6 +2187,9 @@ getAheadSVar st f mrun = do

2102 { outputQueue = outQ

2187 { outputQueue = outQ

2103 , remainingWork = yl

2188 , remainingWork = yl

2104 , maxBufferLimit = getMaxBuffer st

2189 , maxBufferLimit = getMaxBuffer st

2190+ , pushBufferSpace = undefined

2191+ , pushBufferPolicy = undefined

2192+ , pushBufferMVar = undefined

2105 , maxWorkerLimit = min (getMaxThreads st) (getMaxBuffer st)

2193 , maxWorkerLimit = min (getMaxThreads st) (getMaxBuffer st)

2106 , yieldRateInfo = rateInfo

2194 , yieldRateInfo = rateInfo

2107 , outputDoorBell = outQMv

2195 , outputDoorBell = outQMv

@@ -2173,6 +2261,12 @@ getParallelSVar ss st mrun = do

2173 Nothing -> return Nothing

2261 Nothing -> return Nothing

2174 Just x -> Just <$> newIORef x

2262 Just x -> Just <$> newIORef x

2175 rateInfo <- getYieldRateInfo st

2263 rateInfo <- getYieldRateInfo st

2264+ let bufLim =

2265+ case getMaxBuffer st of

2266+ Unlimited -> undefined

2267+ Limited x -> (fromIntegral x)

2268+ remBuf <- newIORef bufLim

2269+ pbMVar <- newMVar ()

2176

2270

2177 stats <- newSVarStats

2271 stats <- newSVarStats

2178 tid <- myThreadId

2272 tid <- myThreadId

@@ -2184,8 +2278,11 @@ getParallelSVar ss st mrun = do

2184

2278

2185 let sv =

2279 let sv =

2186 SVar { outputQueue = outQ

2280 SVar { outputQueue = outQ

2187− , remainingWork = yl

2281+ , remainingWork = yl

2188− , maxBufferLimit = Unlimited

2282+ , maxBufferLimit = getMaxBuffer st

2283+ , pushBufferSpace = remBuf

2284+ , pushBufferPolicy = PushBufferBlock

2285+ , pushBufferMVar = pbMVar

2189 , maxWorkerLimit = Unlimited

2286 , maxWorkerLimit = Unlimited

2190 -- Used only for diagnostics

2287 -- Used only for diagnostics

2191 , yieldRateInfo = rateInfo

2288 , yieldRateInfo = rateInfo

@@ -2222,7 +2319,13 @@ getParallelSVar ss st mrun = do

2222 case yieldRateInfo sv of

2319 case yieldRateInfo sv of

2223 Nothing -> return ()

2320 Nothing -> return ()

2224 Just yinfo -> void $ collectLatency sv yinfo False

2321 Just yinfo -> void $ collectLatency sv yinfo False

2225− fst `fmap` readOutputQRaw sv

2322+ r <- fst `fmap` readOutputQRaw sv

2323+ liftIO $ do

2324+ void $ tryTakeMVar (pushBufferMVar sv)

2325+ resetBufferLimit sv

2326+ writeBarrier

2327+ void $ tryPutMVar (pushBufferMVar sv) ()

2328+ return r

2226

2329

2227sendFirstWorker :: MonadAsync m => SVar t m a -> t m a -> m (SVar t m a)

2330sendFirstWorker :: MonadAsync m => SVar t m a -> t m a -> m (SVar t m a)

2228sendFirstWorker sv m = do

2331sendFirstWorker sv m = do

src/Streamly/Streams/Async.hs+7−1

@@ -294,6 +294,9 @@ getLifoSVar st mrun = do

294 { outputQueue = outQ

295 , remainingWork = yl

296 , maxBufferLimit = getMaxBuffer st

297+ , pushBufferSpace = undefined

298+ , pushBufferPolicy = undefined

299+ , pushBufferMVar = undefined

297 , maxWorkerLimit = min (getMaxThreads st) (getMaxBuffer st)

300 , maxWorkerLimit = min (getMaxThreads st) (getMaxBuffer st)

298 , yieldRateInfo = rateInfo

301 , yieldRateInfo = rateInfo

299 , outputDoorBell = outQMv

302 , outputDoorBell = outQMv

@@ -384,8 +387,11 @@ getFifoSVar st mrun = do

384 -> SVar Stream m a

387 -> SVar Stream m a

385 getSVar sv readOutput postProc workDone wloop = SVar

388 getSVar sv readOutput postProc workDone wloop = SVar

386 { outputQueue = outQ

389 { outputQueue = outQ

387− , remainingWork = yl

390+ , remainingWork = yl

388 , maxBufferLimit = getMaxBuffer st

391 , maxBufferLimit = getMaxBuffer st

392+ , pushBufferSpace = undefined

393+ , pushBufferPolicy = undefined

394+ , pushBufferMVar = undefined

389 , maxWorkerLimit = min (getMaxThreads st) (getMaxBuffer st)

395 , maxWorkerLimit = min (getMaxThreads st) (getMaxBuffer st)

390 , yieldRateInfo = rateInfo

396 , yieldRateInfo = rateInfo

391 , outputDoorBell = outQMv

397 , outputDoorBell = outQMv

src/Streamly/Streams/Parallel.hs+14−17

@@ -68,29 +68,26 @@ import qualified Streamly.Streams.StreamK as K

68runOne

69 :: MonadIO m

70 => State Stream m a -> Stream m a -> Maybe WorkerInfo -> m ()

71+runOne st m winfo = do

72+ yieldLimitOk <- liftIO $ decrementYieldLimit sv

73+ if yieldLimitOk

74+ then do

75+ liftIO $ decrementBufferLimit sv

71−runOne st m winfo = foldStreamShared st yieldk single stop m

76+ foldStreamShared st yieldk single stop m

77+ else liftIO $ cleanupSVarFromWorker sv

73 where

79 where

75 sv = fromJust $ streamVar st

81 sv = fromJust $ streamVar st

76 mrun = runInIO $ svarMrun sv

82 mrun = runInIO $ svarMrun sv

78− withLimitCheck action = do

84+ stop = liftIO $ do

79− yieldLimitOk <- liftIO $ decrementYieldLimitPost sv

85+ incrementBufferLimit sv

80− if yieldLimitOk

86+ incrementYieldLimit sv

81− then action

82− else liftIO $ cleanupSVarFromWorker sv

83−

84− stop = liftIO $ sendStop sv winfo

87+ sendStop sv winfo

85− sendit a = liftIO $ sendYield sv winfo (ChildYield a)

88+ sendit a = liftIO $ void $ send sv (ChildYield a)

86− single a = sendit a >> withLimitCheck stop

89+ single a = sendit a >> (liftIO $ sendStop sv winfo)

87−

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

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

90− -- overflows. Parallel is dangerous because it can accumulate unbounded

91− -- output in the buffer.

92− yieldk a r = void (sendit a)

93− >> withLimitCheck (void $ liftIO $ mrun $ runOne st r winfo)

90+ yieldk a r = sendit a >> (void $ liftIO $ mrun $ runOne st r winfo)

95{-# NOINLINE forkSVarPar #-}

92{-# NOINLINE forkSVarPar #-}

96forkSVarPar :: (IsStream t, MonadAsync m)

93forkSVarPar :: (IsStream t, MonadAsync m)

test/Prop.hs+36−7

@@ -951,20 +951,38 @@ main = hspec

951 _ -> S.foldMapWith (<>) return xs

952 )

953

954− let makeOps :: IsStream t => (t m a -> c) -> [(String, t m a -> c)]

954+ let makeCommonOps :: IsStream t => (t m a -> c) -> [(String, t m a -> c)]

955− makeOps t =

955+ makeCommonOps t =

956 [ ("default", t)

957#ifndef COVERAGE_BUILD

958 , ("rate AvgRate 10000", t . avgRate 10000)

959 , ("rate Nothing", t . rate Nothing)

960 , ("maxBuffer 0", t . maxBuffer 0)

961− , ("maxBuffer 1", t . maxBuffer 1)

962 , ("maxThreads 0", t . maxThreads 0)

961 , ("maxThreads 0", t . maxThreads 0)

963 , ("maxThreads 1", t . maxThreads 1)

962 , ("maxThreads 1", t . maxThreads 1)

964 , ("maxThreads -1", t . maxThreads (-1))

963 , ("maxThreads -1", t . maxThreads (-1))

965#endif

964#endif

966 ]

965 ]

967

966

967+ let makeOps :: IsStream t => (t m a -> c) -> [(String, t m a -> c)]

968+ makeOps t = makeCommonOps t ++

969+ [

970+#ifndef COVERAGE_BUILD

971+ ("maxBuffer 1", t . maxBuffer 1)

972+#endif

973+ ]

974+

975+ -- For concurrent application test we need a buffer of at least size 2 to

976+ -- allow two threads to run.

977+ let makeConcurrentAppOps :: IsStream t

978+ => (t m a -> c) -> [(String, t m a -> c)]

979+ makeConcurrentAppOps t = makeCommonOps t ++

980+ [

981+#ifndef COVERAGE_BUILD

982+ ("maxBuffer 2", t . maxBuffer 2)

983+#endif

984+ ]

985+

968 let mapOps spec = mapM_ (\(desc, f) -> describe desc $ spec f)

986 let mapOps spec = mapM_ (\(desc, f) -> describe desc $ spec f)

969 let serialOps :: IsStream t => ((SerialT IO a -> t IO a) -> Spec) -> Spec

987 let serialOps :: IsStream t => ((SerialT IO a -> t IO a) -> Spec) -> Spec

970 serialOps spec = mapOps spec $ makeOps serially

988 serialOps spec = mapOps spec $ makeOps serially

@@ -993,13 +1011,23 @@ main = hspec

993#ifndef COVERAGE_BUILD

1011#ifndef COVERAGE_BUILD

994 <> [("maxBuffer (-1)", aheadly . maxBuffer (-1))]

1012 <> [("maxBuffer (-1)", aheadly . maxBuffer (-1))]

995#endif

1013#endif

996− let parallelOps :: IsStream t => ((ParallelT IO a -> t IO a) -> Spec) -> Spec

1014+ let parallelCommonOps :: IsStream t => [(String, ParallelT m a -> t m a)]

997− parallelOps spec = mapOps spec $ makeOps parallely

1015+ parallelCommonOps = []

998#ifndef COVERAGE_BUILD

1016#ifndef COVERAGE_BUILD

999 <> [("rate AvgRate 0.00000001", parallely . avgRate 0.00000001)]

1017 <> [("rate AvgRate 0.00000001", parallely . avgRate 0.00000001)]

1000 <> [("maxBuffer (-1)", parallely . maxBuffer (-1))]

1018 <> [("maxBuffer (-1)", parallely . maxBuffer (-1))]

1001#endif

1019#endif

1020+ let parallelOps :: IsStream t

1021+ => ((ParallelT IO a -> t IO a) -> Spec) -> Spec

1022+ parallelOps spec = mapOps spec $ makeOps parallely <> parallelCommonOps

1023+

1024+ let parallelConcurrentAppOps :: IsStream t

1025+ => ((ParallelT IO a -> t IO a) -> Spec) -> Spec

1026+ parallelConcurrentAppOps spec =

1027+ mapOps spec $ makeConcurrentAppOps parallely <> parallelCommonOps

1028+

1029+ let zipSerialOps :: IsStream t

1002− let zipSerialOps :: IsStream t => ((ZipSerialM IO a -> t IO a) -> Spec) -> Spec

1030+ => ((ZipSerialM IO a -> t IO a) -> Spec) -> Spec

1003 zipSerialOps spec = mapOps spec $ makeOps zipSerially

1031 zipSerialOps spec = mapOps spec $ makeOps zipSerially

1004#ifndef COVERAGE_BUILD

1032#ifndef COVERAGE_BUILD

1005 <> [("rate AvgRate 0.00000001", zipSerially . avgRate 0.00000001)]

1033 <> [("rate AvgRate 0.00000001", zipSerially . avgRate 0.00000001)]

@@ -1174,7 +1202,8 @@ main = hspec

1174 serialOps $ prop "serial" . concurrentApplication (==)

1202 serialOps $ prop "serial" . concurrentApplication (==)

1175 asyncOps $ prop "async" . concurrentApplication sortEq

1203 asyncOps $ prop "async" . concurrentApplication sortEq

1176 aheadOps $ prop "ahead" . concurrentApplication (==)

1204 aheadOps $ prop "ahead" . concurrentApplication (==)

1205+ parallelConcurrentAppOps $

1177− parallelOps $ prop "parallel" . concurrentApplication sortEq

1206+ prop "parallel" . concurrentApplication sortEq

1178

1207

1179 prop "concurrent foldr application" $ withMaxSuccess maxTestCount

1208 prop "concurrent foldr application" $ withMaxSuccess maxTestCount

1180 concurrentFoldrApplication

1209 concurrentFoldrApplication