git.iko.soy/streamly
on master help sign in
Use explicit arguments in the concurrency config options
41fb073Adithya Kumar
benchmark/Streamly/Benchmark/Data/Stream/ConcurrentEager.hs+4−1
compare with master
unifiedsplit
@@ -19,4 +19,7 @@ moduleName = "Data.Stream.ConcurrentEager"
19-------------------------------------------------------------------------------
19-------------------------------------------------------------------------------
20
20
21main :: IO ()
21main :: IO ()
22+main =
23+ runWithCLIOpts
24+ defaultStreamSize
22−main = runWithCLIOpts defaultStreamSize (allBenchmarks moduleName Async.eager)
25+ (allBenchmarks moduleName (Async.eager True))
benchmark/Streamly/Benchmark/Data/Stream/ConcurrentInterleaved.hs+2−1
@@ -21,4 +21,5 @@ moduleName = "Data.Stream.ConcurrentInterleaved"
21main :: IO ()
21main :: IO ()
22main =
22main =
23 runWithCLIOpts
23 runWithCLIOpts
24+ defaultStreamSize
24− defaultStreamSize (allBenchmarks moduleName Async.interleaved)
25+ (allBenchmarks moduleName (Async.interleaved True))
benchmark/Streamly/Benchmark/Data/Stream/ConcurrentOrdered.hs+3−1
@@ -20,4 +20,6 @@ moduleName = "Data.Stream.ConcurrentOrdered"
20
20
21main :: IO ()
21main :: IO ()
22main =
22main =
23+ runWithCLIOpts
24+ defaultStreamSize
23− runWithCLIOpts defaultStreamSize (allBenchmarks moduleName Async.ordered)
25+ (allBenchmarks moduleName (Async.ordered True))
src/Streamly/Internal/Data/Stream/Channel/Types.hs+8−8
@@ -489,8 +489,8 @@ getStreamLatency = _streamLatency
489
489
490-- | Print debug information about the 'Channel' when the stream ends.
490-- | Print debug information about the 'Channel' when the stream ends.
491--
491--
492−inspect :: Config -> Config
492+inspect :: Bool -> Config -> Config
493−inspect st = st { _inspect = True }
493+inspect flag st = st { _inspect = flag }
494
494
495getInspectMode :: Config -> Bool
495getInspectMode :: Config -> Bool
496getInspectMode = _inspect
496getInspectMode = _inspect
@@ -505,8 +505,8 @@ getInspectMode = _inspect
505-- /Note:/ This option has no effect when rate has been specified.
505-- /Note:/ This option has no effect when rate has been specified.
506-- /Note:/ Not supported with 'interleaved'.
506-- /Note:/ Not supported with 'interleaved'.
507--
507--
508−eager :: Config -> Config
508+eager :: Bool -> Config -> Config
509−eager st = st { _eagerDispatch = True }
509+eager flag st = st { _eagerDispatch = flag }
510
510
511getEagerDispatch :: Config -> Bool
511getEagerDispatch :: Config -> Bool
512getEagerDispatch = _eagerDispatch
512getEagerDispatch = _eagerDispatch
@@ -523,8 +523,8 @@ getStopWhen = _stopWhen
523--
523--
524-- /Note:/ Not supported with 'interleaved'.
524-- /Note:/ Not supported with 'interleaved'.
525--
525--
526−ordered :: Config -> Config
526+ordered :: Bool -> Config -> Config
527−ordered st = st { _ordered = True }
527+ordered flag st = st { _ordered = flag }
528
528
529getOrdered :: Config -> Bool
529getOrdered :: Config -> Bool
530getOrdered = _ordered
530getOrdered = _ordered
@@ -536,8 +536,8 @@ getOrdered = _ordered
536-- /Note:/ Can only be used on finite number of streams.
536-- /Note:/ Can only be used on finite number of streams.
537-- /Note:/ Not supported with 'ordered'.
537-- /Note:/ Not supported with 'ordered'.
538--
538--
539−interleaved :: Config -> Config
539+interleaved :: Bool -> Config -> Config
540−interleaved st = st { _interleaved = True }
540+interleaved flag st = st { _interleaved = flag }
541
541
542getInterleaved :: Config -> Bool
542getInterleaved :: Config -> Bool
543getInterleaved = _interleaved
543getInterleaved = _interleaved
src/Streamly/Internal/Data/Stream/Concurrent.hs+24−24
@@ -289,46 +289,46 @@ append2 = combineWith id
289-- the left stream. This schedules all streams in a round robin fashion over
289-- the left stream. This schedules all streams in a round robin fashion over
290-- limited number of threads.
290-- limited number of threads.
291--
291--
292−-- >>> interleave2 = Async.combineWith Async.interleaved
292+-- >>> interleave2 = Async.combineWith (Async.interleaved True)
293--
293--
294{-# INLINE interleave2 #-}
294{-# INLINE interleave2 #-}
295interleave2 :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
295interleave2 :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
296−interleave2 = combineWith interleaved
296+interleave2 = combineWith (interleaved True)
297
297
298-- | Like 'append' but with 'ordered' on.
298-- | Like 'append' but with 'ordered' on.
299--
299--
300−-- >>> ahead2 = Async.combineWith Async.ordered
300+-- >>> ahead2 = Async.combineWith (Async.ordered True)
301--
301--
302{-# INLINE ahead2 #-}
302{-# INLINE ahead2 #-}
303ahead2 :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
303ahead2 :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
304−ahead2 = combineWith ordered
304+ahead2 = combineWith (ordered True)
305
305
306-- | Like 'append2' but with 'eager' on.
306-- | Like 'append2' but with 'eager' on.
307--
307--
308−-- >>> parallel2 = Async.combineWith Async.eager
308+-- >>> parallel2 = Async.combineWith (Async.eager True)
309--
309--
310{-# INLINE parallel2 #-}
310{-# INLINE parallel2 #-}
311parallel2 :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
311parallel2 :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
312−parallel2 = combineWith eager
312+parallel2 = combineWith (eager True)
313
313
314-- | Like 'parallel2' but stops the output as soon as the first stream stops.
314-- | Like 'parallel2' but stops the output as soon as the first stream stops.
315--
315--
316−-- >>> parallelFst2 = Async.combineWith (Async.eager . Async.stopWhen Async.FirstStops)
316+-- >>> parallelFst2 = Async.combineWith (Async.eager True . Async.stopWhen Async.FirstStops)
317--
317--
318-- /Pre-release/
318-- /Pre-release/
319{-# INLINE parallelFst2 #-}
319{-# INLINE parallelFst2 #-}
320parallelFst2 :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
320parallelFst2 :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
321−parallelFst2 = combineWith (eager . stopWhen FirstStops)
321+parallelFst2 = combineWith (eager True . stopWhen FirstStops)
322
322
323-- | Like 'parallel2' but stops the output as soon as any of the two streams
323-- | Like 'parallel2' but stops the output as soon as any of the two streams
324-- stops.
324-- stops.
325--
325--
326−-- >>> parallelMin2 = Async.combineWith (Async.eager . Async.stopWhen Async.AnyStops)
326+-- >>> parallelMin2 = Async.combineWith (Async.eager True . Async.stopWhen Async.AnyStops)
327--
327--
328-- /Pre-release/
328-- /Pre-release/
329{-# INLINE parallelMin2 #-}
329{-# INLINE parallelMin2 #-}
330parallelMin2 :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
330parallelMin2 :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
331−parallelMin2 = combineWith (eager . stopWhen AnyStops)
331+parallelMin2 = combineWith (eager True . stopWhen AnyStops)
332
332
333-------------------------------------------------------------------------------
333-------------------------------------------------------------------------------
334-- concat streams
334-- concat streams
@@ -482,11 +482,11 @@ concatMap = concatMapWith id
482-- evaluate the earlier stream first, this schedules all streams in a round
482-- evaluate the earlier stream first, this schedules all streams in a round
483-- robin fashion over the available threads.
483-- robin fashion over the available threads.
484--
484--
485−-- >>> concatMapInterleave = Async.concatMapWith Async.interleaved
485+-- >>> concatMapInterleave = Async.concatMapWith (Async.interleaved True)
486--
486--
487-- When used with a single thread it behaves like serial interleaving:
487-- When used with a single thread it behaves like serial interleaving:
488--
488--
489−-- >>> f cfg xs = Stream.fold Fold.toList $ Async.concatMapWith (Async.interleaved . cfg) id $ Stream.fromList xs
489+-- >>> f cfg xs = Stream.fold Fold.toList $ Async.concatMapWith (Async.interleaved True . cfg) id $ Stream.fromList xs
490--
490--
491-- >>> stream1 = Stream.fromList [1,2,3]
491-- >>> stream1 = Stream.fromList [1,2,3]
492-- >>> stream2 = Stream.fromList [4,5,6]
492-- >>> stream2 = Stream.fromList [4,5,6]
@@ -497,7 +497,7 @@ concatMap = concatMapWith id
497--
497--
498{-# INLINE concatMapInterleave #-}
498{-# INLINE concatMapInterleave #-}
499concatMapInterleave :: MonadAsync m => (a -> Stream m b) -> Stream m a -> Stream m b
499concatMapInterleave :: MonadAsync m => (a -> Stream m b) -> Stream m a -> Stream m b
500−concatMapInterleave = concatMapWith interleaved
500+concatMapInterleave = concatMapWith (interleaved True)
501
501
502-- | Like 'concat' but we can also specify the concurrent channel's
502-- | Like 'concat' but we can also specify the concurrent channel's
503-- configuration parameters using Config modifiers.
503-- configuration parameters using Config modifiers.
@@ -541,44 +541,44 @@ append = concatListWith id
541-- the left stream. This schedules all streams in a round robin fashion over
541-- the left stream. This schedules all streams in a round robin fashion over
542-- limited number of threads.
542-- limited number of threads.
543--
543--
544−-- >>> interleave = Async.concatListWith Async.interleaved
544+-- >>> interleave = Async.concatListWith (Async.interleaved True)
545--
545--
546{-# INLINE interleave #-}
546{-# INLINE interleave #-}
547interleave :: MonadAsync m => [Stream m a] -> Stream m a
547interleave :: MonadAsync m => [Stream m a] -> Stream m a
548−interleave = concatListWith interleaved
548+interleave = concatListWith (interleaved True)
549
549
550-- | Like 'append' but with 'ordered' on.
550-- | Like 'append' but with 'ordered' on.
551--
551--
552−-- >>> ahead = Async.concatListWith Async.ordered
552+-- >>> ahead = Async.concatListWith (Async.ordered True)
553--
553--
554{-# INLINE ahead #-}
554{-# INLINE ahead #-}
555ahead :: MonadAsync m => [Stream m a] -> Stream m a
555ahead :: MonadAsync m => [Stream m a] -> Stream m a
556−ahead = concatListWith ordered
556+ahead = concatListWith (ordered True)
557
557
558-- | Like 'append' but with 'eager' on.
558-- | Like 'append' but with 'eager' on.
559--
559--
560−-- >>> parallel = Async.concatListWith Async.eager
560+-- >>> parallel = Async.concatListWith (Async.eager True)
561--
561--
562{-# INLINE parallel #-}
562{-# INLINE parallel #-}
563parallel :: MonadAsync m => [Stream m a] -> Stream m a
563parallel :: MonadAsync m => [Stream m a] -> Stream m a
564−parallel = concatListWith eager
564+parallel = concatListWith (eager True)
565
565
566-- | Like 'parallel' but stops the output as soon as the first stream stops.
566-- | Like 'parallel' but stops the output as soon as the first stream stops.
567--
567--
568−-- >>> parallelFst = Async.concatListWith (Async.eager . Async.stopWhen Async.FirstStops)
568+-- >>> parallelFst = Async.concatListWith (Async.eager True . Async.stopWhen Async.FirstStops)
569--
569--
570{-# INLINE parallelFst #-}
570{-# INLINE parallelFst #-}
571parallelFst :: MonadAsync m => [Stream m a] -> Stream m a
571parallelFst :: MonadAsync m => [Stream m a] -> Stream m a
572−parallelFst = concatListWith (eager . stopWhen FirstStops)
572+parallelFst = concatListWith (eager True . stopWhen FirstStops)
573
573
574-- | Like 'parallel' but stops the output as soon as any of the two streams
574-- | Like 'parallel' but stops the output as soon as any of the two streams
575-- stops.
575-- stops.
576--
576--
577−-- >>> parallelMin = Async.concatListWith (Async.eager . Async.stopWhen Async.AnyStops)
577+-- >>> parallelMin = Async.concatListWith (Async.eager True . Async.stopWhen Async.AnyStops)
578--
578--
579{-# INLINE parallelMin #-}
579{-# INLINE parallelMin #-}
580parallelMin :: MonadAsync m => [Stream m a] -> Stream m a
580parallelMin :: MonadAsync m => [Stream m a] -> Stream m a
581−parallelMin = concatListWith (eager . stopWhen AnyStops)
581+parallelMin = concatListWith (eager True . stopWhen AnyStops)
582
582
583-------------------------------------------------------------------------------
583-------------------------------------------------------------------------------
584-- Applicative
584-- Applicative
@@ -671,7 +671,7 @@ zipWith f = zipWithM (\a b -> return $ f a b)
671{-# INLINE_NORMAL newCallbackStream #-}
671{-# INLINE_NORMAL newCallbackStream #-}
672newCallbackStream :: MonadAsync m => m (a -> m (), Stream m a)
672newCallbackStream :: MonadAsync m => m (a -> m (), Stream m a)
673newCallbackStream = do
673newCallbackStream = do
674− chan <- newChannel eager
674+ chan <- newChannel (eager True)
675
675
676 -- XXX Add our own thread-id to the SVar as we can not know the callback's
676 -- XXX Add our own thread-id to the SVar as we can not know the callback's
677 -- thread-id and the callback is not run in a managed worker. We need to
677 -- thread-id and the callback is not run in a managed worker. We need to
src/Streamly/Internal/FileSystem/Event/Windows.hs+1−1
@@ -471,7 +471,7 @@ closePathHandleStream =
471--
471--
472watchWith :: (Config -> Config) -> NonEmpty (Array Word8) -> Stream IO Event
472watchWith :: (Config -> Config) -> NonEmpty (Array Word8) -> Stream IO Event
473watchWith f paths =
473watchWith f paths =
474− S.bracketIO before after (Concur.concatMapWith eager eventStreamAggr)
474+ S.bracketIO before after (Concur.concatMapWith (eager True) eventStreamAggr)
475
475
476 where
476 where
477
477
test/Streamly/Test/Data/Stream/Concurrent.hs+2−2
@@ -83,9 +83,9 @@ commonOpConfigs =
83 , ("maxBuffer 0", Async.maxBuffer 0)
83 , ("maxBuffer 0", Async.maxBuffer 0)
84 , ("maxThreads 0", Async.maxThreads 0)
84 , ("maxThreads 0", Async.maxThreads 0)
85 , ("maxThreads 1", Async.maxThreads 1)
85 , ("maxThreads 1", Async.maxThreads 1)
86− , ("eager", Async.eager)
86+ , ("eager", Async.eager True)
87 -- XXX Need to use an unsorted eq operation for ahead
87 -- XXX Need to use an unsorted eq operation for ahead
88− , ("ordered", Async.ordered)
88+ , ("ordered", Async.ordered True)
89#ifdef USE_LARGE_MEMORY
89#ifdef USE_LARGE_MEMORY
90 , ("maxThreads -1", Async.maxThreads (-1))
90 , ("maxThreads -1", Async.maxThreads (-1))
91#endif
91#endif