Provide a better event-loop to context message transfer queue

[vietj]

The Vert.x event-loop thread to context thread message transfer relies on the InboundBuffer implementation.

InboundBuffer performs the add/dispatch in the same method call assuming that the same thread actually handles the dispatch of the message, forcing the dispatch to then schedule the message delivery on the context thread.

Inbound read queue

The InboundReadQueue design actually split this operation in two separate operations providing control to the caller of the message dispatch.

• the add operation queues a message and let the producer knows whether messages should be drained from the queue (e.g. if a drain operation is already in progress, then there is no need to schedule another drain).
• the drain operation let consumer consume messages from the queue until needed

The event-loop to context message dispatch then becomes:

1. add the message to the queue
2. ping the context thread to drain the queue

Event-loop thread dispatch

In this use case, the InboundReadQueue assumes the same thread is producing/consuming messages and therefore no memory visibility is actually required. In practice the event-loop add to the queue and then drain delivers the message to the connection.

Generic context thread dispatch

In this use case, the queue will be drained by the context thread and an SPSC + volatile WIP is used. This behaviour also can optimise the message delivery since we don't need anymore a message received ⇒ scheduling a task, the event-loop thread and the context thread can work in an SPSC consumer design.

This use case holds for:

• another event-loop thread (pooled HTTP client connection)
• worker thread
• virtual thread

Back pressure

The context thread message deals controls the back-pressure and cannot control the inbound channel back-pressure without races. Like Like the OutboundWriteQueue, the InboundReadQueue relies on an internal buffer and the queue acts as an intermediary for back-pressure.

• the producer add operation signals when the queue becomes un-writable (e.g. it turns off Netty auto-read)
• the consumer consumes messages, when messages are consumed it is responsible to signal the queue is writable again

Inbound message queue

The InboundMessageQueue is a construct integrating the InboundReadQueue with the Vert.x context and a demand counter (ReadStream). The consumer deals with pause/resume/fetch to control demand, the producer implements the message flow pause/resume. This queue is the InboundBuffer replacement.

ConnectionBase message flow changes

doPause/doResume of ConnectionBase has been rewritten to be strict concerning the delivery of messages. Previously these operations were turning on/off channel auto read, however this was not controlling the reads in progress. This changes these operations to control a paused flag and buffer any messages to be handled by the connection base when paused.

This change has been made to let the Http1xServerConnection control precisely the message flow when processing an HTTP pipelined request content. Previously the content was poured in the pending queue of the pipelined request which is complicated to deliver when the pipelined request is processed and requires some hacks with respect to the request demand. After this change, Http1xServerConnection can immediately pause the connection after receiving a pipelined request and have the guarantee that no message will be processed until it is resumed. When the pipelined request is processed, the connection is resumed and will deliver the messages with the regular handleMessage flow.

Performance

Plaintext

• master : 4.5.7 baseline
• 5.0.0 : 5.0.0 baseline
• 5.0.0-irq : this PR branch
• 5.0.0-irq-nv : this PR branch using non volatile but synchronised inbound buffer demand (relying on biased locking)

Microbenchmark

> mvn clean package -DskipTests -Pbenchmarks
> java -jar target/vertx-core-5.0.0-SNAPSHOT-benchmarks.jar HttpServerHandlerBenchmark

InboundMessageQueue (synchronised long)

Benchmark                                      Mode  Cnt     Score    Error   Units
HttpServerHandlerBenchmark.netty              thrpt   10  2398.380 ± 19.954  ops/ms
HttpServerHandlerBenchmark.vertx              thrpt   10  1769.388 ±  6.817  ops/ms
HttpServerHandlerBenchmark.vertxOpt           thrpt   10  2241.688 ±  4.130  ops/ms
HttpServerHandlerBenchmark.vertxOptMetricsOn  thrpt   10  1906.300 ± 28.674  ops/ms

InboundMessageQueue (volatile long)

Benchmark                                      Mode  Cnt     Score    Error   Units
HttpServerHandlerBenchmark.netty              thrpt   10  2445.317 ± 17.642  ops/ms
HttpServerHandlerBenchmark.vertx              thrpt   10  1976.744 ±  7.916  ops/ms
HttpServerHandlerBenchmark.vertxOpt           thrpt   10  1951.065 ±  4.861  ops/ms
HttpServerHandlerBenchmark.vertxOptMetricsOn  thrpt   10  2192.769 ± 26.459  ops/ms

InboundBuffer (master)

Benchmark                                      Mode  Cnt     Score    Error   Units
HttpServerHandlerBenchmark.netty              thrpt   10  2423.540 ± 35.876  ops/ms
HttpServerHandlerBenchmark.vertx              thrpt   10  1860.064 ± 16.155  ops/ms
HttpServerHandlerBenchmark.vertxOpt           thrpt   10  2209.176 ± 11.710  ops/ms
HttpServerHandlerBenchmark.vertxOptMetricsOn  thrpt   10  1896.875 ± 14.296  ops/ms

InboundBuffer (4.x)

Benchmark                                      Mode  Cnt     Score    Error   Units
HttpServerHandlerBenchmark.netty              thrpt   10  2368.750 ± 30.001  ops/ms
HttpServerHandlerBenchmark.vertx              thrpt   10  1892.064 ± 21.788  ops/ms
HttpServerHandlerBenchmark.vertxOpt           thrpt   10  2097.548 ±  9.652  ops/ms
HttpServerHandlerBenchmark.vertxOptMetricsOn  thrpt   10  2187.718 ± 10.779  ops/ms

Integration with Vert.x Core

This integrates with

• HTTP/1.1 client/server chunk
• NetSocket
• WebSocket

Past attempts to solve this:

• Inbound message composite bytebuf #5177
• Inbound read queue work + usage in Http1xServerRequest #5164

[vietj]

@franz1981 @jponge mostly ready for review

[vietj]

This is now ready for review @tsegismont @franz1981 @jponge @cescoffier and anyone else who wants to review this

[jponge]

(still on my todo)

[vietj]

@jponge ok

[jponge]

You might want to handle 0 as well and reject it.

[jponge]

(or accept it in the other branch, but in Reactive Streams land 0 is not accepted)

[vietj]

I think it's fine to tolerate 0 is a no-op, any reactive-streams implementation that uses it will never pass 0