9d08a58 · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Disringuish the heap clearing and waiting states

9d08a58Harendra Kumar8 years ago

src/Streamly/SVar.hs+58−34

compare with master

unifiedsplit

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

71 , queueEmptyAhead

72 , dequeueAhead

73+

74+ , HeapDequeueResult(..)

73 , dequeueFromHeap

75 , dequeueFromHeap

74 , dequeueFromHeapSeq

76 , dequeueFromHeapSeq

75 , requeueOnHeapTop

77 , requeueOnHeapTop

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

366 , svarRef :: Maybe (IORef ())

368 , svarRef :: Maybe (IORef ())

367#ifdef DIAGNOSTICS

369#ifdef DIAGNOSTICS

368 , svarCreator :: ThreadId

370 , 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)

370 -- Shared work queue (stream, seqNo)

373 -- Shared work queue (stream, seqNo)

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

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

372#endif

375#endif

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

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

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

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

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

1139

1142

1140−{-# INLINE dequeueFromHeap #-}

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

1141−dequeueFromHeap

1144+-- Heap manipulation

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

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

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)

1151

1146

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

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

1153withIORef ref f = readIORef ref >>= f

1148withIORef ref f = readIORef ref >>= f

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

1156atomicModifyIORef_ ref f =

1151atomicModifyIORef_ ref f =

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

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

1158

1153

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+

1159{-# INLINE canDequeueFromHeap #-}

1174{-# INLINE canDequeueFromHeap #-}

1160canDequeueFromHeap

1175canDequeueFromHeap

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

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

1162 -> IO Bool

1177 -> IO Bool

1163canDequeueFromHeap hpVar =

1178canDequeueFromHeap hpVar =

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

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

1180+ case snum of

1181+ Nothing -> return True

1182+ Just n -> do

1165− let r = H.uncons hp

1183+ let r = H.uncons hp

1166− case r of

1184+ case r of

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

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

1168− _ -> return False

1186+ _ -> return False

1169

1187

1170{-# INLINE dequeueFromHeapSeq #-}

1188{-# INLINE dequeueFromHeapSeq #-}

1171dequeueFromHeapSeq

1189dequeueFromHeapSeq

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

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

1173 -> Int

1191 -> Int

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

1192+ -> IO (HeapDequeueResult t m a)

1175−dequeueFromHeapSeq hpVar snum =

1193+dequeueFromHeapSeq hpVar i =

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

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

1195+ case snum of

1196+ Nothing -> do

1177− let r = H.uncons hp

1197+ let r = H.uncons hp

1178− case r of

1198+ case r of

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

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

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

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

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

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

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

1182

1203

1183{-# INLINE requeueOnHeapTop #-}

1204{-# INLINE requeueOnHeapTop #-}

1184requeueOnHeapTop

1205requeueOnHeapTop

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

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

1186 -> Entry Int (AheadHeapEntry t m a)

1207 -> Entry Int (AheadHeapEntry t m a)

1187 -> Int

1208 -> Int

1188 -> IO ()

1209 -> IO ()

1189requeueOnHeapTop hpVar ent seqNo =

1210requeueOnHeapTop hpVar ent seqNo =

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

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

1191

1212

1192{-# INLINE updateHeapSeq #-}

1213{-# INLINE updateHeapSeq #-}

1193updateHeapSeq

1214updateHeapSeq

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

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

1195 -> Int

1216 -> Int

1196 -> IO ()

1217 -> IO ()

1197updateHeapSeq hpVar seqNo =

1218updateHeapSeq hpVar seqNo =

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

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

1199

1220

1200-------------------------------------------------------------------------------

1221-------------------------------------------------------------------------------

1201-- WAhead

1222-- WAhead

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

1958getAheadSVar :: MonadAsync m

1979getAheadSVar :: MonadAsync m

1959 => State t m a

1980 => State t m a

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

1981 -> ( 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)

1962 -> State t m a

1983 -> State t m a

1963 -> SVar t m a

1984 -> SVar t m a

1964 -> Maybe WorkerInfo

1985 -> Maybe WorkerInfo

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

1966 -> IO (SVar t m a)

1987 -> IO (SVar t m a)

1967getAheadSVar st f = do

1988getAheadSVar st f = do

1968 outQ <- newIORef ([], 0)

1989 outQ <- newIORef ([], 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.

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

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

1970 outQMv <- newEmptyMVar

1994 outQMv <- newEmptyMVar

1971 active <- newIORef 0

1995 active <- newIORef 0

1972 wfw <- newIORef False

1996 wfw <- newIORef False

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

2157 => State t m a

2181 => State t m a

2158 -> t m a

2182 -> t m a

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

2183 -> ( 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)

2161 -> State t m a

2185 -> State t m a

2162 -> SVar t m a

2186 -> SVar t m a

2163 -> Maybe WorkerInfo

2187 -> Maybe WorkerInfo

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

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

164-- False => continue

165preStopCheck ::

166 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)

168 -> IO Bool

169preStopCheck sv heap = do

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

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

187 if rateOk then continue else stop

188 else stop

189

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+--

190processHeap :: MonadIO m

210processHeap :: MonadIO m

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

211 => 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)

193 -> State Stream m a

213 -> State Stream m a

194 -> SVar Stream m a

214 -> SVar Stream m a

195 -> Maybe WorkerInfo

215 -> Maybe WorkerInfo

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

225 else runStreamWithYieldLimit True seqNo r

245 else runStreamWithYieldLimit True seqNo r

226

246

227 nextHeap prevSeqNo = do

247 nextHeap prevSeqNo = do

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

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

229− case ent of

249+ case res of

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

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

231− Nothing -> do

251+ -- when Clearing then stop

252+ _ -> do

232 if stopping

253 if stopping

233 then do

254 then do

234 r <- liftIO $ preStopCheck sv heap

255 r <- liftIO $ preStopCheck sv heap

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

292{-# NOINLINE drainHeap #-}

313{-# NOINLINE drainHeap #-}

293drainHeap :: MonadIO m

314drainHeap :: MonadIO m

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

315 => 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)

296 -> State Stream m a

317 -> State Stream m a

297 -> SVar Stream m a

318 -> SVar Stream m a

298 -> Maybe WorkerInfo

319 -> Maybe WorkerInfo

299 -> m ()

320 -> m ()

300drainHeap q heap st sv winfo = do

321drainHeap q heap st sv winfo = do

301− ent <- liftIO $ dequeueFromHeap heap

322+ r <- liftIO $ dequeueFromHeap heap

302− case ent of

323+ case r of

303− Nothing -> liftIO $ sendStop sv winfo

304− Just (Entry seqNo hent) ->

324+ Ready (Entry seqNo hent) ->

305 processHeap q heap st sv winfo hent seqNo True

325 processHeap q heap st sv winfo hent seqNo True

326+ _ -> liftIO $ sendStop sv winfo

306

327

307data HeapStatus = HContinue | HStop

328data HeapStatus = HContinue | HStop

308

329

309processWithoutToken :: MonadIO m

330processWithoutToken :: MonadIO m

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

331 => 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)

312 -> State Stream m a

333 -> State Stream m a

313 -> SVar Stream m a

334 -> SVar Stream m a

314 -> Maybe WorkerInfo

335 -> Maybe WorkerInfo

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

368

389

369processWithToken :: MonadIO m

390processWithToken :: MonadIO m

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

391 => 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)

372 -> State Stream m a

393 -> State Stream m a

373 -> SVar Stream m a

394 -> SVar Stream m a

374 -> Maybe WorkerInfo

395 -> Maybe WorkerInfo

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

463

484

464workLoopAhead :: MonadIO m

485workLoopAhead :: MonadIO m

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

486 => 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)

467 -> State Stream m a

488 -> State Stream m a

468 -> SVar Stream m a

489 -> SVar Stream m a

469 -> Maybe WorkerInfo

490 -> Maybe WorkerInfo

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

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

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

477 (H.size hp)

498 (H.size hp)

478#endif

499#endif

479− ent <- liftIO $ dequeueFromHeap heap

500+ r <- liftIO $ dequeueFromHeap heap

480− case ent of

501+ case r of

481− Nothing -> do

502+ Ready (Entry seqNo hent) ->

503+ processHeap q heap st sv winfo hent seqNo False

504+ -- when clearing then stop

505+ _ -> do

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

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

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

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

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

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

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

514 liftIO $ reEnqueueAhead sv q m

538 liftIO $ reEnqueueAhead sv q m

515 incrementYieldLimit sv

539 incrementYieldLimit sv

516 sendStop sv winfo

540 sendStop sv winfo

517− Just (Entry seqNo hent) ->

518− processHeap q heap st sv winfo hent seqNo False

519

541

520-------------------------------------------------------------------------------

542-------------------------------------------------------------------------------

521-- WAhead

543-- WAhead