< prev index next > src/java.base/share/classes/sun/nio/ch/Poller.java
Print this page
* questions.
*/
package sun.nio.ch;
import java.io.IOException;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.locks.LockSupport;
import java.util.function.BooleanSupplier;
import jdk.internal.misc.InnocuousThread;
import jdk.internal.vm.annotation.Stable;
/**
* Polls file descriptors. Virtual threads invoke the poll method to park
* until a given file descriptor is ready for I/O.
*/
public abstract class Poller {
! private static final Pollers POLLERS;
! static {
! try {
! var pollers = new Pollers();
! pollers.start();
! POLLERS = pollers;
! } catch (IOException ioe) {
! throw new ExceptionInInitializerError(ioe);
! }
! }
// the poller or sub-poller thread
private @Stable Thread owner;
// maps file descriptors to parked Thread
* questions.
*/
package sun.nio.ch;
import java.io.IOException;
+ import java.io.UncheckedIOException;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.locks.LockSupport;
import java.util.function.BooleanSupplier;
+ import java.util.function.Supplier;
+ import jdk.internal.access.JavaLangAccess;
+ import jdk.internal.access.SharedSecrets;
import jdk.internal.misc.InnocuousThread;
import jdk.internal.vm.annotation.Stable;
/**
* Polls file descriptors. Virtual threads invoke the poll method to park
* until a given file descriptor is ready for I/O.
*/
public abstract class Poller {
! private static final JavaLangAccess JLA = SharedSecrets.getJavaLangAccess();
!
! private static final PollerProvider PROVIDER = PollerProvider.provider();
!
! private static final Executor DEFAULT_SCHEDULER = JLA.virtualThreadDefaultScheduler();
!
! private static Supplier<Mode> POLLER_MODE = StableValue.supplier(Poller::pollerMode);
!
! private static Supplier<PollerGroup> DEFAULT_POLLER_GROUP = StableValue.supplier(PollerGroup::create);
!
+ // maps scheduler to PollerGroup, custom schedulers can't be GC'ed at this time
+ private static final Map<Executor, PollerGroup> POLLER_GROUPS = new ConcurrentHashMap<>();
// the poller or sub-poller thread
private @Stable Thread owner;
// maps file descriptors to parked Thread
*/
static void poll(int fdVal, int event, long nanos, BooleanSupplier supplier)
throws IOException
{
assert nanos >= 0L;
if (event == Net.POLLIN) {
! POLLERS.readPoller(fdVal).poll(fdVal, nanos, supplier);
} else if (event == Net.POLLOUT) {
! POLLERS.writePoller(fdVal).poll(fdVal, nanos, supplier);
} else {
assert false;
}
}
*/
static void poll(int fdVal, int event, long nanos, BooleanSupplier supplier)
throws IOException
{
assert nanos >= 0L;
+ PollerGroup pollerGroup = PollerGroup.groupFor(Thread.currentThread());
if (event == Net.POLLIN) {
! pollerGroup.readPoller(fdVal).poll(fdVal, nanos, supplier);
} else if (event == Net.POLLOUT) {
! pollerGroup.writePoller(fdVal).poll(fdVal, nanos, supplier);
} else {
assert false;
}
}
* @param fdVal the Selector's file descriptor
* @param nanos the waiting time or 0 to wait indefinitely
*/
static void pollSelector(int fdVal, long nanos) throws IOException {
assert nanos >= 0L;
! Poller poller = POLLERS.masterPoller();
if (poller == null) {
! poller = POLLERS.readPoller(fdVal);
}
poller.poll(fdVal, nanos, () -> true);
}
/**
! * If there is a thread polling the given file descriptor for the given event then
- * the thread is unparked.
*/
! static void stopPoll(int fdVal, int event) {
! if (event == Net.POLLIN) {
- POLLERS.readPoller(fdVal).wakeup(fdVal);
- } else if (event == Net.POLLOUT) {
- POLLERS.writePoller(fdVal).wakeup(fdVal);
- } else {
- throw new IllegalArgumentException();
- }
- }
-
- /**
- * If there are any threads polling the given file descriptor then they are unparked.
- */
- static void stopPoll(int fdVal) {
- stopPoll(fdVal, Net.POLLIN);
- stopPoll(fdVal, Net.POLLOUT);
}
/**
* Parks the current thread until a file descriptor is ready.
*/
* @param fdVal the Selector's file descriptor
* @param nanos the waiting time or 0 to wait indefinitely
*/
static void pollSelector(int fdVal, long nanos) throws IOException {
assert nanos >= 0L;
! PollerGroup pollerGroup = PollerGroup.groupFor(Thread.currentThread());
+ Poller poller = pollerGroup.masterPoller();
if (poller == null) {
! poller = pollerGroup.readPoller(fdVal);
}
poller.poll(fdVal, nanos, () -> true);
}
/**
! * Unpark the given thread so that it stops polling.
*/
! static void stopPoll(Thread thread) {
! LockSupport.unpark(thread);
}
/**
* Parks the current thread until a file descriptor is ready.
*/
} catch (Exception e) {
e.printStackTrace();
}
}
- /**
- * Returns the number I/O operations currently registered with this poller.
- */
- public int registered() {
- return map.size();
- }
-
@Override
public String toString() {
return String.format("%s [registered = %d, owner = %s]",
! Objects.toIdentityString(this), registered(), owner);
}
/**
! * The Pollers used for read and write events.
*/
! private static class Pollers {
! private final PollerProvider provider;
- private final Poller.Mode pollerMode;
- private final Poller masterPoller;
private final Poller[] readPollers;
private final Poller[] writePollers;
!
! // used by start method to executor is kept alive
! private Executor executor;
!
! /**
! * Creates the Poller instances based on configuration.
! */
! Pollers() throws IOException {
! PollerProvider provider = PollerProvider.provider();
! Poller.Mode mode;
! String s = System.getProperty("jdk.pollerMode");
! if (s != null) {
! if (s.equalsIgnoreCase(Mode.SYSTEM_THREADS.name()) || s.equals("1")) {
! mode = Mode.SYSTEM_THREADS;
! } else if (s.equalsIgnoreCase(Mode.VTHREAD_POLLERS.name()) || s.equals("2")) {
! mode = Mode.VTHREAD_POLLERS;
} else {
! throw new RuntimeException("Can't parse '" + s + "' as polling mode");
}
- } else {
- mode = provider.defaultPollerMode();
}
! // vthread poller mode needs a master poller
! Poller masterPoller = (mode == Mode.VTHREAD_POLLERS)
! ? provider.readPoller(false)
! : null;
// read pollers (or sub-pollers)
- int readPollerCount = pollerCount("jdk.readPollers", provider.defaultReadPollers(mode));
Poller[] readPollers = new Poller[readPollerCount];
for (int i = 0; i < readPollerCount; i++) {
! readPollers[i] = provider.readPoller(mode == Mode.VTHREAD_POLLERS);
}
// write pollers (or sub-pollers)
- int writePollerCount = pollerCount("jdk.writePollers", provider.defaultWritePollers(mode));
Poller[] writePollers = new Poller[writePollerCount];
for (int i = 0; i < writePollerCount; i++) {
! writePollers[i] = provider.writePoller(mode == Mode.VTHREAD_POLLERS);
}
! this.provider = provider;
- this.pollerMode = mode;
this.masterPoller = masterPoller;
this.readPollers = readPollers;
this.writePollers = writePollers;
}
/**
! * Starts the Poller threads.
*/
! void start() {
! if (pollerMode == Mode.VTHREAD_POLLERS) {
! startPlatformThread("MasterPoller", masterPoller::pollerLoop);
! ThreadFactory factory = Thread.ofVirtual()
! .inheritInheritableThreadLocals(false)
- .name("SubPoller-", 0)
- .uncaughtExceptionHandler((t, e) -> e.printStackTrace())
- .factory();
- executor = Executors.newThreadPerTaskExecutor(factory);
Arrays.stream(readPollers).forEach(p -> {
executor.execute(() -> p.subPollerLoop(masterPoller));
});
Arrays.stream(writePollers).forEach(p -> {
executor.execute(() -> p.subPollerLoop(masterPoller));
});
} else {
Arrays.stream(readPollers).forEach(p -> {
startPlatformThread("Read-Poller", p::pollerLoop);
});
Arrays.stream(writePollers).forEach(p -> {
startPlatformThread("Write-Poller", p::pollerLoop);
});
}
}
- /**
- * Returns the master poller, or null if there is no master poller.
- */
Poller masterPoller() {
return masterPoller;
}
/**
* Returns the read poller for the given file descriptor.
*/
Poller readPoller(int fdVal) {
! int index = provider.fdValToIndex(fdVal, readPollers.length);
return readPollers[index];
}
/**
* Returns the write poller for the given file descriptor.
*/
Poller writePoller(int fdVal) {
! int index = provider.fdValToIndex(fdVal, writePollers.length);
return writePollers[index];
}
- /**
- * Return the list of read pollers.
- */
- List<Poller> readPollers() {
- return List.of(readPollers);
- }
-
- /**
- * Return the list of write pollers.
- */
- List<Poller> writePollers() {
- return List.of(writePollers);
- }
-
-
/**
* Reads the given property name to get the poller count. If the property is
* set then the value must be a power of 2. Returns 1 if the property is not
* set.
* @throws IllegalArgumentException if the property is set to a value that
} catch (Exception e) {
e.printStackTrace();
}
}
@Override
public String toString() {
return String.format("%s [registered = %d, owner = %s]",
! Objects.toIdentityString(this), map.size(), owner);
}
/**
! * The read/write pollers.
*/
! private static class PollerGroup {
! private final Executor scheduler;
private final Poller[] readPollers;
private final Poller[] writePollers;
! private final Poller masterPoller;
! private final Executor executor;
!
! PollerGroup(Executor scheduler) throws IOException {
! Mode mode = Poller.POLLER_MODE.get();
! int readPollerCount, writePollerCount;
! Poller masterPoller;
! if (scheduler == DEFAULT_SCHEDULER) {
! readPollerCount = pollerCount("jdk.readPollers", PROVIDER.defaultReadPollers(mode));
! writePollerCount = pollerCount("jdk.writePollers", PROVIDER.defaultWritePollers(mode));
! masterPoller = (mode == Mode.VTHREAD_POLLERS)
! ? PROVIDER.readPoller(false)
! : null;
! } else {
! readPollerCount = 1;
! writePollerCount = 1;
+ if (mode == Mode.VTHREAD_POLLERS) {
+ masterPoller = DEFAULT_POLLER_GROUP.get().masterPoller();
} else {
! masterPoller = null;
}
}
! Executor executor = null;
! if (mode == Mode.VTHREAD_POLLERS) {
! String namePrefix;
! if (scheduler == DEFAULT_SCHEDULER) {
+ namePrefix = "SubPoller-";
+ } else {
+ namePrefix = Objects.toIdentityString(scheduler) + "-SubPoller-";
+ }
+ @SuppressWarnings("restricted")
+ ThreadFactory factory = Thread.ofVirtual()
+ .scheduler(scheduler)
+ .inheritInheritableThreadLocals(false)
+ .name(namePrefix, 0)
+ .uncaughtExceptionHandler((_, e) -> e.printStackTrace())
+ .factory();
+ executor = Executors.newThreadPerTaskExecutor(factory);
+ }
// read pollers (or sub-pollers)
Poller[] readPollers = new Poller[readPollerCount];
for (int i = 0; i < readPollerCount; i++) {
! readPollers[i] = PROVIDER.readPoller(mode == Mode.VTHREAD_POLLERS);
}
// write pollers (or sub-pollers)
Poller[] writePollers = new Poller[writePollerCount];
for (int i = 0; i < writePollerCount; i++) {
! writePollers[i] = PROVIDER.writePoller(mode == Mode.VTHREAD_POLLERS);
}
! this.scheduler = scheduler;
this.masterPoller = masterPoller;
this.readPollers = readPollers;
this.writePollers = writePollers;
+ this.executor = executor;
+ }
+
+ static PollerGroup create(Executor scheduler) {
+ try {
+ return new PollerGroup(scheduler).start();
+ } catch (IOException ioe) {
+ throw new UncheckedIOException(ioe);
+ }
+ }
+
+ static PollerGroup create() {
+ return create(DEFAULT_SCHEDULER);
}
/**
! * Start poller threads.
*/
! private PollerGroup start() {
! if (POLLER_MODE.get() == Mode.VTHREAD_POLLERS) {
! if (scheduler == DEFAULT_SCHEDULER) {
! startPlatformThread("Master-Poller", masterPoller::pollerLoop);
! }
Arrays.stream(readPollers).forEach(p -> {
executor.execute(() -> p.subPollerLoop(masterPoller));
});
Arrays.stream(writePollers).forEach(p -> {
executor.execute(() -> p.subPollerLoop(masterPoller));
});
} else {
+ // Mode.SYSTEM_THREADS
Arrays.stream(readPollers).forEach(p -> {
startPlatformThread("Read-Poller", p::pollerLoop);
});
Arrays.stream(writePollers).forEach(p -> {
startPlatformThread("Write-Poller", p::pollerLoop);
});
}
+ return this;
}
Poller masterPoller() {
return masterPoller;
}
+ List<Poller> readPollers() {
+ return List.of(readPollers);
+ }
+
+ List<Poller> writePollers() {
+ return List.of(writePollers);
+ }
+
/**
* Returns the read poller for the given file descriptor.
*/
Poller readPoller(int fdVal) {
! int index = PROVIDER.fdValToIndex(fdVal, readPollers.length);
return readPollers[index];
}
/**
* Returns the write poller for the given file descriptor.
*/
Poller writePoller(int fdVal) {
! int index = PROVIDER.fdValToIndex(fdVal, writePollers.length);
return writePollers[index];
}
/**
* Reads the given property name to get the poller count. If the property is
* set then the value must be a power of 2. Returns 1 if the property is not
* set.
* @throws IllegalArgumentException if the property is set to a value that
thread.start();
} catch (Exception e) {
throw new InternalError(e);
}
}
}
/**
* Return the master poller or null if there is no master poller.
*/
public static Poller masterPoller() {
! return POLLERS.masterPoller();
}
/**
* Return the list of read pollers.
*/
public static List<Poller> readPollers() {
! return POLLERS.readPollers();
}
/**
* Return the list of write pollers.
*/
public static List<Poller> writePollers() {
! return POLLERS.writePollers();
}
}
thread.start();
} catch (Exception e) {
throw new InternalError(e);
}
}
+
+ /**
+ * Returns the PollerGroup that the given thread uses to poll file descriptors.
+ */
+ static PollerGroup groupFor(Thread thread) {
+ if (POLLER_MODE.get() == Mode.SYSTEM_THREADS) {
+ return DEFAULT_POLLER_GROUP.get();
+ }
+ Executor scheduler;
+ if (thread.isVirtual()) {
+ scheduler = JLA.virtualThreadScheduler(thread);
+ } else {
+ scheduler = DEFAULT_SCHEDULER;
+ }
+ return POLLER_GROUPS.computeIfAbsent(scheduler, _ -> PollerGroup.create(scheduler));
+ }
+ }
+
+ /**
+ * Returns the poller mode.
+ */
+ private static Mode pollerMode() {
+ String s = System.getProperty("jdk.pollerMode");
+ if (s != null) {
+ if (s.equalsIgnoreCase(Mode.SYSTEM_THREADS.name()) || s.equals("1")) {
+ return Mode.SYSTEM_THREADS;
+ } else if (s.equalsIgnoreCase(Mode.VTHREAD_POLLERS.name()) || s.equals("2")) {
+ return Mode.VTHREAD_POLLERS;
+ } else {
+ throw new RuntimeException("Can't parse '" + s + "' as polling mode");
+ }
+ } else {
+ return PROVIDER.defaultPollerMode();
+ }
}
/**
* Return the master poller or null if there is no master poller.
*/
public static Poller masterPoller() {
! return DEFAULT_POLLER_GROUP.get().masterPoller();
}
/**
* Return the list of read pollers.
*/
public static List<Poller> readPollers() {
! return DEFAULT_POLLER_GROUP.get().readPollers();
}
/**
* Return the list of write pollers.
*/
public static List<Poller> writePollers() {
! return DEFAULT_POLLER_GROUP.get().writePollers();
}
}
< prev index next >