< prev index next >

src/java.base/share/classes/sun/nio/ch/Poller.java

Print this page
*** 23,38 ***
   * 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
--- 23,44 ---
   * 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

*** 136,14 ***
       */
      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;
          }
      }
  
--- 142,15 ---
       */
      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;
          }
      }
  

*** 152,37 ***
       * @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.
       */
--- 159,23 ---
       * @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.
       */

*** 277,147 ***
          } 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
--- 270,149 ---
          } 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

*** 446,28 ***
                  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();
      }
  }
--- 441,62 ---
                  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 >