a7556e8 · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Add some locking notes

a7556e8Harendra Kumar7 years ago

src/Streamly/SVar.hs+41−4

compare with master

unifiedsplit

@@ -258,6 +258,9 @@ data YieldRateInfo = YieldRateInfo

258 { svarLatencyTarget :: NanoSecond64

259 , svarLatencyRange :: LatencyRange

260 , svarRateBuffer :: Int

261+

262+ -- [LOCKING] Unlocked access. Modified by the consumer thread and unsafely

263+ -- read by the worker threads

261 , svarGainedLostYields :: IORef Count

264 , svarGainedLostYields :: IORef Count

262

265

263 -- Actual latency/througput as seen from the consumer side, we count the

266 -- Actual latency/througput as seen from the consumer side, we count the

@@ -266,6 +269,8 @@ data YieldRateInfo = YieldRateInfo

266 -- rate. The idle time of workers is adjusted in this, so that we only

269 -- rate. The idle time of workers is adjusted in this, so that we only

267 -- account for the rate when the consumer actually demands data.

270 -- account for the rate when the consumer actually demands data.

268 -- XXX interval latency is enough, we can move this under diagnostics build

271 -- XXX interval latency is enough, we can move this under diagnostics build

272+ -- [LOCKING] Unlocked access. Modified by the consumer thread and unsafely

273+ -- read by the worker threads

269 , svarAllTimeLatency :: IORef (Count, AbsTime)

274 , svarAllTimeLatency :: IORef (Count, AbsTime)

270

275

271 -- XXX Worker latency specified by the user to be used before the first

276 -- XXX Worker latency specified by the user to be used before the first

@@ -278,6 +283,8 @@ data YieldRateInfo = YieldRateInfo

278 -- long time, in such cases the consumer can change it.

283 -- long time, in such cases the consumer can change it.

279 -- 0 means no latency computation

284 -- 0 means no latency computation

280 -- XXX this is derivable from workerMeasuredLatency, can be removed.

285 -- XXX this is derivable from workerMeasuredLatency, can be removed.

286+ -- [LOCKING] Unlocked access. Modified by the consumer thread and unsafely

287+ -- read by the worker threads

281 , workerPollingInterval :: IORef Count

288 , workerPollingInterval :: IORef Count

282

289

283 -- This is in progress latency stats maintained by the workers which we

290 -- This is in progress latency stats maintained by the workers which we

@@ -286,6 +293,10 @@ data YieldRateInfo = YieldRateInfo

286 -- is all yields, the second count is only those yields for which the

293 -- is all yields, the second count is only those yields for which the

287 -- latency was measured to be non-zero (note that if the timer resolution

294 -- latency was measured to be non-zero (note that if the timer resolution

288 -- is low the measured latency may be zero e.g. on JS platform).

295 -- is low the measured latency may be zero e.g. on JS platform).

296+ -- [LOCKING] Locked access. Modified by the consumer thread as well as

297+ -- worker threads. Workers modify it periodically based on

298+ -- workerPollingInterval and not on every yield to reduce the locking

299+ -- overhead.

289 -- (allYieldCount, yieldCount, timeTaken)

300 -- (allYieldCount, yieldCount, timeTaken)

290 , workerPendingLatency :: IORef (Count, Count, NanoSecond64)

301 , workerPendingLatency :: IORef (Count, Count, NanoSecond64)

291

302

@@ -294,10 +305,14 @@ data YieldRateInfo = YieldRateInfo

294 -- bucket until we have stats for a sufficient period and then we reset it

305 -- bucket until we have stats for a sufficient period and then we reset it

295 -- to start collecting for the next period and retain the computed average

306 -- to start collecting for the next period and retain the computed average

296 -- latency for the last period in workerMeasuredLatency.

307 -- latency for the last period in workerMeasuredLatency.

308+ -- [LOCKING] Unlocked access. Modified by the consumer thread and unsafely

309+ -- read by the worker threads

297 -- (allYieldCount, yieldCount, timeTaken)

310 -- (allYieldCount, yieldCount, timeTaken)

298 , workerCollectedLatency :: IORef (Count, Count, NanoSecond64)

311 , workerCollectedLatency :: IORef (Count, Count, NanoSecond64)

299

312

300 -- Latency as measured by workers, aggregated for the last period.

313 -- Latency as measured by workers, aggregated for the last period.

314+ -- [LOCKING] Unlocked access. Modified by the consumer thread and unsafely

315+ -- read by the worker threads

301 , workerMeasuredLatency :: IORef NanoSecond64

316 , workerMeasuredLatency :: IORef NanoSecond64

302 }

317 }

303

318

@@ -313,6 +328,7 @@ data SVarStats = SVarStats {

313 , svarStopTime :: IORef (Maybe AbsTime)

328 , svarStopTime :: IORef (Maybe AbsTime)

314}

329}

315

330

331+-- This is essentially a 'Maybe Word' type

316data Limit = Unlimited | Limited Word deriving Show

332data Limit = Unlimited | Limited Word deriving Show

317

333

318-- When to stop the composed stream.

334-- When to stop the composed stream.

@@ -341,7 +357,19 @@ data SVar t m a = SVar

341 -- avoid constructing and reversing the list. Possibly we can also avoid

357 -- avoid constructing and reversing the list. Possibly we can also avoid

342 -- the GC copying overhead. When the size increases we should be able to

358 -- the GC copying overhead. When the size increases we should be able to

343 -- allocate the array in chunks.

359 -- allocate the array in chunks.

360+ --

361+ -- [LOCKING] Frequent locked access. This is updated by workers on each

362+ -- yield and once in a while read by the consumer thread. This could have

363+ -- big locking overhead if the number of workers is high.

364+ --

365+ -- XXX We can use a per-CPU data structure to reduce the locking overhead.

366+ -- However, a per-cpu structure cannot guarantee the exact sequence in

367+ -- which the elements were added, though that may not be important.

344 , outputQueue :: IORef ([ChildEvent a], Int)

368 , outputQueue :: IORef ([ChildEvent a], Int)

369+

370+ -- [LOCKING] Infrequent MVar. Used when the outputQ transitions from empty

371+ -- to non-empty, or a work item is queued by a worker to the work queue and

372+ -- needDoorBell is set by the consumer.

345 , outputDoorBell :: MVar () -- signal the consumer about output

373 , outputDoorBell :: MVar () -- signal the consumer about output

346 , readOutputQ :: m [ChildEvent a]

374 , readOutputQ :: m [ChildEvent a]

347 , postProcess :: m Bool

375 , postProcess :: m Bool

@@ -360,7 +388,13 @@ data SVar t m a = SVar

360 , workLoop :: Maybe WorkerInfo -> m ()

388 , workLoop :: Maybe WorkerInfo -> m ()

361

389

362 -- Shared, thread tracking

390 -- Shared, thread tracking

391+ -- [LOCKING] Updated unlocked only by consumer thread in case of

392+ -- Async/Ahead style SVars. Updated locked by worker threads in case of

393+ -- Parallel style.

363 , workerThreads :: IORef (Set ThreadId)

394 , workerThreads :: IORef (Set ThreadId)

395+ -- [LOCKING] Updated locked by consumer thread when dispatching a worker

396+ -- and by the worker threads when the thread stops. This is read unsafely

397+ -- at several places where we want to rely on an approximate value.

364 , workerCount :: IORef Int

398 , workerCount :: IORef Int

365 , accountThread :: ThreadId -> m ()

399 , accountThread :: ThreadId -> m ()

366 , workerStopMVar :: MVar ()

400 , workerStopMVar :: MVar ()

@@ -852,8 +886,12 @@ doFork action (RunInIO mrun) exHandler =

852-- if the application is distributed then inc/dec of a shared variable may be

886-- if the application is distributed then inc/dec of a shared variable may be

853-- very costly.

887-- very costly.

854--

888--

889+-- A worker decrements the yield limit before it executes an action. However,

890+-- the action may not result in an element being yielded, in that case we have

891+-- to increment the yield limit.

892+--

855-- Note that we need it to be an Int type so that we have the ability to undo a

893-- Note that we need it to be an Int type so that we have the ability to undo a

856−-- decrement that takes below zero.

894+-- decrement that takes it below zero.

857{-# INLINE decrementYieldLimit #-}

895{-# INLINE decrementYieldLimit #-}

858decrementYieldLimit :: SVar t m a -> IO Bool

896decrementYieldLimit :: SVar t m a -> IO Bool

859decrementYieldLimit sv =

897decrementYieldLimit sv =

@@ -1426,9 +1464,8 @@ dispatchWorker yieldCount sv = do

1426 if not done

1464 if not done

1427 then do

1465 then do

1428 qDone <- liftIO $ isQueueDone sv

1466 qDone <- liftIO $ isQueueDone sv

1429− -- Note that the worker count is only decremented during event

1467+ -- This count may not be accurate as it is decremented by the workers

1430− -- processing in fromStreamVar and therefore it is safe to read and

1468+ -- and we have no synchronization with that decrement.

1431− -- use it without a lock.

1432 active <- liftIO $ readIORef $ workerCount sv

1469 active <- liftIO $ readIORef $ workerCount sv

1433 if not qDone

1470 if not qDone

1434 then do

1471 then do