Skip to content

Commit 5ddd2d8

Browse files
Azaheclaude
andcommitted
common: Multiplex virtual-thread polling onto a shared poller thread
VirtualThreadPollerNaive occupies a platform thread per waiting file descriptor, taken from a bounded pool. With >= pool-size interdependent waiters, the fd that would release them is never polled: deadlock (Java <= 24 masked this on 1-2 CPUs via CompletableFuture's thread-per-task fallback for a parallelism-1 commonPool). Add VirtualThreadPollerShared (default): virtual threads register (fd, ops, thread) and park; a single platform thread polls all fds plus a wakeup pipe and unparks ready threads. ~3x throughput at 16+ connections, constant platform thread count. The naive poller remains available via -Dorg.newsclub.net.unix.VirtualThreadPoller.naive=true and now uses ForkJoinPool.managedBlock so it cannot starve either. #172 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
1 parent 656dca2 commit 5ddd2d8

5 files changed

Lines changed: 705 additions & 28 deletions

File tree

‎junixsocket-common/src/main/java/org/newsclub/net/unix/VirtualThreadPoller.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -33,7 +33,7 @@ interface VirtualThreadPoller {
3333
/**
3434
* Returns the default instance best suited for the current system.
3535
*/
36-
VirtualThreadPoller INSTANCE = new VirtualThreadPollerNaive();
36+
VirtualThreadPoller INSTANCE = VirtualThreadPollerShared.newDefaultInstance();
3737

3838
/**
3939
* Parks the current thread until the given file descriptor is ready, with respect to the given

‎junixsocket-common/src/main/java/org/newsclub/net/unix/VirtualThreadPollerNaive.java‎

Lines changed: 65 additions & 27 deletions
Original file line numberDiff line numberDiff line change
@@ -27,6 +27,7 @@
2727
import java.util.Map;
2828
import java.util.concurrent.ConcurrentHashMap;
2929
import java.util.concurrent.ExecutionException;
30+
import java.util.concurrent.ForkJoinPool;
3031
import java.util.concurrent.locks.LockSupport;
3132

3233
import org.eclipse.jdt.annotation.Nullable;
@@ -70,39 +71,76 @@ private static final class PollJob {
7071
this.timeout = timeout;
7172
}
7273

73-
@SuppressWarnings("PMD.CognitiveComplexity")
74+
private final class PollBlocker implements ForkJoinPool.ManagedBlocker {
75+
private @Nullable IOException result;
76+
private boolean done;
77+
78+
@Override
79+
public boolean block() {
80+
result = pollUntilReady();
81+
done = true;
82+
return true;
83+
}
84+
85+
@Override
86+
public boolean isReleasable() {
87+
return done;
88+
}
89+
}
90+
91+
/**
92+
* Polls until the file descriptor is ready, the timeout has elapsed, or the thread is
93+
* interrupted.
94+
*
95+
* @return {@code null} if ready or timed out, or an exception.
96+
*/
97+
private @Nullable IOException pollUntilReady() {
98+
Thread thread = Thread.currentThread();
99+
PollFd pfd = new PollFd(new FileDescriptor[] {fd}, new int[] {mode});
100+
do {
101+
if (thread.isInterrupted() || !fd.valid()) {
102+
return POLL_INTERRUPTED_SENTINEL;
103+
}
104+
try {
105+
NativeUnixSocket.poll(pfd, POLL_INTERVAL_MILLIS);
106+
} catch (IOException e) {
107+
return e;
108+
}
109+
if (thread.isInterrupted() || !fd.valid()) {
110+
return POLL_INTERRUPTED_SENTINEL;
111+
}
112+
if (pfd.rops[0] != 0) {
113+
return null;
114+
}
115+
116+
int timeoutMillis = timeout.get();
117+
if (timeoutMillis > 0) {
118+
if ((System.currentTimeMillis() - now) >= timeoutMillis) {
119+
// handle in calling thread
120+
return null;
121+
}
122+
}
123+
} while (true); // NOPMD.WhileLoopWithLiteralBoolean
124+
}
125+
74126
AFFuture<@Nullable IOException> trigger(Thread waitingThread) {
75127
synchronized (fd) {
76128
waitingThreads.add(waitingThread);
77129
}
78130
return AFFuture.supplyAsync(() -> {
79131
try {
80-
Thread thread = Thread.currentThread();
81-
PollFd pfd = new PollFd(new FileDescriptor[] {fd}, new int[] {mode});
82-
do {
83-
if (thread.isInterrupted() || !fd.valid()) {
84-
return POLL_INTERRUPTED_SENTINEL;
85-
}
86-
try {
87-
NativeUnixSocket.poll(pfd, POLL_INTERVAL_MILLIS);
88-
} catch (IOException e) {
89-
return e;
90-
}
91-
if (thread.isInterrupted() || !fd.valid()) {
92-
return POLL_INTERRUPTED_SENTINEL;
93-
}
94-
if (pfd.rops[0] != 0) {
95-
break;
96-
}
97-
98-
int timeoutMillis = timeout.get();
99-
if (timeoutMillis > 0) {
100-
if ((System.currentTimeMillis() - now) >= timeoutMillis) {
101-
// handle in calling thread
102-
break;
103-
}
104-
}
105-
} while (true); // NOPMD.WhileLoopWithLiteralBoolean
132+
// poll() may block this worker for a long time; let the pool compensate, otherwise
133+
// as few as parallelism waiting threads can starve the one we're waiting for
134+
// see https://github.com/kohlschutter/junixsocket/issues/172
135+
PollBlocker blocker = new PollBlocker();
136+
try {
137+
ForkJoinPool.managedBlock(blocker);
138+
} catch (InterruptedException e) {
139+
return POLL_INTERRUPTED_SENTINEL;
140+
}
141+
if (blocker.result != null) {
142+
return blocker.result;
143+
}
106144
} finally {
107145
Thread threadToWake = null;
108146
try {

0 commit comments

Comments
 (0)