Fix thread-safety of RouteImpl

[visualage]

When multiple concurrent requests is processed by the same route at the same time, due to the fact that the RouteImpl instance is shared across multiple requests, it creates a thread-unsafe situation. The main cause is that actualHandlerIndex and actualFailureHandlerIndex are states of a request, not the state of the route instance itself.
Moved those two indexes into RoutingContext instead since that is the place where the state belonging to a request lives.

[vietj]

have you observed this issue ?

a route should be single threaded and at most one request at a time should be processed

[visualage]

Not necessarily. The vertx event loop may be single-threaded, but each handler are free to do whatever they want, including spawn a thread, and in that thread, call RoutingContext.next, which in turn increase the RouteImpl.actualHandlerIndex. In our case, we use RxJava 2. One of the handler is Single.fromCallerable(....).subscribeOn(Schedulers.io()).subscribe((v, ex)-> context.next()); This piece of code would call RoutingContext.next() in RxJava’s io threadpool, and is causing racing conditions.

[vietj]

I don't know if the solution is valid or not (I haven't checked) however I'm concerned by the fact that the index seems to be internal and should not be exposed on the RoutingContext interface (i.e the implementation should likely cast to RoutingContextImpl to obtain these)

[visualage]

Agree that ideally we should not expose these as part of the public RoutingContext. I did not spend too much time on reading the code to see if it is OK to cast it to RoutingContextImpl or RoutingContextImplBase.
At least this is a problem we need to find a proper fix. Agree?

[visualage]

I removed the methods from RoutingContext and moved them into RoutingContextImplBase. Sounds reasonable?

[slinkydeveloper]

This is related #729

[slinkydeveloper]

Hi @visualage , I'm the guy that have created this problems 😄 I was working on a fix similar to yours and for me sounds good. I'm writing some tests to check your solution. I update you ASAP

[vietj]

we need tests indeed for this patch

[slinkydeveloper]

Ok so I've created a pr inside @visualage 's fork. It has tests for:

• The multiple connections problem
• The time delayed and multiple connections problem
• And I've also mixed ok requests with failing ones

I've also switched indexes to AtomicInteger to don't occur in other bugs

[visualage]

@slinkydeveloper, your PR is merged into my fork.

[azagniotov]

@slinkydeveloper / @vietj is the current fix going to be a part of 3.5.1 release?

[azagniotov]

Thank you for giving this attention, @slinkydeveloper

[slinkydeveloper]

I've added another test, for me this pr is ok 😄

[sahlone]

@slinkydeveloper Looks like its still not fixed. I am using 3.5.3 with this code asn its still failing with the same error

val server = vertx
                    .createHttpServer()
                server.requestStream().toFlowable().parallel().runOn(rxScheduler).map {
                    router.accept(it.delegate)
                }.sequential().subscribe()

Trace:

j.l.IllegalStateException: Request has already been read
	at i.v.c.h.i.HttpServerRequestImpl.checkEnded(HttpServerRequestImpl.java:438)
	at i.v.c.h.i.HttpServerRequestImpl.handler(HttpServerRequestImpl.java:203)
	at i.v.e.w.i.HttpServerRequestWrapper.handler(HttpServerRequestWrapper.java:39)
	at i.v.e.w.h.i.BodyHandlerImpl.handle(BodyHandlerImpl.java:73)
	at i.v.e.w.h.i.BodyHandlerImpl.handle(BodyHandlerImpl.java:42)
	at i.v.e.w.i.RouteImpl.handleContext(RouteImpl.java:219)
	at i.v.e.w.i.RoutingContextImplBase.iterateNext(RoutingContextImplBase.java:120)
	... 14 common frames omitted
Wrapped by: c.a.c.w.ErrorResponse: Internal Server Error
	at c.a.c.w.ResponseHandlingKt.withFailureHandling(ResponseHandling.kt:123)
	at c.a.c.w.ResponseHandlingKt.access$withFailureHandling(ResponseHandling.kt:1)
	at c.a.c.w.DefaultRouterErrorHandler.handle(ResponseHandling.kt:114)
	at c.a.c.w.DefaultRouterErrorHandler.handle(ResponseHandling.kt:112)
	at i.v.e.w.i.RouteImpl.handleFailure(RouteImpl.java:223)
	at i.v.e.w.i.RoutingContextImplBase.iterateNext(RoutingContextImplBase.java:117)
	at i.v.e.w.i.RoutingContextImpl.next(RoutingContextImpl.java:133)
	at i.v.e.w.i.RoutingContextImpl.doFail(RoutingContextImpl.java:465)
	at i.v.e.w.i.RoutingContextImpl.fail(RoutingContextImpl.java:166)
	at i.v.e.w.i.RoutingContextImplBase.iterateNext(RoutingContextImplBase.java:128)
	at i.v.e.w.i.RoutingContextImpl.next(RoutingContextImpl.java:133)
	at i.v.e.w.i.RouterImpl.accept(RouterImpl.java:79)
	at c.a.s.MasterDataAPIVerticle$2$2.apply(MasterDataAPIVerticle.kt:58)
	at c.a.s.MasterDataAPIVerticle$2$2.apply(MasterDataAPIVerticle.kt:26)
	at i.r.i.o.p.ParallelMap$ParallelMapSubscriber.onNext(ParallelMap.java:113)
	at i.r.i.o.p.ParallelRunOn$RunOnSubscriber.run(ParallelRunOn.java:273)
	at i.v.r.ContextScheduler$ContextWorker$TimedAction.run(ContextScheduler.java:184)
	at i.v.c.i.ContextImpl.lambda$wrapTask$2(ContextImpl.java:339)
	at i.n.u.c.AbstractEventExecutor.safeExecute(AbstractEventExecutor.java:163)
	at i.n.u.c.SingleThreadEventExecutor.r..."}

[pmlopes]

Your exception tells that your body handler is too late to process the request body. Body handling must start running on the same call that received the request. Any async call in between will render this exception. Check the manual for this, either call the handler sooner or pause and resume the request on demand.

[sahlone]

@pmlopes I think you are right about being late for processing the request. I tried pause and resume methodology but didn't work.
Body handling must start running on the same call that received the request did you mean the same thread ?

The intention is to process the request as a stream in parallel and if i go with vertx standards and use the single thread to read the body then I am not able to scale the server and I dont want to go the way of having multiple verticles.
Can you help me in any way to fix this so I can achive the parallelism ?

I think its still the multi threading issue with router implementation , I say so because if I replace the router with my simple implementation then the above code works fine.