1 /*
  2  * Copyright (c) 2011, 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 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.KQueue.EVFILT_READ;
 41 import static sun.nio.ch.KQueue.EVFILT_WRITE;
 42 import static sun.nio.ch.KQueue.EV_ADD;
 43 import static sun.nio.ch.KQueue.EV_DELETE;
 44 
 45 /**
 46  * KQueue based Selector implementation for macOS
 47  */
 48 
 49 class KQueueSelectorImpl extends SelectorImpl {
 50 
 51     // maximum number of events to poll in one call to kqueue
 52     private static final int MAX_KEVENTS = 256;
 53 
 54     // kqueue file descriptor
 55     private final int kqfd;
 56 
 57     // address of poll array (event list) when polling for pending events
 58     private final long pollArrayAddress;
 59 
 60     // file descriptors used for interrupt
 61     private final int fd0;
 62     private final int fd1;
 63 
 64     // maps file descriptor to selection key, synchronize on selector
 65     private final Map<Integer, SelectionKeyImpl> fdToKey = new HashMap<>();
 66 
 67     // pending new registrations/updates, queued by setEventOps
 68     private final Object updateLock = new Object();
 69     private final Deque<SelectionKeyImpl> updateKeys = new ArrayDeque<>();
 70 
 71     // interrupt triggering and clearing
 72     private final Object interruptLock = new Object();
 73     private boolean interruptTriggered;
 74 
 75     // used by updateSelectedKeys to handle cases where the same file
 76     // descriptor is polled by more than one filter
 77     private int pollCount;
 78 
 79     KQueueSelectorImpl(SelectorProvider sp) throws IOException {
 80         super(sp);
 81 
 82         this.kqfd = KQueue.create();
 83         this.pollArrayAddress = KQueue.allocatePollArray(MAX_KEVENTS);
 84 
 85         try {
 86             long fds = IOUtil.makePipe(false);
 87             this.fd0 = (int) (fds >>> 32);
 88             this.fd1 = (int) fds;
 89         } catch (IOException ioe) {
 90             KQueue.freePollArray(pollArrayAddress);
 91             FileDispatcherImpl.closeIntFD(kqfd);
 92             throw ioe;
 93         }
 94 
 95         // register one end of the socket pair for wakeups
 96         KQueue.register(kqfd, fd0, EVFILT_READ, EV_ADD);
 97     }
 98 
 99     private void ensureOpen() {
100         if (!isOpen())
101             throw new ClosedSelectorException();
102     }
103 
104     @Override
105     protected int doSelect(Consumer<SelectionKey> action, long timeout)
106         throws IOException
107     {
108         assert Thread.holdsLock(this);
109 
110         long to = Math.min(timeout, Integer.MAX_VALUE);  // max kqueue timeout
111         boolean blocking = (to != 0);
112         boolean timedPoll = (to > 0);
113 
114         int numEntries;
115         processUpdateQueue();
116         processDeregisterQueue();
117         try {
118             begin(blocking);
119 
120             do {
121                 long startTime = timedPoll ? System.nanoTime() : 0;
122                 numEntries = poll(to);
123                 if (numEntries == IOStatus.INTERRUPTED && timedPoll) {
124                     // timed poll interrupted so need to adjust timeout
125                     long adjust = System.nanoTime() - startTime;
126                     to -= TimeUnit.MILLISECONDS.convert(adjust, TimeUnit.NANOSECONDS);
127                     if (to <= 0) {
128                         // timeout expired so no retry
129                         numEntries = 0;
130                     }
131                 }
132             } while (numEntries == IOStatus.INTERRUPTED);
133             assert IOStatus.check(numEntries);
134 
135         } finally {
136             end(blocking);
137         }
138         processDeregisterQueue();
139         return processEvents(numEntries, action);
140     }
141 
142     @Override
143     protected int implPoll(long timeout) throws IOException {
144         return KQueue.poll(kqfd, pollArrayAddress, MAX_KEVENTS, timeout);
145     }
146 
147     /**
148      * Process changes to the interest ops.
149      */
150     private void processUpdateQueue() {
151         assert Thread.holdsLock(this);
152 
153         synchronized (updateLock) {
154             SelectionKeyImpl ski;
155             while ((ski = updateKeys.pollFirst()) != null) {
156                 if (ski.isValid()) {
157                     int fd = ski.getFDVal();
158                     // add to fdToKey if needed
159                     SelectionKeyImpl previous = fdToKey.putIfAbsent(fd, ski);
160                     assert (previous == null) || (previous == ski);
161 
162                     int newEvents = ski.translateInterestOps();
163                     int registeredEvents = ski.registeredEvents();
164 
165                     // DatagramChannelImpl::disconnect has reset socket
166                     if (ski.getAndClearReset() && registeredEvents != 0) {
167                         KQueue.register(kqfd, fd, EVFILT_READ, EV_DELETE);
168                         registeredEvents = 0;
169                     }
170 
171                     if (newEvents != registeredEvents) {
172 
173                         // add or delete interest in read events
174                         if ((registeredEvents & Net.POLLIN) != 0) {
175                             if ((newEvents & Net.POLLIN) == 0) {
176                                 KQueue.register(kqfd, fd, EVFILT_READ, EV_DELETE);
177                             }
178                         } else if ((newEvents & Net.POLLIN) != 0) {
179                             KQueue.register(kqfd, fd, EVFILT_READ, EV_ADD);
180                         }
181 
182                         // add or delete interest in write events
183                         if ((registeredEvents & Net.POLLOUT) != 0) {
184                             if ((newEvents & Net.POLLOUT) == 0) {
185                                 KQueue.register(kqfd, fd, EVFILT_WRITE, EV_DELETE);
186                             }
187                         } else if ((newEvents & Net.POLLOUT) != 0) {
188                             KQueue.register(kqfd, fd, EVFILT_WRITE, EV_ADD);
189                         }
190 
191                         ski.registeredEvents(newEvents);
192                     }
193                 }
194             }
195         }
196     }
197 
198     /**
199      * Process the polled events.
200      * If the interrupt fd has been selected, drain it and clear the interrupt.
201      */
202     private int processEvents(int numEntries, Consumer<SelectionKey> action)
203         throws IOException
204     {
205         assert Thread.holdsLock(this);
206 
207         int numKeysUpdated = 0;
208         boolean interrupted = false;
209 
210         // A file descriptor may be registered with kqueue with more than one
211         // filter and so there may be more than one event for a fd. The poll
212         // count is incremented here and compared against the SelectionKey's
213         // "lastPolled" field. This ensures that the ready ops is updated rather
214         // than replaced when a file descriptor is polled by both the read and
215         // write filter.
216         pollCount++;
217 
218         for (int i = 0; i < numEntries; i++) {
219             long kevent = KQueue.getEvent(pollArrayAddress, i);
220             int fd = KQueue.getDescriptor(kevent);
221             if (fd == fd0) {
222                 interrupted = true;
223             } else {
224                 SelectionKeyImpl ski = fdToKey.get(fd);
225                 if (ski != null) {
226                     int rOps = 0;
227                     short filter = KQueue.getFilter(kevent);
228                     if (filter == EVFILT_READ) {
229                         rOps |= Net.POLLIN;
230                     } else if (filter == EVFILT_WRITE) {
231                         rOps |= Net.POLLOUT;
232                     }
233                     int updated = processReadyEvents(rOps, ski, action);
234                     if (updated > 0 && ski.lastPolled != pollCount) {
235                         numKeysUpdated++;
236                         ski.lastPolled = pollCount;
237                     }
238                 }
239             }
240         }
241 
242         if (interrupted) {
243             clearInterrupt();
244         }
245         return numKeysUpdated;
246     }
247 
248     @Override
249     protected void implClose() throws IOException {
250         assert !isOpen();
251         assert Thread.holdsLock(this);
252 
253         // prevent further wakeup
254         synchronized (interruptLock) {
255             interruptTriggered = true;
256         }
257 
258         FileDispatcherImpl.closeIntFD(kqfd);
259         KQueue.freePollArray(pollArrayAddress);
260 
261         FileDispatcherImpl.closeIntFD(fd0);
262         FileDispatcherImpl.closeIntFD(fd1);
263     }
264 
265     @Override
266     protected void implDereg(SelectionKeyImpl ski) throws IOException {
267         assert !ski.isValid();
268         assert Thread.holdsLock(this);
269 
270         int fd = ski.getFDVal();
271         int registeredEvents = ski.registeredEvents();
272         if (fdToKey.remove(fd) != null) {
273             if (registeredEvents != 0) {
274                 if ((registeredEvents & Net.POLLIN) != 0)
275                     KQueue.register(kqfd, fd, EVFILT_READ, EV_DELETE);
276                 if ((registeredEvents & Net.POLLOUT) != 0)
277                     KQueue.register(kqfd, fd, EVFILT_WRITE, EV_DELETE);
278                 ski.registeredEvents(0);
279             }
280         } else {
281             assert registeredEvents == 0;
282         }
283     }
284 
285     @Override
286     public void setEventOps(SelectionKeyImpl ski) {
287         ensureOpen();
288         synchronized (updateLock) {
289             updateKeys.addLast(ski);
290         }
291     }
292 
293     @Override
294     public Selector wakeup() {
295         synchronized (interruptLock) {
296             if (!interruptTriggered) {
297                 try {
298                     IOUtil.write1(fd1, (byte)0);
299                 } catch (IOException ioe) {
300                     throw new InternalError(ioe);
301                 }
302                 interruptTriggered = true;
303             }
304         }
305         return this;
306     }
307 
308     private void clearInterrupt() throws IOException {
309         synchronized (interruptLock) {
310             IOUtil.drain(fd0);
311             interruptTriggered = false;
312         }
313     }
314 }