From 870c2c3412ca98b351f8b1bead9e9ed50f27e675 Mon Sep 17 00:00:00 2001 From: Arturo Bernal Date: Sun, 1 Nov 2026 13:24:01 +0100 Subject: [PATCH] Do not stop the listening reactor when accept() fails SingleCoreListeningIOReactor caught only CancelledKeyException around ServerSocketChannel#accept(). An IOException, typically 'Too many open files', propagated to AbstractSingleCoreIOReactor#execute, which closed the server channel and the selector. The server never accepted again, even after descriptors were freed, while the reactor still reported ACTIVE. A failed accept() is now reported to the exception callback and the listener keeps running. It backs off briefly, as the selector keeps reporting OP_ACCEPT while the failure persists. The accept loop is also bounded, so that a steady stream of connections does not keep the reactor thread from returning to the selector. --- .../reactor/SingleCoreListeningIOReactor.java | 24 +++- .../TestListeningIOReactorAcceptBurst.java | 88 +++++++++++++ .../TestListeningIOReactorAcceptFailure.java | 120 ++++++++++++++++++ 3 files changed, 230 insertions(+), 2 deletions(-) create mode 100644 httpcore5/src/test/java/org/apache/hc/core5/reactor/TestListeningIOReactorAcceptBurst.java create mode 100644 httpcore5/src/test/java/org/apache/hc/core5/reactor/TestListeningIOReactorAcceptFailure.java diff --git a/httpcore5/src/main/java/org/apache/hc/core5/reactor/SingleCoreListeningIOReactor.java b/httpcore5/src/main/java/org/apache/hc/core5/reactor/SingleCoreListeningIOReactor.java index c00f87081a..0df23c2e19 100644 --- a/httpcore5/src/main/java/org/apache/hc/core5/reactor/SingleCoreListeningIOReactor.java +++ b/httpcore5/src/main/java/org/apache/hc/core5/reactor/SingleCoreListeningIOReactor.java @@ -52,6 +52,9 @@ class SingleCoreListeningIOReactor extends AbstractSingleCoreIOReactor implements ConnectionAcceptor { + private static final long ACCEPT_BACK_OFF_MILLIS = 100; + private static final int MAX_ACCEPTS_PER_EVENT = 100; + private final IOReactorConfig reactorConfig; private final Callback callback; private final Queue requestQueue; @@ -119,8 +122,16 @@ private void processEvent(final SelectionKey key) throws IOException { if (key.isAcceptable()) { final ServerSocketChannel serverChannel = (ServerSocketChannel) key.channel(); - for (;;) { - final SocketChannel socketChannel = serverChannel.accept(); + for (int i = 0; i < MAX_ACCEPTS_PER_EVENT; i++) { + final SocketChannel socketChannel; + try { + socketChannel = serverChannel.accept(); + } catch (final IOException ex) { + // Typically descriptor exhaustion; the listener must outlive it. + logException(ex); + backOff(); + break; + } if (socketChannel == null) { break; } @@ -136,6 +147,15 @@ private void processEvent(final SelectionKey key) throws IOException { } } + private static void backOff() { + // The selector keeps reporting OP_ACCEPT while the failure persists. + try { + Thread.sleep(ACCEPT_BACK_OFF_MILLIS); + } catch (final InterruptedException ex) { + Thread.currentThread().interrupt(); + } + } + @Override public Future listen( final SocketAddress address, final Object attachment, final FutureCallback callback) { diff --git a/httpcore5/src/test/java/org/apache/hc/core5/reactor/TestListeningIOReactorAcceptBurst.java b/httpcore5/src/test/java/org/apache/hc/core5/reactor/TestListeningIOReactorAcceptBurst.java new file mode 100644 index 0000000000..065de819e4 --- /dev/null +++ b/httpcore5/src/test/java/org/apache/hc/core5/reactor/TestListeningIOReactorAcceptBurst.java @@ -0,0 +1,88 @@ +/* + * ==================================================================== + * 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.core5.reactor; + +import java.io.IOException; +import java.net.InetAddress; +import java.net.InetSocketAddress; +import java.nio.channels.SocketChannel; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; + +import org.apache.hc.core5.io.Closer; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Assumptions; +import org.junit.jupiter.api.Test; + +class TestListeningIOReactorAcceptBurst { + + // More than the number of connections accepted per selection event + private static final int CONNECTIONS = 150; + + @Test + void testBurstOfConnectionsIsAcceptedInFull() throws Exception { + final CountDownLatch accepted = new CountDownLatch(CONNECTIONS); + final List exceptions = new CopyOnWriteArrayList<>(); + final List clients = new ArrayList<>(); + try (SingleCoreListeningIOReactor reactor = new SingleCoreListeningIOReactor(exceptions::add, IOReactorConfig.DEFAULT, entry -> { + Closer.closeQuietly(entry.channel); + accepted.countDown(); + })) { + final Thread reactorThread = new Thread(reactor::execute, "test-reactor"); + reactorThread.start(); + try { + final ListenerEndpoint endpoint = reactor.listen( + new InetSocketAddress(InetAddress.getLoopbackAddress(), 0), null) + .get(5, TimeUnit.SECONDS); + final InetSocketAddress address = (InetSocketAddress) endpoint.getAddress(); + try { + for (int i = 0; i < CONNECTIONS; i++) { + final SocketChannel client = SocketChannel.open(); + clients.add(client); + client.connect(address); + } + } catch (final IOException ex) { + Assumptions.assumeTrue(false, "Not enough file descriptors: " + ex); + } + if (!accepted.await(10, TimeUnit.SECONDS)) { + Assumptions.assumeTrue(exceptions.isEmpty(), "Accept failed: " + exceptions); + Assertions.fail("Accepted " + (CONNECTIONS - accepted.getCount()) + " of " + CONNECTIONS); + } + } finally { + for (final SocketChannel client : clients) { + Closer.closeQuietly(client); + } + reactor.initiateShutdown(); + reactorThread.join(5000); + } + } + } + +} diff --git a/httpcore5/src/test/java/org/apache/hc/core5/reactor/TestListeningIOReactorAcceptFailure.java b/httpcore5/src/test/java/org/apache/hc/core5/reactor/TestListeningIOReactorAcceptFailure.java new file mode 100644 index 0000000000..20c0b73213 --- /dev/null +++ b/httpcore5/src/test/java/org/apache/hc/core5/reactor/TestListeningIOReactorAcceptFailure.java @@ -0,0 +1,120 @@ +/* + * ==================================================================== + * 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.core5.reactor; + +import java.io.IOException; +import java.net.InetAddress; +import java.net.InetSocketAddress; +import java.nio.channels.SocketChannel; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; + +import org.apache.hc.core5.io.Closer; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Assumptions; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.parallel.Isolated; + +/** + * Exhausts the process file descriptors, hence runs isolated. + */ +@Isolated +class TestListeningIOReactorAcceptFailure { + + private static final int MAX_DESCRIPTORS = 500000; + + @Test + void listenerSurvivesAcceptFailure() throws Exception { + final AtomicReference acceptedRef = new AtomicReference<>(new CountDownLatch(1)); + final List exceptions = new CopyOnWriteArrayList<>(); + final List held = new ArrayList<>(); + try (SingleCoreListeningIOReactor reactor = new SingleCoreListeningIOReactor(exceptions::add, IOReactorConfig.DEFAULT, entry -> { + Closer.closeQuietly(entry.channel); + acceptedRef.get().countDown(); + })) { + final Thread reactorThread = new Thread(reactor::execute, "test-reactor"); + reactorThread.start(); + try { + final ListenerEndpoint endpoint = reactor.listen( + new InetSocketAddress(InetAddress.getLoopbackAddress(), 0), null) + .get(5, TimeUnit.SECONDS); + final InetSocketAddress address = (InetSocketAddress) endpoint.getAddress(); + Assertions.assertTrue(connect(address, acceptedRef)); + + // The last free descriptor goes to the client, so the server side accept() has none left. + IOException exhausted = null; + while (exhausted == null && held.size() < MAX_DESCRIPTORS) { + try { + held.add(SocketChannel.open()); + } catch (final IOException ex) { + exhausted = ex; + } + } + Assumptions.assumeTrue(exhausted != null, "File descriptor limit not reachable"); + held.remove(held.size() - 1).close(); + final SocketChannel client = SocketChannel.open(); + held.add(client); + client.connect(address); + + final long deadline = System.currentTimeMillis() + 5000; + while (exceptions.isEmpty() && System.currentTimeMillis() < deadline) { + Thread.sleep(10); + } + for (final SocketChannel channel : held) { + channel.close(); + } + held.clear(); + Assumptions.assumeFalse(exceptions.isEmpty(), "Accept did not fail"); + Assertions.assertInstanceOf(IOException.class, exceptions.get(0)); + + Assertions.assertTrue(connect(address, acceptedRef), "Listener stopped accepting: " + exceptions); + Assertions.assertEquals(IOReactorStatus.ACTIVE, reactor.getStatus()); + } finally { + for (final SocketChannel channel : held) { + Closer.closeQuietly(channel); + } + reactor.initiateShutdown(); + reactorThread.join(5000); + } + } + } + + private static boolean connect( + final InetSocketAddress address, final AtomicReference acceptedRef) throws Exception { + final CountDownLatch latch = new CountDownLatch(1); + acceptedRef.set(latch); + try (SocketChannel client = SocketChannel.open()) { + client.connect(address); + return latch.await(5, TimeUnit.SECONDS); + } + } + +}