git.iko.soy/streamly
on master help sign in
Fix space leak in async combinator

We run the stream in a worker thread using captured monadic state from the parent thread. The call to mrun caused memory to stack up becuase it is called again and again and never returns. We need this call only once when a stream is picked for execution from the work queue. We do not need to call mrun on every yield. The call on yield is removed by this commit.

commit947d6cc1a875784cc172e32972de7777b5d3c11c authorHarendra Kumar <harendra.kumar@gmail.com> dateWed 25 Dec 2019, 01:48 (7 years ago) parentfb622a6
2 files changed+11−7
compare with master
unifiedsplit
Changelog.md+7−3 src/Streamly/Internal/Data/Stream/Async.hs+4−4
Changelog.md+7−3
@@ -3,9 +3,13 @@
33### Bug Fixes
44
55* Fix a bug that caused `findIndices` to return wrong indices in some
6−pipelines
7−
8−* Fix a bug in `tap` that sometimes increased memory consumption.
6+ cases.
7+* Fix a bug in `tap` that caused memory consumption to increase in some cases.
8+* Fix a space leak in `async` streams that caused memory consumption
9+ to increase with the number of elements in the stream, especially when
10+ built with `-threaded` and used with `-N` RTS option. The issue occurs
11+ only in cases when a worker thread happens to be used continuously for a
12+ long time.
913
1014## 0.7.0
1115
src/Streamly/Internal/Data/Stream/Async.hs+4−4
@@ -95,7 +95,7 @@ workLoopLIFO q st sv winfo = run
9595 yieldk a r = do
9696 res <- liftIO $ sendYield sv winfo (ChildYield a)
9797 if res
98− then foldStreamSVar sv st yieldk single run r
98+ then foldStreamShared st yieldk single run r
9999 else liftIO $ do
100100 enqueueLIFO sv q r
101101 sendStop sv winfo
@@ -155,7 +155,7 @@ workLoopLIFOLimited q st sv winfo = run
155155 yieldLimitOk <- liftIO $ decrementYieldLimit sv
156156 let stop = liftIO (incrementYieldLimit sv) >> run
157157 if res && yieldLimitOk
158− then foldStreamSVar sv st yieldk single stop r
158+ then foldStreamShared st yieldk single stop r
159159 else liftIO $ do
160160 incrementYieldLimit sv
161161 enqueueLIFO sv q r
@@ -196,7 +196,7 @@ workLoopFIFO q st sv winfo = run
196196 yieldk a r = do
197197 res <- liftIO $ sendYield sv winfo (ChildYield a)
198198 if res
199− then foldStreamSVar sv st yieldk single run r
199+ then foldStreamShared st yieldk single run r
200200 else liftIO $ do
201201 enqueueFIFO sv q r
202202 sendStop sv winfo
@@ -237,7 +237,7 @@ workLoopFIFOLimited q st sv winfo = run
237237 yieldLimitOk <- liftIO $ decrementYieldLimit sv
238238 let stop = liftIO (incrementYieldLimit sv) >> run
239239 if res && yieldLimitOk
240− then foldStreamSVar sv st yieldk single stop r
240+ then foldStreamShared st yieldk single stop r
241241 else liftIO $ do
242242 incrementYieldLimit sv
243243 enqueueFIFO sv q r