Reactor-core: Switch to nanosecond precision in Duration-based operators

[william-tran]

Expected behavior

Mono.delay(Duration.ofMillis(1).minusNanos(1)) delays emission by 1 millisecond minus one nanosecond

Actual behavior

The Mono emits immediately without delay, because

    public static Mono<Long> delay(Duration duration, Scheduler timer) {
        return onAssembly(new MonoDelay(duration.toMillis(), TimeUnit.MILLISECONDS, timer));
    }

Proposed fix

    public static Mono<Long> delay(Duration duration, Scheduler timer) {
        return onAssembly(new MonoDelay(duration.toNanos(), TimeUnit.NANOSECONDS, timer));
    }

Use Case

I'm using Reactor to drive high concurrency simulations needing precision better than a millisecond.

[simonbasle]

@william-tran I expect you're not using a default Scheduler nor doing ExecutorService-based scheduling, to get a sub-millisecond precision in Java?

Unfortunately this assumption that the sub-millisecond precision isn't needed is pervasive in Duration-based operators, so this wouldn't be that simple of a fix.

[william-tran]

I've been spending the last half hour looking at how pervasive this is, and I see what you mean (I may have missed a few):

reactor.core.publisher.Mono.delay(Duration, Scheduler)
reactor.core.publisher.Mono.block(Duration)
reactor.core.publisher.Mono.blockOptional(Duration)
reactor.core.publisher.Mono.delayElement(Duration, Scheduler)
reactor.core.publisher.Flux.interval(Duration, Scheduler)
reactor.core.publisher.Flux.interval(Duration, Duration, Scheduler)
reactor.core.publisher.Flux.blockFirst(Duration)
reactor.core.publisher.Flux.blockLast(Duration)
reactor.core.publisher.Flux.bufferTimeout(int, Duration, Scheduler, Supplier<C>) and FluxBufferTimeout
reactor.core.publisher.Flux.replay(int, Duration, Scheduler) and FluxReplay
reactor.core.publisher.Flux.timeout(Duration, Publisher<? extends T>, Scheduler)
reactor.core.publisher.Flux.windowTimeout(int, Duration, Scheduler) and FluxWindowTimeout
reactor.core.publisher.MonoCacheTime.CoordinatorSubscriber.signalCached(Signal<T>)

But it looks like reactor.core.publisher.Flux.delaySequence(Duration) and FluxDelaySequence deals with nanos!

if (delay.compareTo(Duration.ofMinutes(1)) < 0) {
    this.delay = delay.toNanos();
    this.timeUnit = TimeUnit.NANOSECONDS;
}
else {
    this.delay = delay.toMillis();
    this.timeUnit = TimeUnit.MILLISECONDS;
}

I was using Schedulers.parallel() as the scheduler for work and delay. As a work around, would you suggest I create my own ScheduledExecutorService for dealing with delays, and then convert those Futures to CompletableFutures, and then to Monos?

[william-tran]

I also looked at reactor.core.scheduler.ParallelScheduler.schedule(Runnable, long, TimeUnit) but I don't see how I can convert the returned Disposable into a Mono.

[simonbasle]

my main point is that ScheduledExecutorService does not offer real-time execution guarantees (and probably neither does the OS), so nanos is practically irrelevant.

So delaySequence accounting for such small delays is probably an "overoptimization" on my part...

But maybe it would be possible to come up with a Scheduler that uses parkNanos or Thread.sleep(millis, nanos) though... Only _then_ would you actually hit the nanos-to-millis rounding down issue, I think.

[william-tran]

Even without real-time execution guarantees, the aggregate result of using nanos instead of millis for delays under 1 milli, is far better than the milli round-down resulting in no delay. The example Mono.delay(Duration.ofMillis(1).minusNanos(1)) is truly worst case, but even if my delay was half a milli, I would want the average execution time to be as close to that as possible.

[william-tran]

I think I have an easy workaround for Mono.delay: Flux.just(0L).delaySequence(nanoDuration).last(). Please leave that "overoptimization" in there!

[simonbasle]

ok. I think we can switch to a toNanos() conversion without much hurt, especially knowing that Long.MAX_VALUE nanoseconds represents about 292 years (see https://github.com/reactor/reactor-core/issues/1663#issuecomment-495214764)

I'll reword the issue. This would be a slight behavioral change though, and thus would target the 3.3 generation. We are about to release 3.3.0.M2 on monday, but I'm not sure either you and I can make the change in the remaining time @william-tran...

@smaldini wdyt about the switch from toMillis() precision to toNanos()?

[simonbasle]

one way of doing it is by converting anything bigger than a minute to millis, like in FluxDelaySequence:

if (delay.compareTo(Duration.ofMinutes(1)) < 0) {
    this.delay = delay.toNanos();
    this.timeUnit = TimeUnit.NANOSECONDS;
}
else {
    this.delay = delay.toMillis();
    this.timeUnit = TimeUnit.MILLISECONDS;
}

Another approach would be to check for max nanos:

static final Duration MAX_NANOS = Duration.ofNanos(Long.MAX_VALUE);
//then...
if (timeout.compareTo(MAX_NANOS)>=0) {
    this.delay = delay.toMillis();
    this.timeUnit = TimeUnit.MILLISECONDS;
}
else {
    this.delay = delay.toNanos();
    this.timeUnit = TimeUnit.NANOSECONDS;
}

Any better idea? Approach 1 or 2?

[william-tran]

Approach 2 provides the most precision over the widest range of possible values, so I'd go with that. Approach 1 feels like an arbitrary point at which you decide to round off. 3.3 sounds like the right release, will there be an M3 that we can get this into? I'm happy to help with a PR, but not sure if I'll have the time to have it done for Monday.

[simonbasle]

will there be an M3 that we can get this into?

There should be a M3 within a month or so, yes.

[simonbasle]

@william-tran we're still on the fence about making that change in 3.3 at all. Is there a real benefit right now that can be more fully fleshed out?

I still think this makes sense in the near future, because JVMs might become more precise and we would then automatically benefit from that, but since nothing has been done yet and M3 is approaching, I'm just wondering if we can defer to further down the line (ie. 3.4).

[william-tran]

Wow does time ever fly. Because I have a workaround for nanosecond precision delay with Flux.just(0L).delaySequence(nanoDuration).last() and there is no other input from the community, then I am fine to have this part of 3.4. Practically, this workaround did enable me to produce output with the needed precision and frequency, e.g. 10000 "concurrent tasks" at 2Hz and 10 concurrent tasks at 2KHz both produce throughput that is bang on (+/- 0.1%) at 20KHz. I think a contribution from me would be better timed for 3.4 as well if you're depending on that.

[simonbasle]

@william-tran haha yeah, great to hear you have a working workaround 👍 we're not exactly depending on a contribution from you, but it would indeed be greatly appreciated 🎁
I'll mark this issue for 3.4, and ideal for contribution ;)

[simonbasle]

one thing I didn't think about in detail with making this move to NANOS is that it would introduce a discrepancy between what you can supposedly do eg in terms of interval vs the precision of the elapsed operator. Since it is part of the public API contract that this operator gives timings in MILLIS, we cannot really change that without a compelling reason.

And since java scheduling tools like ExecutorService are currently not precise enough to honor nanoseconds precision, we don't really have that compelling reason...

I'm wondering if it would make sense to round up any strictly positive yet sub-millisecond duration to 1 millisecond?

[UgiR]

Ok, it sounds like a bit more trouble than it's worth for now.

As for rounding, I would lean toward more explicitly documenting the fact that sub-millisecond decimals are truncated rather than rounding up for duration's in that range. That way, all duration's 0 to N milliseconds have consistent behavior and the user can decide how to proceed (not using Duration.ofMillis(1).minusNanos(1)). Otherwise, it would not surprise me if the rounding might cause an issue for someone with a niche use case, just like the truncation caused the problem for sub-millisecond delay for the author of the issue. That's my initial thought, but I can see the case for either way of handling it.