You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hc.apache.org by ab...@apache.org on 2026/07/25 10:43:01 UTC
(httpcomponents-client) branch master updated: HTTPCLIENT-2428: Scale reactor capacity in inflating AsyncDataConsumers to match decompressed output
This is an automated email from the ASF dual-hosted git repository.
arturobernalg pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/httpcomponents-client.git
The following commit(s) were added to refs/heads/master by this push:
new 88c62e7cf HTTPCLIENT-2428: Scale reactor capacity in inflating AsyncDataConsumers to match decompressed output
88c62e7cf is described below
commit 88c62e7cfc680ad83096f0165c018272abd28c4b
Author: Arturo Bernal <ab...@apache.org>
AuthorDate: Thu Jul 23 19:32:26 2026 +0200
HTTPCLIENT-2428: Scale reactor capacity in inflating AsyncDataConsumers to match decompressed output
Inflating consumers now request proportionally fewer compressed bytes than the downstream consumer's advertised (decompressed) capacity, so decompression no longer overshoots it; InflatingBrotliDataConsumer also stops overriding the downstream capacity with Integer.MAX_VALUE.
---
.../async/methods/InflatingAsyncDataConsumer.java | 2 +-
.../async/methods/InflatingBrotliDataConsumer.java | 8 +-
.../async/methods/InflatingCapacityChannel.java | 74 ++++++++++++++
.../async/methods/InflatingGzipDataConsumer.java | 2 +-
.../async/methods/InflatingZstdDataConsumer.java | 2 +-
.../methods/InflatingBrotliDataConsumerTest.java | 48 +++++++++
.../methods/InflatingCapacityChannelTest.java | 112 +++++++++++++++++++++
.../AsyncClientInflateCapacityExample.java | 99 ++++++++++++++++++
8 files changed, 337 insertions(+), 10 deletions(-)
diff --git a/httpclient5/src/main/java/org/apache/hc/client5/http/async/methods/InflatingAsyncDataConsumer.java b/httpclient5/src/main/java/org/apache/hc/client5/http/async/methods/InflatingAsyncDataConsumer.java
index 1dad99465..f21d321f4 100644
--- a/httpclient5/src/main/java/org/apache/hc/client5/http/async/methods/InflatingAsyncDataConsumer.java
+++ b/httpclient5/src/main/java/org/apache/hc/client5/http/async/methods/InflatingAsyncDataConsumer.java
@@ -73,7 +73,7 @@ public InflatingAsyncDataConsumer(
@Override
public void updateCapacity(final CapacityChannel ch) throws IOException {
- downstream.updateCapacity(ch);
+ downstream.updateCapacity(new InflatingCapacityChannel(ch));
}
@Override
diff --git a/httpclient5/src/main/java/org/apache/hc/client5/http/async/methods/InflatingBrotliDataConsumer.java b/httpclient5/src/main/java/org/apache/hc/client5/http/async/methods/InflatingBrotliDataConsumer.java
index 52e8f07c2..8da481566 100644
--- a/httpclient5/src/main/java/org/apache/hc/client5/http/async/methods/InflatingBrotliDataConsumer.java
+++ b/httpclient5/src/main/java/org/apache/hc/client5/http/async/methods/InflatingBrotliDataConsumer.java
@@ -67,7 +67,6 @@ public final class InflatingBrotliDataConsumer implements AsyncDataConsumer {
private final AsyncDataConsumer downstream;
private final DecoderJNI.Wrapper decoder;
- private volatile CapacityChannel capacity;
public InflatingBrotliDataConsumer(final AsyncDataConsumer downstream) {
@@ -81,8 +80,7 @@ public InflatingBrotliDataConsumer(final AsyncDataConsumer downstream) {
@Override
public void updateCapacity(final CapacityChannel capacityChannel) throws IOException {
- this.capacity = capacityChannel;
- downstream.updateCapacity(capacityChannel);
+ downstream.updateCapacity(new InflatingCapacityChannel(capacityChannel));
}
@Override
@@ -103,10 +101,6 @@ public void consume(final ByteBuffer src) throws IOException {
decoder.push(xfer);
pump();
}
- final CapacityChannel ch = this.capacity;
- if (ch != null) {
- ch.update(Integer.MAX_VALUE);
- }
}
@Override
diff --git a/httpclient5/src/main/java/org/apache/hc/client5/http/async/methods/InflatingCapacityChannel.java b/httpclient5/src/main/java/org/apache/hc/client5/http/async/methods/InflatingCapacityChannel.java
new file mode 100644
index 000000000..d0e514669
--- /dev/null
+++ b/httpclient5/src/main/java/org/apache/hc/client5/http/async/methods/InflatingCapacityChannel.java
@@ -0,0 +1,74 @@
+/*
+ * ====================================================================
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ * ====================================================================
+ *
+ * This software consists of voluntary contributions made by many
+ * individuals on behalf of the Apache Software Foundation. For more
+ * information on the Apache Software Foundation, please see
+ * <http://www.apache.org/>.
+ *
+ */
+package org.apache.hc.client5.http.async.methods;
+
+import java.io.IOException;
+
+import org.apache.hc.core5.http.nio.CapacityChannel;
+import org.apache.hc.core5.util.Args;
+
+/**
+ * {@link CapacityChannel} decorator used by the inflating {@code AsyncDataConsumer}s. A downstream
+ * consumer advertises capacity in decompressed bytes, but the increment is applied to the compressed
+ * stream read from the I/O reactor. Forwarding it unchanged lets the reactor deliver that many
+ * compressed bytes, which decompress into several times as many bytes and overshoot the downstream
+ * consumer's advertised capacity. This decorator scales the requested increment down by a fixed
+ * expansion factor so the amount of decompressed data delivered per round stays closer to what the
+ * downstream consumer asked for.
+ */
+final class InflatingCapacityChannel implements CapacityChannel {
+
+ /**
+ * Assumed decompressed-to-compressed size ratio. A conservative estimate for typical HTTP
+ * payloads; a larger value trades throughput for tighter capacity predictability.
+ */
+ static final int DEFAULT_EXPANSION_FACTOR = 4;
+
+ private final CapacityChannel delegate;
+ private final int expansionFactor;
+
+ InflatingCapacityChannel(final CapacityChannel delegate) {
+ this(delegate, DEFAULT_EXPANSION_FACTOR);
+ }
+
+ InflatingCapacityChannel(final CapacityChannel delegate, final int expansionFactor) {
+ this.delegate = Args.notNull(delegate, "Capacity channel");
+ this.expansionFactor = Args.positive(expansionFactor, "Expansion factor");
+ }
+
+ @Override
+ public void update(final int increment) throws IOException {
+ // Integer.MAX_VALUE is the idiomatic "unbounded" capacity request; forward it (and any
+ // non-positive value) unchanged so scaling does not turn "no limit" into a finite bound.
+ if (increment <= 0 || increment == Integer.MAX_VALUE) {
+ delegate.update(increment);
+ return;
+ }
+ // Request at least one byte so a small increment still lets the stream make progress.
+ delegate.update(Math.max(1, increment / expansionFactor));
+ }
+}
diff --git a/httpclient5/src/main/java/org/apache/hc/client5/http/async/methods/InflatingGzipDataConsumer.java b/httpclient5/src/main/java/org/apache/hc/client5/http/async/methods/InflatingGzipDataConsumer.java
index 27452a596..9685dbac9 100644
--- a/httpclient5/src/main/java/org/apache/hc/client5/http/async/methods/InflatingGzipDataConsumer.java
+++ b/httpclient5/src/main/java/org/apache/hc/client5/http/async/methods/InflatingGzipDataConsumer.java
@@ -71,7 +71,7 @@ public InflatingGzipDataConsumer(final AsyncDataConsumer downstream) {
@Override
public void updateCapacity(final CapacityChannel c) throws IOException {
- downstream.updateCapacity(c);
+ downstream.updateCapacity(new InflatingCapacityChannel(c));
}
@Override
diff --git a/httpclient5/src/main/java/org/apache/hc/client5/http/async/methods/InflatingZstdDataConsumer.java b/httpclient5/src/main/java/org/apache/hc/client5/http/async/methods/InflatingZstdDataConsumer.java
index 3e633bfe9..891e428c0 100644
--- a/httpclient5/src/main/java/org/apache/hc/client5/http/async/methods/InflatingZstdDataConsumer.java
+++ b/httpclient5/src/main/java/org/apache/hc/client5/http/async/methods/InflatingZstdDataConsumer.java
@@ -81,7 +81,7 @@ public InflatingZstdDataConsumer(final AsyncDataConsumer downstream) {
@Override
public void updateCapacity(final CapacityChannel c) throws IOException {
- downstream.updateCapacity(c);
+ downstream.updateCapacity(new InflatingCapacityChannel(c));
}
@Override
diff --git a/httpclient5/src/test/java/org/apache/hc/client5/http/async/methods/InflatingBrotliDataConsumerTest.java b/httpclient5/src/test/java/org/apache/hc/client5/http/async/methods/InflatingBrotliDataConsumerTest.java
index 230e0936e..38afbf529 100644
--- a/httpclient5/src/test/java/org/apache/hc/client5/http/async/methods/InflatingBrotliDataConsumerTest.java
+++ b/httpclient5/src/test/java/org/apache/hc/client5/http/async/methods/InflatingBrotliDataConsumerTest.java
@@ -34,6 +34,7 @@
import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
import java.util.Collections;
import java.util.HashSet;
import java.util.LinkedHashMap;
@@ -191,6 +192,53 @@ public Set<String> getTrailerNames() {
assertEquals(ORIGINAL, inner.getContent(), "br inflate mismatch");
}
+ @Test
+ void consumeDoesNotOverrideDownstreamCapacity() throws Exception {
+ final byte[] compressed = brCompress();
+ final RecordingCapacityChannel reactor = new RecordingCapacityChannel();
+ final AsyncDataConsumer downstream = new AsyncDataConsumer() {
+ @Override
+ public void updateCapacity(final CapacityChannel ch) throws IOException {
+ ch.update(4096);
+ }
+
+ @Override
+ public void consume(final ByteBuffer src) {
+ }
+
+ @Override
+ public void streamEnd(final List<? extends Header> trailers) {
+ }
+
+ @Override
+ public void releaseResources() {
+ }
+ };
+ final InflatingBrotliDataConsumer inflating = new InflatingBrotliDataConsumer(downstream);
+
+ // The downstream advertises 4096 decompressed bytes, so the reactor must receive a
+ // scaled-down (4096 / 4 = 1024) increment, and consume() must never override it with
+ // Integer.MAX_VALUE as the previous implementation did.
+ inflating.updateCapacity(reactor);
+ for (int off = 0; off < compressed.length; off += 1024) {
+ final int n = Math.min(1024, compressed.length - off);
+ inflating.consume(ByteBuffer.wrap(compressed, off, n));
+ }
+
+ assertEquals(1, reactor.increments.size(), "consume() must not push extra capacity updates");
+ assertEquals(1024, reactor.increments.get(0).intValue());
+ }
+
+ private static final class RecordingCapacityChannel implements CapacityChannel {
+
+ final List<Integer> increments = new ArrayList<>();
+
+ @Override
+ public void update(final int increment) {
+ increments.add(increment);
+ }
+ }
+
@Test
void registerInExec() {
final LinkedHashMap<String, UnaryOperator<AsyncDataConsumer>> map = new LinkedHashMap<>();
diff --git a/httpclient5/src/test/java/org/apache/hc/client5/http/async/methods/InflatingCapacityChannelTest.java b/httpclient5/src/test/java/org/apache/hc/client5/http/async/methods/InflatingCapacityChannelTest.java
new file mode 100644
index 000000000..8854a4e6a
--- /dev/null
+++ b/httpclient5/src/test/java/org/apache/hc/client5/http/async/methods/InflatingCapacityChannelTest.java
@@ -0,0 +1,112 @@
+/*
+ * ====================================================================
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ * ====================================================================
+ *
+ * This software consists of voluntary contributions made by many
+ * individuals on behalf of the Apache Software Foundation. For more
+ * information on the Apache Software Foundation, please see
+ * <http://www.apache.org/>.
+ *
+ */
+package org.apache.hc.client5.http.async.methods;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import org.apache.hc.core5.http.nio.CapacityChannel;
+import org.junit.jupiter.api.Test;
+
+class InflatingCapacityChannelTest {
+
+ private static final class RecordingCapacityChannel implements CapacityChannel {
+
+ final List<Integer> increments = new ArrayList<>();
+
+ @Override
+ public void update(final int increment) {
+ increments.add(increment);
+ }
+ }
+
+ @Test
+ void scalesDownByDefaultExpansionFactor() throws Exception {
+ final RecordingCapacityChannel delegate = new RecordingCapacityChannel();
+ final CapacityChannel channel = new InflatingCapacityChannel(delegate);
+
+ channel.update(400);
+
+ assertEquals(1, delegate.increments.size());
+ assertEquals(400 / InflatingCapacityChannel.DEFAULT_EXPANSION_FACTOR,
+ delegate.increments.get(0).intValue());
+ }
+
+ @Test
+ void scalesDownByExplicitExpansionFactor() throws Exception {
+ final RecordingCapacityChannel delegate = new RecordingCapacityChannel();
+ final CapacityChannel channel = new InflatingCapacityChannel(delegate, 8);
+
+ channel.update(800);
+
+ assertEquals(100, delegate.increments.get(0).intValue());
+ }
+
+ @Test
+ void smallIncrementStillRequestsAtLeastOneByte() throws Exception {
+ final RecordingCapacityChannel delegate = new RecordingCapacityChannel();
+ final CapacityChannel channel = new InflatingCapacityChannel(delegate, 4);
+
+ channel.update(2);
+
+ assertEquals(1, delegate.increments.get(0).intValue());
+ }
+
+ @Test
+ void nonPositiveIncrementIsForwardedUnchanged() throws Exception {
+ final RecordingCapacityChannel delegate = new RecordingCapacityChannel();
+ final CapacityChannel channel = new InflatingCapacityChannel(delegate, 4);
+
+ channel.update(0);
+
+ assertEquals(0, delegate.increments.get(0).intValue());
+ }
+
+ @Test
+ void unboundedCapacityIsForwardedUnchanged() throws Exception {
+ final RecordingCapacityChannel delegate = new RecordingCapacityChannel();
+ final CapacityChannel channel = new InflatingCapacityChannel(delegate, 4);
+
+ channel.update(Integer.MAX_VALUE);
+
+ assertEquals(Integer.MAX_VALUE, delegate.increments.get(0).intValue());
+ }
+
+ @Test
+ void rejectsNonPositiveExpansionFactor() {
+ assertThrows(IllegalArgumentException.class,
+ () -> new InflatingCapacityChannel(new RecordingCapacityChannel(), 0));
+ }
+
+ @Test
+ void rejectsNullDelegate() {
+ assertThrows(NullPointerException.class, () -> new InflatingCapacityChannel(null));
+ }
+}
diff --git a/httpclient5/src/test/java/org/apache/hc/client5/http/examples/AsyncClientInflateCapacityExample.java b/httpclient5/src/test/java/org/apache/hc/client5/http/examples/AsyncClientInflateCapacityExample.java
new file mode 100644
index 000000000..94cd035a1
--- /dev/null
+++ b/httpclient5/src/test/java/org/apache/hc/client5/http/examples/AsyncClientInflateCapacityExample.java
@@ -0,0 +1,99 @@
+/*
+ * ====================================================================
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ * ====================================================================
+ *
+ * This software consists of voluntary contributions made by many
+ * individuals on behalf of the Apache Software Foundation. For more
+ * information on the Apache Software Foundation, please see
+ * <http://www.apache.org/>.
+ *
+ */
+package org.apache.hc.client5.http.examples;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.List;
+
+import org.apache.hc.client5.http.async.methods.InflatingAsyncDataConsumer;
+import org.apache.hc.core5.http.Header;
+import org.apache.hc.core5.http.nio.AsyncDataConsumer;
+import org.apache.hc.core5.http.nio.CapacityChannel;
+
+/**
+ * Demonstrates how the inflating {@code AsyncDataConsumer}s adjust reactor back-pressure.
+ *
+ * <p>A downstream consumer advertises capacity in <em>decompressed</em> bytes, but the increment is
+ * applied to the <em>compressed</em> stream read from the I/O reactor. Forwarding it unchanged would
+ * let the reactor deliver that many compressed bytes, which decompress into several times as many
+ * bytes and overshoot the downstream consumer's advertised capacity. The inflating consumer
+ * therefore scales the requested increment down before handing it to the reactor.</p>
+ *
+ * <p>This example wires an {@link InflatingAsyncDataConsumer} in front of a downstream consumer that
+ * requests 64 KiB of decompressed capacity and prints the (smaller) increment that reaches the
+ * reactor's {@link CapacityChannel}. No network access is required.</p>
+ *
+ * @since 5.7
+ */
+public class AsyncClientInflateCapacityExample {
+
+ public static void main(final String[] args) throws Exception {
+
+ // Stands in for the I/O reactor: reports the number of compressed bytes actually requested.
+ final CapacityChannel reactorChannel = new CapacityChannel() {
+
+ @Override
+ public void update(final int increment) {
+ System.out.println("Reactor asked for " + increment + " compressed bytes");
+ }
+ };
+
+ // A downstream consumer that, once it is handed a channel, asks for 64 KiB of
+ // decompressed capacity.
+ final AsyncDataConsumer downstream = new AsyncDataConsumer() {
+
+ @Override
+ public void updateCapacity(final CapacityChannel channel) throws IOException {
+ final int decompressedCapacity = 64 * 1024;
+ System.out.println("Downstream requested " + decompressedCapacity + " decompressed bytes");
+ channel.update(decompressedCapacity);
+ }
+
+ @Override
+ public void consume(final ByteBuffer src) {
+ }
+
+ @Override
+ public void streamEnd(final List<? extends Header> trailers) {
+ }
+
+ @Override
+ public void releaseResources() {
+ }
+ };
+
+ final InflatingAsyncDataConsumer inflating = new InflatingAsyncDataConsumer(downstream, null);
+ try {
+ // The reactor hands its capacity channel to the inflating consumer, which forwards a
+ // scaled-down wrapper to the downstream consumer.
+ inflating.updateCapacity(reactorChannel);
+ } finally {
+ inflating.releaseResources();
+ }
+ }
+}