diff --git a/junixsocket-common/src/main/java/org/newsclub/net/unix/VirtualThreadPoller.java b/junixsocket-common/src/main/java/org/newsclub/net/unix/VirtualThreadPoller.java index 309a889a9..55ba1c9cd 100644 --- a/junixsocket-common/src/main/java/org/newsclub/net/unix/VirtualThreadPoller.java +++ b/junixsocket-common/src/main/java/org/newsclub/net/unix/VirtualThreadPoller.java @@ -33,7 +33,7 @@ interface VirtualThreadPoller { /** * Returns the default instance best suited for the current system. */ - VirtualThreadPoller INSTANCE = new VirtualThreadPollerNaive(); + VirtualThreadPoller INSTANCE = VirtualThreadPollerShared.newDefaultInstance(); /** * Parks the current thread until the given file descriptor is ready, with respect to the given diff --git a/junixsocket-common/src/main/java/org/newsclub/net/unix/VirtualThreadPollerNaive.java b/junixsocket-common/src/main/java/org/newsclub/net/unix/VirtualThreadPollerNaive.java index c33d938cb..5b193fd47 100644 --- a/junixsocket-common/src/main/java/org/newsclub/net/unix/VirtualThreadPollerNaive.java +++ b/junixsocket-common/src/main/java/org/newsclub/net/unix/VirtualThreadPollerNaive.java @@ -27,6 +27,7 @@ import java.util.Map; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ExecutionException; +import java.util.concurrent.ForkJoinPool; import java.util.concurrent.locks.LockSupport; import org.eclipse.jdt.annotation.Nullable; @@ -70,39 +71,76 @@ private static final class PollJob { this.timeout = timeout; } - @SuppressWarnings("PMD.CognitiveComplexity") + private final class PollBlocker implements ForkJoinPool.ManagedBlocker { + private @Nullable IOException result; + private boolean done; + + @Override + public boolean block() { + result = pollUntilReady(); + done = true; + return true; + } + + @Override + public boolean isReleasable() { + return done; + } + } + + /** + * Polls until the file descriptor is ready, the timeout has elapsed, or the thread is + * interrupted. + * + * @return {@code null} if ready or timed out, or an exception. + */ + private @Nullable IOException pollUntilReady() { + Thread thread = Thread.currentThread(); + PollFd pfd = new PollFd(new FileDescriptor[] {fd}, new int[] {mode}); + do { + if (thread.isInterrupted() || !fd.valid()) { + return POLL_INTERRUPTED_SENTINEL; + } + try { + NativeUnixSocket.poll(pfd, POLL_INTERVAL_MILLIS); + } catch (IOException e) { + return e; + } + if (thread.isInterrupted() || !fd.valid()) { + return POLL_INTERRUPTED_SENTINEL; + } + if (pfd.rops[0] != 0) { + return null; + } + + int timeoutMillis = timeout.get(); + if (timeoutMillis > 0) { + if ((System.currentTimeMillis() - now) >= timeoutMillis) { + // handle in calling thread + return null; + } + } + } while (true); // NOPMD.WhileLoopWithLiteralBoolean + } + AFFuture<@Nullable IOException> trigger(Thread waitingThread) { synchronized (fd) { waitingThreads.add(waitingThread); } return AFFuture.supplyAsync(() -> { try { - Thread thread = Thread.currentThread(); - PollFd pfd = new PollFd(new FileDescriptor[] {fd}, new int[] {mode}); - do { - if (thread.isInterrupted() || !fd.valid()) { - return POLL_INTERRUPTED_SENTINEL; - } - try { - NativeUnixSocket.poll(pfd, POLL_INTERVAL_MILLIS); - } catch (IOException e) { - return e; - } - if (thread.isInterrupted() || !fd.valid()) { - return POLL_INTERRUPTED_SENTINEL; - } - if (pfd.rops[0] != 0) { - break; - } - - int timeoutMillis = timeout.get(); - if (timeoutMillis > 0) { - if ((System.currentTimeMillis() - now) >= timeoutMillis) { - // handle in calling thread - break; - } - } - } while (true); // NOPMD.WhileLoopWithLiteralBoolean + // poll() may block this worker for a long time; let the pool compensate, otherwise + // as few as parallelism waiting threads can starve the one we're waiting for + // see https://github.com/kohlschutter/junixsocket/issues/172 + PollBlocker blocker = new PollBlocker(); + try { + ForkJoinPool.managedBlock(blocker); + } catch (InterruptedException e) { + return POLL_INTERRUPTED_SENTINEL; + } + if (blocker.result != null) { + return blocker.result; + } } finally { Thread threadToWake = null; try { diff --git a/junixsocket-common/src/main/java/org/newsclub/net/unix/VirtualThreadPollerShared.java b/junixsocket-common/src/main/java/org/newsclub/net/unix/VirtualThreadPollerShared.java new file mode 100644 index 000000000..a2d5ff416 --- /dev/null +++ b/junixsocket-common/src/main/java/org/newsclub/net/unix/VirtualThreadPollerShared.java @@ -0,0 +1,356 @@ +/* + * junixsocket + * + * Copyright 2009-2026 Christian Kohlschütter + * + * Licensed 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. + */ +package org.newsclub.net.unix; + +import java.io.Closeable; +import java.io.FileDescriptor; +import java.io.IOException; +import java.net.SocketTimeoutException; +import java.nio.channels.SelectionKey; +import java.util.ArrayList; +import java.util.List; +import java.util.Queue; +import java.util.concurrent.ConcurrentLinkedQueue; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.locks.LockSupport; + +import org.newsclub.net.unix.AFSelector.PollFd; + +/** + * Implementation of {@link VirtualThreadPoller} that multiplexes all waiting file descriptors onto + * a small, fixed number of platform "poller" threads. + * + * A virtual thread that would block registers its file descriptor with a poller and parks. The + * poller calls {@link NativeUnixSocket#poll(PollFd, int)} on all registered file descriptors at + * once (plus a pipe used to wake it up upon new registrations), and unparks the virtual threads + * whose file descriptors are ready. + * + * Unlike {@link VirtualThreadPollerNaive}, waiting does not occupy a platform thread per file + * descriptor, so any number of virtual threads can wait concurrently without starving each other + * (see https://github.com/kohlschutter/junixsocket/issues/172). + * + * @author Jakub Kultys (Azahe) + */ +final class VirtualThreadPollerShared implements VirtualThreadPoller { + /** + * Upper bound for a single poll; ensures that closed file descriptors and cancelled registrations + * are noticed even if nothing else happens. + */ + private static final int POLL_INTERVAL_MILLIS = 1_000; + + private static final String PROP_POLLERS = "org.newsclub.net.unix.VirtualThreadPoller.pollers"; + + private static final String PROP_NAIVE = "org.newsclub.net.unix.VirtualThreadPoller.naive"; + + private final Poller[] pollers; + + VirtualThreadPollerShared() throws IOException { + this(Integer.parseInt(System.getProperty(PROP_POLLERS, "1"))); + } + + VirtualThreadPollerShared(int numPollers) throws IOException { + if (numPollers < 1) { + throw new IllegalArgumentException("numPollers"); + } + Poller[] p = new Poller[numPollers]; + try { + for (int i = 0; i < numPollers; i++) { + p[i] = new Poller(); + } + } catch (IOException | RuntimeException e) { + for (Poller poller : p) { + if (poller != null) { + poller.closeQuietly(); + } + } + throw e; + } + for (int i = 0; i < numPollers; i++) { + ThreadUtil.startNewDaemonThread(false, p[i]).setName("junixsocket VirtualThreadPoller-" + i); + } + this.pollers = p; + } + + /** + * Returns a new {@link VirtualThreadPollerShared}, or a {@link VirtualThreadPollerNaive} if that + * is requested via system property, or if the former cannot be created. + * + * @return The new instance. + */ + static VirtualThreadPoller newDefaultInstance() { + if (Boolean.parseBoolean(System.getProperty(PROP_NAIVE, "false"))) { + return new VirtualThreadPollerNaive(); + } + try { + return new VirtualThreadPollerShared(); + } catch (IOException | RuntimeException e) { + StackTraceUtil.printStackTrace(e); + return new VirtualThreadPollerNaive(); + } + } + + @Override + public void parkThreadUntilReady(FileDescriptor fd, int mode, long now, + AFSupplier timeout, Closeable closeOnInterrupt) throws IOException { + Thread thread = Thread.currentThread(); + if (thread.isInterrupted() || !fd.valid()) { + throw SocketClosedByInterruptException.newInstanceAndClose(closeOnInterrupt); + } + + Registration reg = new Registration(fd, mode, thread); + Poller poller = pollers.length == 1 ? pollers[0] : pollers[(System.identityHashCode(fd) + & Integer.MAX_VALUE) % pollers.length]; + poller.register(reg); + + try { + do { + switch (reg.state.get()) { + case Registration.READY: + return; + case Registration.CLOSED: + throw SocketClosedByInterruptException.newInstanceAndClose(closeOnInterrupt); + default: + break; + } + if (thread.isInterrupted()) { + throw SocketClosedByInterruptException.newInstanceAndClose(closeOnInterrupt); + } + + int timeoutMillis = timeout.get(); + if (timeoutMillis > 0) { + long remainingMillis = now + timeoutMillis - System.currentTimeMillis(); + if (remainingMillis <= 0) { + throw new SocketTimeoutException(); + } + LockSupport.parkNanos(this, TimeUnit.MILLISECONDS.toNanos(remainingMillis)); + } else { + LockSupport.park(this); + } + } while (true); // NOPMD.WhileLoopWithLiteralBoolean + } finally { + // no-op if the poller already completed the registration; + // otherwise the poller discards it in its next iteration + reg.state.compareAndSet(Registration.WAITING, Registration.CANCELLED); + } + } + + /** + * A virtual thread waiting for a file descriptor to become ready. + */ + private static final class Registration { + static final int WAITING = 0; + static final int READY = 1; + static final int CLOSED = 2; + static final int CANCELLED = 3; + + final FileDescriptor fd; + final int mode; + final Thread thread; + final AtomicInteger state = new AtomicInteger(WAITING); + + Registration(FileDescriptor fd, int mode, Thread thread) { + this.fd = fd; + this.mode = mode; + this.thread = thread; + } + + /** + * Completes the registration, and unparks the waiting thread unless the registration had + * already been completed or cancelled. + * + * @param newState The new state. + */ + void complete(int newState) { + if (state.compareAndSet(WAITING, newState)) { + LockSupport.unpark(thread); + } + } + + boolean isWaiting() { + return state.get() == WAITING; + } + } + + /** + * A platform thread polling all file descriptors registered with it. + */ + private static final class Poller implements Runnable { + private final AFPipe wakeupPipe; + private final FileDescriptor wakeupSource; + private final FileDescriptor wakeupSink; + private final int wakeupOptions; + private final byte[] wakeupDrainBuffer = new byte[64]; + + private final Queue incoming = new ConcurrentLinkedQueue<>(); + private final AtomicBoolean wakeupPending = new AtomicBoolean(false); + + // only accessed from the poller thread + private List active = new ArrayList<>(); + private List nextActive = new ArrayList<>(); + + Poller() throws IOException { + this.wakeupPipe = AFUNIXSelectorProvider.getInstance().openSelectablePipe(); + this.wakeupSource = wakeupPipe.sourceFD(); + this.wakeupSink = wakeupPipe.sinkFD(); + this.wakeupOptions = wakeupPipe.getOptions(); + NativeUnixSocket.configureBlocking(wakeupSource, false); + NativeUnixSocket.configureBlocking(wakeupSink, false); + } + + void register(Registration reg) { + incoming.add(reg); + // At most one wakeup byte per poller iteration; see pollOnce() for the ordering requirements. + if (wakeupPending.compareAndSet(false, true)) { + try { + NativeUnixSocket.write(wakeupSink, null, 1, 1, wakeupOptions, null); + } catch (IOException e) { + // the poller will still pick up the registration within POLL_INTERVAL_MILLIS + } + } + } + + @Override + @SuppressWarnings("PMD.AvoidCatchingThrowable") + public void run() { + do { + try { + pollOnce(); + } catch (Throwable t) { // NOPMD + // Must not die, or all registered threads would wait forever. + // Wake up everyone; they will retry their operation and re-register if necessary. + wakeUpAll(); + StackTraceUtil.printStackTrace(t); + LockSupport.parkNanos(TimeUnit.MILLISECONDS.toNanos(10)); + } + } while (true); // NOPMD.WhileLoopWithLiteralBoolean + } + + @SuppressWarnings("PMD.CognitiveComplexity") + private void pollOnce() throws IOException { + // Reset the flag *before* draining the queue: registrations added before this point are + // picked up below; registrations added afterwards will trigger a new wakeup byte. + wakeupPending.set(false); + + Registration reg; + while ((reg = incoming.poll()) != null) { + active.add(reg); + } + + int size = active.size(); + FileDescriptor[] fds = new FileDescriptor[size + 1]; + int[] ops = new int[size + 1]; + fds[0] = wakeupSource; + ops[0] = SelectionKey.OP_READ; + + int n = 1; + for (int i = 0; i < size; i++) { + reg = active.get(i); + if (!reg.isWaiting()) { + continue; + } + if (!reg.fd.valid()) { + reg.complete(Registration.CLOSED); + continue; + } + fds[n] = reg.fd; + ops[n] = reg.mode; + nextActive.add(reg); + n++; + } + swapActive(); + + PollFd pfd; + if (n == fds.length) { + pfd = new PollFd(fds, ops); + } else { + FileDescriptor[] fds0 = new FileDescriptor[n]; + int[] ops0 = new int[n]; + System.arraycopy(fds, 0, fds0, 0, n); + System.arraycopy(ops, 0, ops0, 0, n); + pfd = new PollFd(fds0, ops0); + } + + int numReady; + try { + numReady = NativeUnixSocket.poll(pfd, POLL_INTERVAL_MILLIS); + } catch (IOException e) { + // e.g., EINTR. Wake up everyone; they will retry their operation (and encounter any + // persistent error condition themselves), and re-register if necessary. + wakeUpAll(); + return; + } + if (numReady <= 0) { + return; + } + + if (pfd.rops[0] != 0) { + drainWakeupPipe(); + } + + // Registrations are in the same order as file descriptors 1..n + size = active.size(); + for (int i = 0; i < size; i++) { + reg = active.get(i); + if (pfd.rops[i + 1] != 0) { + // also when OP_INVALID (error/hangup): the caller retries and sees the actual condition + reg.complete(Registration.READY); + } else if (reg.isWaiting()) { + nextActive.add(reg); + } + } + swapActive(); + } + + private void swapActive() { + List tmp = active; + active = nextActive; + nextActive = tmp; + nextActive.clear(); + } + + private void drainWakeupPipe() throws IOException { + int read; + do { + read = NativeUnixSocket.read(wakeupSource, wakeupDrainBuffer, 0, wakeupDrainBuffer.length, + wakeupOptions, null, 0); + } while (read == wakeupDrainBuffer.length); + } + + private void wakeUpAll() { + for (Registration reg : active) { + reg.complete(Registration.READY); + } + active.clear(); + nextActive.clear(); + Registration reg; + while ((reg = incoming.poll()) != null) { + reg.complete(Registration.READY); + } + } + + void closeQuietly() { + try { + wakeupPipe.close(); + } catch (IOException e) { + // ignore + } + } + } +} diff --git a/junixsocket-common/src/test/java/org/newsclub/net/unix/SelftestProvider.java b/junixsocket-common/src/test/java/org/newsclub/net/unix/SelftestProvider.java index 153b52dae..a0cc9ea5a 100644 --- a/junixsocket-common/src/test/java/org/newsclub/net/unix/SelftestProvider.java +++ b/junixsocket-common/src/test/java/org/newsclub/net/unix/SelftestProvider.java @@ -128,6 +128,7 @@ public SelftestProvider() { registerTest(COMMON, org.newsclub.net.unix.domain.VirtualThreadConnectTest.class); registerTest(COMMON, org.newsclub.net.unix.domain.VirtualThreadPollerNaiveDeadlockTest.class); + registerTest(COMMON, org.newsclub.net.unix.domain.VirtualThreadPollerTest.class); } public Set modulesDisabledByDefault() { diff --git a/junixsocket-common/src/test/java/org/newsclub/net/unix/domain/VirtualThreadPollerTest.java b/junixsocket-common/src/test/java/org/newsclub/net/unix/domain/VirtualThreadPollerTest.java new file mode 100644 index 000000000..a32295608 --- /dev/null +++ b/junixsocket-common/src/test/java/org/newsclub/net/unix/domain/VirtualThreadPollerTest.java @@ -0,0 +1,284 @@ +/* + * junixsocket + * + * Copyright 2009-2026 Christian Kohlschütter + * + * Licensed 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. + */ +package org.newsclub.net.unix.domain; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.io.IOException; +import java.net.SocketException; +import java.net.SocketTimeoutException; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.Map; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; + +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.newsclub.net.unix.AFUNIXServerSocket; +import org.newsclub.net.unix.AFUNIXSocket; +import org.newsclub.net.unix.AFUNIXSocketAddress; +import org.newsclub.net.unix.TestUtil; +import org.newsclub.net.unix.ThreadUtil; + +import com.kohlschutter.testutil.TestAbortedNotAnIssueException; + +/** + * Tests blocking socket operations on virtual threads, which are handled by the + * {@code VirtualThreadPoller}. + * + * See https://github.com/kohlschutter/junixsocket/issues/172 + * + * @author Jakub Kultys (Azahe) + */ +@SuppressWarnings({"PMD.DoNotUseThreads", "PMD.AvoidInstantiatingObjectsInsideLoops"}) +public class VirtualThreadPollerTest { + private static final int TRIGGER = 0; + private static final int WAITER = 1; + private static final int REPLY = 42; + + @BeforeEach + public void requireVirtualThreads() { + if (!ThreadUtil.isVirtualThreadSupported()) { + throw new TestAbortedNotAnIssueException("Virtual threads not supported"); + } + } + + /** + * Ensures that virtual-thread socket reads cannot starve each other, regardless of how many + * virtual threads are waiting for data at the same time. + * + * Many "waiter" connections block in {@code read()} until a "trigger" connection has received its + * reply (think: transactions waiting for a lock held by another transaction). If waiting for a + * file descriptor occupies a platform thread from a bounded pool, the waiters exhaust that pool + * and the trigger's reply is never noticed: deadlock. + */ + @Test + public void testMoreWaitersThanProcessors() throws Exception { // NOPMD.CognitiveComplexity + final int waiters = Runtime.getRuntime().availableProcessors() * 2; + + Path socketPath = TestUtil.newPathForUnixDomainSocket(); + CountDownLatch waitersConnected = new CountDownLatch(waiters); + CountDownLatch release = new CountDownLatch(1); + CountDownLatch done = new CountDownLatch(waiters + 1); + AtomicInteger failures = new AtomicInteger(); + + try (AFUNIXServerSocket server = AFUNIXServerSocket.newInstance()) { + AFUNIXSocketAddress address = AFUNIXSocketAddress.of(socketPath.toFile()); + server.bind(address); + + ThreadUtil.startNewDaemonThread(false, () -> { + while (!server.isClosed()) { + AFUNIXSocket socket; + try { + socket = server.accept(); + } catch (IOException e) { + return; + } + ThreadUtil.startNewDaemonThread(false, () -> { + try (AFUNIXSocket s = socket) { + if (s.getInputStream().read() == WAITER) { + waitersConnected.countDown(); + release.await(); + } + s.getOutputStream().write(REPLY); + s.getInputStream().read(); // wait for client to close + } catch (Exception e) { // NOPMD + // ignore + } + }); + } + }); + + ExecutorService virtualEs = ThreadUtil.newVirtualThreadPerTaskExecutor(); + try { + for (int i = 0; i < waiters; i++) { + virtualEs.submit(() -> connectAndRead(address, WAITER, failures, done, null)); + } + assertTrue(waitersConnected.await(5, TimeUnit.SECONDS)); + waitForPoller(waiters); + + virtualEs.submit(() -> connectAndRead(address, TRIGGER, failures, done, release)); + + assertTrue(done.await(10, TimeUnit.SECONDS), "Deadlock detected: " + done.getCount() + + " of " + (waiters + 1) + " connections did not complete"); + assertEquals(0, failures.get()); + } finally { + release.countDown(); // unblock any stuck waiters + virtualEs.shutdown(); + } + } finally { + Files.deleteIfExists(socketPath); + } + } + + @Test + public void testReadTimeout() throws Exception { + withConnectedPair((client, server) -> { + client.setSoTimeout(200); + Future f = ThreadUtil.newVirtualThreadPerTaskExecutor().submit(() -> { + long start = System.nanoTime(); + try { + client.getInputStream().read(); + return null; + } catch (SocketTimeoutException e) { + long elapsedMillis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start); + assertTrue(elapsedMillis >= 150, "Timed out too early: " + elapsedMillis + "ms"); + return e; + } + }); + assertTrue(f.get(5, TimeUnit.SECONDS) instanceof SocketTimeoutException); + }); + } + + @Test + public void testReadAfterTimeoutStillWorks() throws Exception { + withConnectedPair((client, server) -> { + client.setSoTimeout(100); + ExecutorService virtualEs = ThreadUtil.newVirtualThreadPerTaskExecutor(); + Future f = virtualEs.submit(() -> { + try { + client.getInputStream().read(); + throw new IllegalStateException("Expected timeout"); + } catch (SocketTimeoutException e) { + // expected + } + client.setSoTimeout(0); + return client.getInputStream().read(); + }); + Thread.sleep(300); + server.getOutputStream().write(REPLY); + assertEquals(REPLY, f.get(5, TimeUnit.SECONDS)); + }); + } + + @Test + public void testCloseWhileReading() throws Exception { + withConnectedPair((client, server) -> { + CountDownLatch reading = new CountDownLatch(1); + AtomicReference thrown = new AtomicReference<>(); + Future f = ThreadUtil.newVirtualThreadPerTaskExecutor().submit(() -> { + reading.countDown(); + try { + int r = client.getInputStream().read(); + thrown.set(new IllegalStateException("Unexpected read result: " + r)); + } catch (IOException e) { + thrown.set(e); + } + }); + assertTrue(reading.await(5, TimeUnit.SECONDS)); + Thread.sleep(200); // let the thread park + client.close(); + try { + f.get(5, TimeUnit.SECONDS); + } catch (ExecutionException e) { + throw new IllegalStateException(e); + } + assertTrue(thrown.get() instanceof SocketException, "Expected SocketException: " + thrown + .get()); + }); + } + + @FunctionalInterface + private interface PairConsumer { + void accept(AFUNIXSocket client, AFUNIXSocket server) throws Exception; + } + + private static void withConnectedPair(PairConsumer consumer) throws Exception { + Path socketPath = TestUtil.newPathForUnixDomainSocket(); + try (AFUNIXServerSocket server = AFUNIXServerSocket.newInstance()) { + AFUNIXSocketAddress address = AFUNIXSocketAddress.of(socketPath.toFile()); + server.bind(address); + try (AFUNIXSocket client = AFUNIXSocket.connectTo(address); + AFUNIXSocket serverSide = server.accept()) { + consumer.accept(client, serverSide); + } + } finally { + Files.deleteIfExists(socketPath); + } + } + + private static void connectAndRead(AFUNIXSocketAddress address, int role, AtomicInteger failures, + CountDownLatch done, CountDownLatch onReply) { + try (AFUNIXSocket socket = AFUNIXSocket.newInstance()) { + socket.connect(address); + socket.getOutputStream().write(role); + if (socket.getInputStream().read() != REPLY) { + failures.incrementAndGet(); + } + if (onReply != null) { + onReply.countDown(); + } + } catch (Exception e) { // NOPMD + failures.incrementAndGet(); + } finally { + done.countDown(); + } + } + + /** + * Waits until the waiters are likely parked in the poller: until there are as many platform + * threads inside {@code NativeUnixSocket.poll} as there are waiters (one-thread-per-fd pollers), + * or until the number stops growing (multiplexing pollers). + */ + private static void waitForPoller(int waiters) throws InterruptedException { + long deadlineNanos = System.nanoTime() + TimeUnit.SECONDS.toNanos(5); + + int previous = -1; + int stableRounds = 0; + while (System.nanoTime() < deadlineNanos) { + int polling = countPlatformThreadsInPoll(); + if (polling >= waiters) { + return; + } + if (polling > 0 && polling == previous) { + if (++stableRounds >= 10) { + return; + } + } else { + stableRounds = 0; + } + previous = polling; + Thread.sleep(25); + } + } + + private static int countPlatformThreadsInPoll() { + final String nativeUnixSocketClassName = TestUtil.getNativeUnixSocketClassName(); + int polling = 0; + for (Map.Entry entry : Thread.getAllStackTraces().entrySet()) { + if (ThreadUtil.isTrulyAVirtualThread(entry.getKey())) { + continue; + } + for (StackTraceElement element : entry.getValue()) { + if ("poll".equals(element.getMethodName()) && nativeUnixSocketClassName.equals(element + .getClassName())) { + polling++; + break; + } + } + } + return polling; + } +}