While running an application that uses a complex combination of delaySuscriptions and buffers, we started getting warnings about HttpClient being probably shared between verticles and some random concurrent modification exceptions that should never happen.
As it turns out, rx observables that were using a specific scheduler where being run ina different event loop than the expected.
Here's a simple reproducer:
public class Test extends AbstractVerticle {
private final String name;
public Test(final String name) {
super();
this.name = name;
}
@SuppressWarnings("uncommentedmain")
public static void main(final String[] args) throws Throwable {
final Vertx vertx = Vertx.vertx();
final RxJavaSchedulersHook hook = RxHelper.schedulerHook(vertx);
RxJavaPlugins.getInstance().registerSchedulersHook(hook);
vertx.deployVerticle(new Test("a"));
vertx.deployVerticle(new Test("b"));
Thread.currentThread().sleep(1000);
vertx.close();
}
@Override
public void start() throws Exception {
log("Verticle on thread %s", Thread.currentThread().getName());
vertx.setTimer(100, this::timer);
}
private void timer(final long id) {
log("Timer exec on thread %s", Thread.currentThread().getName());
Observable.just("1", "2", "3").delaySubscription(20, TimeUnit.MILLISECONDS)
.subscribe(item -> log("iteration %s on thread %s", item, Thread.currentThread().getName()));
}
private void log(final String fmt, final Object... args) {
System.out.println("[" + name + "] " + String.format(fmt, args));
}
}If you run it, it produces something along this lines
[a] Verticle on thread vert.x-eventloop-thread-1
[b] Verticle on thread vert.x-eventloop-thread-3
[a] Timer exec on thread vert.x-eventloop-thread-1
[b] Timer exec on thread vert.x-eventloop-thread-3
[a] iteration 1 on thread vert.x-eventloop-thread-1
[a] iteration 2 on thread vert.x-eventloop-thread-1
[a] iteration 3 on thread vert.x-eventloop-thread-1
[b] iteration 1 on thread vert.x-eventloop-thread-1
[b] iteration 2 on thread vert.x-eventloop-thread-1
[b] iteration 3 on thread vert.x-eventloop-thread-1
Check that while we run each verticle on a different event-loop thread, the suscription is run on the same eventloop.
The problem seems to be on io.vertx.rx.java.ContextScheduler.TimedAction.execute:
private void execute(Object o) {
if (blocking) {
vertx.executeBlocking(this::run, NOOP);
} else {
context.runOnContext(this::run);
}
}As the context is determined when the scheduler is created, it is fixed, and RxHelper.schedulerHook(Vertx) creates the schedulers once and uses them everywhere.
Changing the line context.runOnContext(this::run); to vertx.runOnContext(this::run); fixed my use case, but I'm not sure if there are other cases that break, since the this is the only place where the context field is used.
React
Typescript
Django
Laravel
Facebook
Microsoft
Google
Alibaba
D3
Tencent