9d08a58 · streamly · git.iko.soy

git.iko.soy/streamly

Disringuish the heap clearing and waiting states

commit9d08a586b030e3205704408ef9396219a15ddea9 authorHarendra Kumar <harendra.kumar@gmail.com> dateMon 10 Sep 2018, 18:50 (8 years ago) parent3020130

2 files changed+99−53

compare with master

unifiedsplit

src/Streamly/SVar.hs+58−34 src/Streamly/Streams/Ahead.hs+41−19

src/Streamly/SVar.hs+58−34

@@ -70,6 +70,8 @@ module Streamly.SVar

7070

7171 , queueEmptyAhead

7272 , dequeueAhead

73+

74+ , HeapDequeueResult(..)

7375 , dequeueFromHeap

7476 , dequeueFromHeapSeq

7577 , requeueOnHeapTop

@@ -366,7 +368,8 @@ data SVar t m a = SVar

366368 , svarRef :: Maybe (IORef ())

367369#ifdef DIAGNOSTICS

368370 , svarCreator :: ThreadId

369− , outputHeap :: IORef (Heap (Entry Int (AheadHeapEntry t m a)) , Int)

371+ , outputHeap :: IORef ( Heap (Entry Int (AheadHeapEntry t m a))

372+ , Maybe Int)

370373 -- Shared work queue (stream, seqNo)

371374 , aheadWorkQueue :: IORef ([t m a], Int)

372375#endif

@@ -1137,17 +1140,9 @@ dequeueAhead q = liftIO $ do

11371140 (x : [], n) -> (([], n), Just (x, n))

11381141 _ -> error "more than one item on queue"

11391142

1140−{-# INLINE dequeueFromHeap #-}

1141−dequeueFromHeap

1142− :: IORef (Heap (Entry Int (AheadHeapEntry t m a)), Int)

1143− -> IO (Maybe (Entry Int (AheadHeapEntry t m a)))

1144−dequeueFromHeap hpVar =

1145− atomicModifyIORef hpVar $ \pair@(hp, snum) -> do

1146− let r = H.uncons hp

1147− case r of

1148− Just (ent@(Entry seqNo _ev), hp') | seqNo == snum ->

1149− ((hp', snum), Just ent)

1150− _ -> (pair, Nothing)

1143+-------------------------------------------------------------------------------

1144+-- Heap manipulation

1145+-------------------------------------------------------------------------------

11511146

11521147withIORef :: IORef a -> (a -> IO b) -> IO b

11531148withIORef ref f = readIORef ref >>= f

@@ -1156,46 +1151,72 @@ atomicModifyIORef_ :: IORef a -> (a -> a) -> IO ()

11561151atomicModifyIORef_ ref f =

11571152 atomicModifyIORef ref $ \x -> (f x, ())

11581153

1154+data HeapDequeueResult t m a =

1155+ Clearing

1156+ | Waiting Int

1157+ | Ready (Entry Int (AheadHeapEntry t m a))

1158+

1159+{-# INLINE dequeueFromHeap #-}

1160+dequeueFromHeap

1161+ :: IORef (Heap (Entry Int (AheadHeapEntry t m a)), Maybe Int)

1162+ -> IO (HeapDequeueResult t m a)

1163+dequeueFromHeap hpVar =

1164+ atomicModifyIORef hpVar $ \pair@(hp, snum) ->

1165+ case snum of

1166+ Nothing -> (pair, Clearing)

1167+ Just n -> do

1168+ let r = H.uncons hp

1169+ case r of

1170+ Just (ent@(Entry seqNo _ev), hp') | seqNo == n ->

1171+ ((hp', Nothing), Ready ent)

1172+ _ -> (pair, Waiting n)

1173+

11591174{-# INLINE canDequeueFromHeap #-}

11601175canDequeueFromHeap

1161− :: IORef (Heap (Entry Int (AheadHeapEntry t m a)), Int)

1176+ :: IORef (Heap (Entry Int (AheadHeapEntry t m a)), Maybe Int)

11621177 -> IO Bool

11631178canDequeueFromHeap hpVar =

1164− withIORef hpVar $ \(hp, snum) -> do

1165− let r = H.uncons hp

1166− case r of

1167− Just ((Entry seqNo _ev), _) -> return $ seqNo == snum

1168− _ -> return False

1179+ withIORef hpVar $ \(hp, snum) ->

1180+ case snum of

1181+ Nothing -> return True

1182+ Just n -> do

1183+ let r = H.uncons hp

1184+ case r of

1185+ Just ((Entry seqNo _ev), _) -> return $ seqNo == n

1186+ _ -> return False

11691187

11701188{-# INLINE dequeueFromHeapSeq #-}

11711189dequeueFromHeapSeq

1172− :: IORef (Heap (Entry Int (AheadHeapEntry t m a)), Int)

1190+ :: IORef (Heap (Entry Int (AheadHeapEntry t m a)), Maybe Int)

11731191 -> Int

1174− -> IO (Maybe (Entry Int (AheadHeapEntry t m a)))

1175−dequeueFromHeapSeq hpVar snum =

1176− atomicModifyIORef hpVar $ \(hp, _) -> do

1177− let r = H.uncons hp

1178− case r of

1179− Just (ent@(Entry seqNo _ev), hp') | seqNo == snum ->

1180− ((hp', snum), Just ent)

1181− _ -> ((hp, snum), Nothing)

1192+ -> IO (HeapDequeueResult t m a)

1193+dequeueFromHeapSeq hpVar i =

1194+ atomicModifyIORef hpVar $ \(hp, snum) ->

1195+ case snum of

1196+ Nothing -> do

1197+ let r = H.uncons hp

1198+ case r of

1199+ Just (ent@(Entry seqNo _ev), hp') | seqNo == i ->

1200+ ((hp', Nothing), Ready ent)

1201+ _ -> ((hp, Just i), Waiting i)

1202+ Just _ -> error "dequeueFromHeapSeq: unreachable"

11821203

11831204{-# INLINE requeueOnHeapTop #-}

11841205requeueOnHeapTop

1185− :: IORef (Heap (Entry Int (AheadHeapEntry t m a)), Int)

1206+ :: IORef (Heap (Entry Int (AheadHeapEntry t m a)), Maybe Int)

11861207 -> Entry Int (AheadHeapEntry t m a)

11871208 -> Int

11881209 -> IO ()

11891210requeueOnHeapTop hpVar ent seqNo =

1190− atomicModifyIORef_ hpVar $ \(hp, _) -> (H.insert ent hp, seqNo)

1211+ atomicModifyIORef_ hpVar $ \(hp, _) -> (H.insert ent hp, Just seqNo)

11911212

11921213{-# INLINE updateHeapSeq #-}

11931214updateHeapSeq

1194− :: IORef (Heap (Entry Int (AheadHeapEntry t m a)), Int)

1215+ :: IORef (Heap (Entry Int (AheadHeapEntry t m a)), Maybe Int)

11951216 -> Int

11961217 -> IO ()

11971218updateHeapSeq hpVar seqNo =

1198− atomicModifyIORef_ hpVar $ \(hp, _) -> (hp, seqNo)

1219+ atomicModifyIORef_ hpVar $ \(hp, _) -> (hp, Just seqNo)

11991220

12001221-------------------------------------------------------------------------------

12011222-- WAhead

@@ -1958,7 +1979,7 @@ getYieldRateInfo st = do

19581979getAheadSVar :: MonadAsync m

19591980 => State t m a

19601981 -> ( IORef ([t m a], Int)

1961− -> IORef (Heap (Entry Int (AheadHeapEntry t m a)), Int)

1982+ -> IORef (Heap (Entry Int (AheadHeapEntry t m a)), Maybe Int)

19621983 -> State t m a

19631984 -> SVar t m a

19641985 -> Maybe WorkerInfo

@@ -1966,7 +1987,10 @@ getAheadSVar :: MonadAsync m

19661987 -> IO (SVar t m a)

19671988getAheadSVar st f = do

19681989 outQ <- newIORef ([], 0)

1969− outH <- newIORef (H.empty, 0)

1990+ -- the second component of the tuple is "Nothing" when heap is being

1991+ -- cleared, "Just n" when we are expecting sequence number n to arrive

1992+ -- before we can start clearing the heap.

1993+ outH <- newIORef (H.empty, Just 0)

19701994 outQMv <- newEmptyMVar

19711995 active <- newIORef 0

19721996 wfw <- newIORef False

@@ -2157,7 +2181,7 @@ newAheadVar :: MonadAsync m

21572181 => State t m a

21582182 -> t m a

21592183 -> ( IORef ([t m a], Int)

2160− -> IORef (Heap (Entry Int (AheadHeapEntry t m a)), Int)

2184+ -> IORef (Heap (Entry Int (AheadHeapEntry t m a)), Maybe Int)

21612185 -> State t m a

21622186 -> SVar t m a

21632187 -> Maybe WorkerInfo

src/Streamly/Streams/Ahead.hs+41−19

@@ -164,7 +164,7 @@ underMaxHeap sv hp = do

164164-- False => continue

165165preStopCheck ::

166166 SVar Stream m a

167− -> IORef (Heap (Entry Int (AheadHeapEntry Stream m a)) , Int)

167+ -> IORef (Heap (Entry Int (AheadHeapEntry Stream m a)) , Maybe Int)

168168 -> IO Bool

169169preStopCheck sv heap = do

170170 -- check the stop condition under a lock before actually

@@ -187,9 +187,29 @@ preStopCheck sv heap = do

187187 if rateOk then continue else stop

188188 else stop

189189

190+-- XXX In absence of a "noyield" primitive (i.e. do not pre-empt inside a

191+-- critical section) from GHC RTS, we have a difficult problem. Assume we have

192+-- a 100,000 threads producing output and queuing it to the heap for

193+-- sequencing. The heap can be drained only by one thread at a time, any thread

194+-- that finds that heap can be drained now, takes a lock and starts draining

195+-- it, however the thread may get prempted in the middle of it holding the

196+-- lock. Since that thread is holding the lock, the other threads cannot pick

197+-- up the draining task, therefore they proceed to picking up the next task to

198+-- execute. If the draining thread could yield voluntarily at a point where it

199+-- has released the lock, then the next threads could pick up the draining

200+-- instead of executing more tasks. When there are 100,000 threads the drainer

201+-- gets a cpu share to run only 1:100000 of the time. This makes the heap

202+-- accumulate a lot of output when we the buffer size is large.

203+--

204+-- The solutions to this problem are:

205+-- 1) make the other threads wait in a queue until the draining finishes

206+-- 2) make the other threads queue and go away if draining is in progress

207+--

208+-- In both cases we give the drainer a chance to run more often.

209+--

190210processHeap :: MonadIO m

191211 => IORef ([Stream m a], Int)

192− -> IORef (Heap (Entry Int (AheadHeapEntry Stream m a)) , Int)

212+ -> IORef (Heap (Entry Int (AheadHeapEntry Stream m a)), Maybe Int)

193213 -> State Stream m a

194214 -> SVar Stream m a

195215 -> Maybe WorkerInfo

@@ -225,10 +245,11 @@ processHeap q heap st sv winfo entry sno stopping = loopHeap sno entry

225245 else runStreamWithYieldLimit True seqNo r

226246

227247 nextHeap prevSeqNo = do

228− ent <- liftIO $ dequeueFromHeapSeq heap (prevSeqNo + 1)

229− case ent of

230− Just (Entry seqNo hent) -> loopHeap seqNo hent

231− Nothing -> do

248+ res <- liftIO $ dequeueFromHeapSeq heap (prevSeqNo + 1)

249+ case res of

250+ Ready (Entry seqNo hent) -> loopHeap seqNo hent

251+ -- when Clearing then stop

252+ _ -> do

232253 if stopping

233254 then do

234255 r <- liftIO $ preStopCheck sv heap

@@ -292,23 +313,23 @@ processHeap q heap st sv winfo entry sno stopping = loopHeap sno entry

292313{-# NOINLINE drainHeap #-}

293314drainHeap :: MonadIO m

294315 => IORef ([Stream m a], Int)

295− -> IORef (Heap (Entry Int (AheadHeapEntry Stream m a)) , Int)

316+ -> IORef (Heap (Entry Int (AheadHeapEntry Stream m a)), Maybe Int)

296317 -> State Stream m a

297318 -> SVar Stream m a

298319 -> Maybe WorkerInfo

299320 -> m ()

300321drainHeap q heap st sv winfo = do

301− ent <- liftIO $ dequeueFromHeap heap

302− case ent of

303− Nothing -> liftIO $ sendStop sv winfo

304− Just (Entry seqNo hent) ->

322+ r <- liftIO $ dequeueFromHeap heap

323+ case r of

324+ Ready (Entry seqNo hent) ->

305325 processHeap q heap st sv winfo hent seqNo True

326+ _ -> liftIO $ sendStop sv winfo

306327

307328data HeapStatus = HContinue | HStop

308329

309330processWithoutToken :: MonadIO m

310331 => IORef ([Stream m a], Int)

311− -> IORef (Heap (Entry Int (AheadHeapEntry Stream m a)) , Int)

332+ -> IORef (Heap (Entry Int (AheadHeapEntry Stream m a)), Maybe Int)

312333 -> State Stream m a

313334 -> SVar Stream m a

314335 -> Maybe WorkerInfo

@@ -368,7 +389,7 @@ processWithoutToken q heap st sv winfo m sno = do

368389

369390processWithToken :: MonadIO m

370391 => IORef ([Stream m a], Int)

371− -> IORef (Heap (Entry Int (AheadHeapEntry Stream m a)) , Int)

392+ -> IORef (Heap (Entry Int (AheadHeapEntry Stream m a)), Maybe Int)

372393 -> State Stream m a

373394 -> SVar Stream m a

374395 -> Maybe WorkerInfo

@@ -463,7 +484,7 @@ processWithToken q heap st sv winfo action sno = do

463484

464485workLoopAhead :: MonadIO m

465486 => IORef ([Stream m a], Int)

466− -> IORef (Heap (Entry Int (AheadHeapEntry Stream m a)) , Int)

487+ -> IORef (Heap (Entry Int (AheadHeapEntry Stream m a)), Maybe Int)

467488 -> State Stream m a

468489 -> SVar Stream m a

469490 -> Maybe WorkerInfo

@@ -476,9 +497,12 @@ workLoopAhead q heap st sv winfo = do

476497 when (H.size hp > maxHp) $ writeIORef (maxHeapSize $ svarStats sv)

477498 (H.size hp)

478499#endif

479− ent <- liftIO $ dequeueFromHeap heap

480− case ent of

481− Nothing -> do

500+ r <- liftIO $ dequeueFromHeap heap

501+ case r of

502+ Ready (Entry seqNo hent) ->

503+ processHeap q heap st sv winfo hent seqNo False

504+ -- when clearing then stop

505+ _ -> do

482506 -- Before we execute the next item from the work queue we check

483507 -- if we are beyond the yield limit. It is better to check the

484508 -- yield limit before we pick up the next item. Otherwise we

@@ -514,8 +538,6 @@ workLoopAhead q heap st sv winfo = do

514538 liftIO $ reEnqueueAhead sv q m

515539 incrementYieldLimit sv

516540 sendStop sv winfo

517− Just (Entry seqNo hent) ->

518− processHeap q heap st sv winfo hent seqNo False

519541

520542-------------------------------------------------------------------------------

521543-- WAhead