ts_future_util: init and write Debounce

[npry]

Write a Debounce stream wrapper that debounces an inner stream with both leading and trailing-edge behavior.

In many cases, we have streams where each item is a complete state snapshot – lots of cases of this in runtime, e.g.: the netmap, the derp map, the set of all peer routes, etc. Because each element of these streams encodes the full information about its part of the system, we don't actually have to process every element of the stream: we just need to end up eventually consistent. More precisely, we must always eventually process the most recent item in the stream to achieve steady-state consistency.

The flip side of this is that if we're not seeing bursty behavior on a stream, we want to process each item immediately as it comes in to reduce latency.

So we want a stream combinator that, when idle, immediately yields an incoming item, then enters a debouncing window. While in the debounce window, it suppresses all items on the stream, and at the end of the window, it emits the last item it received, if any. If there was a debounced item, it opens a new debounce window when it does this. If not, it returns to idle.

This gets us the behavior we want: we throttle bursty incoming traffic when it's bursty, but update immediately if we're not in the middle of a burst. In both cases we're eventually consistent.

Existing alternatives I came across were unsuitable:

• tokio_stream::StreamExt::throttle just adds a time delay between items
• futures_time::StreamExt::throttle only provides leading-edge behavior; it's not eventually consistent
• futures_time::StreamExt::debounce may never become ready; the debounce interval resets with each item received
• futures_time::StreamExt::sample only has trailing-edge behavior and runs on an interval timer (rather than as-needed and triggered on an item from the underlying stream)

crate

I stuck this in a new crate ts_future_util because:

• Generic/unscoped *util crates tend to get bloated/it tends to be unclear what belongs there, and it lowers the barrier to just dumping code in a common spot
• There doesn't seem to be anywhere else obvious it should go
• I need it in both ts_netmon and ts_runtime (that I know of right now), and will probably want it elsewhere

Happy to put it somewhere else if desired

[danderson]

Expected assuming the poll wakers are precise, right? The assertion in the branch is still correct, just making sure I follow the flow of all the branching.

[npry]

Hmm, not sure what you mean. The timer future doesn't need to precisely wake the waker, we'll get Poll::Pending above and fall through to this case (assuming !*slf.stream_done). We actually want that to occur, though it should be the stream that wakes, because we need to continue to make progress on it. The stream can also wake whenever it wants, including eagerly, we just bail below in the ready!. But regardless, we'll get here unless we've bailed with a done stream (Poll::Ready(None)) or the timer expired and we returned a slot item (in which case it's the caller's responsibility to call us again, because we returned Poll::Ready(Some(_))).

[danderson]

I mean that I took "getting to this point means that an item is expected from the stream" to mean that if you get to this point the stream will definitely either be done, or will yield 1+ item when polled in the while loop.

That is true if the timer+stream wakers are precise: we got polled, and to reach this point we had Poll::Pending from the timer, therefore the stream must be Poll::Ready. But in a runtime where the waking is imprecise (worst case: just busy loop polling every future), we might get polled and reach this point and get a Poll::Pending from the stream drain as well.

If that happens, if I'm reading the code correctly, we'll busy loop in Debounce::poll_next until one of the timer or stream becomes ready. Is that the correct behavior? Should we be returning Poll::Pending instead and let the caller repoll later?

I accept this is a niche concern to have, and I may just not grok the semantics of polling stream futures.

[npry]

Ah, ok, two points: first, the comment is misleading -- it should actually say something along the lines that we're trying to get items from the stream, i.e. we need to poll it for it to make progress (including setting a waker if needed). This is why we don't return Poll::Pending when checking the timer unless the stream is over: we must always poll the stream to ensure we're woken if it makes progress in the future.

Second, core::task::ready! is essentially ? for Poll, i.e. it returns with Poll::Pending if the expr produced Poll::Pending, otherwise unwrapping to the Poll::Ready value. The function never busy loops (which is always true for poll* style functions) -- it just does as much work as is synchronously available

[npry]

rerequested you @danderson for debounce_fold -- not too much to it but figured it could use a look

[danderson]

How come? I guess it's fine, but if we allow switching fold functions after creation my assumption was that folding would begin starting at the next received value, with whatever's already in the slot being the accumulator. Fine either way, just curious what the reasoning was.

[npry]

Changing the fold function changes the type of the slot value -- for an old fold fn FnMut(&mut Option<T>, S::Item) to a new fold fn FnMut(&mut Option<U>, S::Item), we would need a map from T -> U to preserve the value. Adding that extra function to the signature felt like it overcomplicated the API with a detail most people wouldn't care about; the thought is that this is typically going to be called before the stream is ever polled, so throwing out the slot value is a no-op in the typical case.

[danderson]

I mean that I took "getting to this point means that an item is expected from the stream" to mean that if you get to this point the stream will definitely either be done, or will yield 1+ item when polled in the while loop.

That is true if the timer+stream wakers are precise: we got polled, and to reach this point we had Poll::Pending from the timer, therefore the stream must be Poll::Ready. But in a runtime where the waking is imprecise (worst case: just busy loop polling every future), we might get polled and reach this point and get a Poll::Pending from the stream drain as well.

If that happens, if I'm reading the code correctly, we'll busy loop in Debounce::poll_next until one of the timer or stream becomes ready. Is that the correct behavior? Should we be returning Poll::Pending instead and let the caller repoll later?

I accept this is a niche concern to have, and I may just not grok the semantics of polling stream futures.