/* * Copyright (c) 2005, 2018, Oracle and/or its affiliates. All rights reserved. * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER. * * This code is free software; you can redistribute it and/or modify it * under the terms of the GNU General Public License version 2 only, as * published by the Free Software Foundation. Oracle designates this * particular file as subject to the "Classpath" exception as provided * by Oracle in the LICENSE file that accompanied this code. * * This code is distributed in the hope that it will be useful, but WITHOUT * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License * version 2 for more details (a copy is included in the LICENSE file that * accompanied this code). * * You should have received a copy of the GNU General Public License version * 2 along with this work; if not, write to the Free Software Foundation, * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA. * * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA * or visit www.oracle.com if you need additional information or have any * questions. */ package sun.nio.ch; import java.io.IOException; import java.nio.channels.ClosedSelectorException; import java.nio.channels.SelectionKey; import java.nio.channels.Selector; import java.nio.channels.spi.SelectorProvider; import java.util.ArrayDeque; import java.util.Deque; import java.util.HashMap; import java.util.Map; import java.util.concurrent.TimeUnit; import java.util.function.Consumer; import static sun.nio.ch.EPoll.EPOLLIN; import static sun.nio.ch.EPoll.EPOLL_CTL_ADD; import static sun.nio.ch.EPoll.EPOLL_CTL_DEL; import static sun.nio.ch.EPoll.EPOLL_CTL_MOD; /** * Linux epoll based Selector implementation */ class EPollSelectorImpl extends SelectorImpl { // maximum number of events to poll in one call to epoll_wait private static final int NUM_EPOLLEVENTS = Math.min(IOUtil.fdLimit(), 1024); // epoll file descriptor private final int epfd; // address of poll array when polling with epoll_wait private final long pollArrayAddress; // file descriptors used for interrupt private final int fd0; private final int fd1; // maps file descriptor to selection key, synchronize on selector private final Map fdToKey = new HashMap<>(); // pending new registrations/updates, queued by setEventOps private final Object updateLock = new Object(); private final Deque updateKeys = new ArrayDeque<>(); // interrupt triggering and clearing private final Object interruptLock = new Object(); private boolean interruptTriggered; EPollSelectorImpl(SelectorProvider sp) throws IOException { super(sp); this.epfd = EPoll.create(); this.pollArrayAddress = EPoll.allocatePollArray(NUM_EPOLLEVENTS); try { long fds = IOUtil.makePipe(false); this.fd0 = (int) (fds >>> 32); this.fd1 = (int) fds; } catch (IOException ioe) { EPoll.freePollArray(pollArrayAddress); FileDispatcherImpl.closeIntFD(epfd); throw ioe; } // register one end of the socket pair for wakeups EPoll.ctl(epfd, EPOLL_CTL_ADD, fd0, EPOLLIN); } private void ensureOpen() { if (!isOpen()) throw new ClosedSelectorException(); } @Override protected int doSelect(Consumer action, long timeout) throws IOException { assert Thread.holdsLock(this); // epoll_wait timeout is int int to = (int) Math.min(timeout, Integer.MAX_VALUE); boolean blocking = (to != 0); boolean timedPoll = (to > 0); int numEntries; processUpdateQueue(); processDeregisterQueue(); try { begin(blocking); do { long startTime = timedPoll ? System.nanoTime() : 0; numEntries = EPoll.wait(epfd, pollArrayAddress, NUM_EPOLLEVENTS, to); if (numEntries == IOStatus.INTERRUPTED && timedPoll) { // timed poll interrupted so need to adjust timeout long adjust = System.nanoTime() - startTime; to -= TimeUnit.MILLISECONDS.convert(adjust, TimeUnit.NANOSECONDS); if (to <= 0) { // timeout expired so no retry numEntries = 0; } } } while (numEntries == IOStatus.INTERRUPTED); assert IOStatus.check(numEntries); } finally { end(blocking); } processDeregisterQueue(); return processEvents(numEntries, action); } /** * Process changes to the interest ops. */ private void processUpdateQueue() { assert Thread.holdsLock(this); synchronized (updateLock) { SelectionKeyImpl ski; while ((ski = updateKeys.pollFirst()) != null) { if (ski.isValid()) { int fd = ski.getFDVal(); // add to fdToKey if needed SelectionKeyImpl previous = fdToKey.putIfAbsent(fd, ski); assert (previous == null) || (previous == ski); int newEvents = ski.translateInterestOps(); int registeredEvents = ski.registeredEvents(); if (newEvents != registeredEvents) { if (newEvents == 0) { // remove from epoll EPoll.ctl(epfd, EPOLL_CTL_DEL, fd, 0); } else { if (registeredEvents == 0) { // add to epoll EPoll.ctl(epfd, EPOLL_CTL_ADD, fd, newEvents); } else { // modify events EPoll.ctl(epfd, EPOLL_CTL_MOD, fd, newEvents); } } ski.registeredEvents(newEvents); } } } } } /** * Process the polled events. * If the interrupt fd has been selected, drain it and clear the interrupt. */ private int processEvents(int numEntries, Consumer action) throws IOException { assert Thread.holdsLock(this); boolean interrupted = false; int numKeysUpdated = 0; for (int i=0; i