tdcornish opened a new pull request, #726: URL: https://github.com/apache/httpcomponents-core/pull/726
`IOSessionImpl#getEventMask` now takes the session lock and returns 0 once the session is closed, matching `setEvent`, `clearEvent` and `setEventMask`. This completes [HTTPCORE-659](https://issues.apache.org/jira/browse/HTTPCORE-659) and [HTTPCORE-753](https://issues.apache.org/jira/browse/HTTPCORE-753). HTTPCORE-659 moved the closed-status check inside the lock for the three setters, and HTTPCORE-753 made `close()` cancel the key under that lock, which fixed a consumer thread's `CapacityWindow#update` racing the I/O reactor closing the session. On TLS connections that update can still fail: `SSLIOSession#setEvent` calls `updateEventMask`, which reads `getEventMask` first, and the getter had no closed check. The `CancelledKeyException` escapes `ReactiveDataConsumer`'s `Subscription#request`, and the response stream never reaches `onComplete`. Which HttpClient versions reach it: with HttpClient 5.6.5 and 5.7-alpha1 the reproducer below hits the race on both released HttpCore and current master. From HttpClient [e750f52dd](https://github.com/apache/httpcomponents-client/commit/e750f52dd9beedb34d039a2212cb5c05b6183640) ("Simplifcation of the message exchange handling logic in the main async exec interceptors") onwards it no longer does, so 5.7-beta1 and HttpClient master come out clean. That commit doesn't touch this code path directly; it looks like it changes when the exchange completes relative to the connection close, which hides the race rather than removing it. I haven't traced that part, and I couldn't reproduce with HttpCore alone, without HttpClient. The getter is still inconsistent with the setters on master, and the unit test below shows the exception directly. Design notes: - The guard is in `IOSessionImpl` rather than `SSLIOSession#updateEventMask` so that it covers every caller, as the setters' guard does. The other callers are the `SSLIOSession` constructor, where the session is open, and `InternalDataChannel#onIOEvent`, which on a closed session now skips `outputReady` instead of throwing. - The check has to be under the lock: `close()` cancels the key while holding it, so a check outside the lock could still race the cancellation. `SSLIOSession` already holds this reentrant lock when it calls `getEventMask`, so the TLS path takes no additional lock. - Returning 0 is safe for `updateEventMask`: it then calls `setEventMask`, which already does nothing on a closed session. Testing: - New `TestIOSessionImpl#eventMaskIsEmptyOnceClosed`; it fails with `CancelledKeyException` without the change. - `httpcore5`, `httpcore5-h2`, `httpcore5-reactive` and `httpcore5-testing` test suites pass (1,975 / 397 / 11 / 375 tests, 4 skipped). - Reproducer, run on x86-64 Linux (4 CPUs, JDK 21) with master 63c0556f4 and then with this change installed into the local Maven repository: | HttpClient | HttpCore | hung streams | `request()` threw | |---|---|---:|---:| | 5.6.5 | 5.4.4 | 17 | 27 | | 5.7-alpha1 | 5.5-beta2 | 29 | 63 | | 5.7-beta1 | 5.5-beta4 | 0 | 0 | | 5.6.5 | master | 23 | 37 | | 5.7-alpha1 | master | 36 | 58 | | 5.6.5 | master + this change | 0 | 0 | | 5.7-alpha1 | master + this change | 0 | 0 | Each row is 10,000 streams. macOS gave the same pattern. The race is timing-dependent, so counts vary from run to run. How to reproduce: The attached Gradle project, `closerace-repro-gradle.zip`, runs a local TLS server that streams 240 KB chunked responses, each ending with `Connection: close`, after which the server closes the connection. The client is `CloseableHttpAsyncClient` with `ReactiveResponseConsumer`; each subscriber requests random batches of 1-16 items and issues every next request from a separate thread pool. A stream that receives nothing for 5 seconds without a terminal signal counts as hung. Gradle needs JDK 17 or later. ``` ./gradlew matrix # released HttpClient/HttpCore pairs # HttpCore from a local checkout (cd ../httpcomponents-core && ./mvnw install -DskipTests -Drat.skip -Djapicmp.skip -pl httpcore5,httpcore5-h2,httpcore5-reactive -am) ./gradlew matrix -PcoreVersion=5.5-beta5-SNAPSHOT ``` `matrix` prints one row per HttpClient/HttpCore pair and marks it REPRODUCED or clean. <details> <summary>CloseRaceRepro.java (the reproducer in the zip)</summary> ```java import java.io.*; import java.net.*; import java.nio.ByteBuffer; import java.nio.charset.StandardCharsets; import java.security.KeyStore; import java.security.cert.X509Certificate; import java.util.*; import java.util.concurrent.*; import java.util.concurrent.atomic.*; import javax.net.ssl.*; import org.apache.hc.client5.http.impl.async.CloseableHttpAsyncClient; import org.apache.hc.client5.http.impl.async.HttpAsyncClients; import org.apache.hc.client5.http.impl.nio.PoolingAsyncClientConnectionManager; import org.apache.hc.client5.http.impl.nio.PoolingAsyncClientConnectionManagerBuilder; import org.apache.hc.client5.http.ssl.ClientTlsStrategyBuilder; import org.apache.hc.client5.http.ssl.NoopHostnameVerifier; import org.apache.hc.core5.concurrent.FutureCallback; import org.apache.hc.core5.http.HttpResponse; import org.apache.hc.core5.http.Message; import org.apache.hc.core5.http.nio.support.AsyncRequestBuilder; import org.apache.hc.core5.http2.HttpVersionPolicy; import org.apache.hc.core5.io.CloseMode; import org.apache.hc.core5.reactive.ReactiveResponseConsumer; import org.reactivestreams.Publisher; import org.reactivestreams.Subscriber; import org.reactivestreams.Subscription; /** * TLS server streaming chunked responses over keep-alive connections; once a connection is older * than ageLimitMs (or has served 100 requests) its next response carries "Connection: close" and * the server closes the connection after it. The client is the HttpClient 5 async client with ReactiveResponseConsumer; * each subscriber requests a random batch of 1-16 items, and for slowQuarters/4 of the streams it * issues each next request from a separate thread pool after a 1 ms pause. Reports streams that * never reach a terminal signal and calls to Subscription#request that threw. * * args: total concurrency ageLimitMs closeMode(notify|fin|rst) chunks chunkSize [slowQuarters 0-4, default 1] * example: CloseRaceRepro 10000 64 0 notify 400 600 4 * The server keystore is read from ks.p12, or from the closerace.keystore system property. */ @SuppressWarnings("deprecation") public class CloseRaceRepro { public static final class Result { public final int total, ok, okClose, err, errClose, hang, serverResponses, serverClose, requestThrew; public final List<String> hangs; public final String firstThrown; Result(int total, int ok, int okClose, int err, int errClose, int hang, int serverResponses, int serverClose, int requestThrew, List<String> hangs, String firstThrown) { this.total = total; this.ok = ok; this.okClose = okClose; this.err = err; this.errClose = errClose; this.hang = hang; this.serverResponses = serverResponses; this.serverClose = serverClose; this.requestThrew = requestThrew; this.hangs = hangs; this.firstThrown = firstThrown; } public String summary() { return String.format("total=%d ok=%d (close %d) err=%d (close %d) hang=%d serverResponses=%d serverClose=%d request()-threw=%d", total, ok, okClose, err, errClose, hang, serverResponses, serverClose, requestThrew); } } final AtomicInteger serverCloseResponses = new AtomicInteger(); final AtomicInteger serverResponses = new AtomicInteger(); final String closeMode; final long ageLimitMs; final int chunks, chunkSize; CloseRaceRepro(long ageLimitMs, String closeMode, int chunks, int chunkSize) { this.ageLimitMs = ageLimitMs; this.closeMode = closeMode; this.chunks = chunks; this.chunkSize = chunkSize; } public static void main(String[] a) throws Exception { Result r = run(a); System.out.println(r.summary()); r.hangs.stream().limit(20).forEach(h -> System.out.println("HANG " + h)); if (r.firstThrown != null) System.out.println("FIRST request() EXCEPTION\n" + r.firstThrown); System.exit(0); } public static Result run(String... a) throws Exception { int total = Integer.parseInt(a[0]); int concurrency = Integer.parseInt(a[1]); int slowQuarters = a.length > 6 ? Integer.parseInt(a[6]) : 1; return new CloseRaceRepro(Long.parseLong(a[2]), a[3], Integer.parseInt(a[4]), Integer.parseInt(a[5])) .execute(total, concurrency, slowQuarters); } Result execute(int total, int concurrency, int slowQuarters) throws Exception { long idleHangMs = 5000; SSLContext serverCtx = SSLContext.getInstance("TLS"); KeyStore ks = KeyStore.getInstance("PKCS12"); try (InputStream in = new FileInputStream(System.getProperty("closerace.keystore", "ks.p12"))) { ks.load(in, "changeit".toCharArray()); } KeyManagerFactory kmf = KeyManagerFactory.getInstance(KeyManagerFactory.getDefaultAlgorithm()); kmf.init(ks, "changeit".toCharArray()); serverCtx.init(kmf.getKeyManagers(), null, null); ServerSocket ss = new ServerSocket(0, 1024, InetAddress.getLoopbackAddress()); int port = ss.getLocalPort(); Thread acceptor = new Thread(() -> { while (true) { try { Socket plain = ss.accept(); Thread t = new Thread(() -> serve(serverCtx, plain)); t.setDaemon(true); t.start(); } catch (IOException e) { return; } } }); acceptor.setDaemon(true); acceptor.start(); SSLContext clientCtx = SSLContext.getInstance("TLS"); clientCtx.init(null, new TrustManager[]{new X509TrustManager() { public void checkClientTrusted(X509Certificate[] c, String s) {} public void checkServerTrusted(X509Certificate[] c, String s) {} public X509Certificate[] getAcceptedIssuers() { return new X509Certificate[0]; } }}, null); PoolingAsyncClientConnectionManager cm = PoolingAsyncClientConnectionManagerBuilder.create() .setTlsStrategy(ClientTlsStrategyBuilder.create().setSslContext(clientCtx) .setHostnameVerifier(NoopHostnameVerifier.INSTANCE).build()) .setMaxConnTotal(concurrency * 2).setMaxConnPerRoute(concurrency * 2).build(); CloseableHttpAsyncClient client = HttpAsyncClients.custom().setConnectionManager(cm) .setVersionPolicy(HttpVersionPolicy.FORCE_HTTP_1).build(); client.start(); ExecutorService consumers = Executors.newFixedThreadPool(16); ScheduledExecutorService sched = Executors.newScheduledThreadPool(2); Semaphore inflight = new Semaphore(concurrency); AtomicInteger ok = new AtomicInteger(), err = new AtomicInteger(), hang = new AtomicInteger(); AtomicInteger okClose = new AtomicInteger(), errClose = new AtomicInteger(); AtomicInteger requestThrew = new AtomicInteger(); AtomicReference<String> firstThrown = new AtomicReference<>(); List<String> hangs = Collections.synchronizedList(new ArrayList<>()); long expected = (long) chunks * chunkSize; Random rnd = new Random(42); for (int i = 0; i < total; i++) { inflight.acquire(); final int id = i; final AtomicLong bytes = new AtomicLong(); final AtomicLong lastActivity = new AtomicLong(System.currentTimeMillis()); final AtomicBoolean done = new AtomicBoolean(); final AtomicReference<String> conn = new AtomicReference<>("?"); final int batch = 1 + rnd.nextInt(16); final boolean slow = rnd.nextInt(4) < slowQuarters; ReactiveResponseConsumer consumer = new ReactiveResponseConsumer(new FutureCallback<Message<HttpResponse, Publisher<ByteBuffer>>>() { public void completed(Message<HttpResponse, Publisher<ByteBuffer>> m) { conn.set(m.getHead().getFirstHeader("Connection") == null ? "keep-alive" : m.getHead().getFirstHeader("Connection").getValue()); m.getBody().subscribe(new Subscriber<ByteBuffer>() { Subscription sub; int outstanding; void request() { try { sub.request(batch); } catch (RuntimeException e) { requestThrew.incrementAndGet(); StringWriter sw = new StringWriter(); e.printStackTrace(new PrintWriter(sw)); firstThrown.compareAndSet(null, sw.toString()); } } public void onSubscribe(Subscription s) { sub = s; outstanding = batch; request(); } public void onNext(ByteBuffer b) { bytes.addAndGet(b.remaining()); lastActivity.set(System.currentTimeMillis()); if (--outstanding == 0) { outstanding = batch; if (slow) { consumers.execute(() -> { try { Thread.sleep(1); } catch (InterruptedException e) {} request(); }); } else { request(); } } } public void onError(Throwable t) { if (done.compareAndSet(false, true)) { err.incrementAndGet(); if ("close".equalsIgnoreCase(conn.get())) errClose.incrementAndGet(); inflight.release(); } } public void onComplete() { if (done.compareAndSet(false, true)) { if (bytes.get() != expected) { System.out.println("SHORT id=" + id + " bytes=" + bytes.get()); } ok.incrementAndGet(); if ("close".equalsIgnoreCase(conn.get())) okClose.incrementAndGet(); inflight.release(); } } }); } public void failed(Exception e) { if (done.compareAndSet(false, true)) { err.incrementAndGet(); inflight.release(); } } public void cancelled() { if (done.compareAndSet(false, true)) { err.incrementAndGet(); inflight.release(); } } }); client.execute(AsyncRequestBuilder.get("https://localhost:"; + port + "/stream?id=" + id).build(), consumer, null, null); sched.schedule(new Runnable() { public void run() { if (done.get()) return; if (System.currentTimeMillis() - lastActivity.get() > idleHangMs) { if (done.compareAndSet(false, true)) { hang.incrementAndGet(); hangs.add("id=" + id + " bytes=" + bytes.get() + "/" + expected + " conn=" + conn.get() + " batch=" + batch + " slow=" + slow); inflight.release(); } } else { sched.schedule(this, 1000, TimeUnit.MILLISECONDS); } } }, idleHangMs, TimeUnit.MILLISECONDS); } inflight.acquire(concurrency); client.close(CloseMode.IMMEDIATE); ss.close(); consumers.shutdownNow(); sched.shutdownNow(); return new Result(total, ok.get(), okClose.get(), err.get(), errClose.get(), hang.get(), serverResponses.get(), serverCloseResponses.get(), requestThrew.get(), new ArrayList<>(hangs), firstThrown.get()); } void serve(SSLContext ctx, Socket plain) { long born = System.currentTimeMillis(); int reqNo = 0; try { SSLSocket tls = (SSLSocket) ctx.getSocketFactory().createSocket(plain, null, plain.getPort(), true); tls.setUseClientMode(false); tls.startHandshake(); InputStream in = new BufferedInputStream(tls.getInputStream()); OutputStream out = tls.getOutputStream(); byte[] payload = new byte[chunkSize]; Arrays.fill(payload, (byte) 'x'); byte[] chunkHead = (Integer.toHexString(chunkSize) + "\r\n").getBytes(StandardCharsets.US_ASCII); while (true) { if (!readHead(in)) return; reqNo++; boolean close = ageLimitMs >= 0 && (System.currentTimeMillis() - born >= ageLimitMs || reqNo >= 100); serverResponses.incrementAndGet(); if (close) serverCloseResponses.incrementAndGet(); String head = "HTTP/1.1 200 OK\r\nContent-Type: application/octet-stream\r\n" + "Transfer-Encoding: chunked\r\n" + (close ? "Connection: close\r\n" : "") + "\r\n"; ByteArrayOutputStream buf = new ByteArrayOutputStream(); buf.write(head.getBytes(StandardCharsets.US_ASCII)); for (int c = 0; c < chunks; c++) { buf.write(chunkHead); buf.write(payload); buf.write('\r'); buf.write('\n'); if (buf.size() > 16384 || c == chunks - 1) { out.write(buf.toByteArray()); out.flush(); buf.reset(); } } out.write("0\r\n\r\n".getBytes(StandardCharsets.US_ASCII)); out.flush(); if (close) { switch (closeMode) { case "notify": tls.close(); break; case "fin": plain.shutdownOutput(); plain.close(); break; case "rst": plain.setSoLinger(true, 0); plain.close(); break; default: throw new IllegalArgumentException(closeMode); } return; } } } catch (IOException e) { try { plain.close(); } catch (IOException ignore) {} } } static boolean readHead(InputStream in) throws IOException { int state = 0, b; while ((b = in.read()) != -1) { if ((state == 0 || state == 2) && b == '\r') state++; else if ((state == 1 || state == 3) && b == '\n') { state++; if (state == 4) return true; } else state = 0; } return false; } } ``` </details> -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
