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.SelectionKey;
 30 import java.nio.channels.Selector;
 31 import java.nio.channels.spi.SelectorProvider;
 32 import java.util.ArrayDeque;
 33 import java.util.Deque;
 34 import java.util.HashMap;
 35 import java.util.Map;
 36 import java.util.concurrent.TimeUnit;
 37 import java.util.function.Consumer;

 38 
 39 import static sun.nio.ch.KQueue.EVFILT_READ;
 40 import static sun.nio.ch.KQueue.EVFILT_WRITE;
 41 import static sun.nio.ch.KQueue.EV_ADD;
 42 import static sun.nio.ch.KQueue.EV_DELETE;
 43 
 44 /**
 45  * KQueue based Selector implementation for macOS
 46  */
 47 
 48 class KQueueSelectorImpl extends SelectorImpl {
 49 
 50     // maximum number of events to poll in one call to kqueue
 51     private static final int MAX_KEVENTS = 256;
 52 
 53     // kqueue file descriptor
 54     private final int kqfd;
 55 
 56     // address of poll array (event list) when polling for pending events
 57     private final long pollArrayAddress;
 58 
 59     // file descriptors used for interrupt
 60     private final int fd0;
 61     private final int fd1;
 62 
 63     // maps file descriptor to selection key, synchronize on selector
 64     private final Map<Integer, SelectionKeyImpl> fdToKey = new HashMap<>();
 65 
 66     // pending new registrations/updates, queued by setEventOps
 67     private final Object updateLock = new Object();
 68     private final Deque<SelectionKeyImpl> updateKeys = new ArrayDeque<>();
 69 
 70     // interrupt triggering and clearing
 71     private final Object interruptLock = new Object();
 72     private boolean interruptTriggered;
 73 
 74     // used by updateSelectedKeys to handle cases where the same file
 75     // descriptor is polled by more than one filter
 76     private int pollCount;
 77 
 78     KQueueSelectorImpl(SelectorProvider sp) throws IOException {
 79         super(sp);
 80 
 81         this.kqfd = KQueue.create();
 82         this.pollArrayAddress = KQueue.allocatePollArray(MAX_KEVENTS);
 83 
 84         try {
 85             long fds = IOUtil.makePipe(false);
 86             this.fd0 = (int) (fds >>> 32);
 87             this.fd1 = (int) fds;
 88         } catch (IOException ioe) {
 89             KQueue.freePollArray(pollArrayAddress);
 90             FileDispatcherImpl.closeIntFD(kqfd);
 91             throw ioe;
 92         }
 93 
 94         // register one end of the socket pair for wakeups
 95         KQueue.register(kqfd, fd0, EVFILT_READ, EV_ADD);
 96     }
 97 
 98     @Override
 99     protected int doSelect(Consumer<SelectionKey> action, long timeout)
100         throws IOException
101     {
102         assert Thread.holdsLock(this);
103 
104         long to = Math.min(timeout, Integer.MAX_VALUE);  // max kqueue timeout
105         boolean blocking = (to != 0);
106         boolean timedPoll = (to > 0);
107 
108         int numEntries;
109         processUpdateQueue();
110         processDeregisterQueue();


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


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


135         }
136         assert IOStatus.check(numEntries);
137 
138         processDeregisterQueue();
139         return processEvents(numEntries, action);
140     }
141 
142     /**
143      * If blocking, parks the current virtual thread until a file descriptor is polled
144      * or the thread is interrupted.
145      */
146     private int untimedPoll(boolean block) throws IOException {
147         int numEntries = KQueue.poll(kqfd, pollArrayAddress, MAX_KEVENTS, 0);
148         if (block) {
149             while (numEntries == 0 && !Thread.currentThread().isInterrupted()) {
150                 Poller.pollSelector(kqfd, 0);
151                 numEntries = KQueue.poll(kqfd, pollArrayAddress, MAX_KEVENTS, 0);
152             }
153         }
154         return numEntries;
155     }
156 
157     /**
158      * Parks the current virtual thread until a file descriptor is polled, or the thread
159      * is interrupted, for up to the specified waiting time.
160      */
161     private int timedPoll(long nanos) throws IOException {
162         long startNanos = System.nanoTime();
163         int numEntries = KQueue.poll(kqfd, pollArrayAddress, MAX_KEVENTS, 0);
164         while (numEntries == 0 && !Thread.currentThread().isInterrupted()) {
165             long remainingNanos = nanos - (System.nanoTime() - startNanos);
166             if (remainingNanos <= 0) {
167                 // timeout
168                 break;
169             }
170             Poller.pollSelector(kqfd, remainingNanos);
171             numEntries = KQueue.poll(kqfd, pollArrayAddress, MAX_KEVENTS, 0);
172         }
173         return numEntries;
174     }
175 
176     /**
177      * Process changes to the interest ops.
178      */
179     private void processUpdateQueue() {
180         assert Thread.holdsLock(this);
181 
182         synchronized (updateLock) {
183             SelectionKeyImpl ski;
184             while ((ski = updateKeys.pollFirst()) != null) {
185                 if (ski.isValid()) {
186                     int fd = ski.getFDVal();
187                     // add to fdToKey if needed
188                     SelectionKeyImpl previous = fdToKey.putIfAbsent(fd, ski);
189                     assert (previous == null) || (previous == ski);
190 
191                     int newEvents = ski.translateInterestOps();
192                     int registeredEvents = ski.registeredEvents();
193 
194                     // DatagramChannelImpl::disconnect has reset socket
195                     if (ski.getAndClearReset() && registeredEvents != 0) {
196                         KQueue.register(kqfd, fd, EVFILT_READ, EV_DELETE);
197                         registeredEvents = 0;
198                     }
199 
200                     if (newEvents != registeredEvents) {
201 
202                         // add or delete interest in read events
203                         if ((registeredEvents & Net.POLLIN) != 0) {
204                             if ((newEvents & Net.POLLIN) == 0) {
205                                 KQueue.register(kqfd, fd, EVFILT_READ, EV_DELETE);
206                             }
207                         } else if ((newEvents & Net.POLLIN) != 0) {
208                             KQueue.register(kqfd, fd, EVFILT_READ, EV_ADD);
209                         }
210 
211                         // add or delete interest in write events
212                         if ((registeredEvents & Net.POLLOUT) != 0) {
213                             if ((newEvents & Net.POLLOUT) == 0) {
214                                 KQueue.register(kqfd, fd, EVFILT_WRITE, EV_DELETE);
215                             }
216                         } else if ((newEvents & Net.POLLOUT) != 0) {
217                             KQueue.register(kqfd, fd, EVFILT_WRITE, EV_ADD);
218                         }
219 
220                         ski.registeredEvents(newEvents);
221                     }
222                 }
223             }
224         }
225     }
226 
227     /**
228      * Process the polled events.
229      * If the interrupt fd has been selected, drain it and clear the interrupt.
230      */
231     private int processEvents(int numEntries, Consumer<SelectionKey> action)
232         throws IOException
233     {
234         assert Thread.holdsLock(this);
235 
236         int numKeysUpdated = 0;
237         boolean interrupted = false;
238 
239         // A file descriptor may be registered with kqueue with more than one
240         // filter and so there may be more than one event for a fd. The poll
241         // count is incremented here and compared against the SelectionKey's
242         // "lastPolled" field. This ensures that the ready ops is updated rather
243         // than replaced when a file descriptor is polled by both the read and
244         // write filter.
245         pollCount++;
246 
247         for (int i = 0; i < numEntries; i++) {
248             long kevent = KQueue.getEvent(pollArrayAddress, i);
249             int fd = KQueue.getDescriptor(kevent);
250             if (fd == fd0) {
251                 interrupted = true;
252             } else {
253                 SelectionKeyImpl ski = fdToKey.get(fd);
254                 if (ski != null) {
255                     int rOps = 0;
256                     short filter = KQueue.getFilter(kevent);
257                     if (filter == EVFILT_READ) {
258                         rOps |= Net.POLLIN;
259                     } else if (filter == EVFILT_WRITE) {
260                         rOps |= Net.POLLOUT;
261                     }
262                     int updated = processReadyEvents(rOps, ski, action);
263                     if (updated > 0 && ski.lastPolled != pollCount) {
264                         numKeysUpdated++;
265                         ski.lastPolled = pollCount;
266                     }
267                 }
268             }
269         }
270 
271         if (interrupted) {
272             clearInterrupt();
273         }
274         return numKeysUpdated;
275     }
276 
277     @Override
278     protected void implClose() throws IOException {
279         assert !isOpen();
280         assert Thread.holdsLock(this);
281 
282         // prevent further wakeup
283         synchronized (interruptLock) {
284             interruptTriggered = true;
285         }
286 
287         FileDispatcherImpl.closeIntFD(kqfd);
288         KQueue.freePollArray(pollArrayAddress);
289 
290         FileDispatcherImpl.closeIntFD(fd0);
291         FileDispatcherImpl.closeIntFD(fd1);
292     }
293 
294     @Override
295     protected void implDereg(SelectionKeyImpl ski) throws IOException {
296         assert !ski.isValid();
297         assert Thread.holdsLock(this);
298 
299         int fd = ski.getFDVal();
300         int registeredEvents = ski.registeredEvents();
301         if (fdToKey.remove(fd) != null) {
302             if (registeredEvents != 0) {
303                 if ((registeredEvents & Net.POLLIN) != 0)
304                     KQueue.register(kqfd, fd, EVFILT_READ, EV_DELETE);
305                 if ((registeredEvents & Net.POLLOUT) != 0)
306                     KQueue.register(kqfd, fd, EVFILT_WRITE, EV_DELETE);
307                 ski.registeredEvents(0);
308             }
309         } else {
310             assert registeredEvents == 0;
311         }
312     }
313 
314     @Override
315     public void setEventOps(SelectionKeyImpl ski) {
316         synchronized (updateLock) {
317             updateKeys.addLast(ski);
318         }
319     }
320 
321     @Override
322     public Selector wakeup() {
323         synchronized (interruptLock) {
324             if (!interruptTriggered) {
325                 try {
326                     IOUtil.write1(fd1, (byte)0);
327                 } catch (IOException ioe) {
328                     throw new InternalError(ioe);
329                 }
330                 interruptTriggered = true;
331             }
332         }
333         return this;
334     }
335 
336     private void clearInterrupt() throws IOException {
337         synchronized (interruptLock) {
338             IOUtil.drain(fd0);
339             interruptTriggered = false;
340         }
341     }
342 }
--- EOF ---