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 {
|
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.Builder.OfVirtual builder = Thread.ofVirtual()
676 .inheritInheritableThreadLocals(false)
677 .name(carrier.getName() + "-Read-Poller")
678 .uncaughtExceptionHandler((_, e) -> e.printStackTrace());
679 Thread thread = JLA.defaultVirtualThreadScheduler()
680 .newThread(builder, carrier, () -> subPollerLoop(readPoller))
681 .thread();
682 thread.start();
683 return readPoller;
684 }
685
686 /**
687 * Returns the read poller for the current carrier, starting it if required.
688 */
689 private Poller readPoller() throws IOException {
690 assert Thread.currentThread().isVirtual() && ContinuationSupport.isSupported();
691 Continuation.pin();
692 try {
693 CarrierPoller carrierPoller = CARRIER_POLLER.get();
694 if (carrierPoller != null) {
695 return carrierPoller.readPoller();
696 } else {
697 // first poll on this carrier will start poller
698 Poller readPoller = startReadPoller();
699 CARRIER_POLLER.set(new CarrierPoller(this, readPoller));
700 return readPoller;
701 }
702 } finally {
|