Skip to content

MultiBufferEntity add loop to produce content. #1366

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
l-trotta merged 7 commits into from
Nov 5, 2026
Merged

MultiBufferEntity add loop to produce content. #1366

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
bce10bb
keep feeding the encoder while it accepts bytes in multibuffer entity
l-trotta Sep 17, 2026
e8e46e8
rest5 MultiBufferEntity: coalesce small buffers into a 16 KiB staging…
l-trotta Sep 17, 2026
09095a9
avoid buffering single body requests in staging buffer
l-trotta Nov 5, 2026
5a7db31
better explain rest5 mbe produce
l-trotta Nov 5, 2026
6d88a18
unit tests
l-trotta Nov 5, 2026
31067b2
apply same staging buffer logic to legacy rest client
l-trotta Nov 5, 2026
3f2e4b9
rest5 MultiBufferEntityTest: use a fresh entity per retry attempt
l-trotta Nov 5, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
93 changes: 77 additions & 16 deletions java-client/src/main/java/co/elastic/clients/transport/rest5_client/MultiBufferEntity.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -39,10 +39,20 @@
*/
class MultiBufferEntity extends AbstractHttpEntity implements AsyncEntityProducer {

/**
* Small input buffers are coalesced up to this size before writing. Matches the default HTTP/2
* frame payload size (RFC 9113 §4.2), since each {@link DataStreamChannel#write} becomes one frame.
* The staging buffer is never allocated for single-buffer bodies, such as most non-bulk requests.
*/
static final int STAGING_BUFFER_SIZE = 16 * 1024;

private final Iterable<ByteBuffer> buffers;

private Iterator<ByteBuffer> iterator;
/** Input buffer currently being consumed, {@code null} once the input is exhausted. Never empty. */
private volatile ByteBuffer currentBuffer;
/** Coalesced bytes not yet accepted by the channel, always in read mode. Allocated on first use. */
private volatile ByteBuffer staging;

private final AtomicReference<Exception> exceptionRef;

Expand All @@ -61,11 +71,24 @@ public void close() throws IOException {

private void init() {
this.iterator = buffers.iterator();
if (this.iterator.hasNext()) {
this.currentBuffer = this.iterator.next().duplicate();
} else {
this.currentBuffer = null;
this.currentBuffer = null;
advance();
ByteBuffer staging = this.staging;
if (staging != null) {
staging.position(0).limit(0);
}
}

/** Moves {@link #currentBuffer} to the next non-empty input buffer, or {@code null} if there is none. */
private void advance() {
while (iterator.hasNext()) {
ByteBuffer next = iterator.next();
if (next.hasRemaining()) {
currentBuffer = next.duplicate();
return;
}
}
currentBuffer = null;
}

@Override
Expand Down Expand Up @@ -108,26 +131,64 @@ public void writeTo(OutputStream out) throws IOException {

@Override
public int available() {
return currentBuffer.remaining();
ByteBuffer staging = this.staging;
ByteBuffer current = this.currentBuffer;
return (staging == null ? 0 : staging.remaining()) + (current == null ? 0 : current.remaining());
}

@Override
public void produce(DataStreamChannel channel) throws IOException {
if (currentBuffer == null) {
channel.endStream();
return;
}
// Each write() is one HTTP/2 frame or HTTP/1.1 chunk, so small buffers are coalesced. Each iteration:
// 1. Staging has unsent bytes: send those first, they come before anything in currentBuffer.
// 2. Staging is empty and there is no current buffer: the body is complete, end the stream.
// 3. The current buffer is large, or it's the last one: write it directly, copying is useless.
// 4. Otherwise small buffers follow: coalesce them into staging and write that.
// Loops while the channel accepts everything, returns on a refused or partial write to resume later.
while (true) {
ByteBuffer out;
if (staging != null && staging.hasRemaining()) {
out = staging;
} else if (currentBuffer == null) {
channel.endStream();
return;
} else if (currentBuffer.remaining() >= STAGING_BUFFER_SIZE || !iterator.hasNext()) {
out = currentBuffer;
} else {
out = fillStaging();
}

channel.write(currentBuffer);
channel.write(out);

if (!currentBuffer.hasRemaining()) {
if (iterator.hasNext()) {
currentBuffer = iterator.next().duplicate();
} else {
currentBuffer = null;
channel.endStream();
if (out == currentBuffer && !out.hasRemaining()) {
advance();
}
if (out.hasRemaining()) {
return;
}
}
}

/** Copies as much input as fits into the (empty) staging buffer and returns it, ready to be read. */
private ByteBuffer fillStaging() {
ByteBuffer staging = this.staging;
if (staging == null) {
staging = ByteBuffer.allocate(STAGING_BUFFER_SIZE);
this.staging = staging;
}
staging.clear();
while (currentBuffer != null && staging.hasRemaining()) {
ByteBuffer src = currentBuffer;
int n = Math.min(src.remaining(), staging.remaining());
int srcLimit = src.limit();
src.limit(src.position() + n);
staging.put(src);
src.limit(srcLimit);
if (!src.hasRemaining()) {
advance();
}
}
staging.flip();
return staging;
}

@Override
Expand Down
92 changes: 76 additions & 16 deletions java-client/src/main/java/co/elastic/clients/transport/rest_client/MultiBufferEntity.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -39,10 +39,19 @@
*/
class MultiBufferEntity extends AbstractHttpEntity implements HttpAsyncContentProducer {

/**
* Small input buffers are coalesced up to this size before writing, since each write is one HTTP chunk.
* The staging buffer is never allocated for single-buffer bodies, such as most non-bulk requests.
*/
static final int STAGING_BUFFER_SIZE = 16 * 1024;

private final Iterable<ByteBuffer> buffers;

private Iterator<ByteBuffer> iterator;
/** Input buffer currently being consumed, {@code null} once the input is exhausted. Never empty. */
private volatile ByteBuffer currentBuffer;
/** Coalesced bytes not yet accepted by the encoder, always in read mode. Allocated on first use. */
private volatile ByteBuffer staging;

MultiBufferEntity(Iterable<ByteBuffer> buffers, ContentType contentType) {
this.buffers = buffers;
Expand All @@ -61,12 +70,27 @@ public void close() throws IOException {

private void init() {
Iterator<ByteBuffer> localIterator = this.buffers.iterator();
if (localIterator.hasNext()) {
this.currentBuffer = localIterator.next().duplicate();
} else {
this.currentBuffer = null;
}
this.currentBuffer = nextNonEmpty(localIterator);
this.iterator = localIterator;
ByteBuffer staging = this.staging;
if (staging != null) {
staging.position(0).limit(0);
}
}

/** Moves {@link #currentBuffer} to the next non-empty input buffer, or {@code null} if there is none. */
private void advance() {
currentBuffer = nextNonEmpty(iterator);
}

private static ByteBuffer nextNonEmpty(Iterator<ByteBuffer> iterator) {
while (iterator.hasNext()) {
ByteBuffer next = iterator.next();
if (next.hasRemaining()) {
return next.duplicate();
}
}
return null;
}

@Override
Expand Down Expand Up @@ -102,20 +126,56 @@ public void writeTo(OutputStream out) throws IOException {

@Override
public void produceContent(ContentEncoder encoder, IOControl ioControl) throws IOException {
if (currentBuffer == null) {
encoder.complete();
return;
}
// Each write() is one HTTP chunk, so small buffers are coalesced. Each iteration:
// 1. Staging has unsent bytes: send those first, they come before anything in currentBuffer.
// 2. Staging is empty and there is no current buffer: the body is complete, complete the encoder.
// 3. The current buffer is large, or it's the last one: write it directly, copying is useless.
// 4. Otherwise small buffers follow: coalesce them into staging and write that.
// Loops while the encoder accepts everything, returns on a refused or partial write to resume later.
while (true) {
ByteBuffer out;
if (staging != null && staging.hasRemaining()) {
out = staging;
} else if (currentBuffer == null) {
encoder.complete();
return;
} else if (currentBuffer.remaining() >= STAGING_BUFFER_SIZE || !iterator.hasNext()) {
out = currentBuffer;
} else {
out = fillStaging();
}

encoder.write(currentBuffer);
encoder.write(out);

if (!currentBuffer.hasRemaining()) {
if (iterator.hasNext()) {
currentBuffer = iterator.next().duplicate();
} else {
currentBuffer = null;
encoder.complete();
if (out == currentBuffer && !out.hasRemaining()) {
advance();
}
if (out.hasRemaining()) {
return;
}
}
}

/** Copies as much input as fits into the (empty) staging buffer and returns it, ready to be read. */
private ByteBuffer fillStaging() {
ByteBuffer staging = this.staging;
if (staging == null) {
staging = ByteBuffer.allocate(STAGING_BUFFER_SIZE);
this.staging = staging;
}
staging.clear();
while (currentBuffer != null && staging.hasRemaining()) {
ByteBuffer src = currentBuffer;
int n = Math.min(src.remaining(), staging.remaining());
int srcLimit = src.limit();
src.limit(src.position() + n);
staging.put(src);
src.limit(srcLimit);
if (!src.hasRemaining()) {
advance();
}
}
staging.flip();
return staging;
}
}
Loading
Loading