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/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/System.java b/src/java.base/share/classes/java/lang/System.java index 3aede3570f8..761e52cacce 100644 --- a/src/java.base/share/classes/java/lang/System.java +++ b/src/java.base/share/classes/java/lang/System.java @@ -2252,6 +2252,18 @@ 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 boolean isMpscScheduler() { + return VirtualThread.builtinScheduler(true) instanceof MpscVirtualThreadScheduler; + } + public T getCarrierThreadLocal(CarrierThreadLocal local) { return ((ThreadLocal)local).getCarrierThreadLocal(); } @@ -2322,6 +2334,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/java/lang/Thread.java b/src/java.base/share/classes/java/lang/Thread.java index 06d69413bad..47e211fccd7 100644 --- a/src/java.base/share/classes/java/lang/Thread.java +++ b/src/java.base/share/classes/java/lang/Thread.java @@ -704,6 +704,20 @@ 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; + + /** + * 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. */ @@ -1266,6 +1280,40 @@ 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(); + + /** + * 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 f08de1d8276..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; @@ -96,6 +97,14 @@ void setInheritInheritableThreadLocals(boolean inherit) { } } + void setStickyAffinity() { + characteristics |= Thread.STICKY_AFFINITY; + } + + void setRoundRobinAffinity() { + characteristics |= Thread.ROUND_ROBIN_AFFINITY; + } + void setUncaughtExceptionHandler(UncaughtExceptionHandler ueh) { this.uhe = Objects.requireNonNull(ueh); } @@ -237,6 +246,18 @@ public OfVirtual uncaughtExceptionHandler(UncaughtExceptionHandler ueh) { return this; } + @Override + public OfVirtual stickyAffinity() { + setStickyAffinity(); + return this; + } + + @Override + public OfVirtual roundRobinAffinity() { + setRoundRobinAffinity(); + return this; + } + Thread unstarted(Runnable task, Thread preferredCarrier) { Objects.requireNonNull(task); var thread = newVirtualThread(scheduler, @@ -262,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(), @@ -368,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, @@ -377,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 @@ -384,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 c6f110ab3c2..b15ba4b3ab0 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,25 @@ 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 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. */ @@ -271,6 +291,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) { @@ -670,8 +691,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(); @@ -777,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"); } @@ -800,13 +822,18 @@ 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); - pool.externalSubmit(task); + if (useLazy) { + pool.lazySubmit(task); + } else { + pool.externalSubmit(task); + } } else { scheduler.onStart(runContinuation); } @@ -829,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 @@ -976,7 +1015,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 @@ -1022,6 +1061,11 @@ void unpark() { unpark(false); } + @Override + void lazyUnpark() { + unpark(true); + } + /** * Invoked by unblocker thread to unblock this virtual thread. */ @@ -1492,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); 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..fb385b5a468 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,16 @@ public interface JavaLangAccess { */ Thread currentCarrierThread(); + /** + * Returns the CarrierLocalPoller for the current carrier thread, or null. + */ + 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. */ @@ -627,6 +637,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 4209eff0e74..14c129c9eb8 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,13 @@ private static PollerGroup createPollerGroup() { case "1" -> Mode.SYSTEM_THREADS; case "2" -> Mode.VTHREAD_POLLERS; case "3" -> Mode.POLLER_PER_CARRIER; + 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"); } @@ -117,9 +131,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 +222,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 +417,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. */ @@ -656,6 +682,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() @@ -751,6 +778,11 @@ List readPollers() { List writePollers() { return List.of(writePollers); } + + @Override + boolean useLazyUnpark() { + return true; + } } /** @@ -792,4 +824,97 @@ 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 with carrier's local poller + if (event == Net.POLLIN + && Thread.currentThread().isVirtual() + && ContinuationSupport.isSupported()) { + 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 non-VT POLLIN: write poller + 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; + } + } } 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..b0c1e72a1db --- /dev/null +++ b/test/jdk/java/lang/Thread/virtual/StickyAffinityTest.java @@ -0,0 +1,109 @@ +/* + * 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.util.concurrent.*; +import java.util.concurrent.atomic.*; +import java.util.concurrent.locks.LockSupport; + +import jdk.test.lib.thread.VThreadScheduler; + +import org.junit.jupiter.api.Test; +import static org.junit.jupiter.api.Assertions.*; + +class StickyAffinityTest { + + /** + * 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(); + targetCarrierAfterUnpark.set(VThreadScheduler.currentCarrierThread()); + done.countDown(); + }); + parked.await(); + + Thread sticky = Thread.ofVirtual() + .stickyAffinity() + .start(() -> { + stickyCarrier.set(VThreadScheduler.currentCarrierThread()); + 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"); + } + +} diff --git a/test/lib/jdk/test/lib/thread/VThreadScheduler.java b/test/lib/jdk/test/lib/thread/VThreadScheduler.java index f64533582a0..9f1bfca2ad5 100644 --- a/test/lib/jdk/test/lib/thread/VThreadScheduler.java +++ b/test/lib/jdk/test/lib/thread/VThreadScheduler.java @@ -131,6 +131,25 @@ public static ThreadFactory virtualThreadFactory(Thread.VirtualThreadScheduler s return virtualThreadBuilder(scheduler).factory(); } + /** + * Returns the carrier thread for the current virtual thread. + */ + public static Thread currentCarrierThread() { + try { + Method m = Thread.class.getDeclaredMethod("currentCarrierThread"); + m.setAccessible(true); + return (Thread) m.invoke(null); + } catch (InvocationTargetException e) { + Throwable cause = e.getCause(); + if (cause instanceof RuntimeException re) { + throw re; + } + throw new RuntimeException(e); + } catch (Exception e) { + throw new RuntimeException(e); + } + } + public static ThreadFactory virtualThreadFactory(Executor executor) { return virtualThreadBuilder(executor).factory(); }