Vertx web Route concurrency bug reproducer

README.md

Run wrk to reproduce high load. To use path async_log with incremental request number

wrk -t2 -c100 -d1s -R500 --latency -s request_script.lua http://localhost:8080

MainVerticle.java

import io.vertx.core.AbstractVerticle;
import io.vertx.core.Future;
import io.vertx.core.Vertx;
import io.vertx.ext.web.Router;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class MainVerticle extends AbstractVerticle {

  public final static Logger log = LoggerFactory.getLogger(MainVerticle.class);

  @Override
  public void start(Future<Void> startFuture) throws Exception {
    Router router = Router.router(vertx);
    router.get("/simple").handler(rc -> {
      vertx.executeBlocking(fut -> {
        try {
          Thread.sleep(20); // Simulate a blocking operation that last 20 ms
          fut.complete();
        } catch (InterruptedException e) {
          fut.fail(e);
        }
      }, ar -> {
        if (ar.succeeded()) rc.response().setStatusCode(200).setStatusMessage("OK").end();
        else rc.response().setStatusCode(500).setStatusMessage("Internal Server Error").end(ar.cause().toString());
      });
    });

    router.get("/async").handler(rc -> {
      vertx.setTimer(20, id -> {
        rc.response().setStatusCode(200).setStatusMessage("OK").end();
      });
    });

    router.get("/async_log/:p1").handler(rc -> {
      final String param = rc.pathParam("p1");
      log.info("Starting timer for request {}", param);
      vertx.setTimer(20, id -> {
        log.info("Timer completed for request {}", param);
        rc.response().setStatusCode(200).setStatusMessage("OK").end();
        log.info("Sent response for request {}", param);
      });
    });

    router.get("/multi").blockingHandler(rc -> {
      try {
        Thread.sleep(20); // Simulate a blocking operation that last 20 ms
        rc.next();
      } catch (InterruptedException e) {
        rc.fail(e);
      }
    })
        .handler(rc -> rc.response().setStatusCode(200).setStatusMessage("OK").end())
        .failureHandler(rc -> rc.response().setStatusCode(500).setStatusMessage("Internal Server Error").end(rc.failure().toString()));

    vertx
        .createHttpServer()
        .requestHandler(router)
        .listen(8080, ar -> {
          if (ar.succeeded()) startFuture.complete();
          else startFuture.fail(ar.cause());
        });
  }

  public static void main(String[] args) {
    Vertx vertx = Vertx.vertx();
    vertx.deployVerticle(new MainVerticle());
  }

}

request_script.lua

i = 1

request = function()
  i = i + 1
  param_value = i
  path = "/async_log/" .. param_value
  return wrk.format("GET", path)
end