VirtualTimeScheduler does not advance time if there is nothing on the drain queue

[a701440]

Expected behavior

VirtualTimeScheduler.advanceTimeBy actually advances the time for the following operations using the scheduler.

Actual behavior

When timedTask is scheduled it uses nanoTime as a starting point.

final TimedRunnable timedTask = new TimedRunnable(this,
nanoTime + unit.toNanos(delayTime),

However now() uses nanoTime + deferredNanoTime.

Also nanoTime is only updated to the sum of nanoTime and deferredNanoTime in the drain method, but only if the queue is not empty.

So advancing time between running timed reactor operations does not work. The following test loop iteration does not take the advanced time into account.

Steps to reproduce

for (int i = 0; i <test.size(); i++) {
Mono mono = createSomeTimedOperation(test.get(i));
assertValue(test.get(i), mono.block());
// at this point there is nothing in the drain queue, so nanoTime is not updated
// next loop iteration will take original nanoTime as base for the timeout.
vts.advanceTimeBy(Duration.ofMillis(delay));
}

Reactor Core version

3.1.7

JVM version (e.g. java -version)

[a701440]

I had to use a hack in the unit test to actually cause time advancement:

vts.advanceTimeBy(Duration.ofMillis(millis));
vts.schedule(() -> { }).dispose();

[simonbasle]

I'm not sure your test makes sense to me. If you create some timed operation, shouldn't you advance the time before you block? At this point, there's a timed operation in the queue and the advanceTimeBy will effectively move the clock.

[a701440]

I think this test would demo the issue. IMO there nothing in this test that should fail, but it does. My test cases are more complex, but involve the same issue.

@Test
public void testVirtualTime() {
    VirtualTimeScheduler vts = VirtualTimeScheduler.create();
    Mono.create(sink -> {
        vts.advanceTimeBy(Duration.ofMillis(100));
        sink.success(1);
    }).timeout(Duration.ofMillis(1000), vts).block();
    vts.advanceTimeBy(Duration.ofMillis(20000));
    // vts.schedule(() -> { }).dispose(); // uncomment for work-around.
    Mono.create(sink -> {
        vts.advanceTimeBy(Duration.ofMillis(100));
        sink.success(1);
    }).timeout(Duration.ofMillis(1000), vts).block();
}

[lestarcdog]

I have a similar issue but I don't know if they are connected with the previous comment.

The problem is that when using VirtualTimeScheduler#schedule and the Runnable is executed the VirtualTimeScheduler#now(...) does not reflect the previously set date-time value. I would expect it returns the actual virtual time not from the epoch.

Library is being used: reactor-test-3.2.3.RELEASE.jar

public class VirtualSchedulerTest {

  private VirtualTimeScheduler underTest;

  @Before
  public void setUp() throws Exception {
    underTest = VirtualTimeScheduler.create();
  }

  @Test
  public void nowDoesNotReflectVirtualTimeInSchedule() {
    ZonedDateTime start = ZonedDateTime.of(2019, 3, 10, 8, 0, 0, 0, ZoneOffset.UTC);
    underTest.advanceTimeTo(start.toInstant());

    // Good: 2019-03-10T08:00Z
    printSchedulerTime();

    underTest.advanceTimeBy(Duration.ofHours(1));
    // Good: 2019-03-10T09:00Z
    printSchedulerTime();

    // Wrong: 1970-01-01T01:00Z
    // Expected : 2019-03-10T10:00Z
    underTest.schedule(this::printSchedulerTime, 1, TimeUnit.HOURS);
    underTest.advanceTimeBy(Duration.ofHours(1));

    // 2019-03-10T10:00Z
    printSchedulerTime();
  }


  private void printSchedulerTime() {
    ZonedDateTime now = Instant.ofEpochMilli(underTest.now(TimeUnit.MILLISECONDS)).atZone(ZoneOffset.UTC);
    System.out.println(now);
  }
}

[nurkiewicz]

A simpler test case based on @lestarcdog 's sample:

package pl.allegro.badges.bargains.badgeradmin.domain.campaign;

import org.junit.Test;
import reactor.test.scheduler.VirtualTimeScheduler;

import java.time.Duration;
import java.time.Instant;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;

import static java.util.concurrent.TimeUnit.MILLISECONDS;
import static org.junit.Assert.assertEquals;

public class VirtualSchedulerTest {

  @Test
  public void nowDoesNotReflectVirtualTimeInSchedule() {
    VirtualTimeScheduler underTest = VirtualTimeScheduler.create();
    underTest.advanceTimeTo(Instant.ofEpochMilli(0));
    assertEquals(0, underTest.now(MILLISECONDS));

    underTest.advanceTimeBy(Duration.ofSeconds(1));
    assertEquals(1_000, underTest.now(MILLISECONDS));

    AtomicLong inJob = new AtomicLong();
    underTest.schedule(() -> inJob.set(underTest.now(MILLISECONDS)), 1, TimeUnit.SECONDS);

    underTest.advanceTimeBy(Duration.ofSeconds(1));
    assertEquals(2_000, underTest.now(MILLISECONDS));
    assertEquals(2_000, inJob.get());
  }

}

The last assertion fails on 3.1.7 upwards, test passes on 3.1.6 (before deferredNanoTime was introduced).

[yarosla]

Original problem still reproduced as of 3.3.1. Here is the test that does not pass without 'the hack' line:

@Test
public void virtualTimeHack() {
    VirtualTimeScheduler scheduler = VirtualTimeScheduler.create();
    assertEquals(0, scheduler.now(TimeUnit.DAYS));
    scheduler.advanceTimeBy(Duration.ofDays(2000));
    assertEquals(2000, scheduler.now(TimeUnit.DAYS));

    scheduler.schedule(() -> {}); // the hack

    AtomicInteger counter = new AtomicInteger();
    scheduler.schedulePeriodically(counter::incrementAndGet, 1, 1, TimeUnit.DAYS);
    assertEquals(0, counter.get()); // without the hack the actual value is 2000
    scheduler.advanceTimeBy(Duration.ofDays(2));
    assertEquals(2, counter.get()); // without the hack the actual value is 2002
}

The problem is that nanoTime is only advanced in drain() method under if (!queue.isEmpty()) condition. So it does not actually advance until you put something into the drain queue.

[yarosla]

Looks like this issue was introduced intentionally in 4b4f20a

This commit changes the way clock advanceTime is performed when the
VirtualTimeScheduler's task queue is empty, deferring the clock move
until a task is scheduled.

This eliminates ordering issues where a scheduling task (like interval)
would happen AFTER the clock was moved (eg. by a StepVerifier's
thenAwait), resulting in the interval being scheduled too far away in
the "future" and never happening.

Although the fix does not seem to be ideal.

Perhaps just need to document the hack.

[simonbasle]

I suspect this behavior would be important for any case that involves using the VTS in a non-trivial Flux composition.
However perhaps it would help to introduce an overload with a boolean parameter to opt out of the behavior? (ie. implement the "hack" of submitting an empty task when advancing time if the opt-out is activated)? That and better documenting the current behavior...
Wdyt @yarosla ?

[yarosla]

I would add overload with additional boolean defer parameter and replace with it all the invocations from StepVerifier. Then change original method to default to defer=false. This logic would look clearer to me.

Although need to dig further and look through those invocations first.

I can prepare the pull request if you agree.

[simonbasle]

@yarosla yeah let's explore how that looks, please prepare that PR 👍
To sum up:

• VirtualTimeScheduler defaults to defer = false
• StepVerifier.withVirtualTime defaults to defer = true

There is probably a bunch of tests that use StepVerifier but manually create their VirtualTimeScheduler, which might stop working. So be sure to run the full test suite to identify the problematic ones and to ensure switching to an explicit defer = true fixes these.

[simonbasle]

For reference, the idea of putting the parameter at the advanceTimeBy method level doesn't work. That is because it needs to influence the drain() loops - including those that are triggered from other codepath. It would make sense to explore an alternate solution where the defer nature is set upon creation of the VirtualTimeScheduler.

[yarosla]

Created pull request #2017

[simonbasle]

⚠️ behavior change: creating a VirtualTimeScheduler directly won`t defer advanceTime tasks