java/nio/SelectorImpl.java

/* Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package java.nio;

import java.io.FileDescriptor;
import java.io.IOException;
import java.nio.channels.ClosedSelectorException;
import java.nio.channels.IllegalSelectorException;
import java.nio.channels.SelectableChannel;
import java.nio.channels.SelectionKey;
import static java.nio.channels.SelectionKey.*;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.nio.channels.spi.AbstractSelectableChannel;
import java.nio.channels.spi.AbstractSelectionKey;
import java.nio.channels.spi.AbstractSelector;
import java.nio.channels.spi.SelectorProvider;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.HashSet;
import java.util.Iterator;
import java.util.Set;
import java.util.UnsafeArrayList;
import libcore.io.ErrnoException;
import libcore.io.IoBridge;
import libcore.io.IoUtils;
import libcore.io.Libcore;
import libcore.io.StructPollfd;
import libcore.util.EmptyArray;
import static libcore.io.OsConstants.*;

/*
 * Default implementation of java.nio.channels.Selector
 */
final class SelectorImpl extends AbstractSelector {

    /**
     * Used to synchronize when a key's interest ops change.
     */
    final Object keysLock = new Object();

    private final Set<SelectionKeyImpl> mutableKeys = new HashSet<SelectionKeyImpl>();

    /**
     * The unmodifiable set of keys as exposed to the user. This object is used
     * for synchronization.
     */
    private final Set<SelectionKey> unmodifiableKeys = Collections
            .<SelectionKey>unmodifiableSet(mutableKeys);

    private final Set<SelectionKey> mutableSelectedKeys = new HashSet<SelectionKey>();

    /**
     * The unmodifiable set of selectable keys as seen by the user. This object
     * is used for synchronization.
     */
    private final Set<SelectionKey> selectedKeys
            = new UnaddableSet<SelectionKey>(mutableSelectedKeys);

    /**
     * The wakeup pipe. To trigger a wakeup, write a byte to wakeupOut. Each
     * time select returns, wakeupIn is drained.
     */
    private final FileDescriptor wakeupIn;
    private final FileDescriptor wakeupOut;

    private final UnsafeArrayList<StructPollfd> pollFds = new UnsafeArrayList<StructPollfd>(StructPollfd.class, 8);

    public SelectorImpl(SelectorProvider selectorProvider) throws IOException {
        super(selectorProvider);

        /*
         * Create a pipes to trigger wakeup. We can't use a NIO pipe because it
         * would be closed if the selecting thread is interrupted. Also
         * configure the pipe so we can fully drain it without blocking.
         */
        try {
            FileDescriptor[] pipeFds = Libcore.os.pipe();
            wakeupIn = pipeFds[0];
            wakeupOut = pipeFds[1];
            IoUtils.setBlocking(wakeupIn, false);
            pollFds.add(new StructPollfd());
            setPollFd(0, wakeupIn, POLLIN, null);
        } catch (ErrnoException errnoException) {
            throw errnoException.rethrowAsIOException();
        }
    }

    @Override protected void implCloseSelector() throws IOException {
        wakeup();
        synchronized (this) {
            synchronized (unmodifiableKeys) {
                synchronized (selectedKeys) {
                    IoUtils.close(wakeupIn);
                    IoUtils.close(wakeupOut);
                    doCancel();
                    for (SelectionKey sk : mutableKeys) {
                        deregister((AbstractSelectionKey) sk);
                    }
                }
            }
        }
    }

    @Override protected SelectionKey register(AbstractSelectableChannel channel,
            int operations, Object attachment) {
        if (!provider().equals(channel.provider())) {
            throw new IllegalSelectorException();
        }
        synchronized (this) {
            synchronized (unmodifiableKeys) {
                SelectionKeyImpl selectionKey = new SelectionKeyImpl(channel, operations,
                        attachment, this);
                mutableKeys.add(selectionKey);
                ensurePollFdsCapacity();
                return selectionKey;
            }
        }
    }

    @Override public synchronized Set<SelectionKey> keys() {
        checkClosed();
        return unmodifiableKeys;
    }

    private void checkClosed() {
        if (!isOpen()) {
            throw new ClosedSelectorException();
        }
    }

    @Override public int select() throws IOException {
        // Blocks until some fd is ready.
        return selectInternal(-1);
    }

    @Override public int select(long timeout) throws IOException {
        if (timeout < 0) {
            throw new IllegalArgumentException();
        }
        // Our timeout is interpreted differently to Unix's --- 0 means block. See selectNow.
        return selectInternal((timeout == 0) ? -1 : timeout);
    }

    @Override public int selectNow() throws IOException {
        return selectInternal(0);
    }

    private int selectInternal(long timeout) throws IOException {
        checkClosed();
        synchronized (this) {
            synchronized (unmodifiableKeys) {
                synchronized (selectedKeys) {
                    doCancel();
                    boolean isBlock = (timeout != 0);
                    synchronized (keysLock) {
                        preparePollFds();
                    }
                    int rc = -1;
                    try {
                        if (isBlock) {
                            begin();
                        }
                        try {
                            rc = Libcore.os.poll(pollFds.array(), (int) timeout);
                        } catch (ErrnoException errnoException) {
                            if (errnoException.errno != EINTR) {
                                throw errnoException.rethrowAsIOException();
                            }
                        }
                    } finally {
                        if (isBlock) {
                            end();
                        }
                    }

                    int readyCount = (rc > 0) ? processPollFds() : 0;
                    readyCount -= doCancel();
                    return readyCount;
                }
            }
        }
    }

    private void setPollFd(int i, FileDescriptor fd, int events, Object object) {
        StructPollfd pollFd = pollFds.get(i);
        pollFd.fd = fd;
        pollFd.events = (short) events;
        pollFd.userData = object;
    }

    private void preparePollFds() {
        int i = 1; // Our wakeup pipe comes before all the user's fds.
        for (SelectionKeyImpl key : mutableKeys) {
            int interestOps = key.interestOpsNoCheck();
            short eventMask = 0;
            if (((OP_ACCEPT | OP_READ) & interestOps) != 0) {
                eventMask |= POLLIN;
            }
            if (((OP_CONNECT | OP_WRITE) & interestOps) != 0) {
                eventMask |= POLLOUT;
            }
            if (eventMask != 0) {
                setPollFd(i++, ((FileDescriptorChannel) key.channel()).getFD(), eventMask, key);
            }
        }
    }

    private void ensurePollFdsCapacity() {
        // We need one slot for each element of mutableKeys, plus one for the wakeup pipe.
        while (pollFds.size() < mutableKeys.size() + 1) {
            pollFds.add(new StructPollfd());
        }
    }

    /**
     * Updates the key ready ops and selected key set.
     */
    private int processPollFds() throws IOException {
        if (pollFds.get(0).revents == POLLIN) {
            // Read bytes from the wakeup pipe until the pipe is empty.
            byte[] buffer = new byte[8];
            while (IoBridge.read(wakeupIn, buffer, 0, 1) > 0) {
            }
        }

        int readyKeyCount = 0;
        for (int i = 1; i < pollFds.size(); ++i) {
            StructPollfd pollFd = pollFds.get(i);
            if (pollFd.revents == 0) {
                continue;
            }
            if (pollFd.fd == null) {
                break;
            }

            SelectionKeyImpl key = (SelectionKeyImpl) pollFd.userData;

            pollFd.fd = null;
            pollFd.userData = null;

            int ops = key.interestOpsNoCheck();
            int selectedOp = 0;
            if ((pollFd.revents & POLLIN) != 0) {
                selectedOp = ops & (OP_ACCEPT | OP_READ);
            } else if ((pollFd.revents & POLLOUT) != 0) {
                if (key.isConnected()) {
                    selectedOp = ops & OP_WRITE;
                } else {
                    selectedOp = ops & OP_CONNECT;
                }
            }

            if (selectedOp != 0) {
                boolean wasSelected = mutableSelectedKeys.contains(key);
                if (wasSelected && key.readyOps() != selectedOp) {
                    key.setReadyOps(key.readyOps() | selectedOp);
                    ++readyKeyCount;
                } else if (!wasSelected) {
                    key.setReadyOps(selectedOp);
                    mutableSelectedKeys.add(key);
                    ++readyKeyCount;
                }
            }
        }

        return readyKeyCount;
    }

    @Override public synchronized Set<SelectionKey> selectedKeys() {
        checkClosed();
        return selectedKeys;
    }

    /**
     * Removes cancelled keys from the key set and selected key set, and
     * deregisters the corresponding channels. Returns the number of keys
     * removed from the selected key set.
     */
    private int doCancel() {
        int deselected = 0;

        Set<SelectionKey> cancelledKeys = cancelledKeys();
        synchronized (cancelledKeys) {
            if (cancelledKeys.size() > 0) {
                for (SelectionKey currentKey : cancelledKeys) {
                    mutableKeys.remove(currentKey);
                    deregister((AbstractSelectionKey) currentKey);
                    if (mutableSelectedKeys.remove(currentKey)) {
                        deselected++;
                    }
                }
                cancelledKeys.clear();
            }
        }

        return deselected;
    }

    @Override public Selector wakeup() {
        try {
            Libcore.os.write(wakeupOut, new byte[] { 1 }, 0, 1);
        } catch (ErrnoException ignored) {
        }
        return this;
    }

    private static class UnaddableSet<E> implements Set<E> {

        private final Set<E> set;

        UnaddableSet(Set<E> set) {
            this.set = set;
        }

        @Override
        public boolean equals(Object object) {
            return set.equals(object);
        }

        @Override
        public int hashCode() {
            return set.hashCode();
        }

        public boolean add(E object) {
            throw new UnsupportedOperationException();
        }

        public boolean addAll(Collection<? extends E> c) {
            throw new UnsupportedOperationException();
        }

        public void clear() {
            set.clear();
        }

        public boolean contains(Object object) {
            return set.contains(object);
        }

        public boolean containsAll(Collection<?> c) {
            return set.containsAll(c);
        }

        public boolean isEmpty() {
            return set.isEmpty();
        }

        public Iterator<E> iterator() {
            return set.iterator();
        }

        public boolean remove(Object object) {
            return set.remove(object);
        }

        public boolean removeAll(Collection<?> c) {
            return set.removeAll(c);
        }

        public boolean retainAll(Collection<?> c) {
            return set.retainAll(c);
        }

        public int size() {
            return set.size();
        }

        public Object[] toArray() {
            return set.toArray();
        }

        public <T> T[] toArray(T[] a) {
            return set.toArray(a);
        }
    }
}