1 /*
2 * Copyright (c) 2017, 2026, Oracle and/or its affiliates. All rights reserved.
3 * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
4 *
5 * This code is free software; you can redistribute it and/or modify it
6 * under the terms of the GNU General Public License version 2 only, as
7 * published by the Free Software Foundation. Oracle designates this
8 * particular file as subject to the "Classpath" exception as provided
9 * by Oracle in the LICENSE file that accompanied this code.
10 *
11 * This code is distributed in the hope that it will be useful, but WITHOUT
12 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
13 * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License
14 * version 2 for more details (a copy is included in the LICENSE file that
15 * accompanied this code).
16 *
17 * You should have received a copy of the GNU General Public License version
18 * 2 along with this work; if not, write to the Free Software Foundation,
19 * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
20 *
21 * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA
22 * or visit www.oracle.com if you need additional information or have any
23 * questions.
24 */
25 package sun.nio.ch;
26
27 import java.io.IOException;
28 import java.io.UncheckedIOException;
29 import java.lang.ref.Reference;
30 import java.util.Arrays;
31 import java.util.List;
32 import java.util.Map;
33 import java.util.Objects;
34 import java.util.Set;
35 import java.util.concurrent.ConcurrentHashMap;
36 import java.util.concurrent.Executor;
37 import java.util.concurrent.Executors;
38 import java.util.concurrent.ThreadFactory;
39 import java.util.concurrent.ThreadLocalRandom;
40 import java.util.concurrent.locks.LockSupport;
41 import java.util.function.BooleanSupplier;
42 import jdk.internal.access.JavaLangAccess;
43 import jdk.internal.access.SharedSecrets;
44 import jdk.internal.misc.InnocuousThread;
45 import jdk.internal.misc.TerminatingThreadLocal;
46 import jdk.internal.vm.Continuation;
47 import jdk.internal.vm.ContinuationSupport;
48 import jdk.internal.vm.annotation.Stable;
49
50 /**
51 * I/O poller to allow virtual threads park until a file descriptor is ready for I/O.
52 */
53 public abstract class Poller {
54 private static final JavaLangAccess JLA = SharedSecrets.getJavaLangAccess();
55
56 // the poller group for the I/O pollers and poller threads
57 private static final PollerGroup POLLER_GROUP = createPollerGroup();
58
59 // the poller or sub-poller thread (used for observability only)
60 private @Stable Thread owner;
61
62 // maps file descriptors to parked Thread
63 private final Map<Integer, Thread> map = new ConcurrentHashMap<>();
64
65 // shutdown (if supported by poller group)
66 private volatile boolean shutdown;
67
68 /**
69 * Poller mode.
70 */
71 enum Mode {
72 /**
73 * Read and write pollers are platform threads that block waiting for events and
74 * unpark virtual threads when file descriptors are ready for I/O.
75 */
76 SYSTEM_THREADS,
77
78 /**
79 * Read and write pollers are virtual threads that poll for events, yielding
80 * between polls and unparking virtual threads when file descriptors are
81 * ready for I/O. If there are no events then the poller threads park until there
82 * are I/O events to poll. This mode helps to integrate polling with virtual
83 * thread scheduling. The approach is similar to the default scheme in "User-level
84 * Threading: Have Your Cake and Eat It Too" by Karsten and Barghi 2020
85 * (https://dl.acm.org/doi/10.1145/3379483).
86 */
87 VTHREAD_POLLERS,
88
89 /**
90 * Read pollers are per-carrier virtual threads that poll for events, yielding
91 * between polls and unparking virtual threads when file descriptors are ready
92 * for I/O. If there are no events then the poller threads park until there
93 * are I/O events to poll. The write poller is a system-wide platform thread.
94 */
95 POLLER_PER_CARRIER
96 }
97
98 /**
99 * Create and return the PollerGroup.
100 */
101 private static PollerGroup createPollerGroup() {
102 try {
103 PollerProvider provider;
104 if (System.getProperty("jdk.pollerMode") instanceof String s) {
105 Mode mode = switch (s) {
106 case "1" -> Mode.SYSTEM_THREADS;
107 case "2" -> Mode.VTHREAD_POLLERS;
108 case "3" -> Mode.POLLER_PER_CARRIER;
109 default -> {
110 throw new RuntimeException(s + " is not a valid polling mode");
111 }
112 };
113 provider = PollerProvider.createProvider(mode);
114 } else {
115 provider = PollerProvider.createProvider();
116 }
117
118 int readPollers = pollerCount("jdk.readPollers", provider.defaultReadPollers());
119 int writePollers = pollerCount("jdk.writePollers", provider.defaultWritePollers());
120 PollerGroup group = switch (provider.pollerMode()) {
121 case SYSTEM_THREADS -> new SystemThreadsPollerGroup(provider, readPollers, writePollers);
122 case VTHREAD_POLLERS -> new VThreadsPollerGroup(provider, readPollers, writePollers);
123 case POLLER_PER_CARRIER -> new PollerPerCarrierPollerGroup(provider, writePollers);
124 };
125 group.start();
126 return group;
127 } catch (IOException ioe) {
128 throw new UncheckedIOException(ioe);
129 }
130 }
131
132 /**
133 * Initialize a Poller.
134 */
135 protected Poller() {
136 }
137
138 /**
139 * Closes the poller and release resources. This method can only be used to cleanup
140 * when creating a poller group fails.
141 */
142 abstract void close() throws IOException;
143
144 /**
145 * Sets the poller's thread owner.
146 */
147 private void setOwner() {
148 owner = Thread.currentThread();
149 }
150
151 /**
152 * Returns true if this poller is marked for shutdown.
153 */
154 boolean isShutdown() {
155 return shutdown;
156 }
157
158 /**
159 * Marks this poller for shutdown.
160 */
161 private void setShutdown() {
162 shutdown = true;
163 }
164
165 /**
166 * Returns the poller's file descriptor to use when polling with the master poller.
167 * @throws UnsupportedOperationException if not supported
168 */
169 int fdVal() {
170 throw new UnsupportedOperationException();
171 }
172
173 /**
174 * Register the file descriptor with the I/O event management facility so that it is
175 * polled when the file descriptor is ready for I/O. The registration is "one shot",
176 * meaning it should be polled at most once.
177 */
178 abstract void implStartPoll(int fdVal) throws IOException;
179
180 /**
181 * Deregister a file descriptor from the I/O event management facility. This may be
182 * a no-op in some implementations when the file descriptor has already been polled.
183 * @param polled true if the file descriptor has already been polled
184 */
185 abstract void implStopPoll(int fdVal, boolean polled) throws IOException;
186
187 /**
188 * Poll for events. The {@link #polled(int)} method is invoked for each
189 * polled file descriptor.
190 *
191 * @param timeout if positive then block for up to {@code timeout} milliseconds,
192 * if zero then don't block, if -1 then block indefinitely
193 * @return >0 if file descriptors are polled, 0 if no file descriptor polled
194 */
195 abstract int poll(int timeout) throws IOException;
196
197 /**
198 * Wakeup the poller thread if blocked in poll so it can shutdown.
199 * @throws UnsupportedOperationException if not supported
200 */
201 void wakeupPoller() throws IOException {
202 throw new UnsupportedOperationException();
203 }
204
205 /**
206 * Callback by the poll method when a file descriptor is polled.
207 */
208 final void polled(int fdVal) {
209 Thread t = map.remove(fdVal);
210 if (t != null) {
211 LockSupport.unpark(t);
212 }
213 }
214
215 /**
216 * Parks the current thread until a file descriptor is ready for the given op.
217 * @param fdVal the file descriptor
218 * @param event POLLIN or POLLOUT
219 * @param nanos the waiting time or 0 to wait indefinitely
220 * @param isOpen supplies a boolean to indicate if the enclosing object is open
221 */
222 public static void poll(int fdVal, int event, long nanos, BooleanSupplier isOpen) throws IOException {
223 POLLER_GROUP.poll(fdVal, event, nanos, isOpen);
224 }
225
226 /**
227 * Parks the current thread until a Selector's file descriptor is ready.
228 * @param fdVal the Selector's file descriptor
229 * @param nanos the waiting time or 0 to wait indefinitely
230 */
231 public static void pollSelector(int fdVal, long nanos) throws IOException {
232 POLLER_GROUP.pollSelector(fdVal, nanos);
233 }
234
235 /**
236 * Unpark the given thread so that it stops polling.
237 */
238 public static void stopPoll(Thread thread) {
239 LockSupport.unpark(thread);
240 }
241
242 /**
243 * Parks the current thread until a file descriptor is ready.
244 */
245 private void poll(int fdVal, long nanos, BooleanSupplier isOpen) throws IOException {
246 startPoll(fdVal);
247 try {
248 if (isOpen.getAsBoolean() && !isShutdown()) {
249 if (nanos > 0) {
250 LockSupport.parkNanos(nanos);
251 } else {
252 LockSupport.park();
253 }
254 }
255 } finally {
256 stopPoll(fdVal);
257 }
258 }
259
260 /**
261 * Register a file descriptor with the I/O event management facility so that it is
262 * polled when the file descriptor is ready for I/O.
263 */
264 private void startPoll(int fdVal) throws IOException {
265 Thread previous = map.put(fdVal, Thread.currentThread());
266 assert previous == null;
267 try {
268 implStartPoll(fdVal);
269 } catch (Throwable t) {
270 map.remove(fdVal);
271 throw t;
272 } finally {
273 Reference.reachabilityFence(this);
274 }
275 }
276
277 /**
278 * Deregister a file descriptor from the I/O event management facility.
279 */
280 private void stopPoll(int fdVal) throws IOException {
281 Thread previous = map.remove(fdVal);
282 boolean polled = (previous == null);
283 assert polled || previous == Thread.currentThread();
284 try {
285 implStopPoll(fdVal, polled);
286 } finally {
287 Reference.reachabilityFence(this);
288 }
289 }
290
291 /**
292 * Master polling loop. The {@link #polled(int)} method is invoked for each file
293 * descriptor that is polled.
294 */
295 private void pollerLoop() {
296 setOwner();
297 try {
298 while (!isShutdown()) {
299 poll(-1);
300 }
301 } catch (Exception e) {
302 e.printStackTrace();
303 }
304 }
305
306 /**
307 * Sub-poller polling loop. The {@link #polled(int)} method is invoked for each file
308 * descriptor that is polled.
309 *
310 * The sub-poller registers its file descriptor with the master poller to park until
311 * there are events to poll. When unparked, it does non-blocking polls and parks
312 * again when there are no more events. The sub-poller yields after each poll to help
313 * with fairness and to avoid re-registering with the master poller where possible.
314 */
315 private void subPollerLoop(Poller masterPoller) {
316 assert Thread.currentThread().isVirtual();
317 setOwner();
318 try {
319 int polled = 0;
320 while (!isShutdown()) {
321 if (polled == 0) {
322 masterPoller.poll(fdVal(), 0, () -> true); // park
323 } else {
324 Thread.yield();
325 }
326 polled = poll(0);
327 }
328 } catch (Exception e) {
329 e.printStackTrace();
330 }
331 }
332
333 /**
334 * Unparks all threads waiting on a file descriptor registered with this poller.
335 */
336 private void wakeupAll() {
337 map.values().forEach(LockSupport::unpark);
338 }
339
340 @Override
341 public String toString() {
342 return String.format("%s [registered = %d, owner = %s]",
343 Objects.toIdentityString(this), map.size(), owner);
344 }
345
346 /**
347 * A group of poller threads that support virtual threads polling file descriptors.
348 */
349 private static abstract class PollerGroup {
350 private final PollerProvider provider;
351
352 PollerGroup(PollerProvider provider) {
353 this.provider = provider;
354 }
355
356 final PollerProvider provider() {
357 return provider;
358 }
359
360 /**
361 * Starts the poller group and any system-wide poller threads.
362 */
363 abstract void start();
364
365 /**
366 * Parks the current thread until a file descriptor is ready for the given op.
367 */
368 abstract void poll(int fdVal, int event, long nanos, BooleanSupplier isOpen) throws IOException;
369
370 /**
371 * Parks the current thread until a Selector's file descriptor is ready.
372 */
373 void pollSelector(int fdVal, long nanos) throws IOException {
374 poll(fdVal, Net.POLLIN, nanos, () -> true);
375 }
376
377 /**
378 * Starts a platform thread to run the given task.
379 */
380 protected final void startPlatformThread(String name, Runnable task) {
381 Thread thread = InnocuousThread.newSystemThread(name, task);
382 thread.setDaemon(true);
383 thread.setUncaughtExceptionHandler((t, e) -> e.printStackTrace());
384 thread.start();
385 }
386
387 /**
388 * Return the master poller, or null if no master poller.
389 */
390 abstract Poller masterPoller();
391
392 /**
393 * Return the read pollers.
394 */
395 abstract List<Poller> readPollers();
396
397 /**
398 * Return the write pollers.
399 */
400 abstract List<Poller> writePollers();
401
402 /**
403 * Close the given pollers.
404 */
405 static void closeAll(Poller... pollers) {
406 for (Poller poller : pollers) {
407 if (poller != null) {
408 try {
409 poller.close();
410 } catch (IOException _) { }
411 }
412 }
413 }
414 }
415
416 /**
417 * SYSTEM_THREADS poller group. The read and write pollers are system-wide platform
418 * threads. To avoid a file descriptor being registered with both a read and write
419 * poller at the same time, the poller group can be configured to run without a
420 * write poller, in which case polling for POLLOUT does timed-parked to force the
421 * caller to retry write ops until they succeed (or fail).
422 */
423 private static class SystemThreadsPollerGroup extends PollerGroup {
424 // system-wide read and write pollers
425 private final Poller[] readPollers;
426 private final Poller[] writePollers;
427
428 SystemThreadsPollerGroup(PollerProvider provider,
429 int readPollerCount,
430 int writePollerCount) throws IOException {
431 super(provider);
432 Poller[] readPollers = new Poller[readPollerCount];
433 Poller[] writePollers = new Poller[writePollerCount];
434 try {
435 for (int i = 0; i < readPollerCount; i++) {
436 readPollers[i] = provider.readPoller(false);
437 }
438 for (int i = 0; i < writePollerCount; i++) {
439 writePollers[i] = provider.writePoller(false);
440 }
441 } catch (Throwable e) {
442 closeAll(readPollers);
443 closeAll(writePollers);
444 throw e;
445 }
446
447 this.readPollers = readPollers;
448 this.writePollers = writePollers;
449 }
450
451 @Override
452 void start() {
453 Arrays.stream(readPollers).forEach(p -> {
454 startPlatformThread("Read-Poller", p::pollerLoop);
455 });
456 Arrays.stream(writePollers).forEach(p -> {
457 startPlatformThread("Write-Poller", p::pollerLoop);
458 });
459 }
460
461 private Poller readPoller(int fdVal) {
462 int index = provider().fdValToIndex(fdVal, readPollers.length);
463 return readPollers[index];
464 }
465
466 private Poller writePoller(int fdVal) {
467 if (writePollers.length > 0) {
468 int index = provider().fdValToIndex(fdVal, writePollers.length);
469 return writePollers[index];
470 } else {
471 return null;
472 }
473 }
474
475 @Override
476 void poll(int fdVal, int event, long nanos, BooleanSupplier isOpen) throws IOException {
477 if (event == Net.POLLIN) {
478 readPoller(fdVal).poll(fdVal, nanos, isOpen);
479 } else {
480 Poller writePoller = writePoller(fdVal);
481 if (writePoller != null) {
482 writePoller.poll(fdVal, nanos, isOpen);
483 } else {
484 // random delay in [50ms, 120ms)
485 int max = ThreadLocalRandom.current().nextInt(50_000_000, 120_000_000);
486 long delay = (nanos > 0) ? Math.min(nanos, max) : max;
487 LockSupport.parkNanos(delay);
488 }
489 }
490 }
491
492 @Override
493 Poller masterPoller() {
494 return null;
495 }
496
497 @Override
498 List<Poller> readPollers() {
499 return List.of(readPollers);
500 }
501
502 @Override
503 List<Poller> writePollers() {
504 return List.of(writePollers);
505 }
506 }
507
508 /**
509 * VTHREAD_POLLERS poller group. The read and write pollers are virtual threads.
510 * When read and write pollers need to block then they register with a system-wide
511 * "master poller" that runs in a dedicated platform thread.
512 */
513 private static class VThreadsPollerGroup extends PollerGroup {
514 private final Poller masterPoller;
515 private final Poller[] readPollers;
516 private final Poller[] writePollers;
517
518 // keep virtual thread pollers alive
519 private final Executor executor;
520
521 VThreadsPollerGroup(PollerProvider provider,
522 int readPollerCount,
523 int writePollerCount) throws IOException {
524 super(provider);
525 Poller masterPoller = provider.readPoller(false);
526 Poller[] readPollers = new Poller[readPollerCount];
527 Poller[] writePollers = new Poller[writePollerCount];
528
529 try {
530 for (int i = 0; i < readPollerCount; i++) {
531 readPollers[i] = provider.readPoller(true);
532 }
533 for (int i = 0; i < writePollerCount; i++) {
534 writePollers[i] = provider.writePoller(true);
535 }
536 } catch (Throwable e) {
537 masterPoller.close();
538 closeAll(readPollers);
539 closeAll(writePollers);
540 throw e;
541 }
542
543 this.masterPoller = masterPoller;
544 this.readPollers = readPollers;
545 this.writePollers = writePollers;
546
547 ThreadFactory factory = Thread.ofVirtual()
548 .inheritInheritableThreadLocals(false)
549 .name("SubPoller-", 0)
550 .uncaughtExceptionHandler((_, e) -> e.printStackTrace())
551 .factory();
552 this.executor = Executors.newThreadPerTaskExecutor(factory);
553 }
554
555 @Override
556 void start() {
557 startPlatformThread("Master-Poller", masterPoller::pollerLoop);
558 Arrays.stream(readPollers).forEach(p -> {
559 executor.execute(() -> p.subPollerLoop(masterPoller));
560 });
561 Arrays.stream(writePollers).forEach(p -> {
562 executor.execute(() -> p.subPollerLoop(masterPoller));
563 });
564 }
565
566 private Poller readPoller(int fdVal) {
567 int index = provider().fdValToIndex(fdVal, readPollers.length);
568 return readPollers[index];
569 }
570
571 private Poller writePoller(int fdVal) {
572 int index = provider().fdValToIndex(fdVal, writePollers.length);
573 return writePollers[index];
574 }
575
576 @Override
577 void poll(int fdVal, int event, long nanos, BooleanSupplier isOpen) throws IOException {
578 Poller poller = (event == Net.POLLIN)
579 ? readPoller(fdVal)
580 : writePoller(fdVal);
581 poller.poll(fdVal, nanos, isOpen);
582 }
583
584 @Override
585 void pollSelector(int fdVal, long nanos) throws IOException {
586 masterPoller.poll(fdVal, nanos, () -> true);
587 }
588
589 @Override
590 Poller masterPoller() {
591 return masterPoller;
592 }
593
594 @Override
595 List<Poller> readPollers() {
596 return List.of(readPollers);
597 }
598
599 @Override
600 List<Poller> writePollers() {
601 return List.of(writePollers);
602 }
603 }
604
605 /**
606 * POLLER_PER_CARRIER poller group. The read poller is a per-carrier virtual thread.
607 * When a virtual thread polls a file descriptor for POLLIN, then it will use (almost
608 * always, not guaranteed) the read poller for its carrier. When a read poller needs
609 * to block then it registers with a system-wide "master poller" that runs in a
610 * dedicated platform thread. The read poller terminates if the carrier terminates.
611 * The write pollers are system-wide platform threads (usually one).
612 */
613 private static class PollerPerCarrierPollerGroup extends PollerGroup {
614 private record CarrierPoller(PollerPerCarrierPollerGroup group, Poller readPoller) { }
615 private static final TerminatingThreadLocal<CarrierPoller> CARRIER_POLLER =
616 new TerminatingThreadLocal<>() {
617 @Override
618 protected void threadTerminated(CarrierPoller carrierPoller) {
619 Poller readPoller = carrierPoller.readPoller();
620 carrierPoller.group().carrierTerminated(readPoller);
621 }
622 };
623
624 private final Poller masterPoller;
625 private final Set<Poller> readPollers;
626 private final Poller[] writePollers;
627
628 /**
629 * Create a PollerPerCarrierPollerGroup with the given number of write pollers.
630 */
631 PollerPerCarrierPollerGroup(PollerProvider provider,
632 int writePollerCount) throws IOException {
633 super(provider);
634 Poller masterPoller = provider.readPoller(false);
635 Poller[] writePollers = new Poller[writePollerCount];
636 try {
637 for (int i = 0; i < writePollerCount; i++) {
638 writePollers[i] = provider.writePoller(false);
639 }
640 } catch (Throwable e) {
641 masterPoller.close();
642 closeAll(writePollers);
643 throw e;
644 }
645 this.masterPoller = masterPoller;
646 this.readPollers = ConcurrentHashMap.newKeySet();;
647 this.writePollers = writePollers;
648 }
649
650 @Override
651 void start() {
652 startPlatformThread("Master-Poller", masterPoller::pollerLoop);
653 Arrays.stream(writePollers).forEach(p -> {
654 startPlatformThread("Write-Poller", p::pollerLoop);
655 });
656 }
657
658 private Poller writePoller(int fdVal) {
659 int index = provider().fdValToIndex(fdVal, writePollers.length);
660 return writePollers[index];
661 }
662
663 /**
664 * Starts a read sub-poller in a virtual thread.
665 */
666 private Poller startReadPoller() throws IOException {
667 assert Thread.currentThread().isVirtual() && ContinuationSupport.isSupported();
668
669 // create read sub-poller
670 Poller readPoller = provider().readPoller(true);
671 readPollers.add(readPoller);
672
673 // start virtual thread to execute sub-polling loop
674 Thread carrier = JLA.currentCarrierThread();
675 Thread.ofVirtual()
676 .inheritInheritableThreadLocals(false)
677 .name(carrier.getName() + "-Read-Poller")
678 .uncaughtExceptionHandler((_, e) -> e.printStackTrace())
679 .start(() -> subPollerLoop(readPoller));
680 return readPoller;
681 }
682
683 /**
684 * Returns the read poller for the current carrier, starting it if required.
685 */
686 private Poller readPoller() throws IOException {
687 assert Thread.currentThread().isVirtual() && ContinuationSupport.isSupported();
688 Continuation.pin();
689 try {
690 CarrierPoller carrierPoller = CARRIER_POLLER.get();
691 if (carrierPoller != null) {
692 return carrierPoller.readPoller();
693 } else {
694 // first poll on this carrier will start poller
695 Poller readPoller = startReadPoller();
696 CARRIER_POLLER.set(new CarrierPoller(this, readPoller));
697 return readPoller;
698 }
699 } finally {
700 Continuation.unpin();
701 }
702 }
703
704 @Override
705 void poll(int fdVal, int event, long nanos, BooleanSupplier isOpen) throws IOException {
706 // for POLLIN, get the read poller for this carrier
707 if (event == Net.POLLIN
708 && Thread.currentThread().isVirtual()
709 && ContinuationSupport.isSupported()) {
710 readPoller().poll(fdVal, nanos, isOpen);
711 return;
712 }
713
714 // -XX:-VMContinuations or POLLIN from platform thread does master poller
715 if (event == Net.POLLIN) {
716 masterPoller.poll(fdVal, nanos, isOpen);
717 } else {
718 writePoller(fdVal).poll(fdVal, nanos, isOpen);
719 }
720 }
721
722 @Override
723 void pollSelector(int fdVal, long nanos) throws IOException {
724 masterPoller.poll(fdVal, nanos, () -> true);
725 }
726
727 /**
728 * Sub-poller polling loop.
729 */
730 private void subPollerLoop(Poller readPoller) {
731 try {
732 readPoller.subPollerLoop(masterPoller);
733 } finally {
734 // wakeup all threads waiting on file descriptors registered with the
735 // read poller, these I/O operation will migrate to another carrier.
736 readPoller.wakeupAll();
737
738 // remove from serviceability view
739 readPollers.remove(readPoller);
740 }
741 }
742
743 /**
744 * Invoked by the carrier thread before it terminates.
745 */
746 private void carrierTerminated(Poller readPoller) {
747 readPoller.setShutdown();
748 try {
749 readPoller.wakeupPoller();
750 } catch (Throwable e) {
751 e.printStackTrace();
752 }
753 }
754
755 @Override
756 Poller masterPoller() {
757 return masterPoller;
758 }
759
760 @Override
761 List<Poller> readPollers() {
762 return readPollers.stream().toList();
763 }
764
765 @Override
766 List<Poller> writePollers() {
767 return List.of(writePollers);
768 }
769 }
770
771 /**
772 * Reads the given property name to get the poller count. If the property is
773 * set then the value must be a power of 2. Returns 1 if the property is not
774 * set.
775 * @throws IllegalArgumentException if the property is set to a value that
776 * is not a power of 2.
777 */
778 private static int pollerCount(String propName, int defaultCount) {
779 String s = System.getProperty(propName);
780 int count = (s != null) ? Integer.parseInt(s) : defaultCount;
781
782 // check power of 2
783 if (count != Integer.highestOneBit(count)) {
784 String msg = propName + " is set to a value that is not a power of 2";
785 throw new IllegalArgumentException(msg);
786 }
787 return count;
788 }
789
790 /**
791 * Return the master poller or null if there is no master poller.
792 */
793 public static Poller masterPoller() {
794 return POLLER_GROUP.masterPoller();
795 }
796
797 /**
798 * Return the list of read pollers.
799 */
800 public static List<Poller> readPollers() {
801 return POLLER_GROUP.readPollers();
802 }
803
804 /**
805 * Return the list of write pollers.
806 */
807 public static List<Poller> writePollers() {
808 return POLLER_GROUP.writePollers();
809 }
810 }