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