git.iko.soy/streamly
on master help sign in
Merge branch 'luke-clifton-fixmaxthreads'
e8d0393Harendra Kumar
.gitignore+4−1
compare with master
unifiedsplit
@@ -1 +1,4 @@
1+/.gitignore
1−.stack-work
2+/.stack-work/
3+/dist/
4+/dist-newstyle/
Changelog.md+5
@@ -1,5 +1,10 @@
1## Unreleased
1## Unreleased
2
2
3+### Bug Fixes
4+
5+* Fix a bug that caused `maxThreads` directive to be ignored when rate control
6+ was not used.
7+
3### Enhancements
8### Enhancements
4
9
5* Add GHCJS support
10* Add GHCJS support
src/Streamly/SVar.hs+8−5
@@ -1426,10 +1426,13 @@ dispatchWorker yieldCount sv = do
1426 Nothing -> return workerLimit
1426 Nothing -> return workerLimit
1427 Just ref -> do
1427 Just ref -> do
1428 n <- liftIO $ readIORef ref
1428 n <- liftIO $ readIORef ref
1429+ case yieldRateInfo sv of
1429− return $
1430+ Just _ -> return workerLimit
1431+ Nothing ->
1432+ return $
1430− case workerLimit of
1433+ case workerLimit of
1431− Unlimited -> Limited (fromIntegral n)
1434+ Unlimited -> Limited (fromIntegral n)
1432− Limited lim -> Limited $ min lim (fromIntegral n)
1435+ Limited lim -> Limited $ min lim (fromIntegral n)
1433
1436
1434 -- XXX for ahead streams shall we take the heap yields into account
1437 -- XXX for ahead streams shall we take the heap yields into account
1435 -- for controlling the dispatch? We should not dispatch if the heap
1438 -- for controlling the dispatch? We should not dispatch if the heap
@@ -1442,7 +1445,7 @@ dispatchWorker yieldCount sv = do
1442 -- and using them here, so this is just approximate logic and
1445 -- and using them here, so this is just approximate logic and
1443 -- we cannot rely on it for correctness. We may actually
1446 -- we cannot rely on it for correctness. We may actually
1444 -- dispatch more workers than required.
1447 -- dispatch more workers than required.
1445− Limited lim | lim > 0 -> dispatch
1448+ Limited lim | lim > fromIntegral active -> dispatch
1446 _ -> return False
1449 _ -> return False
1447 else do
1450 else do
1448 when (active <= 0) $ pushWorker 0 sv
1451 when (active <= 0) $ pushWorker 0 sv
stack.yaml+1−5
@@ -1,16 +1,12 @@
1−resolver: lts-12.11
1+resolver: lts-13.13
2packages:
2packages:
3- '.'
3- '.'
4extra-deps:
4extra-deps:
5 - SDL-0.6.6.0
5 - SDL-0.6.6.0
6− - gauge-0.2.4
7 - Chart-1.9
6 - Chart-1.9
8 - Chart-diagrams-1.9
7 - Chart-diagrams-1.9
9 - SVGFonts-1.6.0.3
8 - SVGFonts-1.6.0.3
10 - bench-show-0.2.2
9 - bench-show-0.2.2
11− - statistics-0.15.0.0
12− - dense-linear-algebra-0.1.0.0
13− - math-functions-0.3.0.2
14
10
15flags: {}
11flags: {}
16extra-package-dbs: []
12extra-package-dbs: []
streamly.cabal+6−1
@@ -110,6 +110,11 @@ flag dev
110 manual: True
110 manual: True
111 default: False
111 default: False
112
112
113+flag no-charts
114+ description: Disable chart generation
115+ manual: True
116+ default: False
117+
113flag no-fusion
118flag no-fusion
114 description: Disable rewrite rules
119 description: Disable rewrite rules
115 manual: True
120 manual: True
@@ -628,7 +633,7 @@ executable chart
628 default-language: Haskell2010
633 default-language: Haskell2010
629 hs-source-dirs: benchmark
634 hs-source-dirs: benchmark
630 main-is: Chart.hs
635 main-is: Chart.hs
631− if flag(dev) && !impl(ghcjs)
636+ if flag(dev) && !flag(no-charts) && !impl(ghcjs)
632 buildable: True
637 buildable: True
633 build-Depends:
638 build-Depends:
634 base >= 4.8 && < 5
639 base >= 4.8 && < 5
test/MaxRate.hs+100−31
@@ -26,41 +26,66 @@ toMicroSecs x = x * 10^(6 :: Int)
26measureRate' :: IsStream t
26measureRate' :: IsStream t
27 => String
27 => String
28 -> (t IO Int -> SerialT IO Int)
28 -> (t IO Int -> SerialT IO Int)
29− -> Double
29+ -> Int -- buffers
30+ -> Int -- threads
31+ -> Either Double Int -- either rate or count of actions
30 -> Int
32 -> Int
31 -> (Double, Double)
33 -> (Double, Double)
32 -> (Double, Double)
34 -> (Double, Double)
33 -> Spec
35 -> Spec
34−measureRate' desc t rval consumerDelay producerDelay dur =
36+measureRate' desc t buffers threads rval consumerDelay producerDelay expectedRange = do
35− it (desc <> " rate: " <> show rval
37+
38+ let threadAction =
39+ case rval of
40+ Left r -> S.take (round $ 10 * r) . S.repeatM
41+ Right n -> S.replicateM n
42+
43+ rateDesc = case rval of
44+ Left r -> " rate: " <> show r
45+ Right n -> " count: " <> show n
46+
47+ it (desc <> rateDesc
48+ <> " buffers: " <> show buffers
49+ <> " threads: " <> show threads
36 <> ", consumer latency: " <> show consumerDelay
50 <> ", consumer latency: " <> show consumerDelay
37 <> ", producer latency: " <> show producerDelay)
51 <> ", producer latency: " <> show producerDelay)
38− $ durationShouldBe dur $
52+ $ durationShouldBe expectedRange $
39− runStream
53+ runStream
40− $ (if consumerDelay > 0
54+ $ (if consumerDelay > 0
41− then S.mapM $ \x ->
55+ then S.mapM $ \x ->
42− threadDelay (toMicroSecs consumerDelay) >> return x
56+ threadDelay (toMicroSecs consumerDelay) >> return x
43− else id)
57+ else id)
44− $ t
58+ $ t
45− $ maxBuffer (-1)
59+ $ maxBuffer buffers
46− $ maxThreads (-1)
60+ $ maxThreads threads
47− $ avgRate rval
61+ $ (case rval of {Left r -> avgRate r; Right _ -> rate Nothing})
48− $ S.take (round $ rval * 10)
62+ $ threadAction $ do
49− $ S.repeatM $ do
50− let (t1, t2) = producerDelay
63+ let (t1, t2) = producerDelay
51− r <- if t1 == t2
64+ r <- if t1 == t2
52− then return $ round $ toMicroSecs t1
65+ then return $ round $ toMicroSecs t1
53− else randomRIO ( round $ toMicroSecs t1
66+ else randomRIO ( round $ toMicroSecs t1
54− , round $ toMicroSecs t2)
67+ , round $ toMicroSecs t2)
55− when (r > 0) $ -- do
68+ when (r > 0) $ -- do
56− -- t1 <- getTime Monotonic
69+ -- t1 <- getTime Monotonic
57− threadDelay r
70+ threadDelay r
58− -- t2 <- getTime Monotonic
71+ -- t2 <- getTime Monotonic
59− -- let delta = fromIntegral (toNanoSecs (t2 - t1)) / 1000000000
72+ -- let delta = fromIntegral (toNanoSecs (t2 - t1)) / 1000000000
60− -- putStrLn $ "delay took: " <> show delta
73+ -- putStrLn $ "delay took: " <> show delta
61− -- when (delta > 2) $ do
74+ -- when (delta > 2) $ do
62− -- putStrLn $ "delay took high: " <> show delta
75+ -- putStrLn $ "delay took high: " <> show delta
63− return 1
76+ return 1
77+
78+measureRateVariable :: IsStream t
79+ => String
80+ -> (t IO Int -> SerialT IO Int)
81+ -> Double
82+ -> Int
83+ -> (Double, Double)
84+ -> (Double, Double)
85+ -> Spec
86+measureRateVariable desc t rval consumerDelay producerDelay dur =
87+ measureRate' desc t (-1) (-1) (Left rval)
88+ consumerDelay producerDelay dur
64
89
65measureRate :: IsStream t
90measureRate :: IsStream t
66 => String
91 => String
@@ -72,10 +97,54 @@ measureRate :: IsStream t
72 -> Spec
97 -> Spec
73measureRate desc t rval consumerDelay producerDelay dur =
98measureRate desc t rval consumerDelay producerDelay dur =
74 let d = fromIntegral producerDelay
99 let d = fromIntegral producerDelay
75− in measureRate' desc t rval consumerDelay (d, d) dur
100+ in measureRateVariable desc t rval consumerDelay (d, d) dur
101+
102+measureThreads :: IsStream t
103+ => String
104+ -> (t IO Int -> SerialT IO Int)
105+ -> Int -- threads
106+ -> Int -- count of actions
107+ -> Spec
108+measureThreads desc t threads count = do
109+ let expectedTime =
110+ if threads < 0
111+ then 1.0
112+ else fromIntegral count / fromIntegral threads
113+ duration = (expectedTime * 0.9, expectedTime * 1.1)
114+ measureRate' desc t (-1) threads (Right count) 0 (1,1) duration
115+
116+measureBuffers :: IsStream t
117+ => String
118+ -> (t IO Int -> SerialT IO Int)
119+ -> Int -- buffers
120+ -> Int -- count of actions
121+ -> Spec
122+measureBuffers desc t buffers count = do
123+ let expectedTime =
124+ if buffers < 0
125+ then 1.0
126+ else fromIntegral count / fromIntegral buffers
127+ duration = (expectedTime * 0.9, expectedTime * 1.1)
128+ measureRate' desc t buffers (-1) (Right count) 0 (1,1) duration
76
129
77main :: IO ()
130main :: IO ()
78main = hspec $ do
131main = hspec $ do
132+
133+ describe "maxBuffers" $ do
134+ measureBuffers "asyncly" asyncly (-1) 5
135+ -- XXX this test fails due to a known issue
136+ -- measureBuffers "maxBuffers" asyncly 1 5
137+ measureBuffers "asyncly" asyncly 5 5
138+
139+ describe "maxThreads" $ do
140+ measureThreads "asyncly" asyncly (-1) 5
141+ measureThreads "asyncly" asyncly 1 5
142+ measureThreads "asyncly" asyncly 5 5
143+
144+ measureThreads "aheadly" aheadly (-1) 5
145+ measureThreads "aheadly" aheadly 1 5
146+ measureThreads "aheadly" aheadly 5 5
147+
79 let range = (8,12)
148 let range = (8,12)
80
149
81 -- Note that because after the last yield we don't wait, the last period
150 -- Note that because after the last yield we don't wait, the last period
@@ -108,7 +177,7 @@ main = hspec $ do
108 ]
177 ]
109 in describe "asyncly no consumer delay and variable producer delay" $
178 in describe "asyncly no consumer delay and variable producer delay" $
110 forM_ rates $ \r ->
179 forM_ rates $ \r ->
111− measureRate' "asyncly" asyncly r 0 (0.1, 3) range
180+ measureRateVariable "asyncly" asyncly r 0 (0.1, 3) range
112
181
113 let rates = [1, 10, 100, 1000, 10000
182 let rates = [1, 10, 100, 1000, 10000
114#ifndef __GHCJS__
183#ifndef __GHCJS__