1 /*
2 * Copyright (c) 2002, 2018, 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
26 /*
27 */
28
29
30 package sun.nio.ch;
31
32 import java.nio.channels.spi.SelectorProvider;
33 import java.nio.channels.Selector;
34 import java.nio.channels.ClosedSelectorException;
35 import java.nio.channels.Pipe;
36 import java.nio.channels.SelectableChannel;
37 import java.io.IOException;
38 import java.nio.channels.CancelledKeyException;
39 import java.util.List;
40 import java.util.ArrayList;
41 import java.util.HashMap;
42 import java.util.Iterator;
43
44 /**
45 * A multi-threaded implementation of Selector for Windows.
46 *
47 * @author Konstantin Kladko
48 * @author Mark Reinhold
49 */
50
51 class WindowsSelectorImpl extends SelectorImpl {
52 // Initial capacity of the poll array
53 private final int INIT_CAP = 8;
54 // Maximum number of sockets for select().
55 // Should be INIT_CAP times a power of 2
56 private static final int MAX_SELECTABLE_FDS = 1024;
57
58 // The list of SelectableChannels serviced by this Selector. Every mod
59 // MAX_SELECTABLE_FDS entry is bogus, to align this array with the poll
60 // array, where the corresponding entry is occupied by the wakeupSocket
61 private SelectionKeyImpl[] channelArray = new SelectionKeyImpl[INIT_CAP];
62
63 // The global native poll array holds file decriptors and event masks
64 private PollArrayWrapper pollWrapper;
65
66 // The number of valid entries in poll array, including entries occupied
67 // by wakeup socket handle.
68 private int totalChannels = 1;
69
70 // Number of helper threads needed for select. We need one thread per
71 // each additional set of MAX_SELECTABLE_FDS - 1 channels.
72 private int threadsCount = 0;
73
74 // A list of helper threads for select.
75 private final List<SelectThread> threads = new ArrayList<SelectThread>();
76
77 //Pipe used as a wakeup object.
78 private final Pipe wakeupPipe;
79
80 // File descriptors corresponding to source and sink
81 private final int wakeupSourceFd, wakeupSinkFd;
82
83 // Lock for close cleanup
84 private final Object closeLock = new Object();
85
86 // Maps file descriptors to their indices in pollArray
87 private static final class FdMap extends HashMap<Integer, MapEntry> {
88 static final long serialVersionUID = 0L;
89 private MapEntry get(int desc) {
90 return get(Integer.valueOf(desc));
91 }
92 private MapEntry put(SelectionKeyImpl ski) {
93 return put(Integer.valueOf(ski.channel.getFDVal()), new MapEntry(ski));
94 }
95 private MapEntry remove(SelectionKeyImpl ski) {
96 Integer fd = Integer.valueOf(ski.channel.getFDVal());
97 MapEntry x = get(fd);
98 if ((x != null) && (x.ski.channel == ski.channel))
99 return remove(fd);
100 return null;
101 }
102 }
103
104 // class for fdMap entries
105 private static final class MapEntry {
106 SelectionKeyImpl ski;
107 long updateCount = 0;
108 long clearedCount = 0;
109 MapEntry(SelectionKeyImpl ski) {
110 this.ski = ski;
111 }
112 }
113 private final FdMap fdMap = new FdMap();
114
115 // SubSelector for the main thread
116 private final SubSelector subSelector = new SubSelector();
117
118 private long timeout; //timeout for poll
119
120 // Lock for interrupt triggering and clearing
121 private final Object interruptLock = new Object();
122 private volatile boolean interruptTriggered;
123
124 WindowsSelectorImpl(SelectorProvider sp) throws IOException {
125 super(sp);
126 pollWrapper = new PollArrayWrapper(INIT_CAP);
127 wakeupPipe = Pipe.open();
128 wakeupSourceFd = ((SelChImpl)wakeupPipe.source()).getFDVal();
129
130 // Disable the Nagle algorithm so that the wakeup is more immediate
131 SinkChannelImpl sink = (SinkChannelImpl)wakeupPipe.sink();
132 (sink.sc).socket().setTcpNoDelay(true);
133 wakeupSinkFd = ((SelChImpl)sink).getFDVal();
134
135 pollWrapper.addWakeupSocket(wakeupSourceFd, 0);
136 }
137
138 @Override
139 protected int doSelect(long timeout) throws IOException {
140 if (channelArray == null)
141 throw new ClosedSelectorException();
142 this.timeout = timeout; // set selector timeout
143 processDeregisterQueue();
144 if (interruptTriggered) {
145 resetWakeupSocket();
146 return 0;
147 }
148 // Calculate number of helper threads needed for poll. If necessary
149 // threads are created here and start waiting on startLock
150 adjustThreadsCount();
151 finishLock.reset(); // reset finishLock
152 // Wakeup helper threads, waiting on startLock, so they start polling.
153 // Redundant threads will exit here after wakeup.
154 startLock.startThreads();
155 // do polling in the main thread. Main thread is responsible for
156 // first MAX_SELECTABLE_FDS entries in pollArray.
157 try {
158 begin();
159 try {
160 subSelector.poll();
161 } catch (IOException e) {
162 finishLock.setException(e); // Save this exception
163 }
164 // Main thread is out of poll(). Wakeup others and wait for them
165 if (threads.size() > 0)
166 finishLock.waitForHelperThreads();
167 } finally {
168 end();
169 }
170 // Done with poll(). Set wakeupSocket to nonsignaled for the next run.
171 finishLock.checkForException();
172 processDeregisterQueue();
173 int updated = updateSelectedKeys();
174 // Done with poll(). Set wakeupSocket to nonsignaled for the next run.
175 resetWakeupSocket();
176 return updated;
177 }
178
179 // Helper threads wait on this lock for the next poll.
180 private final StartLock startLock = new StartLock();
181
182 private final class StartLock {
183 // A variable which distinguishes the current run of doSelect from the
184 // previous one. Incrementing runsCounter and notifying threads will
185 // trigger another round of poll.
186 private long runsCounter;
187 // Triggers threads, waiting on this lock to start polling.
188 private synchronized void startThreads() {
189 runsCounter++; // next run
190 notifyAll(); // wake up threads.
191 }
192 // This function is called by a helper thread to wait for the
193 // next round of poll(). It also checks, if this thread became
194 // redundant. If yes, it returns true, notifying the thread
195 // that it should exit.
196 private synchronized boolean waitForStart(SelectThread thread) {
197 while (true) {
198 while (runsCounter == thread.lastRun) {
199 try {
200 startLock.wait();
201 } catch (InterruptedException e) {
202 Thread.currentThread().interrupt();
203 }
204 }
205 if (thread.isZombie()) { // redundant thread
206 return true; // will cause run() to exit.
207 } else {
208 thread.lastRun = runsCounter; // update lastRun
209 return false; // will cause run() to poll.
210 }
211 }
212 }
213 }
214
215 // Main thread waits on this lock, until all helper threads are done
216 // with poll().
217 private final FinishLock finishLock = new FinishLock();
218
219 private final class FinishLock {
220 // Number of helper threads, that did not finish yet.
221 private int threadsToFinish;
222
223 // IOException which occurred during the last run.
224 IOException exception = null;
225
226 // Called before polling.
227 private void reset() {
228 threadsToFinish = threads.size(); // helper threads
229 }
230
231 // Each helper thread invokes this function on finishLock, when
232 // the thread is done with poll().
233 private synchronized void threadFinished() {
234 if (threadsToFinish == threads.size()) { // finished poll() first
235 // if finished first, wakeup others
236 wakeup();
237 }
238 threadsToFinish--;
239 if (threadsToFinish == 0) // all helper threads finished poll().
240 notify(); // notify the main thread
241 }
242
243 // The main thread invokes this function on finishLock to wait
244 // for helper threads to finish poll().
245 private synchronized void waitForHelperThreads() {
246 if (threadsToFinish == threads.size()) {
247 // no helper threads finished yet. Wakeup them up.
248 wakeup();
249 }
250 while (threadsToFinish != 0) {
251 try {
252 finishLock.wait();
253 } catch (InterruptedException e) {
254 // Interrupted - set interrupted state.
255 Thread.currentThread().interrupt();
256 }
257 }
258 }
259
260 // sets IOException for this run
261 private synchronized void setException(IOException e) {
262 exception = e;
263 }
264
265 // Checks if there was any exception during the last run.
266 // If yes, throws it
267 private void checkForException() throws IOException {
268 if (exception == null)
269 return;
270 StringBuffer message = new StringBuffer("An exception occurred" +
271 " during the execution of select(): \n");
272 message.append(exception);
273 message.append('\n');
274 exception = null;
275 throw new IOException(message.toString());
276 }
277 }
278
279 private final class SubSelector {
280 private final int pollArrayIndex; // starting index in pollArray to poll
281 // These arrays will hold result of native select().
282 // The first element of each array is the number of selected sockets.
283 // Other elements are file descriptors of selected sockets.
284 private final int[] readFds = new int [MAX_SELECTABLE_FDS + 1];
285 private final int[] writeFds = new int [MAX_SELECTABLE_FDS + 1];
286 private final int[] exceptFds = new int [MAX_SELECTABLE_FDS + 1];
287
288 private SubSelector() {
289 this.pollArrayIndex = 0; // main thread
290 }
291
292 private SubSelector(int threadIndex) { // helper threads
293 this.pollArrayIndex = (threadIndex + 1) * MAX_SELECTABLE_FDS;
294 }
295
296 private int poll() throws IOException{ // poll for the main thread
297 return poll0(pollWrapper.pollArrayAddress,
298 Math.min(totalChannels, MAX_SELECTABLE_FDS),
299 readFds, writeFds, exceptFds, timeout);
300 }
301
302 private int poll(int index) throws IOException {
303 // poll for helper threads
304 return poll0(pollWrapper.pollArrayAddress +
305 (pollArrayIndex * PollArrayWrapper.SIZE_POLLFD),
306 Math.min(MAX_SELECTABLE_FDS,
307 totalChannels - (index + 1) * MAX_SELECTABLE_FDS),
308 readFds, writeFds, exceptFds, timeout);
309 }
310
311 private native int poll0(long pollAddress, int numfds,
312 int[] readFds, int[] writeFds, int[] exceptFds, long timeout);
313
314 private int processSelectedKeys(long updateCount) {
315 int numKeysUpdated = 0;
316 numKeysUpdated += processFDSet(updateCount, readFds,
317 Net.POLLIN,
318 false);
319 numKeysUpdated += processFDSet(updateCount, writeFds,
320 Net.POLLCONN |
321 Net.POLLOUT,
322 false);
323 numKeysUpdated += processFDSet(updateCount, exceptFds,
324 Net.POLLIN |
325 Net.POLLCONN |
326 Net.POLLOUT,
327 true);
328 return numKeysUpdated;
329 }
330
331 /**
332 * Note, clearedCount is used to determine if the readyOps have
333 * been reset in this select operation. updateCount is used to
334 * tell if a key has been counted as updated in this select
335 * operation.
336 *
337 * me.updateCount <= me.clearedCount <= updateCount
338 */
339 private int processFDSet(long updateCount, int[] fds, int rOps,
340 boolean isExceptFds)
341 {
342 int numKeysUpdated = 0;
343 for (int i = 1; i <= fds[0]; i++) {
344 int desc = fds[i];
345 if (desc == wakeupSourceFd) {
346 synchronized (interruptLock) {
347 interruptTriggered = true;
348 }
349 continue;
350 }
351 MapEntry me = fdMap.get(desc);
352 // If me is null, the key was deregistered in the previous
353 // processDeregisterQueue.
354 if (me == null)
355 continue;
356 SelectionKeyImpl sk = me.ski;
357
358 // The descriptor may be in the exceptfds set because there is
359 // OOB data queued to the socket. If there is OOB data then it
360 // is discarded and the key is not added to the selected set.
361 if (isExceptFds &&
362 (sk.channel() instanceof SocketChannelImpl) &&
363 discardUrgentData(desc))
364 {
365 continue;
366 }
367
368 if (selectedKeys.contains(sk)) { // Key in selected set
369 if (me.clearedCount != updateCount) {
370 if (sk.channel.translateAndSetReadyOps(rOps, sk) &&
371 (me.updateCount != updateCount)) {
372 me.updateCount = updateCount;
373 numKeysUpdated++;
374 }
375 } else { // The readyOps have been set; now add
376 if (sk.channel.translateAndUpdateReadyOps(rOps, sk) &&
377 (me.updateCount != updateCount)) {
378 me.updateCount = updateCount;
379 numKeysUpdated++;
380 }
381 }
382 me.clearedCount = updateCount;
383 } else { // Key is not in selected set yet
384 if (me.clearedCount != updateCount) {
385 sk.channel.translateAndSetReadyOps(rOps, sk);
386 if ((sk.nioReadyOps() & sk.nioInterestOps()) != 0) {
387 selectedKeys.add(sk);
388 me.updateCount = updateCount;
389 numKeysUpdated++;
390 }
391 } else { // The readyOps have been set; now add
392 sk.channel.translateAndUpdateReadyOps(rOps, sk);
393 if ((sk.nioReadyOps() & sk.nioInterestOps()) != 0) {
394 selectedKeys.add(sk);
395 me.updateCount = updateCount;
396 numKeysUpdated++;
397 }
398 }
399 me.clearedCount = updateCount;
400 }
401 }
402 return numKeysUpdated;
403 }
404 }
405
406 // Represents a helper thread used for select.
407 private final class SelectThread extends Thread {
408 private final int index; // index of this thread
409 final SubSelector subSelector;
410 private long lastRun = 0; // last run number
411 private volatile boolean zombie;
412 // Creates a new thread
413 private SelectThread(int i) {
414 super(null, null, "SelectorHelper", 0, false);
415 this.index = i;
416 this.subSelector = new SubSelector(i);
417 //make sure we wait for next round of poll
418 this.lastRun = startLock.runsCounter;
419 }
420 void makeZombie() {
421 zombie = true;
422 }
423 boolean isZombie() {
424 return zombie;
425 }
426 public void run() {
427 while (true) { // poll loop
428 // wait for the start of poll. If this thread has become
429 // redundant, then exit.
430 if (startLock.waitForStart(this))
431 return;
432 // call poll()
433 try {
434 subSelector.poll(index);
435 } catch (IOException e) {
436 // Save this exception and let other threads finish.
437 finishLock.setException(e);
438 }
439 // notify main thread, that this thread has finished, and
440 // wakeup others, if this thread is the first to finish.
441 finishLock.threadFinished();
442 }
443 }
444 }
445
446 // After some channels registered/deregistered, the number of required
447 // helper threads may have changed. Adjust this number.
448 private void adjustThreadsCount() {
449 if (threadsCount > threads.size()) {
450 // More threads needed. Start more threads.
451 for (int i = threads.size(); i < threadsCount; i++) {
452 SelectThread newThread = new SelectThread(i);
453 threads.add(newThread);
454 newThread.setDaemon(true);
455 newThread.start();
456 }
457 } else if (threadsCount < threads.size()) {
458 // Some threads become redundant. Remove them from the threads List.
459 for (int i = threads.size() - 1 ; i >= threadsCount; i--)
460 threads.remove(i).makeZombie();
461 }
462 }
463
464 // Sets Windows wakeup socket to a signaled state.
465 private void setWakeupSocket() {
466 setWakeupSocket0(wakeupSinkFd);
467 }
468 private native void setWakeupSocket0(int wakeupSinkFd);
469
470 // Sets Windows wakeup socket to a non-signaled state.
471 private void resetWakeupSocket() {
472 synchronized (interruptLock) {
473 if (interruptTriggered == false)
474 return;
475 resetWakeupSocket0(wakeupSourceFd);
476 interruptTriggered = false;
477 }
478 }
479
480 private native void resetWakeupSocket0(int wakeupSourceFd);
481
482 private native boolean discardUrgentData(int fd);
483
484 // We increment this counter on each call to updateSelectedKeys()
485 // each entry in SubSelector.fdsMap has a memorized value of
486 // updateCount. When we increment numKeysUpdated we set updateCount
487 // for the corresponding entry to its current value. This is used to
488 // avoid counting the same key more than once - the same key can
489 // appear in readfds and writefds.
490 private long updateCount = 0;
491
492 // Update ops of the corresponding Channels. Add the ready keys to the
493 // ready queue.
494 private int updateSelectedKeys() {
495 updateCount++;
496 int numKeysUpdated = 0;
497 numKeysUpdated += subSelector.processSelectedKeys(updateCount);
498 for (SelectThread t: threads) {
499 numKeysUpdated += t.subSelector.processSelectedKeys(updateCount);
500 }
501 return numKeysUpdated;
502 }
503
504 @Override
505 protected void implClose() throws IOException {
506 synchronized (closeLock) {
507 if (channelArray != null) {
508 if (pollWrapper != null) {
509 // prevent further wakeup
510 synchronized (interruptLock) {
511 interruptTriggered = true;
512 }
513 wakeupPipe.sink().close();
514 wakeupPipe.source().close();
515 for(int i = 1; i < totalChannels; i++) { // Deregister channels
516 if (i % MAX_SELECTABLE_FDS != 0) { // skip wakeupEvent
517 deregister(channelArray[i]);
518 SelectableChannel selch = channelArray[i].channel();
519 if (!selch.isOpen() && !selch.isRegistered())
520 ((SelChImpl)selch).kill();
521 }
522 }
523 pollWrapper.free();
524 pollWrapper = null;
525 channelArray = null;
526 // Make all remaining helper threads exit
527 for (SelectThread t: threads)
528 t.makeZombie();
529 startLock.startThreads();
530 }
531 }
532 }
533 }
534
535 protected void implRegister(SelectionKeyImpl ski) {
536 synchronized (closeLock) {
537 if (pollWrapper == null)
538 throw new ClosedSelectorException();
539 growIfNeeded();
540 channelArray[totalChannels] = ski;
541 ski.setIndex(totalChannels);
542 fdMap.put(ski);
543 keys.add(ski);
544 pollWrapper.addEntry(totalChannels, ski);
545 totalChannels++;
546 }
547 }
548
549 private void growIfNeeded() {
550 if (channelArray.length == totalChannels) {
551 int newSize = totalChannels * 2; // Make a larger array
552 SelectionKeyImpl temp[] = new SelectionKeyImpl[newSize];
553 System.arraycopy(channelArray, 1, temp, 1, totalChannels - 1);
554 channelArray = temp;
555 pollWrapper.grow(newSize);
556 }
557 if (totalChannels % MAX_SELECTABLE_FDS == 0) { // more threads needed
558 pollWrapper.addWakeupSocket(wakeupSourceFd, totalChannels);
559 totalChannels++;
560 threadsCount++;
561 }
562 }
563
564 protected void implDereg(SelectionKeyImpl ski) throws IOException{
565 int i = ski.getIndex();
566 assert (i >= 0);
567 synchronized (closeLock) {
568 if (i != totalChannels - 1) {
569 // Copy end one over it
570 SelectionKeyImpl endChannel = channelArray[totalChannels-1];
571 channelArray[i] = endChannel;
572 endChannel.setIndex(i);
573 pollWrapper.replaceEntry(pollWrapper, totalChannels - 1,
574 pollWrapper, i);
575 }
576 ski.setIndex(-1);
577 }
578 channelArray[totalChannels - 1] = null;
579 totalChannels--;
580 if ( totalChannels != 1 && totalChannels % MAX_SELECTABLE_FDS == 1) {
581 totalChannels--;
582 threadsCount--; // The last thread has become redundant.
583 }
584 fdMap.remove(ski); // Remove the key from fdMap, keys and selectedKeys
585 keys.remove(ski);
586 selectedKeys.remove(ski);
587 deregister(ski);
588 SelectableChannel selch = ski.channel();
589 if (!selch.isOpen() && !selch.isRegistered())
590 ((SelChImpl)selch).kill();
591 }
592
593 public void putEventOps(SelectionKeyImpl sk, int ops) {
594 synchronized (closeLock) {
595 if (pollWrapper == null)
596 throw new ClosedSelectorException();
597 // make sure this sk has not been removed yet
598 int index = sk.getIndex();
599 if (index == -1)
600 throw new CancelledKeyException();
601 pollWrapper.putEventOps(index, ops);
602 }
603 }
604
605 public Selector wakeup() {
606 synchronized (interruptLock) {
607 if (!interruptTriggered) {
608 setWakeupSocket();
609 interruptTriggered = true;
610 }
611 }
612 return this;
613 }
614
615 static {
616 IOUtil.load();
617 }
618 }