Skip to content

Alternative channel flow control implementation - #6462

Draft
vietj wants to merge 6 commits into
masterfrom
message-write-cont
Draft

vietj wants to merge 6 commits into
masterfrom
message-write-cont

Conversation

@vietj

@vietj vietj commented Oct 22, 2026 •

Copy link
Copy Markdown
Member

Motivation

WriteStream flow control relies on writeQueueFull/drainHandler methods, only its usage in a context is race free, there can be several types of races depending on the implementation of the stream, the thread doing the operation, the concurrency, etc...

This aims to improve and replace this API by providing an atomic view of a write operation. A WriteResult API result extending Future<Void> provides an atomic view of the write operation:

  • a writable field signaling whether the channel remains writable (replaces WriteStream#writeQueueFull)
  • a drainHandler signaling when the channel becomes writable again (replaces WriteStream#drainHandler)
  • extends Future providing the outcome of the result operation

Sample usage pattern, implementing a pipe:

void pipe(ReadStream<Buffer> src, WriteStream<Buffer> dst) {
  src.handler(chunk -> {
    WriteResult result = dst.write(chunk);
    if (!result.isWritable()) {
      src.pause();
      result.drainHandler(v -> {
        src.resume();
      });
    }
  });
}

It looks very similar to the actual pipe implementation but is actually race free.

Goals

  • provide a base implementation in vertx-core and gradually to the stack.
  • WriteStream#write returns a WriteResult, so do other write operations like HttpServerResponse#end(Buffer), this is a breaking change and will likely occur in Vert.x 6
  • deprecate WriteStream#drainHandler and `WriteStream#writeQueueFull
  • reimplement flow controlled write using the new API, ReadStream#pipe, RxJava WriteStream support, Reactive Streams bridge.

Todo

  • provide support for correct implementation OutboundWriteQueue
    • signal queue close to write result
    • improve handle drained signal, it should be triggered by element depth crossing the low watermark during drain
  • WriteStream flow control emulation using WriteResult
  • WriteResult API
  • Implementations
    • HTTP/1.1
    • HTTP/2
    • HTTP/3
    • TCP
    • QUIC
    • File system
    • ...

@vietj vietj added this to the 5.3.0 milestone Oct 22, 2026
@vietj
vietj marked this pull request as draft October 22, 2026 18:45
@vietj
vietj force-pushed the message-write-cont branch from dfd90fa to 330ad62 Compare October 22, 2026 18:53
@vietj vietj self-assigned this Oct 22, 2026
@vietj
vietj force-pushed the message-write-cont branch 10 times, most recently from 09dfcee to b4008ff Compare October 28, 2026 20:11
vietj added 5 commits October 28, 2026 22:29
Motivation:

The dispose terminology is reserved for the upcoming per-write drain signaling, where a write result is either drained or disposed. The queue hook releases the messages of a closed queue, as its caller releaseMessages already states.

Changes:

Rename OutboundMessageQueue/InboundMessageQueue handleDispose to handleRelease and adapt the overriders.

Assisted-by: Anthropic Claude (claude-fable-5)
Signed-off-by: Julien Viet <julien@julienviet.com>
Motivation:

The writability signal is edge-triggered (one UNWRITABLE_MASK per crossing): the consumer has no per-element knowledge at drain time to apply a specific treatment based on the queue depth at which an element was added, for example whether it was added at or above the high-water mark. In addition, concurrent adds are published to the queue out of their add order, so the consumer cannot perform add-order accounting from the delivery order.

Modifications:

The wip value packs the add sequence (high 32 bits, incremented by every add) with the queue depth (low 32 bits): a single atomic addition maintains both halves, since the depth is bounded by the queue capacity (no carry) and a consume never exceeds the depth (no borrow). The wip operations reduce to addAndGet and get (the compare-and-set and the increment/decrement are dead code).

add() increments wip before publishing the element, the increment selecting the overflow path, and stamps the element depth and sequence through an injectable contract before the queue add or the overflow store publishes it, so the consumer reads the values the producer observed: the depth for a per-element treatment, the sequence (the add order, a 32-bit modular counter) for add-order accounting such as determining when the queue becomes writable again from the perspective of a given element. The values carried on the element are the sole channel for this knowledge, there is no producer-side return flag. The drain spins when a poll comes up empty, since the wip bound guarantees the element was accounted by a producer that has not yet added it. clear() waits overflow-aware: a racing producer whose increment selected the overflow path publishes to the overflow slot, never to the queue, so a queue-only wait would spin forever.

Stress tests check the stamp correctness and its visibility across the DRAIN_REQUIRED hand-off and under producer-side draining, and that the sequences of all consumed elements are exactly distinct; a deterministic test parks a producer between its increment and its publication and asserts clear() completes.

Result:

The consumer knows, per element at drain time, the queue depth at which the element was added and its add order, in all queue flavors under concurrency.

Assisted-by: Anthropic Claude (claude-fable-5)
Signed-off-by: Julien Viet <julien@julienviet.com>
Motivation:

Polling writeQueueFull after a write is racy: it reads level state the event loop updates concurrently, and the drain handler is a single slot. A write can instead learn whether its element reached the high-water mark and obtain a drain signal for that very element.

Changes:

The verdict is stamped on the element through the MessagePassingQueue depth contract before publication: producer and consumer read the same depth. OutboundMessageQueue#add keeps its counter-based return for writeQueueFull, now incremented from the original add flags (the counter could drift below zero when both water marks are equal).

Introduce OutboundWriteQueue: a drain collects written elements added at or above the high-water mark and releases them per element (fireDrain) at drain end, once the drain emptied the queue or a later element — wrap-aware seq comparison, so a stale concurrent stamp cannot release a batch it predates — was added below the low-water mark. A batch from a refusal-ended drain is held: no wake onto a stuck downstream. Signals fire exactly once, on the consumer thread, after the element was written.

Assisted-by: Anthropic Claude (claude-fable-5)
Signed-off-by: Julien Viet <julien@julienviet.com>
Motivation:

An HTTP write operation requires currently a MessageWrite implementation along with a Promise to report the write result.

We can consolidate both in a single class to save allocations.

Changes:

Introduce a WritePromise object that extends PromiseImpl and implements MessageWrite.

Upon an HTTP write operation, that is passed to the channel and used as return of the write operation.
…ction.

Motivation:

VertxConnection relis on OutboundMessageQueue to relay the channel write operations to the actual Netty channel.

The new OutboundWriteQueue can replace it and enable the return of a WriteResult future.
@vietj
vietj force-pushed the message-write-cont branch from b4008ff to 52667ed Compare October 28, 2026 21:33
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant