< prev index next >

src/java.base/linux/classes/sun/nio/ch/EPollSelectorImpl.java

Print this page

 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 
 26 package sun.nio.ch;
 27 
 28 import java.io.IOException;
 29 import java.nio.channels.ClosedSelectorException;
 30 import java.nio.channels.SelectionKey;
 31 import java.nio.channels.Selector;
 32 import java.nio.channels.spi.SelectorProvider;
 33 import java.util.ArrayDeque;
 34 import java.util.Deque;
 35 import java.util.HashMap;
 36 import java.util.Map;
 37 import java.util.concurrent.TimeUnit;
 38 import java.util.function.Consumer;
 39 import jdk.internal.misc.Blocker;
 40 
 41 import static sun.nio.ch.EPoll.EPOLLIN;
 42 import static sun.nio.ch.EPoll.EPOLL_CTL_ADD;
 43 import static sun.nio.ch.EPoll.EPOLL_CTL_DEL;
 44 import static sun.nio.ch.EPoll.EPOLL_CTL_MOD;
 45 
 46 
 47 /**
 48  * Linux epoll based Selector implementation
 49  */
 50 
 51 class EPollSelectorImpl extends SelectorImpl {
 52 
 53     // maximum number of events to poll in one call to epoll_wait
 54     private static final int NUM_EPOLLEVENTS = Math.min(IOUtil.fdLimit(), 1024);
 55 
 56     // epoll file descriptor
 57     private final int epfd;
 58 
 59     // address of poll array when polling with epoll_wait

 94 
 95     private void ensureOpen() {
 96         if (!isOpen())
 97             throw new ClosedSelectorException();
 98     }
 99 
100     @Override
101     protected int doSelect(Consumer<SelectionKey> action, long timeout)
102         throws IOException
103     {
104         assert Thread.holdsLock(this);
105 
106         // epoll_wait timeout is int
107         int to = (int) Math.min(timeout, Integer.MAX_VALUE);
108         boolean blocking = (to != 0);
109         boolean timedPoll = (to > 0);
110 
111         int numEntries;
112         processUpdateQueue();
113         processDeregisterQueue();
114         try {
115             begin(blocking);
116 
117             do {
118                 long startTime = timedPoll ? System.nanoTime() : 0;
119                 long comp = Blocker.begin(blocking);
120                 try {





121                     numEntries = EPoll.wait(epfd, pollArrayAddress, NUM_EPOLLEVENTS, to);
122                 } finally {
123                     Blocker.end(comp);
124                 }
125                 if (numEntries == IOStatus.INTERRUPTED && timedPoll) {
126                     // timed poll interrupted so need to adjust timeout
127                     long adjust = System.nanoTime() - startTime;
128                     to -= (int) TimeUnit.NANOSECONDS.toMillis(adjust);
129                     if (to <= 0) {
130                         // timeout expired so no retry
131                         numEntries = 0;
132                     }
133                 }
134             } while (numEntries == IOStatus.INTERRUPTED);
135             assert IOStatus.check(numEntries);
136 
137         } finally {
138             end(blocking);
139         }


140         processDeregisterQueue();
141         return processEvents(numEntries, action);
142     }
143 


































144     /**
145      * Process changes to the interest ops.
146      */
147     private void processUpdateQueue() {
148         assert Thread.holdsLock(this);
149 
150         synchronized (updateLock) {
151             SelectionKeyImpl ski;
152             while ((ski = updateKeys.pollFirst()) != null) {
153                 if (ski.isValid()) {
154                     int fd = ski.getFDVal();
155                     // add to fdToKey if needed
156                     SelectionKeyImpl previous = fdToKey.putIfAbsent(fd, ski);
157                     assert (previous == null) || (previous == ski);
158 
159                     int newEvents = ski.translateInterestOps();
160                     int registeredEvents = ski.registeredEvents();
161                     if (newEvents != registeredEvents) {
162                         if (newEvents == 0) {
163                             // remove from epoll

 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 
 26 package sun.nio.ch;
 27 
 28 import java.io.IOException;
 29 import java.nio.channels.ClosedSelectorException;
 30 import java.nio.channels.SelectionKey;
 31 import java.nio.channels.Selector;
 32 import java.nio.channels.spi.SelectorProvider;
 33 import java.util.ArrayDeque;
 34 import java.util.Deque;
 35 import java.util.HashMap;
 36 import java.util.Map;
 37 import java.util.concurrent.TimeUnit;
 38 import java.util.function.Consumer;

 39 
 40 import static sun.nio.ch.EPoll.EPOLLIN;
 41 import static sun.nio.ch.EPoll.EPOLL_CTL_ADD;
 42 import static sun.nio.ch.EPoll.EPOLL_CTL_DEL;
 43 import static sun.nio.ch.EPoll.EPOLL_CTL_MOD;
 44 
 45 
 46 /**
 47  * Linux epoll based Selector implementation
 48  */
 49 
 50 class EPollSelectorImpl extends SelectorImpl {
 51 
 52     // maximum number of events to poll in one call to epoll_wait
 53     private static final int NUM_EPOLLEVENTS = Math.min(IOUtil.fdLimit(), 1024);
 54 
 55     // epoll file descriptor
 56     private final int epfd;
 57 
 58     // address of poll array when polling with epoll_wait

 93 
 94     private void ensureOpen() {
 95         if (!isOpen())
 96             throw new ClosedSelectorException();
 97     }
 98 
 99     @Override
100     protected int doSelect(Consumer<SelectionKey> action, long timeout)
101         throws IOException
102     {
103         assert Thread.holdsLock(this);
104 
105         // epoll_wait timeout is int
106         int to = (int) Math.min(timeout, Integer.MAX_VALUE);
107         boolean blocking = (to != 0);
108         boolean timedPoll = (to > 0);
109 
110         int numEntries;
111         processUpdateQueue();
112         processDeregisterQueue();


113 
114         if (Thread.currentThread().isVirtual()) {
115             numEntries = (timedPoll)
116                     ? timedPoll(TimeUnit.MILLISECONDS.toNanos(to))
117                     : untimedPoll(blocking);
118         } else {
119             try {
120                 begin(blocking);
121                 do {
122                     long startTime = timedPoll ? System.nanoTime() : 0;
123                     numEntries = EPoll.wait(epfd, pollArrayAddress, NUM_EPOLLEVENTS, to);
124                     if (numEntries == IOStatus.INTERRUPTED && timedPoll) {
125                         // timed poll interrupted so need to adjust timeout
126                         long adjust = System.nanoTime() - startTime;
127                         to -= (int) TimeUnit.NANOSECONDS.toMillis(adjust);
128                         if (to <= 0) {
129                             // timeout expired so no retry
130                             numEntries = 0;
131                         }


132                     }
133                 } while (numEntries == IOStatus.INTERRUPTED);
134             } finally {
135                 end(blocking);
136             }


137         }
138         assert IOStatus.check(numEntries);
139 
140         processDeregisterQueue();
141         return processEvents(numEntries, action);
142     }
143 
144     /**
145      * If blocking, parks the current virtual thread until a file descriptor is polled
146      * or the thread is interrupted.
147      */
148     private int untimedPoll(boolean block) throws IOException {
149         int numEntries = EPoll.wait(epfd, pollArrayAddress, NUM_EPOLLEVENTS, 0);
150         if (block) {
151             while (numEntries == 0 && !Thread.currentThread().isInterrupted()) {
152                 Poller.pollSelector(epfd, 0);
153                 numEntries = EPoll.wait(epfd, pollArrayAddress, NUM_EPOLLEVENTS, 0);
154             }
155         }
156         return numEntries;
157     }
158 
159     /**
160      * Parks the current virtual thread until a file descriptor is polled, or the thread
161      * is interrupted, for up to the specified waiting time.
162      */
163     private int timedPoll(long nanos) throws IOException {
164         long startNanos = System.nanoTime();
165         int numEntries = EPoll.wait(epfd, pollArrayAddress, NUM_EPOLLEVENTS, 0);
166         while (numEntries == 0 && !Thread.currentThread().isInterrupted()) {
167             long remainingNanos = nanos - (System.nanoTime() - startNanos);
168             if (remainingNanos <= 0) {
169                 // timeout
170                 break;
171             }
172             Poller.pollSelector(epfd, remainingNanos);
173             numEntries = EPoll.wait(epfd, pollArrayAddress, NUM_EPOLLEVENTS, 0);
174         }
175         return numEntries;
176     }
177 
178     /**
179      * Process changes to the interest ops.
180      */
181     private void processUpdateQueue() {
182         assert Thread.holdsLock(this);
183 
184         synchronized (updateLock) {
185             SelectionKeyImpl ski;
186             while ((ski = updateKeys.pollFirst()) != null) {
187                 if (ski.isValid()) {
188                     int fd = ski.getFDVal();
189                     // add to fdToKey if needed
190                     SelectionKeyImpl previous = fdToKey.putIfAbsent(fd, ski);
191                     assert (previous == null) || (previous == ski);
192 
193                     int newEvents = ski.translateInterestOps();
194                     int registeredEvents = ski.registeredEvents();
195                     if (newEvents != registeredEvents) {
196                         if (newEvents == 0) {
197                             // remove from epoll
< prev index next >