72766cb · streamly · git.iko.soy

src/Streamly.hs+15−3

compare with master

unifiedsplit

@@ -101,8 +101,9 @@ module Streamly

101 , parallel

102

103 -- * Concurrency Control

104− , threads

104+ -- $concurrency

105− , buffer

105+ , maxThreads

106+ , maxBuffer

106

107

107 -- * Folding Containers of Streams

108 -- * Folding Containers of Streams

108 -- $foldutils

109 -- $foldutils

@@ -163,7 +164,7 @@ import Streamly.Streams.Ahead

163import Streamly.Streams.Parallel

164import Streamly.Streams.Parallel

164import Streamly.Streams.Zip

165import Streamly.Streams.Zip

165import Streamly.Streams.Prelude

166import Streamly.Streams.Prelude

166−import Streamly.Streams.SVar (threads, buffer)

167+import Streamly.Streams.SVar (maxThreads, maxBuffer)

167import Streamly.SVar (MonadAsync)

168import Streamly.SVar (MonadAsync)

168import Data.Semigroup (Semigroup(..))

169import Data.Semigroup (Semigroup(..))

169

170

@@ -214,6 +215,17 @@ import Data.Semigroup (Semigroup(..))

214-- which can be used to combine two streams in a predetermined way irrespective

215-- which can be used to combine two streams in a predetermined way irrespective

215-- of the type.

216-- of the type.

216

217

218+-- $concurrency

219+--

220+-- These combinators can be used at any point in the composition to control

221+-- concurrency, and they affect only the enclosed stream. When nested

222+-- combinators are used, the nearest enclosing combinator overrides the outer

223+-- ones. These combinators have no effect on 'Parallel' streams, concurrency

224+-- for 'Parallel' streams is always unbounded.

225+-- Note that the use of these combinators does not enable concurrency, to

226+-- enable concurrency you have to use one of the concurrent stream type

227+-- combinators.

228+

217-- $adapters

229-- $adapters

218--

230--

219-- You may want to use different stream composition styles at different points

231-- You may want to use different stream composition styles at different points

src/Streamly/SVar.hs+5−5

@@ -188,8 +188,8 @@ data SVar t m a =

188

189data State t m a = State

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

191− , maxThreads :: Int

191+ , threadsHigh :: Int

192− , maxBuffer :: Int

192+ , bufferHigh :: Int

193 }

194

195defaultMaxThreads, defaultMaxBuffer :: Int

@@ -199,8 +199,8 @@ defaultMaxBuffer = 1500

199defState :: State t m a

200defState = State

201 { streamVar = Nothing

202− , maxThreads = defaultMaxThreads

202+ , threadsHigh = defaultMaxThreads

203− , maxBuffer = defaultMaxBuffer

203+ , bufferHigh = defaultMaxBuffer

204 }

205

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

@@ -827,7 +827,7 @@ getAheadSVar st f = do

827 let sv =

828 SVar { outputQueue = outQ

829 , outputDoorBell = outQMv

830− , readOutputQ = readOutputQBounded (maxThreads st) sv

830+ , readOutputQ = readOutputQBounded (threadsHigh st) sv

831 , postProcess = postProcessBounded sv

832 , workerThreads = running

833 -- , workLoop = workLoopAhead sv q outH

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

@@ -124,7 +124,7 @@ workLoopAhead st q heap = runHeap

124 where

125

126 sv = fromJust $ streamVar st

127− maxBuf = maxBuffer st

127+ maxBuf = bufferHigh st

128 toHeap seqNo ent = do

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

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

src/Streamly/Streams/Async.hs+4−4

@@ -70,7 +70,7 @@ runStreamLIFO :: MonadIO m

70runStreamLIFO st q m stop = unStream m st stop single yieldk

71 where

72 sv = fromJust $ streamVar st

73− maxBuf = maxBuffer st

73+ maxBuf = bufferHigh st

74 single a = do

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

76 if res then stop else liftIO $ sendStop sv

@@ -95,7 +95,7 @@ runStreamFIFO

95runStreamFIFO st q m stop = unStream m st stop single yieldk

96 where

97 sv = fromJust $ streamVar st

98− maxBuf = maxBuffer st

98+ maxBuf = bufferHigh st

99 single a = do

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

101 if res then stop else liftIO $ sendStop sv

@@ -132,7 +132,7 @@ getLifoSVar st = do

132 let sv =

133 SVar { outputQueue = outQ

134 , outputDoorBell = outQMv

135− , readOutputQ = readOutputQBounded (maxThreads st) sv

135+ , readOutputQ = readOutputQBounded (threadsHigh st) sv

136 , postProcess = postProcessBounded sv

137 , workerThreads = running

138 , workLoop = workLoopLIFO runStreamLIFO

@@ -173,7 +173,7 @@ getFifoSVar st = do

173 let sv =

174 SVar { outputQueue = outQ

175 , outputDoorBell = outQMv

176− , readOutputQ = readOutputQBounded (maxThreads st) sv

176+ , readOutputQ = readOutputQBounded (threadsHigh st) sv

177 , postProcess = postProcessBounded sv

178 , workerThreads = running

179 , workLoop = workLoopFIFO runStreamFIFO

src/Streamly/Streams/SVar.hs+10−20

@@ -23,8 +23,8 @@ module Streamly.Streams.SVar

23 (

24 fromSVar

25 , toSVar

26− , threads

26+ , maxThreads

27− , buffer

27+ , maxBuffer

28 )

29where

30

@@ -89,32 +89,22 @@ toSVar sv m = toStreamVar sv (toStream m)

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

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

91-- A value of 0 resets the thread limit to default, a negative value means

92−-- there is no limit. Note that this primitive has no effect on 'Parallel'

92+-- there is no limit.

93−-- streams, the number of threads for 'Parallel' streams are always unbounded.

94−--

95−-- This primitive can be used at any point in the composition, and it affects

96−-- only the enclosed stream. When nested primitives are used the nearest

97−-- enclosing primitive overrides the outer ones.

98−-- Note that the use of this primitive does not enable concurrency, to enable

99−-- concurrency you have to use one of the concurrent stream type combinators.

100--

93--

101-- @since 0.4.0

94-- @since 0.4.0

102−threads :: IsStream t => Int -> t m a -> t m a

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

103−threads n m = fromStream $ Stream $ \st stp sng yld -> do

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

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

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

105− unStream (toStream m) (st {maxThreads = n'}) stp sng yld

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

106

99

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

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

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

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

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

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

110-- A value of 0 resets the buffer size to default, a negative value means

103-- A value of 0 resets the buffer size to default, a negative value means

111−-- there is no limit. Note that this primitive has no effect on 'Parallel'

104+-- there is no limit.

112−-- streams, the buffer size for 'Parallel' streams is always unbounded.

113−--

114−-- The same scoping rules apply as for the 'threads' primitive.

115--

105--

116-- @since 0.4.0

106-- @since 0.4.0

117−buffer :: IsStream t => Int -> t m a -> t m a

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

118−buffer n m = fromStream $ Stream $ \st stp sng yld -> do

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

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

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

120− unStream (toStream m) (st {maxBuffer = n'}) stp sng yld

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

test/Prop.hs+1−1

@@ -56,7 +56,7 @@ constructWithReplicateM

56constructWithReplicateM op thr buf len =

57 monadicIO $ do

58 let x = return (1 :: Int)

59− stream <- run $ (S.toList . op) (threads thr $ buffer buf $

59+ stream <- run $ (S.toList . op) (maxThreads thr $ maxBuffer buf $

60 S.replicateM (fromIntegral len) x)

61 list <- run $ replicateM (fromIntegral len) x

62 equals (==) stream list