From e7d76eec688b5f86e5d7956772f6d080b3ab05b8 Mon Sep 17 00:00:00 2001 From: Francesco Nigro Date: Mon, 29 Jun 2026 14:37:51 +0200 Subject: [PATCH 1/6] Add stickyAffinity() to Thread.Builder.OfVirtual Virtual threads with sticky affinity preserve carrier locality: - start/unpark from a sticky VT uses lazy submission (local queue, no signal) - sub-pollers (mode 2 and 3) are made sticky, simplifying Poller.polled() --- .../share/classes/java/lang/Thread.java | 17 +++ .../classes/java/lang/ThreadBuilders.java | 10 ++ .../classes/java/lang/VirtualThread.java | 14 ++- .../share/classes/sun/nio/ch/Poller.java | 8 +- .../Thread/virtual/StickyAffinityTest.java | 119 ++++++++++++++++++ 5 files changed, 161 insertions(+), 7 deletions(-) create mode 100644 test/jdk/java/lang/Thread/virtual/StickyAffinityTest.java diff --git a/src/java.base/share/classes/java/lang/Thread.java b/src/java.base/share/classes/java/lang/Thread.java index 87029dd4771..877594626e7 100644 --- a/src/java.base/share/classes/java/lang/Thread.java +++ b/src/java.base/share/classes/java/lang/Thread.java @@ -704,6 +704,13 @@ public static void onSpinWait() {} */ static final int NO_INHERIT_THREAD_LOCALS = 1 << 2; + /** + * Characteristic value signifying that this virtual thread has sticky affinity. + * When a sticky virtual thread starts or unparks another virtual thread, + * the runtime uses lazy submission to preserve thread locality. + */ + static final int STICKY_AFFINITY = 1 << 3; + /** * Thread identifier assigned to the primordial thread. */ @@ -1267,6 +1274,16 @@ sealed interface OfVirtual extends Builder @Override OfVirtual inheritInheritableThreadLocals(boolean inherit); @Override OfVirtual uncaughtExceptionHandler(UncaughtExceptionHandler ueh); + /** + * Sets this builder to create virtual threads with sticky affinity. + * When a sticky virtual thread starts or unparks another virtual thread, + * the runtime uses lazy submission to preserve thread locality. + * + * @return this builder + * @since 99 + */ + OfVirtual stickyAffinity(); + /** * Creates a new {@code Thread} from the current state of the builder and * schedules it without guaranteeing that it will eventually execute. diff --git a/src/java.base/share/classes/java/lang/ThreadBuilders.java b/src/java.base/share/classes/java/lang/ThreadBuilders.java index 4a538f2389e..b73a82b0906 100644 --- a/src/java.base/share/classes/java/lang/ThreadBuilders.java +++ b/src/java.base/share/classes/java/lang/ThreadBuilders.java @@ -96,6 +96,10 @@ void setInheritInheritableThreadLocals(boolean inherit) { } } + void setStickyAffinity() { + characteristics |= Thread.STICKY_AFFINITY; + } + void setUncaughtExceptionHandler(UncaughtExceptionHandler ueh) { this.uhe = Objects.requireNonNull(ueh); } @@ -237,6 +241,12 @@ public OfVirtual uncaughtExceptionHandler(UncaughtExceptionHandler ueh) { return this; } + @Override + public OfVirtual stickyAffinity() { + setStickyAffinity(); + return this; + } + Thread unstarted(Runnable task, Thread preferredCarrier) { Objects.requireNonNull(task); var thread = newVirtualThread(scheduler, diff --git a/src/java.base/share/classes/java/lang/VirtualThread.java b/src/java.base/share/classes/java/lang/VirtualThread.java index d647735b83b..4288ab82072 100644 --- a/src/java.base/share/classes/java/lang/VirtualThread.java +++ b/src/java.base/share/classes/java/lang/VirtualThread.java @@ -106,6 +106,7 @@ final class VirtualThread extends BaseVirtualThread { private final VirtualThreadScheduler scheduler; private final Continuation cont; private final VThreadTask runContinuation; + private final boolean stickyAffinity; // virtual thread state, accessed by VM private volatile int state; @@ -233,6 +234,13 @@ static VirtualThreadScheduler defaultScheduler() { return DEFAULT_SCHEDULER; } + /** + * Returns true if the current thread is a virtual thread with sticky affinity. + */ + static boolean currentThreadIsSticky() { + return currentThread() instanceof VirtualThread vt && vt.stickyAffinity; + } + /** * Returns the continuation scope used for virtual threads. */ @@ -271,6 +279,7 @@ VirtualThreadTask virtualThreadTask() { throw new UnsupportedOperationException(); } this.scheduler = scheduler; + this.stickyAffinity = (characteristics & Thread.STICKY_AFFINITY) != 0; this.cont = new VThreadContinuation(this, task); if (scheduler == BUILTIN_SCHEDULER) { @@ -799,13 +808,14 @@ private void start(ThreadContainer container, boolean lazy) { // submit task to schedule try { if (currentThread().isVirtual()) { + boolean useLazy = lazy || currentThreadIsSticky(); Continuation.pin(); try { if (scheduler == BUILTIN_SCHEDULER && currentCarrierThread() instanceof CarrierThread ct) { ForkJoinPool pool = ct.getPool(); ForkJoinTask task = ForkJoinTask.adapt(runContinuation); - if (lazy) { + if (useLazy) { pool.lazySubmit(task); } else { pool.externalSubmit(task); @@ -991,7 +1001,7 @@ private void unpark(boolean lazySubmit) { // unparked while parked if ((s == PARKED || s == TIMED_PARKED) && compareAndSetState(s, UNPARKED)) { - if (lazySubmit && currentThread().isVirtual()) { + if ((lazySubmit || currentThreadIsSticky()) && currentThread().isVirtual()) { Continuation.pin(); try { if (scheduler == BUILTIN_SCHEDULER diff --git a/src/java.base/share/classes/sun/nio/ch/Poller.java b/src/java.base/share/classes/sun/nio/ch/Poller.java index 6762f939c90..fdcdf355aa4 100644 --- a/src/java.base/share/classes/sun/nio/ch/Poller.java +++ b/src/java.base/share/classes/sun/nio/ch/Poller.java @@ -207,11 +207,7 @@ void wakeupPoller() throws IOException { final void polled(int fdVal) { Thread t = map.remove(fdVal); if (t != null) { - if (POLLER_GROUP.useLazyUnpark() && Thread.currentThread().isVirtual()) { - JLA.lazyUnparkVirtualThread(t); - } else { - LockSupport.unpark(t); - } + LockSupport.unpark(t); } } @@ -539,6 +535,7 @@ private static class VThreadsPollerGroup extends PollerGroup { ThreadFactory factory = Thread.ofVirtual() .inheritInheritableThreadLocals(false) + .stickyAffinity() .name("SubPoller-", 0) .uncaughtExceptionHandler((_, e) -> e.printStackTrace()) .factory(); @@ -667,6 +664,7 @@ private Poller startReadPoller() throws IOException { Thread carrier = JLA.currentCarrierThread(); Thread.Builder.OfVirtual builder = Thread.ofVirtual() .inheritInheritableThreadLocals(false) + .stickyAffinity() .name(carrier.getName() + "-Read-Poller") .uncaughtExceptionHandler((_, e) -> e.printStackTrace()); Thread thread = JLA.defaultVirtualThreadScheduler() diff --git a/test/jdk/java/lang/Thread/virtual/StickyAffinityTest.java b/test/jdk/java/lang/Thread/virtual/StickyAffinityTest.java new file mode 100644 index 00000000000..42d082bc6cf --- /dev/null +++ b/test/jdk/java/lang/Thread/virtual/StickyAffinityTest.java @@ -0,0 +1,119 @@ +/* + * Copyright (c) 2025, Oracle and/or its affiliates. All rights reserved. + * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER. + * + * This code is free software; you can redistribute it and/or modify it + * under the terms of the GNU General Public License version 2 only, as + * published by the Free Software Foundation. + * + * This code is distributed in the hope that it will be useful, but WITHOUT + * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or + * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License + * version 2 for more details (a copy is included in the LICENSE file that + * accompanied this code). + * + * You should have received a copy of the GNU General Public License version + * 2 along with this work; if not, write to the Free Software Foundation, + * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA. + * + * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA + * or visit www.oracle.com if you need additional information or have any + * questions. + */ + +/** + * @test + * @summary Test virtual threads with sticky affinity + * @requires vm.continuations + * @modules java.base/java.lang:+open + * @library /test/lib + * @run junit StickyAffinityTest + */ + +import java.lang.reflect.Field; +import java.util.concurrent.*; +import java.util.concurrent.atomic.*; +import java.util.concurrent.locks.LockSupport; + +import org.junit.jupiter.api.Test; +import static org.junit.jupiter.api.Assertions.*; + +class StickyAffinityTest { + + private static Thread currentCarrierThread() throws Exception { + Field f = Class.forName("java.lang.VirtualThread") + .getDeclaredField("carrierThread"); + f.setAccessible(true); + return (Thread) f.get(Thread.currentThread()); + } + + /** + * Test that stickyAffinity() builder method creates and starts a thread. + */ + @Test + void testBuilderApi() throws Exception { + var ran = new AtomicBoolean(); + Thread thread = Thread.ofVirtual() + .stickyAffinity() + .start(() -> ran.set(true)); + thread.join(); + assertTrue(ran.get()); + } + + /** + * Test that stickyAffinity works with factory(). + */ + @Test + void testFactoryApi() throws Exception { + ThreadFactory factory = Thread.ofVirtual() + .stickyAffinity() + .name("sticky-", 0) + .factory(); + var ran = new AtomicBoolean(); + Thread thread = factory.newThread(() -> ran.set(true)); + thread.start(); + thread.join(); + assertTrue(ran.get()); + assertTrue(thread.getName().startsWith("sticky-")); + } + + /** + * Test that when a sticky VT unparks another VT, the unparked VT resumes + * on the same carrier as the sticky VT (builtin scheduler). + */ + @Test + void testStickyUnparkPreservesCarrier() throws Exception { + var stickyCarrier = new AtomicReference(); + var targetCarrierAfterUnpark = new AtomicReference(); + var parked = new CountDownLatch(1); + var done = new CountDownLatch(1); + + Thread target = Thread.ofVirtual().start(() -> { + parked.countDown(); + LockSupport.park(); + try { + targetCarrierAfterUnpark.set(currentCarrierThread()); + } catch (Exception e) { throw new RuntimeException(e); } + done.countDown(); + }); + parked.await(); + + Thread sticky = Thread.ofVirtual() + .stickyAffinity() + .start(() -> { + try { + stickyCarrier.set(currentCarrierThread()); + } catch (Exception e) { throw new RuntimeException(e); } + LockSupport.unpark(target); + }); + sticky.join(); + assertTrue(done.await(5, TimeUnit.SECONDS)); + target.join(); + + assertNotNull(stickyCarrier.get()); + assertNotNull(targetCarrierAfterUnpark.get()); + assertEquals(stickyCarrier.get(), targetCarrierAfterUnpark.get(), + "unparked VT should resume on the sticky VT's carrier"); + } + +} From bbd44c55c1494d54a50c8fcb157248bbae271b72 Mon Sep 17 00:00:00 2001 From: Francesco Nigro Date: Tue, 30 Jun 2026 10:16:18 +0200 Subject: [PATCH 2/6] Sticky VTs skip externalSubmit in afterYield --- src/java.base/share/classes/java/lang/VirtualThread.java | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/src/java.base/share/classes/java/lang/VirtualThread.java b/src/java.base/share/classes/java/lang/VirtualThread.java index 4288ab82072..508e0a2a6f5 100644 --- a/src/java.base/share/classes/java/lang/VirtualThread.java +++ b/src/java.base/share/classes/java/lang/VirtualThread.java @@ -679,8 +679,10 @@ private void afterYield() { if (s == YIELDING) { setState(YIELDED); - // external submit if there are no tasks in the local task queue - if (currentThread() instanceof CarrierThread ct && ct.getQueuedTaskCount() == 0) { + // sticky VTs stay on the current carrier — skip external submit + if (!stickyAffinity + && currentThread() instanceof CarrierThread ct + && ct.getQueuedTaskCount() == 0) { externalSubmitRunContinuation(); } else { submitRunContinuation(); From 6e7e35e0e9e5bd80adc883c446869ef178701c58 Mon Sep 17 00:00:00 2001 From: Francesco Nigro Date: Tue, 30 Jun 2026 10:16:19 +0200 Subject: [PATCH 3/6] Restore useLazyUnpark dispatch, limit stickyAffinity to mode 3 sub-pollers MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Mode 2 sub-pollers no longer use stickyAffinity — they revert to the original useLazyUnpark dispatch in polled(). Only mode 3 (per-carrier) sub-pollers are sticky. --- src/java.base/share/classes/sun/nio/ch/Poller.java | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/src/java.base/share/classes/sun/nio/ch/Poller.java b/src/java.base/share/classes/sun/nio/ch/Poller.java index fdcdf355aa4..0757412232a 100644 --- a/src/java.base/share/classes/sun/nio/ch/Poller.java +++ b/src/java.base/share/classes/sun/nio/ch/Poller.java @@ -207,7 +207,11 @@ void wakeupPoller() throws IOException { final void polled(int fdVal) { Thread t = map.remove(fdVal); if (t != null) { - LockSupport.unpark(t); + if (POLLER_GROUP.useLazyUnpark() && Thread.currentThread().isVirtual()) { + JLA.lazyUnparkVirtualThread(t); + } else { + LockSupport.unpark(t); + } } } @@ -535,7 +539,6 @@ private static class VThreadsPollerGroup extends PollerGroup { ThreadFactory factory = Thread.ofVirtual() .inheritInheritableThreadLocals(false) - .stickyAffinity() .name("SubPoller-", 0) .uncaughtExceptionHandler((_, e) -> e.printStackTrace()) .factory(); From 1d9a237cd2ef1a166cc373b637ed4f32e6a95463 Mon Sep 17 00:00:00 2001 From: Francesco Nigro Date: Fri, 10 Jul 2026 10:01:08 +0200 Subject: [PATCH 4/6] Add MPSC virtual thread scheduler Single MPSC queue per carrier, affinityHint routing, roundRobinAffinity support, preferredCarrier, drain budget, plainLoop fallback. Includes MpscUnboundedQueue with @Contended fields and onSpinWait spin loop, lazyUnpark on BaseVirtualThread, ROUND_ROBIN_AFFINITY characteristic on Thread.OfVirtual, and useMpsc activation in VirtualThread. --- .../classes/java/lang/BaseVirtualThread.java | 8 + .../classes/java/lang/MpscUnboundedQueue.java | 312 ++++++++++++++++++ .../java/lang/MpscVirtualThreadScheduler.java | 250 ++++++++++++++ .../share/classes/java/lang/Thread.java | 33 ++ .../classes/java/lang/ThreadBuilders.java | 32 ++ .../classes/java/lang/VirtualThread.java | 42 ++- 6 files changed, 672 insertions(+), 5 deletions(-) create mode 100644 src/java.base/share/classes/java/lang/MpscUnboundedQueue.java create mode 100644 src/java.base/share/classes/java/lang/MpscVirtualThreadScheduler.java diff --git a/src/java.base/share/classes/java/lang/BaseVirtualThread.java b/src/java.base/share/classes/java/lang/BaseVirtualThread.java index f0d02f5dbf3..36170f776cb 100644 --- a/src/java.base/share/classes/java/lang/BaseVirtualThread.java +++ b/src/java.base/share/classes/java/lang/BaseVirtualThread.java @@ -63,5 +63,13 @@ abstract sealed class BaseVirtualThread extends Thread * Makes available the parking permit to the given this virtual thread. */ abstract void unpark(); + + /** + * Makes available the parking permit to the given this virtual thread. If the + * thread is parked then there is no guarantee that it will continue execution. + */ + void lazyUnpark() { + unpark(); + } } diff --git a/src/java.base/share/classes/java/lang/MpscUnboundedQueue.java b/src/java.base/share/classes/java/lang/MpscUnboundedQueue.java new file mode 100644 index 00000000000..2289a887077 --- /dev/null +++ b/src/java.base/share/classes/java/lang/MpscUnboundedQueue.java @@ -0,0 +1,312 @@ +/* + * Copyright (c) 2026, Oracle and/or its affiliates. All rights reserved. + * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER. + * + * This code is free software; you can redistribute it and/or modify it + * under the terms of the GNU General Public License version 2 only, as + * published by the Free Software Foundation. Oracle designates this + * particular file as subject to the "Classpath" exception as provided + * by Oracle in the LICENSE file that accompanied this code. + * + * This code is distributed in the hope that it will be useful, but WITHOUT + * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or + * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License + * version 2 for more details (a copy is included in the LICENSE file that + * accompanied this code). + * + * You should have received a copy of the GNU General Public License version + * 2 along with this work; if not, write to the Free Software Foundation, + * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA. + * + * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA + * or visit www.oracle.com if you need additional information or have any + * questions. + */ +package java.lang; + +import java.lang.invoke.MethodHandles; +import java.lang.invoke.VarHandle; +import jdk.internal.vm.annotation.Contended; + +/** + * Multi-Producer Single-Consumer unbounded array queue using VarHandles. + * Based on JCTools MpscUnboundedArrayQueue but self-contained. + * + * @param the type of elements held in this queue + */ +final class MpscUnboundedQueue { + + private static final VarHandle PRODUCER_INDEX; + private static final VarHandle CONSUMER_INDEX; + private static final VarHandle PRODUCER_LIMIT; + private static final VarHandle ARRAY; + + static { + try { + MethodHandles.Lookup lookup = MethodHandles.lookup(); + PRODUCER_INDEX = lookup.findVarHandle(MpscUnboundedQueue.class, "producerIndex", long.class); + CONSUMER_INDEX = lookup.findVarHandle(MpscUnboundedQueue.class, "consumerIndex", long.class); + PRODUCER_LIMIT = lookup.findVarHandle(MpscUnboundedQueue.class, "producerLimit", long.class); + ARRAY = MethodHandles.arrayElementVarHandle(Object[].class); + } catch (ReflectiveOperationException e) { + throw new ExceptionInInitializerError(e); + } + } + + private static final Object JUMP = new Object(); + private static final Object BUFFER_CONSUMED = new Object(); + private static final int CONTINUE_TO_P_INDEX_CAS = 0; + private static final int RETRY = 1; + private static final int QUEUE_RESIZE = 3; + + private static final long RESIZE_BIT = 1L; + + // producer-written fields + @Contended("producer") + @SuppressWarnings("FieldMayBeFinal") + private long producerIndex; + @Contended("producer") + @SuppressWarnings("FieldMayBeFinal") + private long producerLimit; + @Contended("producer") + private long producerMask; + @Contended("producer") + private E[] producerBuffer; + + // consumer-written fields + @Contended("consumer") + @SuppressWarnings("FieldMayBeFinal") + private long consumerIndex; + @Contended("consumer") + private long consumerMask; + @Contended("consumer") + private E[] consumerBuffer; + + MpscUnboundedQueue(int initialCapacity) { + if (initialCapacity < 2) { + throw new IllegalArgumentException("Initial capacity must be 2 or more"); + } + int p2capacity = roundToPowerOfTwo(initialCapacity); + long mask = (p2capacity - 1L) << 1; + @SuppressWarnings("unchecked") + E[] buffer = (E[]) new Object[p2capacity + 1]; + producerBuffer = buffer; + consumerBuffer = buffer; + producerMask = mask; + consumerMask = mask; + soProducerLimit(mask); + } + + private static int roundToPowerOfTwo(int value) { + if (value <= 0) { + throw new IllegalArgumentException("Must be positive"); + } + return 1 << (32 - Integer.numberOfLeadingZeros(value - 1)); + } + + private void soProducerLimit(long v) { + PRODUCER_LIMIT.setRelease(this, v); + } + + private long lvProducerLimit() { + return (long) PRODUCER_LIMIT.getAcquire(this); + } + + private long lvProducerIndex() { + return (long) PRODUCER_INDEX.getAcquire(this); + } + + private boolean casProducerIndex(long expect, long newValue) { + return PRODUCER_INDEX.compareAndSet(this, expect, newValue); + } + + private long lvConsumerIndex() { + return (long) CONSUMER_INDEX.getAcquire(this); + } + + private void soConsumerIndex(long v) { + CONSUMER_INDEX.setRelease(this, v); + } + + private void soProducerIndex(long v) { + PRODUCER_INDEX.setRelease(this, v); + } + + private boolean casProducerLimit(long expect, long newValue) { + return PRODUCER_LIMIT.compareAndSet(this, expect, newValue); + } + + private static void soRefElement(E[] buffer, int offset, E e) { + ARRAY.setRelease(buffer, offset, e); + } + + @SuppressWarnings("unchecked") + private static E lvRefElement(E[] buffer, int offset) { + return (E) ARRAY.getAcquire(buffer, offset); + } + + void offer(E e) { + if (null == e) { + throw new NullPointerException(); + } + + long mask; + E[] buffer; + long pIndex; + + while (true) { + long producerLimit = lvProducerLimit(); + pIndex = lvProducerIndex(); + if ((pIndex & RESIZE_BIT) == 1) { + continue; + } + + mask = this.producerMask; + buffer = this.producerBuffer; + + if (producerLimit <= pIndex) { + int result = offerSlowPath(mask, pIndex, producerLimit); + switch (result) { + case CONTINUE_TO_P_INDEX_CAS: + break; + case RETRY: + continue; + case QUEUE_RESIZE: + resize(mask, buffer, pIndex, e); + return; + } + } + + if (casProducerIndex(pIndex, pIndex + 2)) { + break; + } + } + final int offset = modifiedCalcCircularRefElementOffset(pIndex, mask); + soRefElement(buffer, offset, e); + } + + private int offerSlowPath(long mask, long pIndex, long producerLimit) { + final long cIndex = lvConsumerIndex(); + long bufferCapacity = mask; + if (cIndex + bufferCapacity > pIndex) { + if (!casProducerLimit(producerLimit, cIndex + bufferCapacity)) { + return RETRY; + } + return CONTINUE_TO_P_INDEX_CAS; + } + if (casProducerIndex(pIndex, pIndex + 1)) { + return QUEUE_RESIZE; + } + return RETRY; + } + + private void resize(long oldMask, E[] oldBuffer, long pIndex, final E e) { + int newBufferLength = oldBuffer.length; + @SuppressWarnings("unchecked") + final E[] newBuffer = (E[]) new Object[newBufferLength]; + + producerBuffer = newBuffer; + final int newMask = (newBufferLength - 2) << 1; + producerMask = newMask; + + final int offsetInOld = modifiedCalcCircularRefElementOffset(pIndex, oldMask); + final int offsetInNew = modifiedCalcCircularRefElementOffset(pIndex, newMask); + + soRefElement(newBuffer, offsetInNew, e); + soRefElement(oldBuffer, nextArrayOffset(oldMask), newBuffer); + + final long cIndex = lvConsumerIndex(); + final long availableInQueue = Integer.MAX_VALUE - (pIndex - cIndex); + if (availableInQueue <= 0) { + throw new IllegalStateException(); + } + + soProducerLimit(pIndex + Math.min(newMask, availableInQueue)); + soProducerIndex(pIndex + 2); + soRefElement(oldBuffer, offsetInOld, JUMP); + } + + private int nextArrayOffset(final long mask) { + return modifiedCalcCircularRefElementOffset(mask + 2, Long.MAX_VALUE); + } + + private static int modifiedCalcCircularRefElementOffset(long index, long mask) { + return (int) ((index & mask) >> 1); + } + + @SuppressWarnings("unchecked") + E poll() { + final E[] buffer = consumerBuffer; + final long index = consumerIndex; + final long mask = consumerMask; + + final int offset = modifiedCalcCircularRefElementOffset(index, mask); + Object e = lvRefElement(buffer, offset); + if (e == null) { + long pIndex = lvProducerIndex(); + pIndex += (pIndex & RESIZE_BIT); + if (index == pIndex) { + return null; + } + do { + Thread.onSpinWait(); + e = lvRefElement(buffer, offset); + } while (e == null); + } + if (e == JUMP) { + final E[] nextBuffer = nextBuffer(buffer, mask); + return newBufferPoll(nextBuffer, index); + } + soRefElement(buffer, offset, null); + soConsumerIndex(index + 2); + return (E) e; + } + + private E[] nextBuffer(final E[] buffer, final long mask) { + final int nextArrayOffset = nextArrayOffset(mask); + @SuppressWarnings("unchecked") + final E[] nextBuffer = (E[]) lvRefElement(buffer, nextArrayOffset); + consumerBuffer = nextBuffer; + consumerMask = (nextBuffer.length - 2L) << 1; + soRefElement(buffer, nextArrayOffset, BUFFER_CONSUMED); + return nextBuffer; + } + + private E newBufferPoll(E[] nextBuffer, final long index) { + final int offset = modifiedCalcCircularRefElementOffset(index, consumerMask); + final E n = lvRefElement(nextBuffer, offset); + if (n == null) { + throw new IllegalStateException("new buffer must have at least one element"); + } + soRefElement(nextBuffer, offset, null); + soConsumerIndex(index + 2); + return n; + } + + boolean isEmpty() { + long cIndex = lvConsumerIndex(); + long pIndex = lvProducerIndex(); + pIndex += (pIndex & RESIZE_BIT); + return cIndex == pIndex; + } + + int size() { + long after = lvConsumerIndex(); + long size; + while (true) { + final long before = after; + final long currentProducerIndex = lvProducerIndex(); + after = lvConsumerIndex(); + if (before == after) { + long pIndex = currentProducerIndex; + pIndex += (pIndex & RESIZE_BIT); + size = (pIndex - after) >> 1; + break; + } + } + if (size > Integer.MAX_VALUE) { + return Integer.MAX_VALUE; + } + return (int) size; + } +} diff --git a/src/java.base/share/classes/java/lang/MpscVirtualThreadScheduler.java b/src/java.base/share/classes/java/lang/MpscVirtualThreadScheduler.java new file mode 100644 index 00000000000..27303a64e39 --- /dev/null +++ b/src/java.base/share/classes/java/lang/MpscVirtualThreadScheduler.java @@ -0,0 +1,250 @@ +/* + * Copyright (c) 2026, Oracle and/or its affiliates. All rights reserved. + * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER. + * + * This code is free software; you can redistribute it and/or modify it + * under the terms of the GNU General Public License version 2 only, as + * published by the Free Software Foundation. Oracle designates this + * particular file as subject to the "Classpath" exception as provided + * by Oracle in the LICENSE file that accompanied this code. + * + * This code is distributed in the hope that it will be useful, but WITHOUT + * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or + * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License + * version 2 for more details (a copy is included in the LICENSE file that + * accompanied this code). + * + * You should have received a copy of the GNU General Public License version + * 2 along with this work; if not, write to the Free Software Foundation, + * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA. + * + * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA + * or visit www.oracle.com if you need additional information or have any + * questions. + */ +package java.lang; + +import java.io.IOException; +import java.io.UncheckedIOException; +import java.lang.Thread.VirtualThreadScheduler; +import java.lang.Thread.VirtualThreadTask; +import java.util.concurrent.locks.LockSupport; +import jdk.internal.misc.Unsafe; +import jdk.internal.vm.annotation.Contended; +import sun.nio.ch.CarrierLocalPoller; + +/** + * An alternative virtual thread scheduler using a single MPSC queue per carrier. + * No work stealing — each carrier drains only its own queue. + * + *

External submissions use a probe-based hash (FJP-style) to distribute + * across carriers. Carrier affinity is set once at start via affinityHint; + * onContinue always routes back to the same carrier. + * + *

With poller Mode 4 (CARRIER_LOCAL_POLLER), each carrier owns its own + * epoll fd. VT fds register directly — no sub-pollers, no master poller. + * The carrier interleaves task draining with I/O polling. + */ +final class MpscVirtualThreadScheduler implements VirtualThreadScheduler { + + private static final Unsafe U = Unsafe.getUnsafe(); + + private static final long PROBE = + U.objectFieldOffset(Thread.class, "threadLocalRandomProbe"); + + private final CarrierThread[] carriers; + + MpscVirtualThreadScheduler(int parallelism) { + if (parallelism < 1) { + throw new IllegalArgumentException("parallelism must be >= 1"); + } + this.carriers = new CarrierThread[parallelism]; + for (int i = 0; i < parallelism; i++) { + carriers[i] = new CarrierThread(i, this); + } + for (int i = 0; i < parallelism; i++) { + carriers[i].start(); + } + } + + @Override + public void onStart(VirtualThreadTask task) { + VirtualThread vt = (VirtualThread) task.thread(); + CarrierThread target; + if (vt.affinityHint >= 0) { + target = carriers[Math.floorMod(vt.affinityHint, carriers.length)]; + } else { + target = carrierFor(); + } + vt.affinityHint = target.id; + enqueue(target, task); + } + + @Override + public void onContinue(VirtualThreadTask task) { + int hint = ((VirtualThread) task.thread()).affinityHint; + if (hint >= 0 && hint < carriers.length) { + enqueue(carriers[hint], task); + return; + } + onStart(task); + } + + private static void enqueue(CarrierThread carrier, VirtualThreadTask task) { + carrier.queue.offer(task); + if (carrier.carrierState == CarrierThread.PARKED) { + if (carrier.poller != null) { + try { + carrier.poller.wakeup(); + } catch (IOException e) { + LockSupport.unpark(carrier); + } + } else { + LockSupport.unpark(carrier); + } + } + } + + private CarrierThread carrierFor() { + Thread caller = Thread.currentCarrierThread(); + if (caller instanceof CarrierThread ct && ct.scheduler == this) { + return ct; + } + return carriers[Math.floorMod(probe(), carriers.length)]; + } + + private static int probe() { + int p = U.getInt(Thread.currentThread(), PROBE); + if (p == 0) { + long tid = Thread.currentThread().threadId(); + p = (int) (tid ^ (tid >>> 16)); + if (p == 0) p = 1; + U.putInt(Thread.currentThread(), PROBE, p); + } + return p; + } + + // ---- Carrier thread ---- + + static final class CarrierThread extends Thread { + static final int RUNNING = 0; + static final int PARKED = 1; + + final int id; + final MpscUnboundedQueue queue = new MpscUnboundedQueue<>(64); + final MpscVirtualThreadScheduler scheduler; + + // carrier-local poller (Mode 4), null if using Mode 3 or lower + CarrierLocalPoller poller; + + @Contended + volatile int carrierState; + + CarrierThread(int id, MpscVirtualThreadScheduler scheduler) { + super(null, null, "mpsc-carrier-" + id, 0, false); + this.id = id; + this.scheduler = scheduler; + setDaemon(true); + } + + @Override + public void run() { + if ("4".equals(System.getProperty("jdk.pollerMode"))) { + try { + this.poller = new CarrierLocalPoller(); + eventLoop(); + return; + } catch (IOException e) { + throw new UncheckedIOException(e); + } + } + plainLoop(); + } + + /** + * Mode 4: interleave task draining with I/O polling. + * Like Netty's EventLoop: drain tasks → poll I/O → drain → ... + * Block in epoll_wait only when both queue and I/O are idle. + */ + private static final long DRAIN_BUDGET_NS = java.util.concurrent.TimeUnit.MICROSECONDS + .toNanos(Integer.getInteger("jdk.virtualThreadScheduler.drainBudgetUs", 50)); + private static final int TIME_CHECK_INTERVAL = 4; + + private void eventLoop() { + var queue = this.queue; + var poller = this.poller; + for (;;) { + // drain tasks with time budget + int drained = 0; + long drainStart = System.nanoTime(); + VirtualThreadTask task; + while ((task = queue.poll()) != null) { + try { task.run(); } catch (Throwable t) { } + if ((++drained & (TIME_CHECK_INTERVAL - 1)) == 0 + && System.nanoTime() - drainStart >= DRAIN_BUDGET_NS) { + break; + } + } + + // non-blocking I/O poll + int ioEvents = 0; + try { + ioEvents = poller.poll(0); + } catch (IOException e) { } + + if (drained + ioEvents > 0) { + continue; + } + + // one more non-blocking check before parking + try { + if (poller.poll(0) > 0) continue; + } catch (IOException e) { } + + // genuinely idle: blocking poll + carrierState = PARKED; + + if ((task = queue.poll()) != null) { + carrierState = RUNNING; + try { task.run(); } catch (Throwable t) { } + continue; + } + + try { + poller.poll(-1); + } catch (IOException e) { } + carrierState = RUNNING; + } + } + + /** + * Plain loop (Mode 3 or lower): poll tasks, park when idle. + */ + private void plainLoop() { + var queue = this.queue; + for (;;) { + VirtualThreadTask task = queue.poll(); + if (task != null) { + try { task.run(); } catch (Throwable t) { } + continue; + } + + carrierState = PARKED; + + if ((task = queue.poll()) != null) { + carrierState = RUNNING; + try { task.run(); } catch (Throwable t) { } + continue; + } + + LockSupport.park(); + carrierState = RUNNING; + } + } + } + + @Override + public String toString() { + return "MpscVirtualThreadScheduler[carriers=" + carriers.length + "]"; + } +} diff --git a/src/java.base/share/classes/java/lang/Thread.java b/src/java.base/share/classes/java/lang/Thread.java index c91f05b2453..47e211fccd7 100644 --- a/src/java.base/share/classes/java/lang/Thread.java +++ b/src/java.base/share/classes/java/lang/Thread.java @@ -711,6 +711,13 @@ public static void onSpinWait() {} */ static final int STICKY_AFFINITY = 1 << 3; + /** + * Characteristic value signifying that this virtual thread uses round-robin + * carrier affinity. Each thread created by the resulting factory is submitted + * to the next carrier in sequence. + */ + static final int ROUND_ROBIN_AFFINITY = 1 << 4; + /** * Thread identifier assigned to the primordial thread. */ @@ -1276,11 +1283,37 @@ sealed interface OfVirtual extends Builder /** * Sets this builder to create virtual threads with sticky affinity. + * When a sticky virtual thread starts or unparks another virtual thread, + * the runtime uses lazy submission to preserve thread locality. * * @return this builder * @since 99 */ OfVirtual stickyAffinity(); + + /** + * Sets this builder to create virtual threads with round-robin carrier + * affinity. Each thread created by the resulting factory is submitted to + * the next carrier in the scheduler's pool in sequence. + * + *

This is a scheduling hint. The scheduler may ignore it. + * + * @return this builder + * @since 99 + */ + OfVirtual roundRobinAffinity(); + + /** + * Creates a new {@code Thread} from the current state of the builder and + * schedules it without guaranteeing that it will eventually execute. + * + * @param task the object to run when the thread executes + * @return a new started Thread + * + * @see Inheritance when creating threads + * @since 99 + */ + Thread lazyStart(Runnable task); } } diff --git a/src/java.base/share/classes/java/lang/ThreadBuilders.java b/src/java.base/share/classes/java/lang/ThreadBuilders.java index cc660289d25..75522f74660 100644 --- a/src/java.base/share/classes/java/lang/ThreadBuilders.java +++ b/src/java.base/share/classes/java/lang/ThreadBuilders.java @@ -32,6 +32,7 @@ import java.util.Locale; import java.util.Objects; import java.util.concurrent.ThreadFactory; +import java.util.concurrent.atomic.AtomicInteger; import jdk.internal.misc.Unsafe; import jdk.internal.invoke.MhUtil; import jdk.internal.vm.ContinuationSupport; @@ -100,6 +101,10 @@ void setStickyAffinity() { characteristics |= Thread.STICKY_AFFINITY; } + void setRoundRobinAffinity() { + characteristics |= Thread.ROUND_ROBIN_AFFINITY; + } + void setUncaughtExceptionHandler(UncaughtExceptionHandler ueh) { this.uhe = Objects.requireNonNull(ueh); } @@ -247,6 +252,12 @@ public OfVirtual stickyAffinity() { return this; } + @Override + public OfVirtual roundRobinAffinity() { + setRoundRobinAffinity(); + return this; + } + Thread unstarted(Runnable task, Thread preferredCarrier) { Objects.requireNonNull(task); var thread = newVirtualThread(scheduler, @@ -272,6 +283,17 @@ public Thread start(Runnable task) { return thread; } + @Override + public Thread lazyStart(Runnable task) { + Thread thread = unstarted(task); + if (thread instanceof VirtualThread vthread) { + vthread.lazyStart(); + } else { + thread.start(); + } + return thread; + } + @Override public ThreadFactory factory() { return new VirtualThreadFactory(scheduler, name(), counter(), characteristics(), @@ -378,7 +400,13 @@ public Thread newThread(Runnable task) { * ThreadFactory for virtual threads. */ private static class VirtualThreadFactory extends BaseThreadFactory { + private static final VarHandle ROUND_ROBIN_COUNT = MhUtil.findVarHandle( + MethodHandles.lookup(), "roundRobinCount", long.class); + private final Thread.VirtualThreadScheduler scheduler; + private final boolean roundRobin; + @SuppressWarnings("unused") + private volatile long roundRobinCount; VirtualThreadFactory(Thread.VirtualThreadScheduler scheduler, String name, @@ -387,6 +415,7 @@ private static class VirtualThreadFactory extends BaseThreadFactory { UncaughtExceptionHandler uhe) { super(name, start, characteristics, uhe); this.scheduler = scheduler; + this.roundRobin = (characteristics & Thread.ROUND_ROBIN_AFFINITY) != 0; } @Override @@ -394,6 +423,9 @@ public Thread newThread(Runnable task) { Objects.requireNonNull(task); String name = nextThreadName(); Thread thread = newVirtualThread(scheduler, null, name, characteristics(), task); + if (roundRobin && thread instanceof VirtualThread vt) { + vt.affinityHint = (int) (long) ROUND_ROBIN_COUNT.getAndAdd(this, 1L); + } UncaughtExceptionHandler uhe = uncaughtExceptionHandler(); if (uhe != null) thread.uncaughtExceptionHandler(uhe); diff --git a/src/java.base/share/classes/java/lang/VirtualThread.java b/src/java.base/share/classes/java/lang/VirtualThread.java index 10a4b82a68e..b15ba4b3ab0 100644 --- a/src/java.base/share/classes/java/lang/VirtualThread.java +++ b/src/java.base/share/classes/java/lang/VirtualThread.java @@ -241,6 +241,18 @@ static boolean currentThreadIsSticky() { return currentThread() instanceof VirtualThread vt && vt.stickyAffinity; } + /** + * Returns true if this virtual thread has sticky affinity. + */ + boolean hasStickyAffinity() { + return stickyAffinity; + } + + // Carrier affinity hint. Set by the factory (round-robin counter) or by the + // scheduler on first start (resolved carrier id). The scheduler resolves it + // to a carrier via modulus. -1 means no affinity. + int affinityHint = -1; + /** * Returns the continuation scope used for virtual threads. */ @@ -788,8 +800,7 @@ private void afterDone(boolean notifyContainer) { * @throws IllegalThreadStateException if the thread has already been started * @throws RejectedExecutionException if the scheduler cannot accept a task */ - @Override - void start(ThreadContainer container) { + private void start(ThreadContainer container, boolean lazy) { if (!compareAndSetState(NEW, STARTED)) { throw new IllegalThreadStateException("Already started"); } @@ -811,13 +822,14 @@ void start(ThreadContainer container) { // submit task to schedule try { if (currentThread().isVirtual()) { + boolean useLazy = lazy || currentThreadIsSticky(); Continuation.pin(); try { if (scheduler == BUILTIN_SCHEDULER && currentCarrierThread() instanceof CarrierThread ct) { ForkJoinPool pool = ct.getPool(); ForkJoinTask task = ForkJoinTask.adapt(runContinuation); - if (currentThreadIsSticky()) { + if (useLazy) { pool.lazySubmit(task); } else { pool.externalSubmit(task); @@ -844,9 +856,21 @@ && currentCarrierThread() instanceof CarrierThread ct) { } } + @Override + void start(ThreadContainer container) { + start(container, false); + } + @Override public void start() { - start(ThreadContainers.root()); + start(ThreadContainers.root(), false); + } + + /** + * Schedules this thread to begin execution without guarantee that it will execute. + */ + void lazyStart() { + start(ThreadContainers.root(), true); } @Override @@ -1037,6 +1061,11 @@ void unpark() { unpark(false); } + @Override + void lazyUnpark() { + unpark(true); + } + /** * Invoked by unblocker thread to unblock this virtual thread. */ @@ -1507,7 +1536,10 @@ private static VirtualThreadScheduler createBuiltinScheduler(boolean wrapped) { } else { minRunnable = Integer.max(parallelism / 2, 1); } - if (Boolean.getBoolean("jdk.virtualThreadScheduler.useTPE")) { + if (Boolean.getBoolean("jdk.virtualThreadScheduler.useMpsc")) { + System.err.println("WARNING: Using experimental MPSC virtual thread scheduler"); + return new MpscVirtualThreadScheduler(parallelism); + } else if (Boolean.getBoolean("jdk.virtualThreadScheduler.useTPE")) { return new BuiltinThreadPoolExecutorScheduler(parallelism); } else { return new BuiltinForkJoinPoolScheduler(parallelism, maxPoolSize, minRunnable, wrapped); From 081d3800364dff791223ae84f7db05b8803c2012 Mon Sep 17 00:00:00 2001 From: Francesco Nigro Date: Fri, 10 Jul 2026 10:01:30 +0200 Subject: [PATCH 5/6] Add carrier-local poller (Mode 4) Each carrier is its own master poller with EPOLLONESHOT, HashMap fd tracking, isEmpty() skip on non-blocking poll, eventfd wakeup. Adds CARRIER_LOCAL_POLLER mode enum and CarrierLocalPollerGroup in Poller. JLA bridge (System.java + JavaLangAccess.java) for carrierLocalPoller() lookup from NIO layer. --- .../sun/nio/ch/CarrierLocalPoller.java | 126 ++++++++++++++++ .../share/classes/java/lang/System.java | 16 +++ .../jdk/internal/access/JavaLangAccess.java | 12 ++ .../share/classes/sun/nio/ch/Poller.java | 135 +++++++++++++++++- 4 files changed, 284 insertions(+), 5 deletions(-) create mode 100644 src/java.base/linux/classes/sun/nio/ch/CarrierLocalPoller.java diff --git a/src/java.base/linux/classes/sun/nio/ch/CarrierLocalPoller.java b/src/java.base/linux/classes/sun/nio/ch/CarrierLocalPoller.java new file mode 100644 index 00000000000..897ab8ae8c5 --- /dev/null +++ b/src/java.base/linux/classes/sun/nio/ch/CarrierLocalPoller.java @@ -0,0 +1,126 @@ +/* + * Copyright (c) 2026, Oracle and/or its affiliates. All rights reserved. + * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER. + * + * This code is free software; you can redistribute it and/or modify it + * under the terms of the GNU General Public License version 2 only, as + * published by the Free Software Foundation. Oracle designates this + * particular file as subject to the "Classpath" exception as provided + * by Oracle in the LICENSE file that accompanied this code. + * + * This code is distributed in the hope that it will be useful, but WITHOUT + * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or + * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License + * version 2 for more details (a copy is included in the LICENSE file that + * accompanied this code). + * + * You should have received a copy of the GNU General Public License version + * 2 along with this work; if not, write to the Free Software Foundation, + * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA. + * + * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA + * or visit www.oracle.com if you need additional information or have any + * questions. + */ +package sun.nio.ch; + +import java.io.IOException; +import java.util.HashMap; +import java.util.concurrent.locks.LockSupport; +import static sun.nio.ch.EPoll.*; + +/** + * A carrier-local I/O poller using EPOLLONESHOT. Each carrier thread owns + * one instance. VTs that park on I/O register their fd directly with the + * carrier's epoll fd. No sub-pollers, no master poller. + * + *

The carrier calls {@link #poll(int)} when idle. When I/O arrives, + * the blocked VTs are unparked and enqueued to the carrier's own queue. + * External task submissions wake the carrier via the eventfd. + */ +public final class CarrierLocalPoller { + + private static final int ENOENT = 2; + private static final int MAX_EVENTS = 64; + + private final int epfd; + private final long pollAddress; + private final EventFD eventfd; + private final HashMap fdToThread = new HashMap<>(); + + public CarrierLocalPoller() throws IOException { + this.epfd = EPoll.create(); + this.pollAddress = EPoll.allocatePollArray(MAX_EVENTS); + this.eventfd = new EventFD(); + IOUtil.configureBlocking(eventfd.efd(), false); + EPoll.ctl(epfd, EPOLL_CTL_ADD, eventfd.efd(), EPOLLIN); + } + + /** + * Register a file descriptor for read or write polling. Called by VTs + * on this carrier before parking. The VT is unparked when the fd is ready. + */ + public void register(int fdVal, int event, Thread thread) throws IOException { + fdToThread.put(fdVal, thread); + int err = EPoll.ctl(epfd, EPOLL_CTL_MOD, fdVal, (event | EPOLLONESHOT)); + if (err == ENOENT) { + err = EPoll.ctl(epfd, EPOLL_CTL_ADD, fdVal, (event | EPOLLONESHOT)); + } + if (err != 0) { + fdToThread.remove(fdVal); + throw new IOException("epoll_ctl failed: " + err); + } + } + + /** + * Deregister a file descriptor. Called if the VT was unparked by + * something other than I/O readiness (e.g. interrupt, timeout). + */ + public void deregister(int fdVal) { + if (fdToThread.remove(fdVal) != null) { + EPoll.ctl(epfd, EPOLL_CTL_DEL, fdVal, 0); + } + } + + /** + * Poll for I/O events. Returns the number of VTs unparked. + * + * @param timeout milliseconds: -1 to block, 0 for non-blocking + */ + public int poll(int timeout) throws IOException { + if (timeout == 0 && fdToThread.isEmpty()) { + return 0; + } + int n = EPoll.wait(epfd, pollAddress, MAX_EVENTS, timeout); + int unparked = 0; + for (int i = 0; i < n; i++) { + long eventAddress = EPoll.getEvent(pollAddress, i); + int fd = EPoll.getDescriptor(eventAddress); + if (fd == eventfd.efd()) { + eventfd.reset(); + } else { + Thread vt = fdToThread.remove(fd); + if (vt != null) { + LockSupport.unpark(vt); + unparked++; + } + } + } + return unparked; + } + + /** + * Wake the carrier from a blocking {@link #poll} call. + * Called by external threads submitting tasks to this carrier. + */ + public void wakeup() throws IOException { + eventfd.set(); + } + + /** + * Returns true if there are fds registered for polling. + */ + public boolean hasPendingFds() { + return !fdToThread.isEmpty(); + } +} diff --git a/src/java.base/share/classes/java/lang/System.java b/src/java.base/share/classes/java/lang/System.java index 3aede3570f8..0cea3f0091e 100644 --- a/src/java.base/share/classes/java/lang/System.java +++ b/src/java.base/share/classes/java/lang/System.java @@ -2252,6 +2252,14 @@ public Thread currentCarrierThread() { return Thread.currentCarrierThread(); } + public Object carrierLocalPoller() { + Thread carrier = Thread.currentCarrierThread(); + if (carrier instanceof MpscVirtualThreadScheduler.CarrierThread ct) { + return ct.poller; + } + return null; + } + public T getCarrierThreadLocal(CarrierThreadLocal local) { return ((ThreadLocal)local).getCarrierThreadLocal(); } @@ -2322,6 +2330,14 @@ public void unparkVirtualThread(Thread thread) { } } + public void lazyUnparkVirtualThread(Thread thread) { + if (thread instanceof BaseVirtualThread vthread) { + vthread.lazyUnpark(); + } else { + throw new IllegalArgumentException(); + } + } + public Thread.VirtualThreadScheduler builtinVirtualThreadScheduler() { return VirtualThread.builtinScheduler(true); } diff --git a/src/java.base/share/classes/jdk/internal/access/JavaLangAccess.java b/src/java.base/share/classes/jdk/internal/access/JavaLangAccess.java index 45c01889250..f3ea660ecff 100644 --- a/src/java.base/share/classes/jdk/internal/access/JavaLangAccess.java +++ b/src/java.base/share/classes/jdk/internal/access/JavaLangAccess.java @@ -550,6 +550,11 @@ public interface JavaLangAccess { */ Thread currentCarrierThread(); + /** + * Returns the CarrierLocalPoller for the current carrier thread, or null. + */ + Object carrierLocalPoller(); + /** * Returns the value of the current carrier thread's copy of a thread-local. */ @@ -627,6 +632,13 @@ public interface JavaLangAccess { */ void unparkVirtualThread(Thread thread); + /** + * Re-enables a virtual thread for scheduling. If the thread is parked then it will + * be scheduled to continue, without guaranteeing that it will eventually continue + * execution. + */ + void lazyUnparkVirtualThread(Thread thread); + /** * Returns the builtin virtual thread scheduler. */ diff --git a/src/java.base/share/classes/sun/nio/ch/Poller.java b/src/java.base/share/classes/sun/nio/ch/Poller.java index 475284adbeb..828db8c1e68 100644 --- a/src/java.base/share/classes/sun/nio/ch/Poller.java +++ b/src/java.base/share/classes/sun/nio/ch/Poller.java @@ -91,7 +91,14 @@ enum Mode { * for I/O. If there are no events then the poller threads park until there * are I/O events to poll. The write poller is a system-wide platform thread. */ - POLLER_PER_CARRIER + POLLER_PER_CARRIER, + + /** + * Each carrier thread is its own poller. VT fds register directly with the + * carrier's epoll fd. No sub-pollers, no master poller. The carrier calls + * epoll_wait when idle. Write pollers are system-wide platform threads. + */ + CARRIER_LOCAL_POLLER } /** @@ -105,6 +112,7 @@ private static PollerGroup createPollerGroup() { case "1" -> Mode.SYSTEM_THREADS; case "2" -> Mode.VTHREAD_POLLERS; case "3" -> Mode.POLLER_PER_CARRIER; + case "4" -> Mode.CARRIER_LOCAL_POLLER; default -> { throw new RuntimeException(s + " is not a valid polling mode"); } @@ -117,9 +125,10 @@ private static PollerGroup createPollerGroup() { int readPollers = pollerCount("jdk.readPollers", provider.defaultReadPollers()); int writePollers = pollerCount("jdk.writePollers", provider.defaultWritePollers()); PollerGroup group = switch (provider.pollerMode()) { - case SYSTEM_THREADS -> new SystemThreadsPollerGroup(provider, readPollers, writePollers); - case VTHREAD_POLLERS -> new VThreadsPollerGroup(provider, readPollers, writePollers); - case POLLER_PER_CARRIER -> new PollerPerCarrierPollerGroup(provider, writePollers); + case SYSTEM_THREADS -> new SystemThreadsPollerGroup(provider, readPollers, writePollers); + case VTHREAD_POLLERS -> new VThreadsPollerGroup(provider, readPollers, writePollers); + case POLLER_PER_CARRIER -> new PollerPerCarrierPollerGroup(provider, writePollers); + case CARRIER_LOCAL_POLLER -> new CarrierLocalPollerGroup(provider, writePollers); }; group.start(); return group; @@ -207,7 +216,11 @@ void wakeupPoller() throws IOException { final void polled(int fdVal) { Thread t = map.remove(fdVal); if (t != null) { - LockSupport.unpark(t); + if (POLLER_GROUP.useLazyUnpark() && Thread.currentThread().isVirtual()) { + JLA.lazyUnparkVirtualThread(t); + } else { + LockSupport.unpark(t); + } } } @@ -398,6 +411,13 @@ protected final void startPlatformThread(String name, Runnable task) { */ abstract List writePollers(); + /** + * Return true if the unparking threads should use lazyUnpark. + */ + boolean useLazyUnpark() { + return false; + } + /** * Close the given pollers. */ @@ -752,6 +772,11 @@ List readPollers() { List writePollers() { return List.of(writePollers); } + + @Override + boolean useLazyUnpark() { + return true; + } } /** @@ -793,4 +818,104 @@ public static List readPollers() { public static List writePollers() { return POLLER_GROUP.writePollers(); } + + + // ---- CARRIER_LOCAL_POLLER group ---- + + /** + * Each carrier owns its own epoll fd. VT fds register directly with the + * carrier's poller. No sub-pollers, no master poller. Write pollers are + * system-wide platform threads. + */ + private static class CarrierLocalPollerGroup extends PollerGroup { + private final Poller[] writePollers; + + CarrierLocalPollerGroup(PollerProvider provider, + int writePollerCount) throws IOException { + super(provider); + Poller[] writePollers = new Poller[writePollerCount]; + try { + for (int i = 0; i < writePollerCount; i++) { + writePollers[i] = provider.writePoller(false); + } + } catch (Throwable e) { + PollerGroup.closeAll(writePollers); + throw e; + } + this.writePollers = writePollers; + } + + @Override + void start() { + Arrays.stream(writePollers).forEach(p -> { + startPlatformThread("Write-Poller", p::pollerLoop); + }); + } + + CarrierLocalPoller getLocalPoller() { + Object p = JLA.carrierLocalPoller(); + return (p instanceof CarrierLocalPoller clp) ? clp : null; + } + + + private Poller writePoller(int fdVal) { + int index = provider().fdValToIndex(fdVal, writePollers.length); + return writePollers[index]; + } + + @Override + void poll(int fdVal, int event, long nanos, BooleanSupplier isOpen) throws IOException { + // POLLIN from VT: register directly with carrier's local poller + if (event == Net.POLLIN + && Thread.currentThread().isVirtual() + && ContinuationSupport.isSupported()) { + Thread carrier = JLA.currentCarrierThread(); + // read the ThreadLocal from the carrier thread context + CarrierLocalPoller poller = getLocalPoller(); + if (poller != null) { + poller.register(fdVal, event, Thread.currentThread()); + try { + if (isOpen.getAsBoolean()) { + if (nanos > 0) { + LockSupport.parkNanos(nanos); + } else { + LockSupport.park(); + } + } + } finally { + poller.deregister(fdVal); + } + return; + } + } + + // POLLOUT or fallback: use write poller + if (event == Net.POLLOUT) { + writePoller(fdVal).poll(fdVal, nanos, isOpen); + } else { + // platform thread POLLIN fallback + writePoller(fdVal).poll(fdVal, nanos, isOpen); + } + } + + @Override + Poller masterPoller() { + return null; + } + + @Override + List readPollers() { + return List.of(); + } + + @Override + List writePollers() { + return List.of(writePollers); + } + + @Override + boolean useLazyUnpark() { + return true; + } + } } From e20e4a6a00ddb2491b5f3b6394a898f7dd4834f3 Mon Sep 17 00:00:00 2001 From: Francesco Nigro Date: Fri, 10 Jul 2026 16:54:55 +0200 Subject: [PATCH 6/6] Fail-safe: pollerMode=4 falls back to Mode 2 without MPSC scheduler Check actual scheduler instance via JLA.isMpscScheduler() instead of re-reading system property. Warns on fallback. Fixes POLLIN routing to write poller when carrier-local poller is unavailable. --- .../share/classes/java/lang/System.java | 4 ++++ .../jdk/internal/access/JavaLangAccess.java | 5 +++++ .../share/classes/sun/nio/ch/Poller.java | 21 +++++++++---------- 3 files changed, 19 insertions(+), 11 deletions(-) diff --git a/src/java.base/share/classes/java/lang/System.java b/src/java.base/share/classes/java/lang/System.java index 0cea3f0091e..761e52cacce 100644 --- a/src/java.base/share/classes/java/lang/System.java +++ b/src/java.base/share/classes/java/lang/System.java @@ -2260,6 +2260,10 @@ public Object carrierLocalPoller() { return null; } + public boolean isMpscScheduler() { + return VirtualThread.builtinScheduler(true) instanceof MpscVirtualThreadScheduler; + } + public T getCarrierThreadLocal(CarrierThreadLocal local) { return ((ThreadLocal)local).getCarrierThreadLocal(); } diff --git a/src/java.base/share/classes/jdk/internal/access/JavaLangAccess.java b/src/java.base/share/classes/jdk/internal/access/JavaLangAccess.java index f3ea660ecff..fb385b5a468 100644 --- a/src/java.base/share/classes/jdk/internal/access/JavaLangAccess.java +++ b/src/java.base/share/classes/jdk/internal/access/JavaLangAccess.java @@ -555,6 +555,11 @@ public interface JavaLangAccess { */ Object carrierLocalPoller(); + /** + * Returns true if the built-in scheduler is the MPSC scheduler. + */ + boolean isMpscScheduler(); + /** * Returns the value of the current carrier thread's copy of a thread-local. */ diff --git a/src/java.base/share/classes/sun/nio/ch/Poller.java b/src/java.base/share/classes/sun/nio/ch/Poller.java index 828db8c1e68..14c129c9eb8 100644 --- a/src/java.base/share/classes/sun/nio/ch/Poller.java +++ b/src/java.base/share/classes/sun/nio/ch/Poller.java @@ -112,7 +112,13 @@ private static PollerGroup createPollerGroup() { case "1" -> Mode.SYSTEM_THREADS; case "2" -> Mode.VTHREAD_POLLERS; case "3" -> Mode.POLLER_PER_CARRIER; - case "4" -> Mode.CARRIER_LOCAL_POLLER; + case "4" -> { + if (JLA.isMpscScheduler()) { + yield Mode.CARRIER_LOCAL_POLLER; + } + System.err.println("WARNING: pollerMode=4 requires MPSC scheduler, falling back to mode 2"); + yield Mode.VTHREAD_POLLERS; + } default -> { throw new RuntimeException(s + " is not a valid polling mode"); } @@ -865,12 +871,10 @@ private Poller writePoller(int fdVal) { @Override void poll(int fdVal, int event, long nanos, BooleanSupplier isOpen) throws IOException { - // POLLIN from VT: register directly with carrier's local poller + // POLLIN from VT: register with carrier's local poller if (event == Net.POLLIN && Thread.currentThread().isVirtual() && ContinuationSupport.isSupported()) { - Thread carrier = JLA.currentCarrierThread(); - // read the ThreadLocal from the carrier thread context CarrierLocalPoller poller = getLocalPoller(); if (poller != null) { poller.register(fdVal, event, Thread.currentThread()); @@ -889,13 +893,8 @@ void poll(int fdVal, int event, long nanos, BooleanSupplier isOpen) throws IOExc } } - // POLLOUT or fallback: use write poller - if (event == Net.POLLOUT) { - writePoller(fdVal).poll(fdVal, nanos, isOpen); - } else { - // platform thread POLLIN fallback - writePoller(fdVal).poll(fdVal, nanos, isOpen); - } + // POLLOUT or non-VT POLLIN: write poller + writePoller(fdVal).poll(fdVal, nanos, isOpen); } @Override