637 });
638 }
639
640 private Poller writePoller(int fdVal) {
641 int index = provider().fdValToIndex(fdVal, writePollers.length);
642 return writePollers[index];
643 }
644
645 /**
646 * Starts a read sub-poller in a virtual thread.
647 */
648 private Poller startReadPoller() throws IOException {
649 assert Thread.currentThread().isVirtual() && ContinuationSupport.isSupported();
650
651 // create read sub-poller
652 Poller readPoller = provider().readPoller(true);
653 readPollers.add(readPoller);
654
655 // start virtual thread to execute sub-polling loop
656 Thread carrier = JLA.currentCarrierThread();
657 Thread.ofVirtual()
658 .inheritInheritableThreadLocals(false)
659 .name(carrier.getName() + "-Read-Poller")
660 .uncaughtExceptionHandler((_, e) -> e.printStackTrace())
661 .start(() -> subPollerLoop(readPoller));
662 return readPoller;
663 }
664
665 /**
666 * Returns the read poller for the current carrier, starting it if required.
667 */
668 private Poller readPoller() throws IOException {
669 assert Thread.currentThread().isVirtual() && ContinuationSupport.isSupported();
670 Continuation.pin();
671 try {
672 CarrierPoller carrierPoller = CARRIER_POLLER.get();
673 if (carrierPoller != null) {
674 return carrierPoller.readPoller();
675 } else {
676 // first poll on this carrier will start poller
677 Poller readPoller = startReadPoller();
678 CARRIER_POLLER.set(new CarrierPoller(this, readPoller));
679 return readPoller;
680 }
681 } finally {
|
637 });
638 }
639
640 private Poller writePoller(int fdVal) {
641 int index = provider().fdValToIndex(fdVal, writePollers.length);
642 return writePollers[index];
643 }
644
645 /**
646 * Starts a read sub-poller in a virtual thread.
647 */
648 private Poller startReadPoller() throws IOException {
649 assert Thread.currentThread().isVirtual() && ContinuationSupport.isSupported();
650
651 // create read sub-poller
652 Poller readPoller = provider().readPoller(true);
653 readPollers.add(readPoller);
654
655 // start virtual thread to execute sub-polling loop
656 Thread carrier = JLA.currentCarrierThread();
657 Thread.Builder.OfVirtual builder = Thread.ofVirtual()
658 .inheritInheritableThreadLocals(false)
659 .name(carrier.getName() + "-Read-Poller")
660 .uncaughtExceptionHandler((_, e) -> e.printStackTrace());
661 Thread thread = JLA.defaultVirtualThreadScheduler()
662 .newThread(builder, carrier, () -> subPollerLoop(readPoller))
663 .thread();
664 thread.start();
665 return readPoller;
666 }
667
668 /**
669 * Returns the read poller for the current carrier, starting it if required.
670 */
671 private Poller readPoller() throws IOException {
672 assert Thread.currentThread().isVirtual() && ContinuationSupport.isSupported();
673 Continuation.pin();
674 try {
675 CarrierPoller carrierPoller = CARRIER_POLLER.get();
676 if (carrierPoller != null) {
677 return carrierPoller.readPoller();
678 } else {
679 // first poll on this carrier will start poller
680 Poller readPoller = startReadPoller();
681 CARRIER_POLLER.set(new CarrierPoller(this, readPoller));
682 return readPoller;
683 }
684 } finally {
|