https://eclipse.org/jetty/jetty-io: The Eclipse Jetty Project (Webtide) 
//
// ========================================================================
// Copyright (c) 1995-2020 Mort Bay Consulting Pty Ltd and others.
//
// This program and the accompanying materials are made available under
// the terms of the Eclipse Public License 2.0 which is available at
// https://www.eclipse.org/legal/epl-2.0
//
// This Source Code may also be made available under the following
// Secondary Licenses when the conditions for such availability set
// forth in the Eclipse Public License, v. 2.0 are satisfied:
// the Apache License v2.0 which is available at
// https://www.apache.org/licenses/LICENSE-2.0
//
// SPDX-License-Identifier: EPL-2.0 OR Apache-2.0
// ========================================================================
//

package org.eclipse.jetty.io;

import java.io.Closeable;
import java.io.IOException;
import java.net.Socket;
import java.net.SocketAddress;
import java.nio.channels.SelectableChannel;
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.Arrays;
import java.util.EventListener;
import java.util.List;
import java.util.concurrent.Executor;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.IntUnaryOperator;

import org.eclipse.jetty.util.ProcessorUtils;
import org.eclipse.jetty.util.annotation.ManagedAttribute;
import org.eclipse.jetty.util.annotation.ManagedObject;
import org.eclipse.jetty.util.component.Container;
import org.eclipse.jetty.util.component.ContainerLifeCycle;
import org.eclipse.jetty.util.component.Dumpable;
import org.eclipse.jetty.util.thread.Scheduler;
import org.eclipse.jetty.util.thread.ThreadPool;
import org.eclipse.jetty.util.thread.ThreadPoolBudget;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;


SelectorManager manages a number of ManagedSelectors that simplify the non-blocking primitives provided by the JVM via the java.nio package.


SelectorManager subclasses implement methods to return protocol-specific EndPoints and Connections.


/**
 * <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>
 */

@ManagedObject("Manager of the NIO Selectors")
public abstract class SelectorManager extends ContainerLifeCycle implements Dumpable
{
    public static final int DEFAULT_CONNECT_TIMEOUT = 15000;
    protected static final Logger LOG = LoggerFactory.getLogger(SelectorManager.class);

    private final Executor executor;
    private final Scheduler scheduler;
    private final ManagedSelector[] _selectors;
    private final AtomicInteger _selectorIndex = new AtomicInteger();
    private final IntUnaryOperator _selectorIndexUpdate;
    private final List<AcceptListener> _acceptListeners = new ArrayList<>();
    private long _connectTimeout = DEFAULT_CONNECT_TIMEOUT;
    private ThreadPoolBudget.Lease _lease;

    private static int defaultSelectors(Executor executor)
    {
        if (executor instanceof ThreadPool.SizedThreadPool)
        {
            int threads = ((ThreadPool.SizedThreadPool)executor).getMaxThreads();
            int cpus = ProcessorUtils.availableProcessors();
            return Math.max(1, Math.min(cpus / 2, threads / 16));
        }
        return Math.max(1, ProcessorUtils.availableProcessors() / 2);
    }

    protected SelectorManager(Executor executor, Scheduler scheduler)
    {
        this(executor, scheduler, -1);
    }

    
Params:
  • executor – The executor to use for handling selected EndPoints
  • scheduler – The scheduler to use for timing events
  • selectors – The number of selectors to use, or -1 for a default derived
from a heuristic over available CPUs and thread pool size.
/**
     * @param executor The executor to use for handling selected {@link EndPoint}s
     * @param scheduler The scheduler to use for timing events
     * @param selectors The number of selectors to use, or -1 for a default derived
     * from a heuristic over available CPUs and thread pool size.
     */
    protected SelectorManager(Executor executor, Scheduler scheduler, int selectors)
    {
        if (selectors <= 0)
            selectors = defaultSelectors(executor);
        this.executor = executor;
        this.scheduler = scheduler;
        _selectors = new ManagedSelector[selectors];
        _selectorIndexUpdate = index -> (index + 1) % _selectors.length;
    }

    @ManagedAttribute("The Executor")
    public Executor getExecutor()
    {
        return executor;
    }

    @ManagedAttribute("The Scheduler")
    public Scheduler getScheduler()
    {
        return scheduler;
    }

    
Get the connect timeout

Returns:the connect timeout (in milliseconds)
/**
     * Get the connect timeout
     *
     * @return the connect timeout (in milliseconds)
     */
    @ManagedAttribute("The Connection timeout (ms)")
    public long getConnectTimeout()
    {
        return _connectTimeout;
    }

    
Set the connect timeout (in milliseconds)

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

    
Executes the given task in a different thread.

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

    
Returns:the number of selectors in use
/**
     * @return the number of selectors in use
     */
    @ManagedAttribute("The number of NIO Selectors")
    public int getSelectorCount()
    {
        return _selectors.length;
    }

    private ManagedSelector chooseSelector()
    {
        return _selectors[_selectorIndex.updateAndGet(_selectorIndexUpdate)];
    }

    
Registers a channel to perform a non-blocking connect.


The channel must be set in non-blocking mode, SocketChannel.connect(SocketAddress) must be called prior to calling this method, and the connect operation must not be completed (the return value of SocketChannel.connect(SocketAddress) must be false).


Params:
  • channel – the channel to register
  • attachment – the attachment object
See Also:
/**
     * <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(SelectableChannel, Object)
     */
    public void connect(SelectableChannel channel, Object attachment)
    {
        ManagedSelector set = chooseSelector();
        set.submit(set.new Connect(channel, attachment));
    }

    
Params:
  • channel – the channel to accept
See Also:
/**
     * @param channel the channel to accept
     * @see #accept(SelectableChannel, Object)
     */
    public void accept(SelectableChannel channel)
    {
        accept(channel, null);
    }

    
Registers a channel to perform non-blocking read/write operations.


This method is called just after a channel has been accepted by ServerSocketChannel.accept(), or just after having performed a blocking connect via Socket.connect(SocketAddress, int), or just after a non-blocking connect via SocketChannel.connect(SocketAddress) that completed successfully.


Params:
  • channel – the channel to register
  • attachment – the attachment object
/**
     * <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(SelectableChannel channel, Object attachment)
    {
        ManagedSelector selector = chooseSelector();
        selector.submit(selector.new Accept(channel, attachment));
    }

    
Registers a server channel for accept operations. When a SocketChannel is accepted from the given ServerSocketChannel then the accepted(SelectableChannel) method is called, which must be overridden by a derivation of this class to handle the accepted channel 
Params:
  • server – the server channel to register
Returns:A Closable that allows the acceptor to be cancelled
/**
     * <p>Registers a server channel for accept operations.
     * When a {@link SocketChannel} is accepted from the given {@link ServerSocketChannel}
     * then the {@link #accepted(SelectableChannel)} 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
     * @return A Closable that allows the acceptor to be cancelled
     */
    public Closeable acceptor(SelectableChannel server)
    {
        ManagedSelector selector = chooseSelector();
        ManagedSelector.Acceptor acceptor = selector.new Acceptor(server);
        selector.submit(acceptor);
        return acceptor;
    }

    
Callback method when a channel is accepted from the ServerSocketChannel passed to acceptor(SelectableChannel). The default impl throws an UnsupportedOperationException, so it must be overridden by subclasses if a server channel is provided. 
Params:
  • channel – the
Throws:
/**
     * Callback method when a channel is accepted from the {@link ServerSocketChannel}
     * passed to {@link #acceptor(SelectableChannel)}.
     * 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 if unable to accept channel
     */
    protected void accepted(SelectableChannel channel) throws IOException
    {
        throw new UnsupportedOperationException();
    }

    @Override
    protected void doStart() throws Exception
    {
        _lease = ThreadPoolBudget.leaseFrom(getExecutor(), this, _selectors.length);
        for (int i = 0; i < _selectors.length; i++)
        {
            ManagedSelector selector = newSelector(i);
            _selectors[i] = selector;
            addBean(selector);
        }
        super.doStart();
    }

    
Factory method for ManagedSelector.


Params:
Returns:a new ManagedSelector
/**
     * <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(this, id);
    }

    protected Selector newSelector() throws IOException
    {
        return Selector.open();
    }

    @Override
    protected void doStop() throws Exception
    {
        try
        {
            super.doStop();
        }
        finally
        {
            // Cleanup
            for (ManagedSelector selector : _selectors)
            {
                if (selector != null)
                    removeBean(selector);
            }
            Arrays.fill(_selectors, null);
            if (_lease != null)
                _lease.close();
        }
    }

    
Callback method invoked when an endpoint is opened.


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

    
Callback method invoked when an endpoint is closed.


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

    
Callback method invoked when a connection is opened.


Params:
  • connection – the connection just opened
  • context – the attachment associated with the creation of the connection
See Also:
/**
     * <p>Callback method invoked when a connection is opened.</p>
     *
     * @param connection the connection just opened
     * @param context the attachment associated with the creation of the connection
     * @see #newConnection(SelectableChannel, EndPoint, Object)
     */
    public void connectionOpened(Connection connection, Object context)
    {
        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;
        }
    }

    
Callback method invoked when a connection is closed.


Params:
  • connection – the connection just closed
  • cause – the cause of the close or null for normal close
/**
     * <p>Callback method invoked when a connection is closed.</p>
     *
     * @param connection the connection just closed
     * @param cause the cause of the close or null for normal close
     */
    public void connectionClosed(Connection connection, Throwable cause)
    {
        try
        {
            connection.onClose(cause);
        }
        catch (Throwable x)
        {
            if (LOG.isDebugEnabled())
                LOG.debug("Exception while notifying connection {}", connection, x);
        }
    }

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

    protected boolean isConnectionPending(SelectableChannel channel)
    {
        return ((SocketChannel)channel).isConnectionPending();
    }

    protected SelectableChannel doAccept(SelectableChannel server) throws IOException
    {
        return ((ServerSocketChannel)server).accept();
    }

    
Callback method invoked when a non-blocking connect cannot be completed.


By default it just logs with level warning.


Params:
  • channel – the channel that attempted the connect
  • ex – the exception that caused the connect to fail
  • attachment – the attachment object associated at registration
/**
     * <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(SelectableChannel channel, Throwable ex, Object attachment)
    {
        LOG.warn(String.format("%s - %s", channel, attachment), ex);
    }

    
Factory method to create EndPoint.


This method is invoked as a result of the registration of a channel via connect(SelectableChannel, Object) or accept(SelectableChannel).


Params:
  • channel – the channel associated to the endpoint
  • selector – the selector the channel is registered to
  • selectionKey – the selection key
Throws:
See Also:
Returns:a new endpoint
/**
     * <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(SelectableChannel, Object)}
     * or {@link #accept(SelectableChannel)}.</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(SelectableChannel, EndPoint, Object)
     */
    protected abstract EndPoint newEndPoint(SelectableChannel channel, ManagedSelector selector, SelectionKey selectionKey) throws IOException;

    
Factory method to create Connection.


Params:
  • channel – the channel associated to the connection
  • endpoint – the endpoint
  • attachment – the attachment
Throws:
Returns:a new connection
/**
     * <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 if unable to create new connection
     */
    public abstract Connection newConnection(SelectableChannel channel, EndPoint endpoint, Object attachment) throws IOException;

    
Params:
  • listener – An EventListener
See Also:
/**
     * @param listener An EventListener
     * @see AcceptListener
     * @see Container#addEventListener(EventListener)
     */
    @Override
    public boolean addEventListener(EventListener listener)
    {
        if (isRunning())
            throw new IllegalStateException(this.toString());
        if (super.addEventListener(listener))
        {
            if (listener instanceof AcceptListener)
                _acceptListeners.add((AcceptListener)listener);
            return true;
        }
        return false;
    }

    @Override
    public boolean removeEventListener(EventListener listener)
    {
        if (isRunning())
            throw new IllegalStateException(this.toString());
        if (super.removeEventListener(listener))
        {
            if (listener instanceof AcceptListener)
                _acceptListeners.remove(listener);
            return true;
        }
        return false;
    }

    protected void onAccepting(SelectableChannel channel)
    {
        for (AcceptListener l : _acceptListeners)
        {
            try
            {
                l.onAccepting(channel);
            }
            catch (Throwable x)
            {
                LOG.warn("Failed to notify onAccepting on listener {}", l, x);
            }
        }
    }

    protected void onAcceptFailed(SelectableChannel channel, Throwable cause)
    {
        for (AcceptListener l : _acceptListeners)
        {
            try
            {
                l.onAcceptFailed(channel, cause);
            }
            catch (Throwable x)
            {
                LOG.warn("Failed to notify onAcceptFailed on listener {}", l, x);
            }
        }
    }

    protected void onAccepted(SelectableChannel channel)
    {
        for (AcceptListener l : _acceptListeners)
        {
            try
            {
                l.onAccepted(channel);
            }
            catch (Throwable x)
            {
                LOG.warn("Failed to notify onAccepted on listener {}", l, x);
            }
        }
    }

    public interface SelectorManagerListener extends EventListener
    {
    }

    
A listener for accept events.


This listener is called from either the selector or acceptor thread
and implementations must be non blocking and fast.


/**
     * <p>A listener for accept events.</p>
     * <p>This listener is called from either the selector or acceptor thread
     * and implementations must be non blocking and fast.</p>
     */
    public interface AcceptListener extends SelectorManagerListener
    {
        
Called immediately after a new SelectableChannel is accepted, but before it has been submitted to the SelectorManager. 
Params:
  • channel – the accepted channel
/**
         * Called immediately after a new SelectableChannel is accepted, but
         * before it has been submitted to the {@link SelectorManager}.
         *
         * @param channel the accepted channel
         */
        default void onAccepting(SelectableChannel channel)
        {
        }

        
Called if the processing of the accepted channel fails prior to calling onAccepted(SelectableChannel). 
Params:
  • channel – the accepted channel
  • cause – the cause of the failure
/**
         * Called if the processing of the accepted channel fails prior to calling
         * {@link #onAccepted(SelectableChannel)}.
         *
         * @param channel the accepted channel
         * @param cause the cause of the failure
         */
        default void onAcceptFailed(SelectableChannel channel, Throwable cause)
        {
        }

        
Called after the accepted channel has been allocated an EndPoint and associated Connection, and after the onOpen notifications have been called on both endPoint and connection. 
Params:
  • channel – the accepted channel
/**
         * Called after the accepted channel has been allocated an {@link EndPoint}
         * and associated {@link Connection}, and after the onOpen notifications have
         * been called on both endPoint and connection.
         *
         * @param channel the accepted channel
         */
        default void onAccepted(SelectableChannel channel)
        {
        }
    }
}