Streamly.Internal.Data.Channel

A worker decrements the yield limit before it executes an action. However, the action may not result in an element being yielded, in that case we have to increment the yield limit.

Note that we need it to be an Int type so that we have the ability to undo a decrement that takes it below zero.

A magical value for the buffer size arrived at by running the smallest possible task and measuring the optimal value of the buffer for that. This is obviously dependent on hardware, this figure is based on a 2.2GHz intel core-i7 processor.

Read the output queue of the channel. After reading set it to empty list and 0 count.

Same as readOutputQBasic but additionally update the max output queue size channel stat if the new size is more than current max.

RingArray door bell. The IORef is read after adding a store-load barrier. If the IORef was set to True it is atomically reset to False.

MVar diagnostics has some overhead - around 5% on AsyncT null benchmark, we can keep it on in production to debug problems quickly if and when they happen, but it may result in unexpected output when threads are left hanging until they are GCed because the consumer went away.

Events that a child thread may send to a parent thread.

Specifies the stream yield rate in yields per second (Hertz). We keep accumulating yield credits at rateGoal. At any point of time we allow only as many yields as we have accumulated as per rateGoal since the start of time. If the consumer or the producer is slower or faster, the actual rate may fall behind or exceed rateGoal. We try to recover the gap between the two by increasing or decreasing the pull rate from the producer. However, if the yield count gap becomes more than rateBuffer (specified as a yield count) we try to recover only as much as rateBuffer.

rateLow puts a bound on how low the instantaneous rate can go when recovering the rate gap. In other words, it determines the maximum yield latency. Similarly, rateHigh puts a bound on how high the instantaneous rate can go when recovering the rate gap. In other words, it determines the minimum yield latency. We reduce the latency by increasing concurrency, therefore we can say that it puts an upper bound on concurrency.

If the rateGoal is 0 or negative the stream never yields a value. If the rateBuffer is 0 or negative we do not attempt to recover.

Specify when the Channel should stop.

Channel driver throws this exception to all active workers to clean up the channel.

We measure the individual worker latencies to estimate the number of workers needed or the amount of time we have to sleep between dispatches to achieve a particular rate when controlled pace mode it used.

Rate control.

This is safe even if we are adding more threads concurrently because if a child thread is adding another thread then anyway workerThreads will not be empty.

Always moves workerPendingLatency to workerCollectedLatency:

• workerCollectedLatency always incremented by workerPendingLatency

• workerPendingLatency always reset to 0

Moves workerCollectedLatency to svarAllTimeLatency periodically, when the collected batch size hits a limit, or time limit is over, or latency changes beyond a limit. Updates done when the batch is collected:

• svarAllTimeLatency yield count updated

• workerMeasuredLatency set to (new+prev)/2

• workerPollingInterval set using max of new/prev worker latency

• workerCollectedLatency reset to 0

See also getWorkerLatency.

This is a magic number and it is overloaded, and used at several places to achieve batching:

1. If we have to sleep to slowdown this is the minimum period that we accumulate before we sleep. Also, workers do not stop until this much sleep time is accumulated.

3. Collected latencies are computed and transferred to measured latency after a minimum of this period.

Estimate how many workers and yield count (Work) is required to maintian the target yield rate of the channel.

This is used by the worker dispatcher to estimate how many workers to dispatch. It is also used periodically by the workers to decide whether to stop or continue working.

Update the local yield count of the worker and check if:

• the channel yield rate is beyond max limit

• worker's yield count is beyond max limit

Using the channel worker latency and channel yield count stats from the current measurement interval, estimate how many workers are needed to maintain the target rate and compare that with current number of workers. Returns true if we have have more than required workers.

Low level API to add an event on the channel's output queue. Atomically adds the event to the queue and rings the doorbell if needed to wakeup the consumer thread.

Add a ChildStop event with exception to the channel's output queue.

Add a ChildStop event to the channel's output queue. When rate control is enabled, it pushes the worker latency stats to the channel.

Add a ChildYield event to the channel's output queue.

This is a wrapper over sendEvent, it does a few more things:

• performs a buffer limit check, returns False if exceeded

When rate control is enabled and WorkerInfo is supplied::

• increments the worker yield count

• periodically pushes the worker latency stats to the channel

• performs a rate limit check, returns False if exceeded

Describes how to pace the work based on current measurement estimates. If the rate is higher than expected we may have to sleep for some time (BlockWait), or send just one worker with limited yield count (PartialWorker) or send more than one workers with max yield count of each limited to the total maximum target count.