@@ -258,6 +258,9 @@ data YieldRateInfo = YieldRateInfo
258258 { svarLatencyTarget :: NanoSecond64
259259 , svarLatencyRange :: LatencyRange
260260 , svarRateBuffer :: Int
261+
262+ -- [LOCKING] Unlocked access. Modified by the consumer thread and unsafely
263+ -- read by the worker threads
261264 , svarGainedLostYields :: IORef Count
262265
263266 -- Actual latency/througput as seen from the consumer side, we count the
@@ -266,6 +269,8 @@ data YieldRateInfo = YieldRateInfo
266269 -- rate. The idle time of workers is adjusted in this, so that we only
267270 -- account for the rate when the consumer actually demands data.
268271 -- 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
269274 , svarAllTimeLatency :: IORef (Count, AbsTime)
270275
271276 -- XXX Worker latency specified by the user to be used before the first
@@ -278,6 +283,8 @@ data YieldRateInfo = YieldRateInfo
278283 -- long time, in such cases the consumer can change it.
279284 -- 0 means no latency computation
280285 -- 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
281288 , workerPollingInterval :: IORef Count
282289
283290 -- This is in progress latency stats maintained by the workers which we
@@ -286,6 +293,10 @@ data YieldRateInfo = YieldRateInfo
286293 -- is all yields, the second count is only those yields for which the
287294 -- latency was measured to be non-zero (note that if the timer resolution
288295 -- 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.
289300 -- (allYieldCount, yieldCount, timeTaken)
290301 , workerPendingLatency :: IORef (Count, Count, NanoSecond64)
291302
@@ -294,10 +305,14 @@ data YieldRateInfo = YieldRateInfo
294305 -- bucket until we have stats for a sufficient period and then we reset it
295306 -- to start collecting for the next period and retain the computed average
296307 -- latency for the last period in workerMeasuredLatency.
308+ -- [LOCKING] Unlocked access. Modified by the consumer thread and unsafely
309+ -- read by the worker threads
297310 -- (allYieldCount, yieldCount, timeTaken)
298311 , workerCollectedLatency :: IORef (Count, Count, NanoSecond64)
299312
300313 -- 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
301316 , workerMeasuredLatency :: IORef NanoSecond64
302317 }
303318
@@ -313,6 +328,7 @@ data SVarStats = SVarStats {
313328 , svarStopTime :: IORef (Maybe AbsTime)
314329}
315330
331+-- This is essentially a 'Maybe Word' type
316332data Limit = Unlimited | Limited Word deriving Show
317333
318334-- When to stop the composed stream.
@@ -341,7 +357,19 @@ data SVar t m a = SVar
341357 -- avoid constructing and reversing the list. Possibly we can also avoid
342358 -- the GC copying overhead. When the size increases we should be able to
343359 -- 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.
344368 , 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.
345373 , outputDoorBell :: MVar () -- signal the consumer about output
346374 , readOutputQ :: m [ChildEvent a]
347375 , postProcess :: m Bool
@@ -360,7 +388,13 @@ data SVar t m a = SVar
360388 , workLoop :: Maybe WorkerInfo -> m ()
361389
362390 -- 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.
363394 , 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.
364398 , workerCount :: IORef Int
365399 , accountThread :: ThreadId -> m ()
366400 , workerStopMVar :: MVar ()
@@ -852,8 +886,12 @@ doFork action (RunInIO mrun) exHandler =
852886-- if the application is distributed then inc/dec of a shared variable may be
853887-- very costly.
854888--
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+--
855893-- 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.
857895{-# INLINE decrementYieldLimit #-}
858896decrementYieldLimit :: SVar t m a -> IO Bool
859897decrementYieldLimit sv =
@@ -1426,9 +1464,8 @@ dispatchWorker yieldCount sv = do
14261464 if not done
14271465 then do
14281466 qDone <- liftIO $ isQueueDone sv
1429− -- Note that the worker count is only decremented during event
1430− -- processing in fromStreamVar and therefore it is safe to read and
1431− -- use it without a lock.
1467+ -- This count may not be accurate as it is decremented by the workers
1468+ -- and we have no synchronization with that decrement.
14321469 active <- liftIO $ readIORef $ workerCount sv
14331470 if not qDone
14341471 then do