//
//  ========================================================================
//  Copyright (c) 1995-2017 Mort Bay Consulting Pty. Ltd.
//  ------------------------------------------------------------------------
//  All rights reserved. This program and the accompanying materials
//  are made available under the terms of the Eclipse Public License v1.0
//  and Apache License v2.0 which accompanies this distribution.
//
//      The Eclipse Public License is available at
//      http://www.eclipse.org/legal/epl-v10.html
//
//      The Apache License v2.0 is available at
//      http://www.opensource.org/licenses/apache2.0.php
//
//  You may elect to redistribute this code under either of these licenses.
//  ========================================================================
//

package org.eclipse.jetty.io;

import java.io.Closeable;
import java.io.IOException;
import java.net.ConnectException;
import java.net.Socket;
import java.net.SocketAddress;
import java.net.SocketTimeoutException;
import java.nio.channels.CancelledKeyException;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.ArrayList;
import java.util.List;
import java.util.Queue;
import java.util.Set;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;

import org.eclipse.jetty.util.ConcurrentArrayQueue;
import org.eclipse.jetty.util.TypeUtil;
import org.eclipse.jetty.util.annotation.ManagedAttribute;
import org.eclipse.jetty.util.component.AbstractLifeCycle;
import org.eclipse.jetty.util.component.ContainerLifeCycle;
import org.eclipse.jetty.util.component.Dumpable;
import org.eclipse.jetty.util.log.Log;
import org.eclipse.jetty.util.log.Logger;
import org.eclipse.jetty.util.thread.NonBlockingThread;
import org.eclipse.jetty.util.thread.Scheduler;

/**
 * <p>{@link SelectorManager} manages a number of {@link ManagedSelector}s that
 * simplify the non-blocking primitives provided by the JVM via the {@code java.nio} package.</p>
 * <p>{@link SelectorManager} subclasses implement methods to return protocol-specific
 * {@link EndPoint}s and {@link Connection}s.</p>
 */
public abstract class SelectorManager extends AbstractLifeCycle implements Dumpable
{
    public static final String SUBMIT_KEY_UPDATES = "org.eclipse.jetty.io.SelectorManager.submitKeyUpdates";
    public static final int DEFAULT_CONNECT_TIMEOUT = 15000;
    protected static final Logger LOG = Log.getLogger(SelectorManager.class);
    private final static boolean __submitKeyUpdates = Boolean.valueOf(System.getProperty(SUBMIT_KEY_UPDATES, "true"));

    private final Executor executor;
    private final Scheduler scheduler;
    private final ManagedSelector[] _selectors;
    private long _connectTimeout = DEFAULT_CONNECT_TIMEOUT;
    private long _selectorIndex;
    private int _priorityDelta;

    protected SelectorManager(Executor executor, Scheduler scheduler)
    {
        this(executor, scheduler, (Runtime.getRuntime().availableProcessors() + 1) / 2);
    }

    protected SelectorManager(Executor executor, Scheduler scheduler, int selectors)
    {
        if (selectors<=0)
            throw new IllegalArgumentException("No selectors");
        this.executor = executor;
        this.scheduler = scheduler;
        _selectors = new ManagedSelector[selectors];
    }

    public Executor getExecutor()
    {
        return executor;
    }

    public Scheduler getScheduler()
    {
        return scheduler;
    }

    /**
     * Get the connect timeout
     *
     * @return the connect timeout (in milliseconds)
     */
    public long getConnectTimeout()
    {
        return _connectTimeout;
    }

    /**
     * Set the connect timeout (in milliseconds)
     *
     * @param milliseconds the number of milliseconds for the timeout
     */
    public void setConnectTimeout(long milliseconds)
    {
        _connectTimeout = milliseconds;
    }


    @ManagedAttribute("The priority delta to apply to selector threads")
    public int getSelectorPriorityDelta()
    {
        return _priorityDelta;
    }

    /**
     * Sets the selector thread priority delta to the given amount.
     * <p>This allows the selector threads to run at a different priority.
     * Typically this would be used to lower the priority to give preference
     * to handling previously accepted connections rather than accepting
     * new connections.</p>
     *
     * @param selectorPriorityDelta the amount to change the thread priority
     *                              delta to (may be negative)
     * @see Thread#getPriority()
     */
    public void setSelectorPriorityDelta(int selectorPriorityDelta)
    {
        int oldDelta = _priorityDelta;
        _priorityDelta = selectorPriorityDelta;
        if (oldDelta != selectorPriorityDelta && isStarted())
        {
            for (ManagedSelector selector : _selectors)
            {
                Thread thread = selector._thread;
                if (thread != null)
                {
                    int deltaDiff = selectorPriorityDelta - oldDelta;
                    thread.setPriority(Math.max(Thread.MIN_PRIORITY, Math.min(Thread.MAX_PRIORITY, thread.getPriority() - deltaDiff)));
                }
            }
        }
    }

    /**
     * Executes the given task in a different thread.
     *
     * @param task the task to execute
     */
    protected void execute(Runnable task)
    {
        executor.execute(task);
    }

    /**
     * @return the number of selectors in use
     */
    public int getSelectorCount()
    {
        return _selectors.length;
    }

    private ManagedSelector chooseSelector()
    {
        // The ++ increment here is not atomic, but it does not matter,
        // so long as the value changes sometimes, then connections will
        // be distributed over the available selectors.
        long s = _selectorIndex++;
        int index = (int)(s % getSelectorCount());
        return _selectors[index];
    }

    /**
     * <p>Registers a channel to perform a non-blocking connect.</p>
     * <p>The channel must be set in non-blocking mode, {@link SocketChannel#connect(SocketAddress)}
     * must be called prior to calling this method, and the connect operation must not be completed
     * (the return value of {@link SocketChannel#connect(SocketAddress)} must be false).</p>
     *
     * @param channel    the channel to register
     * @param attachment the attachment object
     * @see #accept(SocketChannel, Object)
     */
    public void connect(SocketChannel channel, Object attachment)
    {
        ManagedSelector set = chooseSelector();
        set.submit(set.new Connect(channel, attachment));
    }

    /**
     * @see #accept(SocketChannel, Object)
     */
    public void accept(SocketChannel channel)
    {
        accept(channel, null);
    }

    /**
     * <p>Registers a channel to perform non-blocking read/write operations.</p>
     * <p>This method is called just after a channel has been accepted by {@link ServerSocketChannel#accept()},
     * or just after having performed a blocking connect via {@link Socket#connect(SocketAddress, int)}, or
     * just after a non-blocking connect via {@link SocketChannel#connect(SocketAddress)} that completed
     * successfully.</p>
     *
     * @param channel the channel to register
     * @param attachment the attachment object
     */
    public void accept(SocketChannel channel, Object attachment)
    {
        final ManagedSelector selector = chooseSelector();
        selector.submit(selector.new Accept(channel, attachment));
    }

    /**
     * <p>Registers a server channel for accept operations.
     * When a {@link SocketChannel} is accepted from the given {@link ServerSocketChannel}
     * then the {@link #accepted(SocketChannel)} method is called, which must be
     * overridden by a derivation of this class to handle the accepted channel
     *
     * @param server the server channel to register
     */
    public void acceptor(ServerSocketChannel server)
    {
        final ManagedSelector selector = chooseSelector();
        selector.submit(selector.new Acceptor(server));
    }

    /**
     * Callback method when a channel is accepted from the {@link ServerSocketChannel}
     * passed to {@link #acceptor(ServerSocketChannel)}.
     * The default impl throws an {@link UnsupportedOperationException}, so it must
     * be overridden by subclasses if a server channel is provided.
     *
     * @param channel the
     * @throws IOException
     */
    protected void accepted(SocketChannel channel) throws IOException
    {
        throw new UnsupportedOperationException();
    }

    @Override
    protected void doStart() throws Exception
    {
        super.doStart();
        for (int i = 0; i < _selectors.length; i++)
        {
            ManagedSelector selector = newSelector(i);
            _selectors[i] = selector;
            selector.start();
            execute(new NonBlockingThread(selector));
        }
    }

    /**
     * <p>Factory method for {@link ManagedSelector}.</p>
     *
     * @param id an identifier for the {@link ManagedSelector to create}
     * @return a new {@link ManagedSelector}
     */
    protected ManagedSelector newSelector(int id)
    {
        return new ManagedSelector(id);
    }

    @Override
    protected void doStop() throws Exception
    {
        for (ManagedSelector selector : _selectors)
            selector.stop();
        super.doStop();
    }

    /**
     * <p>Callback method invoked when an endpoint is opened.</p>
     *
     * @param endpoint the endpoint being opened
     */
    protected void endPointOpened(EndPoint endpoint)
    {
        endpoint.onOpen();
    }

    /**
     * <p>Callback method invoked when an endpoint is closed.</p>
     *
     * @param endpoint the endpoint being closed
     */
    protected void endPointClosed(EndPoint endpoint)
    {
        endpoint.onClose();
    }

    /**
     * <p>Callback method invoked when a connection is opened.</p>
     *
     * @param connection the connection just opened
     */
    public void connectionOpened(Connection connection)
    {
        try
        {
            connection.onOpen();
        }
        catch (Throwable x)
        {
            if (isRunning())
                LOG.warn("Exception while notifying connection " + connection, x);
            else
                LOG.debug("Exception while notifying connection " + connection, x);
            throw x;
        }
    }

    /**
     * <p>Callback method invoked when a connection is closed.</p>
     *
     * @param connection the connection just closed
     */
    public void connectionClosed(Connection connection)
    {
        try
        {
            connection.onClose();
        }
        catch (Throwable x)
        {
            LOG.debug("Exception while notifying connection " + connection, x);
        }
    }

    protected boolean finishConnect(SocketChannel channel) throws IOException
    {
        return channel.finishConnect();
    }

    /**
     * <p>Callback method invoked when a non-blocking connect cannot be completed.</p>
     * <p>By default it just logs with level warning.</p>
     *
     * @param channel the channel that attempted the connect
     * @param ex the exception that caused the connect to fail
     * @param attachment the attachment object associated at registration
     */
    protected void connectionFailed(SocketChannel channel, Throwable ex, Object attachment)
    {
        LOG.warn(String.format("%s - %s", channel, attachment), ex);
    }

    /**
     * <p>Factory method to create {@link EndPoint}.</p>
     * <p>This method is invoked as a result of the registration of a channel via {@link #connect(SocketChannel, Object)}
     * or {@link #accept(SocketChannel)}.</p>
     *
     * @param channel   the channel associated to the endpoint
     * @param selector the selector the channel is registered to
     * @param selectionKey      the selection key
     * @return a new endpoint
     * @throws IOException if the endPoint cannot be created
     * @see #newConnection(SocketChannel, EndPoint, Object)
     */
    protected abstract EndPoint newEndPoint(SocketChannel channel, SelectorManager.ManagedSelector selector, SelectionKey selectionKey) throws IOException;

    /**
     * <p>Factory method to create {@link Connection}.</p>
     *
     * @param channel    the channel associated to the connection
     * @param endpoint   the endpoint
     * @param attachment the attachment
     * @return a new connection
     * @throws IOException
     * @see #newEndPoint(SocketChannel, ManagedSelector, SelectionKey)
     */
    public abstract Connection newConnection(SocketChannel channel, EndPoint endpoint, Object attachment) throws IOException;

    @Override
    public String dump()
    {
        return ContainerLifeCycle.dump(this);
    }

    @Override
    public void dump(Appendable out, String indent) throws IOException
    {
        ContainerLifeCycle.dumpObject(out, this);
        ContainerLifeCycle.dump(out, indent, TypeUtil.asList(_selectors));
    }

    private enum State
    {
        CHANGES, MORE_CHANGES, SELECT, WAKEUP, PROCESS
    }

    /**
     * <p>{@link ManagedSelector} wraps a {@link Selector} simplifying non-blocking operations on channels.</p>
     * <p>{@link ManagedSelector} runs the select loop, which waits on {@link Selector#select()} until events
     * happen for registered channels. When events happen, it notifies the {@link EndPoint} associated
     * with the channel.</p>
     */
    public class ManagedSelector extends AbstractLifeCycle implements Runnable, Dumpable
    {
        private final AtomicReference<State> _state= new AtomicReference<>(State.PROCESS);
        private final Queue<Runnable> _changes = new ConcurrentArrayQueue<>();
        private final int _id;
        private Selector _selector;
        private volatile Thread _thread;

        public ManagedSelector(int id)
        {
            _id = id;
            setStopTimeout(5000);
        }

        @Override
        protected void doStart() throws Exception
        {
            super.doStart();
            _selector = Selector.open();
            _state.set(State.PROCESS);
        }

        @Override
        protected void doStop() throws Exception
        {
            if (LOG.isDebugEnabled())
                LOG.debug("Stopping {}", this);
            Stop stop = new Stop();
            submit(stop);
            stop.await(getStopTimeout());
            if (LOG.isDebugEnabled())
                LOG.debug("Stopped {}", this);
        }

        /**
         * Submit a task to update a selector key.  If the System property {@link SelectorManager#SUBMIT_KEY_UPDATES}
         * is set true (default is false), the task is passed to {@link #submit(Runnable)}.   Otherwise it is run immediately and the selector
         * woken up if need be.
         * @param update the update to a key
         */
        public void updateKey(Runnable update)
        {
            if (__submitKeyUpdates)
            {
                submit(update);
            }
            else
            {
                // Run only 1 change at once
                synchronized (this)
                {
                    runChange(update);
                }
                if (_state.compareAndSet(State.SELECT, State.WAKEUP))
                   wakeup();
            }
        }

        /**
         * <p>Submits a change to be executed in the selector thread.</p>
         * <p>Changes may be submitted from any thread, and the selector thread woken up
         * (if necessary) to execute the change.</p>
         *
         * @param change the change to submit
         */
        public void submit(Runnable change)
        {
            // This method may be called from the selector thread, and therefore
            // we could directly run the change without queueing, but this may
            // lead to stack overflows on a busy server, so we always offer the
            // change to the queue and process the state.

            _changes.offer(change);
            if (LOG.isDebugEnabled())
                LOG.debug("Queued change {}", change);

            out: while (true)
            {
                switch (_state.get())
                {
                    case SELECT:
                        // Avoid multiple wakeup() calls if we the CAS fails
                        if (!_state.compareAndSet(State.SELECT, State.WAKEUP))
                            continue;
                        wakeup();
                        break out;
                    case CHANGES:
                        // Tell the selector thread that we have more changes.
                        // If we fail to CAS, we possibly need to wakeup(), so loop.
                        if (_state.compareAndSet(State.CHANGES, State.MORE_CHANGES))
                            break out;
                        continue;
                    case WAKEUP:
                        // Do nothing, we have already a wakeup scheduled
                        break out;
                    case MORE_CHANGES:
                        // Do nothing, we already notified the selector thread of more changes
                        break out;
                    case PROCESS:
                        // Do nothing, the changes will be run after the processing
                        break out;
                    default:
                        throw new IllegalStateException();
                }
            }
        }

        private void runChanges()
        {
            Runnable change;
            while ((change = _changes.poll()) != null)
                runChange(change);
        }

        protected void runChange(Runnable change)
        {
            try
            {
                if (LOG.isDebugEnabled())
                    LOG.debug("Running change {}", change);
                change.run();
            }
            catch (Throwable x)
            {
                LOG.debug("Could not run change " + change, x);
            }
        }

        @Override
        public void run()
        {
            _thread = Thread.currentThread();
            String name = _thread.getName();
            int priority = _thread.getPriority();
            try
            {
                if (_priorityDelta != 0)
                    _thread.setPriority(Math.max(Thread.MIN_PRIORITY, Math.min(Thread.MAX_PRIORITY, priority + _priorityDelta)));

                _thread.setName(String.format("%s-selector-%s@%h/%d", name, SelectorManager.this.getClass().getSimpleName(), SelectorManager.this.hashCode(), _id));
                if (LOG.isDebugEnabled())
                    LOG.debug("Starting {} on {}", _thread, this);
                while (isRunning())
                    select();
                while(isStopping())
                    runChanges();
            }
            finally
            {
                if (LOG.isDebugEnabled())
                    LOG.debug("Stopped {} on {}", _thread, this);
                _thread.setName(name);
                if (_priorityDelta != 0)
                    _thread.setPriority(priority);
            }
        }

        /**
         * <p>Process changes and waits on {@link Selector#select()}.</p>
         *
         * @see #submit(Runnable)
         */
        public void select()
        {
            boolean debug = LOG.isDebugEnabled();
            try
            {
                _state.set(State.CHANGES);

                // Run the changes, and only exit if we ran all changes
                out: while(true)
                {
                    switch (_state.get())
                    {
                        case CHANGES:
                            runChanges();
                            if (_state.compareAndSet(State.CHANGES, State.SELECT))
                                break out;
                            continue;
                        case MORE_CHANGES:
                            runChanges();
                            _state.set(State.CHANGES);
                            continue;
                        default:
                            throw new IllegalStateException();
                    }
                }
                // Must check first for SELECT and *then* for WAKEUP
                // because we read the state twice in the assert, and
                // it could change from SELECT to WAKEUP in between.
                assert _state.get() == State.SELECT || _state.get() == State.WAKEUP;

                if (debug)
                    LOG.debug("Selector loop waiting on select");
                int selected = _selector.select();
                if (debug)
                    LOG.debug("Selector loop woken up from select, {}/{} selected", selected, _selector.keys().size());

                _state.set(State.PROCESS);

                Set<SelectionKey> selectedKeys = _selector.selectedKeys();
                for (SelectionKey key : selectedKeys)
                {
                    if (key.isValid())
                    {
                        processKey(key);
                    }
                    else
                    {
                        if (debug)
                            LOG.debug("Selector loop ignoring invalid key for channel {}", key.channel());
                        Object attachment = key.attachment();
                        if (attachment instanceof EndPoint)
                            ((EndPoint)attachment).close();
                    }
                }
                selectedKeys.clear();
            }
            catch (Throwable x)
            {
                if (isRunning())
                    LOG.warn(x);
                else
                    LOG.ignore(x);
            }
        }

        private void processKey(SelectionKey key)
        {
            Object attachment = key.attachment();
            try
            {
                if (attachment instanceof SelectableEndPoint)
                {
                    ((SelectableEndPoint)attachment).onSelected();
                }
                else if (key.isConnectable())
                {
                    processConnect(key, (Connect)attachment);
                }
                else if (key.isAcceptable())
                {
                    processAccept(key);
                }
                else
                {
                    throw new IllegalStateException();
                }
            }
            catch (CancelledKeyException x)
            {
                LOG.debug("Ignoring cancelled key for channel {}", key.channel());
                if (attachment instanceof EndPoint)
                    closeNoExceptions((EndPoint)attachment);
            }
            catch (Throwable x)
            {
                LOG.warn("Could not process key for channel " + key.channel(), x);
                if (attachment instanceof EndPoint)
                    closeNoExceptions((EndPoint)attachment);
            }
        }

        private void processConnect(SelectionKey key, Connect connect)
        {
            SocketChannel channel = (SocketChannel)key.channel();
            try
            {
                key.attach(connect.attachment);
                boolean connected = finishConnect(channel);
                if (connected)
                {
                    if (connect.timeout.cancel())
                    {
                        key.interestOps(0);
                        EndPoint endpoint = createEndPoint(channel, key);
                        key.attach(endpoint);
                    }
                    else
                    {
                        throw new SocketTimeoutException("Concurrent Connect Timeout");
                    }
                }
                else
                {
                    throw new ConnectException();
                }
            }
            catch (Throwable x)
            {
                connect.failed(x);
            }
        }

        private void processAccept(SelectionKey key)
        {
            ServerSocketChannel server = (ServerSocketChannel)key.channel();
            SocketChannel channel = null;
            try
            {
                while ((channel = server.accept()) != null)
                {
                    accepted(channel);
                }
            }
            catch (Throwable x)
            {
                closeNoExceptions(channel);
                LOG.warn("Accept failed for channel " + channel, x);
            }
        }

        private void closeNoExceptions(Closeable closeable)
        {
            try
            {
                if (closeable != null)
                    closeable.close();
            }
            catch (Throwable x)
            {
                LOG.ignore(x);
            }
        }

        public void wakeup()
        {
            _selector.wakeup();
        }

        public boolean isSelectorThread()
        {
            return Thread.currentThread() == _thread;
        }

        private EndPoint createEndPoint(SocketChannel channel, SelectionKey selectionKey) throws IOException
        {
            EndPoint endPoint = newEndPoint(channel, this, selectionKey);
            endPointOpened(endPoint);
            Connection connection = newConnection(channel, endPoint, selectionKey.attachment());
            endPoint.setConnection(connection);
            connectionOpened(connection);
            if (LOG.isDebugEnabled())
                LOG.debug("Created {}", endPoint);
            return endPoint;
        }

        public void destroyEndPoint(EndPoint endPoint)
        {
            if (LOG.isDebugEnabled())
                LOG.debug("Destroyed {}", endPoint);
            Connection connection = endPoint.getConnection();
            if (connection != null)
                connectionClosed(connection);
            endPointClosed(endPoint);
        }

        @Override
        public String dump()
        {
            return ContainerLifeCycle.dump(this);
        }

        @Override
        public void dump(Appendable out, String indent) throws IOException
        {
            out.append(String.valueOf(this)).append(" id=").append(String.valueOf(_id)).append("\n");

            Thread selecting = _thread;

            Object where = "not selecting";
            StackTraceElement[] trace = selecting == null ? null : selecting.getStackTrace();
            if (trace != null)
            {
                for (StackTraceElement t : trace)
                    if (t.getClassName().startsWith("org.eclipse.jetty."))
                    {
                        where = t;
                        break;
                    }
            }

            Selector selector = _selector;
            if (selector != null && selector.isOpen())
            {
                final ArrayList<Object> dump = new ArrayList<>(selector.keys().size() * 2);
                dump.add(where);

                DumpKeys dumpKeys = new DumpKeys(dump);
                submit(dumpKeys);
                dumpKeys.await(5, TimeUnit.SECONDS);

                ContainerLifeCycle.dump(out, indent, dump);
            }
        }

        public void dumpKeysState(List<Object> dumps)
        {
            Selector selector = _selector;
            Set<SelectionKey> keys = selector.keys();
            dumps.add(selector + " keys=" + keys.size());
            for (SelectionKey key : keys)
            {
                if (key.isValid())
                    dumps.add(key.attachment() + " iOps=" + key.interestOps() + " rOps=" + key.readyOps());
                else
                    dumps.add(key.attachment() + " iOps=-1 rOps=-1");
            }
        }

        @Override
        public String toString()
        {
            Selector selector = _selector;
            return String.format("%s keys=%d selected=%d",
                    super.toString(),
                    selector != null && selector.isOpen() ? selector.keys().size() : -1,
                    selector != null && selector.isOpen() ? selector.selectedKeys().size() : -1);
        }

        private class DumpKeys implements Runnable
        {
            private final CountDownLatch latch = new CountDownLatch(1);
            private final List<Object> _dumps;

            private DumpKeys(List<Object> dumps)
            {
                this._dumps = dumps;
            }

            @Override
            public void run()
            {
                dumpKeysState(_dumps);
                latch.countDown();
            }

            public boolean await(long timeout, TimeUnit unit)
            {
                try
                {
                    return latch.await(timeout, unit);
                }
                catch (InterruptedException x)
                {
                    return false;
                }
            }
        }

        private class Acceptor implements Runnable
        {
            private final ServerSocketChannel _channel;

            public Acceptor(ServerSocketChannel channel)
            {
                this._channel = channel;
            }

            @Override
            public void run()
            {
                try
                {
                    SelectionKey key = _channel.register(_selector, SelectionKey.OP_ACCEPT, null);
                    if (LOG.isDebugEnabled())
                        LOG.debug("{} acceptor={}", this, key);
                }
                catch (Throwable x)
                {
                    closeNoExceptions(_channel);
                    LOG.warn(x);
                }
            }
        }

        private class Accept implements Runnable
        {
            private final SocketChannel channel;
            private final Object attachment;

            private Accept(SocketChannel channel, Object attachment)
            {
                this.channel = channel;
                this.attachment = attachment;
            }

            @Override
            public void run()
            {
                try
                {
                    SelectionKey key = channel.register(_selector, 0, attachment);
                    EndPoint endpoint = createEndPoint(channel, key);
                    key.attach(endpoint);
                }
                catch (Throwable x)
                {
                    closeNoExceptions(channel);
                    LOG.debug(x);
                }
            }
        }

        private class Connect implements Runnable
        {
            private final AtomicBoolean failed = new AtomicBoolean();
            private final SocketChannel channel;
            private final Object attachment;
            private final Scheduler.Task timeout;

            private Connect(SocketChannel channel, Object attachment)
            {
                this.channel = channel;
                this.attachment = attachment;
                this.timeout = scheduler.schedule(new ConnectTimeout(this), getConnectTimeout(), TimeUnit.MILLISECONDS);
            }

            @Override
            public void run()
            {
                try
                {
                    channel.register(_selector, SelectionKey.OP_CONNECT, this);
                }
                catch (Throwable x)
                {
                    failed(x);
                }
            }

            private void failed(Throwable failure)
            {
                if (failed.compareAndSet(false, true))
                {
                    timeout.cancel();
                    closeNoExceptions(channel);
                    connectionFailed(channel, failure, attachment);
                }
            }
        }

        private class ConnectTimeout implements Runnable
        {
            private final Connect connect;

            private ConnectTimeout(Connect connect)
            {
                this.connect = connect;
            }

            @Override
            public void run()
            {
                SocketChannel channel = connect.channel;
                if (channel.isConnectionPending())
                {
                    if (LOG.isDebugEnabled())
                        LOG.debug("Channel {} timed out while connecting, closing it", channel);
                    connect.failed(new SocketTimeoutException("Connect Timeout"));
                }
            }
        }

        private class Stop implements Runnable
        {
            private final CountDownLatch latch = new CountDownLatch(1);

            @Override
            public void run()
            {
                try
                {
                    for (SelectionKey key : _selector.keys())
                    {
                        Object attachment = key.attachment();
                        if (attachment instanceof EndPoint)
                        {
                            EndPointCloser closer = new EndPointCloser((EndPoint)attachment);
                            execute(closer);
                            // We are closing the SelectorManager, so we want to block the
                            // selector thread here until we have closed all EndPoints.
                            // This is different than calling close() directly, because close()
                            // can wait forever, while here we are limited by the stop timeout.
                            closer.await(getStopTimeout());
                        }
                    }

                    closeNoExceptions(_selector);
                }
                finally
                {
                    latch.countDown();
                }
            }

            public boolean await(long timeout)
            {
                try
                {
                    return latch.await(timeout, TimeUnit.MILLISECONDS);
                }
                catch (InterruptedException x)
                {
                    return false;
                }
            }
        }

        private class EndPointCloser implements Runnable
        {
            private final CountDownLatch latch = new CountDownLatch(1);
            private final EndPoint endPoint;

            private EndPointCloser(EndPoint endPoint)
            {
                this.endPoint = endPoint;
            }

            @Override
            public void run()
            {
                try
                {
                    closeNoExceptions(endPoint.getConnection());
                }
                finally
                {
                    latch.countDown();
                }
            }

            private boolean await(long timeout)
            {
                try
                {
                    return latch.await(timeout, TimeUnit.MILLISECONDS);
                }
                catch (InterruptedException x)
                {
                    return false;
                }
            }
        }
    }

    /**
     * A {@link SelectableEndPoint} is an {@link EndPoint} that wish to be notified of
     * non-blocking events by the {@link ManagedSelector}.
     */
    public interface SelectableEndPoint extends EndPoint
    {
        /**
         * <p>Callback method invoked when a read or write events has been detected by the {@link ManagedSelector}
         * for this endpoint.</p>
         */
        void onSelected();
    }
}