Skip to content

Commit a12ef5b

Browse files
arturobernalgarturobernalg
committed
Add H2 multiplexing connection pool
Implement an HTTP/2 stream-capacity-aware connection pool with per-route and global connection limits, reservation-aware leasing, GOAWAY draining, disconnect handling, and route-level wakeup logic. Optimize uncontended lease paths and add coverage for the main pooling and lifecycle edge cases.
1 parent 35e7d53 commit a12ef5b

22 files changed

Lines changed: 4023 additions & 123 deletions

‎httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/AbstractH2IOEventHandler.java‎

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -47,7 +47,16 @@ class AbstractH2IOEventHandler implements HttpConnectionEventHandler {
4747
final AbstractH2StreamMultiplexer streamMultiplexer;
4848

4949
AbstractH2IOEventHandler(final AbstractH2StreamMultiplexer streamMultiplexer) {
50+
this(streamMultiplexer, null);
51+
}
52+
53+
AbstractH2IOEventHandler(
54+
final AbstractH2StreamMultiplexer streamMultiplexer,
55+
final H2PoolSessionSupport sessionSupport) {
5056
this.streamMultiplexer = Args.notNull(streamMultiplexer, "Stream multiplexer");
57+
if (sessionSupport != null) {
58+
streamMultiplexer.setPoolSessionSupport(sessionSupport);
59+
}
5160
}
5261

5362
@Override

‎httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/AbstractH2StreamMultiplexer.java‎

Lines changed: 60 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -143,6 +143,7 @@ enum SettingsHandshake { READY, TRANSMITTED, ACKED }
143143

144144
private EndpointDetails endpointDetails;
145145
private boolean goAwayReceived;
146+
private volatile H2PoolSessionSupport poolSessionSupport;
146147

147148
private volatile boolean peerNoRfc7540Priorities;
148149

@@ -193,6 +194,7 @@ enum SettingsHandshake { READY, TRANSMITTED, ACKED }
193194
this.frameFactory = Args.notNull(frameFactory, "Frame factory");
194195
this.httpProcessor = Args.notNull(httpProcessor, "HTTP processor");
195196
this.streams = new H2Streams(idGenerator);
197+
this.streams.setLocalStreamChangeCallback(this::fireCapacityAvailable);
196198
this.localConfig = h2Config != null ? h2Config : H2Config.DEFAULT;
197199
this.inputMetrics = new BasicH2TransportMetrics();
198200
this.outputMetrics = new BasicH2TransportMetrics();
@@ -674,6 +676,7 @@ public final void onDisconnect() {
674676
}
675677
streams.shutdownAndReleaseAll();
676678
CommandSupport.cancelCommands(ioSession);
679+
fireSessionClosed();
677680
}
678681

679682
private void executeShutdown(final ShutdownCommand shutdownCommand) throws IOException {
@@ -1099,6 +1102,7 @@ private void consumeFrame(final RawFrame frame) throws HttpException, IOExceptio
10991102
streams.shutdownAndReleaseAll();
11001103
connState = ConnectionHandshake.SHUTDOWN;
11011104
}
1105+
fireDraining();
11021106
}
11031107
ioSession.setEvent(SelectionKey.OP_WRITE);
11041108
break;
@@ -1358,6 +1362,7 @@ private void applyRemoteSettings(final H2Config config) throws H2ConnectionExcep
13581362
}
13591363
}
13601364
}
1365+
fireCapacityAvailable();
13611366
}
13621367

13631368
private void applyLocalSettings() throws H2ConnectionException {
@@ -1438,6 +1443,61 @@ public SocketAddress getLocalAddress() {
14381443
return ioSession.getLocalAddress();
14391444
}
14401445

1446+
int getActiveLocalStreams() {
1447+
return streams.getLocalCount();
1448+
}
1449+
1450+
int getPeerMaxConcurrentStreams() {
1451+
return remoteConfig.getMaxConcurrentStreams();
1452+
}
1453+
1454+
boolean isGoAwayReceived() {
1455+
return goAwayReceived;
1456+
}
1457+
1458+
boolean isShutdown() {
1459+
return connState.compareTo(ConnectionHandshake.GRACEFUL_SHUTDOWN) >= 0;
1460+
}
1461+
1462+
void setPoolSessionSupport(final H2PoolSessionSupport poolSessionSupport) {
1463+
this.poolSessionSupport = poolSessionSupport;
1464+
if (poolSessionSupport != null) {
1465+
poolSessionSupport.updateActiveLocalStreams(streams.getLocalCount());
1466+
poolSessionSupport.updatePeerMaxConcurrentStreams(remoteConfig.getMaxConcurrentStreams());
1467+
poolSessionSupport.updateGoAwayReceived(goAwayReceived);
1468+
poolSessionSupport.updateShutdown(
1469+
connState.compareTo(ConnectionHandshake.GRACEFUL_SHUTDOWN) >= 0);
1470+
}
1471+
}
1472+
1473+
void fireCapacityAvailable() {
1474+
final H2PoolSessionSupport support = this.poolSessionSupport;
1475+
if (support != null) {
1476+
support.updateActiveLocalStreams(streams.getLocalCount());
1477+
support.updatePeerMaxConcurrentStreams(remoteConfig.getMaxConcurrentStreams());
1478+
support.updateShutdown(connState.compareTo(ConnectionHandshake.GRACEFUL_SHUTDOWN) >= 0);
1479+
support.fireCapacityAvailable();
1480+
}
1481+
}
1482+
1483+
void fireDraining() {
1484+
final H2PoolSessionSupport support = this.poolSessionSupport;
1485+
if (support != null) {
1486+
support.updateGoAwayReceived(true);
1487+
support.updateShutdown(connState.compareTo(ConnectionHandshake.GRACEFUL_SHUTDOWN) >= 0);
1488+
support.fireDraining();
1489+
}
1490+
}
1491+
1492+
void fireSessionClosed() {
1493+
final H2PoolSessionSupport support = this.poolSessionSupport;
1494+
if (support != null) {
1495+
support.updateActiveLocalStreams(0);
1496+
support.updateShutdown(true);
1497+
support.fireSessionClosed();
1498+
}
1499+
}
1500+
14411501
void appendState(final StringBuilder buf) {
14421502
buf.append("connState=").append(connState)
14431503
.append(", connInputWindow=").append(connInputWindow)

‎httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/ClientH2IOEventHandler.java‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -42,6 +42,12 @@ public ClientH2IOEventHandler(final ClientH2StreamMultiplexer streamMultiplexer)
4242
super(streamMultiplexer);
4343
}
4444

45+
ClientH2IOEventHandler(
46+
final ClientH2StreamMultiplexer streamMultiplexer,
47+
final H2PoolSessionSupport sessionSupport) {
48+
super(streamMultiplexer, sessionSupport);
49+
}
50+
4551
@Override
4652
public String toString() {
4753
final StringBuilder buf = new StringBuilder();

‎httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/ClientH2PrefaceHandler.java‎

Lines changed: 17 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -61,6 +61,7 @@ public class ClientH2PrefaceHandler extends PrefaceHandlerBase {
6161

6262
private final ClientH2StreamMultiplexerFactory http2StreamHandlerFactory;
6363
private final boolean strictALPNHandshake;
64+
private final H2PoolSessionSupport sessionSupport;
6465
private final AtomicBoolean initialized;
6566

6667
private volatile ByteBuffer preface;
@@ -83,9 +84,23 @@ public ClientH2PrefaceHandler(
8384
final boolean strictALPNHandshake,
8485
final FutureCallback<ProtocolIOSession> resultCallback,
8586
final Callback<Exception> exceptionCallback) {
87+
this(ioSession, http2StreamHandlerFactory, strictALPNHandshake, null, resultCallback, exceptionCallback);
88+
}
89+
90+
/**
91+
* @since 5.5
92+
*/
93+
public ClientH2PrefaceHandler(
94+
final ProtocolIOSession ioSession,
95+
final ClientH2StreamMultiplexerFactory http2StreamHandlerFactory,
96+
final boolean strictALPNHandshake,
97+
final H2PoolSessionSupport sessionSupport,
98+
final FutureCallback<ProtocolIOSession> resultCallback,
99+
final Callback<Exception> exceptionCallback) {
86100
super(ioSession, resultCallback, exceptionCallback);
87101
this.http2StreamHandlerFactory = Args.notNull(http2StreamHandlerFactory, "HTTP/2 stream handler factory");
88102
this.strictALPNHandshake = strictALPNHandshake;
103+
this.sessionSupport = sessionSupport;
89104
this.initialized = new AtomicBoolean();
90105
}
91106

@@ -117,7 +132,8 @@ private boolean writeOutPreface(final IOSession session, final ByteBuffer prefac
117132
if (!preface.hasRemaining()) {
118133
session.clearEvent(SelectionKey.OP_WRITE);
119134
final ByteBuffer data = inBuf != null ? inBuf.data() : null;
120-
startProtocol(new ClientH2IOEventHandler(http2StreamHandlerFactory.create(ioSession)), data);
135+
startProtocol(new ClientH2IOEventHandler(
136+
http2StreamHandlerFactory.create(ioSession), sessionSupport), data);
121137
if (inBuf != null) {
122138
inBuf.clear();
123139
}

‎httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/H2PoolSessionSupport.java‎

Lines changed: 163 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,163 @@
1+
/*
2+
* ====================================================================
3+
* Licensed to the Apache Software Foundation (ASF) under one
4+
* or more contributor license agreements. See the NOTICE file
5+
* distributed with this work for additional information
6+
* regarding copyright ownership. The ASF licenses this file
7+
* to you under the Apache License, Version 2.0 (the
8+
* "License"); you may not use this file except in compliance
9+
* with the License. You may obtain a copy of the License at
10+
*
11+
* http://www.apache.org/licenses/LICENSE-2.0
12+
*
13+
* Unless required by applicable law or agreed to in writing,
14+
* software distributed under the License is distributed on an
15+
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
16+
* KIND, either express or implied. See the License for the
17+
* specific language governing permissions and limitations
18+
* under the License.
19+
* ====================================================================
20+
*
21+
* This software consists of voluntary contributions made by many
22+
* individuals on behalf of the Apache Software Foundation. For more
23+
* information on the Apache Software Foundation, please see
24+
* <http://www.apache.org/>.
25+
*
26+
*/
27+
package org.apache.hc.core5.http2.impl.nio;
28+
29+
import java.util.concurrent.atomic.AtomicInteger;
30+
import java.util.concurrent.atomic.AtomicReference;
31+
32+
import org.apache.hc.core5.annotation.Internal;
33+
34+
/**
35+
* Attachment-based bridge between the HTTP/2 multiplexing connection
36+
* pool and the stream multiplexer. The pool creates an instance and
37+
* passes it as the {@code attachment} argument to the connection
38+
* initiator; the I/O event handler detects it and wires it to the
39+
* multiplexer. The multiplexer pushes state updates into this object
40+
* and fires observer callbacks through it; the pool reads state and
41+
* registers its observer here.
42+
*
43+
* @since 5.5
44+
*/
45+
@Internal
46+
public final class H2PoolSessionSupport {
47+
48+
/**
49+
* Callback for capacity, draining and session-closure notifications.
50+
* Implementations are invoked from the I/O reactor thread
51+
* and must not block.
52+
*/
53+
public interface Observer {
54+
55+
/** Stream capacity became available on the connection. */
56+
void onCapacityAvailable();
57+
58+
/** Connection entered a draining state. */
59+
void onDraining();
60+
61+
/**
62+
* Underlying session has been fully disconnected. Fired exactly once
63+
* after the multiplexer has finalised its disconnect bookkeeping, so
64+
* the observer may treat reservations and active streams on this
65+
* session as void. Default is a no-op to preserve source and binary
66+
* compatibility with existing observers.
67+
*
68+
* @since 5.5
69+
*/
70+
default void onSessionClosed() {
71+
}
72+
73+
}
74+
75+
private final AtomicInteger activeLocalStreams;
76+
private final AtomicInteger peerMaxConcurrentStreams;
77+
private final AtomicReference<Observer> observerRef;
78+
79+
private volatile boolean goAwayReceived;
80+
private volatile boolean shutdown;
81+
82+
public H2PoolSessionSupport() {
83+
this.activeLocalStreams = new AtomicInteger();
84+
this.peerMaxConcurrentStreams = new AtomicInteger(Integer.MAX_VALUE);
85+
this.observerRef = new AtomicReference<>();
86+
}
87+
88+
/** Returns the number of locally initiated streams currently open. */
89+
public int getActiveLocalStreams() {
90+
return activeLocalStreams.get();
91+
}
92+
93+
/** Returns the peer's {@code MAX_CONCURRENT_STREAMS} value. */
94+
public int getPeerMaxConcurrentStreams() {
95+
return peerMaxConcurrentStreams.get();
96+
}
97+
98+
/** Returns {@code true} if a {@code GOAWAY} frame has been received. */
99+
public boolean isGoAwayReceived() {
100+
return goAwayReceived;
101+
}
102+
103+
/** Returns {@code true} if the session is shutting down. */
104+
public boolean isShutdown() {
105+
return shutdown;
106+
}
107+
108+
/** Registers the observer for capacity change notifications. */
109+
public void setObserver(final Observer observer) {
110+
observerRef.set(observer);
111+
}
112+
113+
/** Pushes the current active local stream count. */
114+
public void updateActiveLocalStreams(final int count) {
115+
activeLocalStreams.set(count);
116+
}
117+
118+
/** Pushes the peer's {@code MAX_CONCURRENT_STREAMS} value. */
119+
public void updatePeerMaxConcurrentStreams(final int max) {
120+
peerMaxConcurrentStreams.set(max);
121+
}
122+
123+
/** Pushes the GOAWAY-received flag. */
124+
public void updateGoAwayReceived(final boolean value) {
125+
goAwayReceived = value;
126+
}
127+
128+
/** Pushes the shutdown flag. */
129+
public void updateShutdown(final boolean value) {
130+
shutdown = value;
131+
}
132+
133+
/** Fires the capacity-available notification. */
134+
public void fireCapacityAvailable() {
135+
final Observer observer = observerRef.get();
136+
if (observer != null) {
137+
observer.onCapacityAvailable();
138+
}
139+
}
140+
141+
/** Fires the draining notification. */
142+
public void fireDraining() {
143+
final Observer observer = observerRef.get();
144+
if (observer != null) {
145+
observer.onDraining();
146+
}
147+
}
148+
149+
/**
150+
* Fires the session-closed notification. Intended to be invoked exactly
151+
* once per session after the multiplexer has reset its local stream
152+
* state, so the pool sees a coherent final picture.
153+
*
154+
* @since 5.5
155+
*/
156+
public void fireSessionClosed() {
157+
final Observer observer = observerRef.get();
158+
if (observer != null) {
159+
observer.onSessionClosed();
160+
}
161+
}
162+
163+
}

‎httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/H2Streams.java‎

Lines changed: 17 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -51,6 +51,7 @@ class H2Streams {
5151
private final AtomicInteger lastRemoteId;
5252
private final AtomicInteger localCount;
5353
private final AtomicInteger remoteCount;
54+
private volatile Runnable localStreamChangeCallback;
5455

5556
public H2Streams(final StreamIdGenerator idGenerator) {
5657
this.idGenerator = Args.notNull(idGenerator, "Stream id generator");
@@ -62,6 +63,10 @@ public H2Streams(final StreamIdGenerator idGenerator) {
6263
this.remoteCount = new AtomicInteger(0);
6364
}
6465

66+
void setLocalStreamChangeCallback(final Runnable callback) {
67+
this.localStreamChangeCallback = callback;
68+
}
69+
6570
public boolean isEmpty() {
6671
return streams.isEmpty();
6772
}
@@ -94,9 +99,21 @@ private H2Stream createStream(final H2StreamChannel channel, final H2StreamHandl
9499
switch (state) {
95100
case OPEN:
96101
count.incrementAndGet();
102+
if (!remoteStream) {
103+
final Runnable cb = localStreamChangeCallback;
104+
if (cb != null) {
105+
cb.run();
106+
}
107+
}
97108
break;
98109
case CLOSED:
99110
count.decrementAndGet();
111+
if (!remoteStream) {
112+
final Runnable cb = localStreamChangeCallback;
113+
if (cb != null) {
114+
cb.run();
115+
}
116+
}
100117
}
101118
});
102119
if (remoteStream) {

0 commit comments

Comments
 (0)