From 35061f17c16236919c444fb5530fbb7f7aaaa2ff Mon Sep 17 00:00:00 2001 From: Arturo Bernal Date: Thu, 23 Jul 2026 19:32:26 +0200 Subject: [PATCH] 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. --- .../methods/InflatingAsyncDataConsumer.java | 2 +- .../methods/InflatingBrotliDataConsumer.java | 8 +- .../methods/InflatingCapacityChannel.java | 74 ++++++++++++ .../methods/InflatingGzipDataConsumer.java | 2 +- .../methods/InflatingZstdDataConsumer.java | 2 +- .../InflatingBrotliDataConsumerTest.java | 48 ++++++++ .../methods/InflatingCapacityChannelTest.java | 112 ++++++++++++++++++ .../AsyncClientInflateCapacityExample.java | 99 ++++++++++++++++ 8 files changed, 337 insertions(+), 10 deletions(-) create mode 100644 httpclient5/src/main/java/org/apache/hc/client5/http/async/methods/InflatingCapacityChannel.java create mode 100644 httpclient5/src/test/java/org/apache/hc/client5/http/async/methods/InflatingCapacityChannelTest.java create mode 100644 httpclient5/src/test/java/org/apache/hc/client5/http/examples/AsyncClientInflateCapacityExample.java 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 1dad994655..f21d321f45 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 52e8f07c27..8da4815667 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 0000000000..d0e5146693 --- /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 + * . + * + */ +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 27452a596c..9685dbac92 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 3e633bfe96..891e428c00 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 230e0936ec..38afbf5296 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 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 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 increments = new ArrayList<>(); + + @Override + public void update(final int increment) { + increments.add(increment); + } + } + @Test void registerInExec() { final LinkedHashMap> 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 0000000000..8854a4e6a9 --- /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 + * . + * + */ +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 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 0000000000..94cd035a15 --- /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 + * . + * + */ +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. + * + *

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 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.

+ * + *

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.

+ * + * @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 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(); + } + } +}