feat: measured oneshots

[drahnr]

Introduces oneshots that measure the delay between creation, first poll and completion or error (distinguishing between hard timeout and cancellation).

Also prints a warning based on a set soft timeout.

Closes #3648

[drahnr]

The API usage is as follows:

When creating a pair, it has to be given a name as well as two timeouts.

// 1. create
let (tx, rx) = metered::oneshot::channel("foo", Duration::from_secs(500), Duration::from_secs(3));

// 2. store the meter
let meter = rx.meter();

// 3.
let val = rx.await;

reasoning for this rather providing the metrics as part of the error or as tuple, is compat. This is a rather minimal change and we could add metrics as needed, yet just change all use-statements at once and adjust to the additional params without much chore.

[ordian]

The impl looks correct, but a bit too complicated to me. Can we just have a thin abstraction on top of rx.await that will do the following?

let now = Instant::now();
let result = rx.timeout(HARD_TIMEOUT).await?;
match result {
    None => {
        error!("Subsystem {name} is unresponsive", name=name);
    }
    Some(msg) if now.elapsed() > SOFT_TIMEOUT {
        warn!("subsystem {name} is slow", name=name);
    }
<snip>    
}

[drahnr]

The impl looks correct, but a bit too complicated to me. Can we just have a thin abstraction on top of rx.await that will do the following?

let now = Instant::now();
let result = rx.timeout(HARD_TIMEOUT).await?;
match result {
    None => {
        error!("Subsystem {name} is unresponsive", name=name);
    }
    Some(msg) if now.elapsed() > SOFT_TIMEOUT {
        warn!("subsystem {name} is slow", name=name);
    }
<snip>    
}

Essentially that is what is implemented explicitly within poll, since I'd rather not have the above code spread over all cases of oneshot metering ( + the above is not measuring anything yet).

The complexity arises from Option<impl Future>, since we need to measure from the time we poll for the first time rather than creation. Without that, it'd become much simpler, but that'd be slightly less helpful I think.

Another, alternate simplification would be to spawn Delay with a very large timeout. And then once polled for the first time, call reset. But that seems mildly hacky.

I very much appreciate the feedback!

Note that I did not consider the complexity of the above code that much, I was more concerned how invasive introducing it throughout the code base would be.

[ordian]

The complexity arises from Option, since we need to measure from the time we poll for the first time rather than creation. Without that, it'd become much simpler, but that'd be slightly less helpful I think.

What I'm suggesting is implementing a function (or a struct wrapper for more type-safety) with the code above (modulo bikeshed on naming)

async fn timed_await(rx: oneshot::Receiver<T>) -> Result<T> {...}

and replacing rx.await? with timed_await(rx).await?. The time we poll the first time is the same as the time we call this function, so I don't see a difference here.

But maybe I'm missing something?

[drahnr]

The complexity arises from Option, since we need to measure from the time we poll for the first time rather than creation. Without that, it'd become much simpler, but that'd be slightly less helpful I think.

What I'm suggesting is implementing a function (or a struct wrapper for more type-safety) with the code above (modulo bikeshed on naming)

async fn timed_await(rx: oneshot::Receiver<T>) -> Result<T> {...}

and replacing rx.await? with timed_await(rx).await?. The time we poll the first time is the same as the time we call this function, so I don't see a difference here.

But maybe I'm missing something?

What you describe would be the functional approach and at the first glance you indeed obtain similar measurements while avoiding some complexity.

With the struct- wrappers you will always end up with the complexity exhibited in this PR - one has to somehow mingle the futures and provide a Future trait impl for the wrapper type.

The complexity on the other handside allows to better measure the actual time between the first poll and the send rather than the time of the receiver end blocking. In the common case these will be almost identical, but I'd rather not simplify here, especially when one looks at grafana dashboards occluding the odd deltas.

This also allows to simply add a bunch of extra measurements as needed (what's the delay between receiver creation and first poll, the delta between the true send and the true conclusion, so machine overload conditions will not show as wrongly blocking durations, but these are just a few details that will eventually become important in our grafana views).

I'd prefer to merge this with the tad of added complexity rather than measuring almost the right thing :) - in fact this impl allows to measure both as needed.

[rphmeier]

why is this allow(missing_docs)? the measurements function is also without docs.

[drahnr]

This in particular has missing docs, because iff the error is not self describing, there is something wrong already. The error annotation should be enough. This is a pattern we have in more places throughout the codebase, limited to Error types.

[rphmeier]

The code here looks OK, but I wonder about expected usage - if it's meant to be plugged into prometheus, then it feels fairly unwieldy to do so.

[drahnr]

The code here looks OK, but I wonder about expected usage - if it's meant to be plugged into prometheus, then it feels fairly unwieldy to do so.

How so?

Currently we have a struct Metrics per subsystem, with collector fn-implementations specific to the subsystem i.e. fn on_foo(u64) and fn on_bar(Hash).

Now the returned Result type of the receiver ( or more specific: Measurements) contains 3 values, which would need to be forwarded to a prometheus type like (in the most common case) prometheus::Histogram with i.e. fn observe(&self, v: f64).

So yes, the resulting evaluation is not perfect but it's also encapsulated and one does not have to differentiate between Ok and Err at the cost of a local let result = rx.await binding (or not, since Error and OutputWithMesasurements both impl Measurable).

There are trade-offs with this impl, but in a discussion /w @eskimor this seemed the most practical one, happy to be enlightened though

[Lldenaurois]

Even though it's just a test, I think it would be better to make this a constant.

[drahnr]

They are all used exactly once, and are really only relevant per test case (and each of them is super short), so I don't think that's necessary (though I generally agree with you).

[eskimor]

Wasn't that pretty enough? 😛

[drahnr]

Can't handle it, too pretty 🎭

[drahnr]

bot merge

[ghost]

Waiting for commit status.