src/java/org/eclipse/jetty/io/nio/SelectChannelEndPoint.java

//
//  ========================================================================
//  Copyright (c) 1995-2014 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.nio;

import java.io.IOException;
import java.io.InterruptedIOException;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.SelectableChannel;
import java.nio.channels.SelectionKey;
import java.nio.channels.SocketChannel;
import java.util.Locale;

import org.eclipse.jetty.io.AsyncEndPoint;
import org.eclipse.jetty.io.Buffer;
import org.eclipse.jetty.io.ConnectedEndPoint;
import org.eclipse.jetty.io.Connection;
import org.eclipse.jetty.io.EofException;
import org.eclipse.jetty.io.nio.SelectorManager.SelectSet;
import org.eclipse.jetty.util.log.Log;
import org.eclipse.jetty.util.log.Logger;
import org.eclipse.jetty.util.thread.Timeout.Task;

/* ------------------------------------------------------------ */
/**
 * An Endpoint that can be scheduled by {@link SelectorManager}.
 */
public class SelectChannelEndPoint extends ChannelEndPoint implements AsyncEndPoint, ConnectedEndPoint
{
    public static final Logger LOG=Log.getLogger("org.eclipse.jetty.io.nio");

    private final boolean WORK_AROUND_JVM_BUG_6346658 = System.getProperty("os.name").toLowerCase(Locale.ENGLISH).contains("win");
    private final SelectorManager.SelectSet _selectSet;
    private final SelectorManager _manager;
    private  SelectionKey _key;
    private final Runnable _handler = new Runnable()
        {
            public void run() { handle(); }
        };

    /** The desired value for {@link SelectionKey#interestOps()} */
    private int _interestOps;

    /**
     * The connection instance is the handler for any IO activity on the endpoint.
     * There is a different type of connection for HTTP, AJP, WebSocket and
     * ProxyConnect.   The connection may change for an SCEP as it is upgraded
     * from HTTP to proxy connect or websocket.
     */
    private volatile AsyncConnection _connection;

    private static final int STATE_NEEDS_DISPATCH=-1;
    private static final int STATE_UNDISPATCHED=0;
    private static final int STATE_DISPATCHED=1;
    private static final int STATE_ASYNC=2;
    private int _state;
    
    private boolean _onIdle;

    /** true if the last write operation succeed and wrote all offered bytes */
    private volatile boolean _writable = true;


    /** True if a thread has is blocked in {@link #blockReadable(long)} */
    private boolean _readBlocked;

    /** True if a thread has is blocked in {@link #blockWritable(long)} */
    private boolean _writeBlocked;

    /** true if {@link SelectSet#destroyEndPoint(SelectChannelEndPoint)} has not been called */
    private boolean _open;

    private volatile long _idleTimestamp;
    private volatile boolean _checkIdle;
    
    private boolean _interruptable;

    private boolean _ishut;

    /* ------------------------------------------------------------ */
    public SelectChannelEndPoint(SocketChannel channel, SelectSet selectSet, SelectionKey key, int maxIdleTime)
        throws IOException
    {
        super(channel, maxIdleTime);

        _manager = selectSet.getManager();
        _selectSet = selectSet;
        _state=STATE_UNDISPATCHED;
        _onIdle=false;
        _open=true;
        _key = key;

        setCheckForIdle(true);
    }

    /* ------------------------------------------------------------ */
    public SelectionKey getSelectionKey()
    {
        synchronized (this)
        {
            return _key;
        }
    }

    /* ------------------------------------------------------------ */
    public SelectorManager getSelectManager()
    {
        return _manager;
    }

    /* ------------------------------------------------------------ */
    public Connection getConnection()
    {
        return _connection;
    }

    /* ------------------------------------------------------------ */
    public void setConnection(Connection connection)
    {
        Connection old=_connection;
        _connection=(AsyncConnection)connection;
        if (old!=null && old!=_connection)
            _manager.endPointUpgraded(this,old);
    }

    /* ------------------------------------------------------------ */
    public long getIdleTimestamp()
    {
        return _idleTimestamp;
    }

    /* ------------------------------------------------------------ */
    /** Called by selectSet to schedule handling
     *
     */
    public void schedule()
    {
        synchronized (this)
        {
            // If there is no key, then do nothing
            if (_key == null || !_key.isValid())
            {
                _readBlocked=false;
                _writeBlocked=false;
                this.notifyAll();
                return;
            }

            // If there are threads dispatched reading and writing
            if (_readBlocked || _writeBlocked)
            {
                // assert _dispatched;
                if (_readBlocked && _key.isReadable())
                    _readBlocked=false;
                if (_writeBlocked && _key.isWritable())
                    _writeBlocked=false;

                // wake them up is as good as a dispatched.
                this.notifyAll();

                // we are not interested in further selecting
                _key.interestOps(0);
                if (_state<STATE_DISPATCHED)
                    updateKey();
                return;
            }

            // Remove writeable op
            if ((_key.readyOps() & SelectionKey.OP_WRITE) == SelectionKey.OP_WRITE && (_key.interestOps() & SelectionKey.OP_WRITE) == SelectionKey.OP_WRITE)
            {
                // Remove writeable op
                _interestOps = _key.interestOps() & ~SelectionKey.OP_WRITE;
                _key.interestOps(_interestOps);
                _writable = true; // Once writable is in ops, only removed with dispatch.
            }

            // If dispatched, then deregister interest
            if (_state>=STATE_DISPATCHED)
                _key.interestOps(0);
            else
            {
                // other wise do the dispatch
                dispatch();
                if (_state>=STATE_DISPATCHED && !_selectSet.getManager().isDeferringInterestedOps0())
                {
                    _key.interestOps(0);
                }
            }
        }
    }

    /* ------------------------------------------------------------ */
    public void asyncDispatch()
    {
        synchronized(this)
        {
            switch(_state)
            {
                case STATE_NEEDS_DISPATCH:
                case STATE_UNDISPATCHED:
                    dispatch();
                    break;
                    
                case STATE_DISPATCHED:
                case STATE_ASYNC:
                    _state=STATE_ASYNC;
                    break;
            }
        }
    }

    /* ------------------------------------------------------------ */
    public void dispatch()
    {
        synchronized(this)
        {
            if (_state<=STATE_UNDISPATCHED)
            {
                if (_onIdle)
                    _state = STATE_NEEDS_DISPATCH;
                else
                {
                    _state = STATE_DISPATCHED;
                    boolean dispatched = _manager.dispatch(_handler);
                    if(!dispatched)
                    {
                        _state = STATE_NEEDS_DISPATCH;
                        LOG.warn("Dispatched Failed! "+this+" to "+_manager);
                        updateKey();
                    }
                }
            }
        }
    }

    /* ------------------------------------------------------------ */
    /**
     * Called when a dispatched thread is no longer handling the endpoint.
     * The selection key operations are updated.
     * @return If false is returned, the endpoint has been redispatched and
     * thread must keep handling the endpoint.
     */
    protected boolean undispatch()
    {
        synchronized (this)
        {
            switch(_state)
            {
                case STATE_ASYNC:
                    _state=STATE_DISPATCHED;
                    return false;

                default:
                    _state=STATE_UNDISPATCHED;
                    updateKey();
                    return true;
            }
        }
    }

    /* ------------------------------------------------------------ */
    public void cancelTimeout(Task task)
    {
        getSelectSet().cancelTimeout(task);
    }

    /* ------------------------------------------------------------ */
    public void scheduleTimeout(Task task, long timeoutMs)
    {
        getSelectSet().scheduleTimeout(task,timeoutMs);
    }

    /* ------------------------------------------------------------ */
    public void setCheckForIdle(boolean check)
    {
        if (check)
        {
            _idleTimestamp=System.currentTimeMillis();
            _checkIdle=true;
        }
        else
            _checkIdle=false;
    }

    /* ------------------------------------------------------------ */
    public boolean isCheckForIdle()
    {
        return _checkIdle;
    }

    /* ------------------------------------------------------------ */
    protected void notIdle()
    {
        _idleTimestamp=System.currentTimeMillis();
    }

    /* ------------------------------------------------------------ */
    public void checkIdleTimestamp(long now)
    {
        if (isCheckForIdle() && _maxIdleTime>0)
        {
            final long idleForMs=now-_idleTimestamp;

            if (idleForMs>_maxIdleTime)
            {
                // Don't idle out again until onIdleExpired task completes.
                setCheckForIdle(false);
                _manager.dispatch(new Runnable()
                {
                    public void run()
                    {
                        try
                        {
                            onIdleExpired(idleForMs);
                        }
                        finally
                        {
                            setCheckForIdle(true);
                        }
                    }
                });
            }
        }
    }

    /* ------------------------------------------------------------ */
    public void onIdleExpired(long idleForMs)
    {
        try
        {
            synchronized (this)
            {
                _onIdle=true;
            }

            _connection.onIdleExpired(idleForMs);
        }
        finally
        {
            synchronized (this)
            {
                _onIdle=false;
                if (_state==STATE_NEEDS_DISPATCH)
                    dispatch();
            }
        }
    }

    /* ------------------------------------------------------------ */
    @Override
    public int fill(Buffer buffer) throws IOException
    {
        int fill=super.fill(buffer);
        if (fill>0)
            notIdle();
        return fill;
    }

    /* ------------------------------------------------------------ */
    @Override
    public int flush(Buffer header, Buffer buffer, Buffer trailer) throws IOException
    {
        int l = super.flush(header, buffer, trailer);

        // If there was something to write and it wasn't written, then we are not writable.
        if (l==0 && ( header!=null && header.hasContent() || buffer!=null && buffer.hasContent() || trailer!=null && trailer.hasContent()))
        {
            synchronized (this)
            {   
                _writable=false;
                if (_state<STATE_DISPATCHED)
                    updateKey();
            }
        }
        else if (l>0)
        {
            _writable=true;
            notIdle();
        }
        return l;
    }

    /* ------------------------------------------------------------ */
    /*
     */
    @Override
    public int flush(Buffer buffer) throws IOException
    {
        int l = super.flush(buffer);

        // If there was something to write and it wasn't written, then we are not writable.
        if (l==0 && buffer!=null && buffer.hasContent())
        {
            synchronized (this)
            {   
                _writable=false;
                if (_state<STATE_DISPATCHED)
                    updateKey();
            }
        }
        else if (l>0)
        {
            _writable=true;
            notIdle();
        }

        return l;
    }

    /* ------------------------------------------------------------ */
    /*
     * Allows thread to block waiting for further events.
     */
    @Override
    public boolean blockReadable(long timeoutMs) throws IOException
    {
        synchronized (this)
        {
            if (isInputShutdown())
                throw new EofException();

            long now=_selectSet.getNow();
            long end=now+timeoutMs;
            boolean check=isCheckForIdle();
            setCheckForIdle(true);
            try
            {
                _readBlocked=true;
                while (!isInputShutdown() && _readBlocked)
                {
                    try
                    {
                        updateKey();
                        this.wait(timeoutMs>0?(end-now):10000);
                    }
                    catch (final InterruptedException e)
                    {
                        LOG.warn(e);
                        if (_interruptable)
                            throw new InterruptedIOException(){{this.initCause(e);}};
                    }
                    finally
                    {
                        now=_selectSet.getNow();
                    }

                    if (_readBlocked && timeoutMs>0 && now>=end)
                        return false;
                }
            }
            finally
            {
                _readBlocked=false;
                setCheckForIdle(check);
            }
        }
        return true;
    }

    /* ------------------------------------------------------------ */
    /*
     * Allows thread to block waiting for further events.
     */
    @Override
    public boolean blockWritable(long timeoutMs) throws IOException
    {
        synchronized (this)
        {
            if (isOutputShutdown())
                throw new EofException();

            long now=_selectSet.getNow();
            long end=now+timeoutMs;
            boolean check=isCheckForIdle();
            setCheckForIdle(true);
            try
            {
                _writeBlocked=true;
                while (_writeBlocked && !isOutputShutdown())
                {
                    try
                    {
                        updateKey();
                        this.wait(timeoutMs>0?(end-now):10000);
                    }
                    catch (final InterruptedException e)
                    {
                        LOG.warn(e);
                        if (_interruptable)
                            throw new InterruptedIOException(){{this.initCause(e);}};
                    }
                    finally
                    {
                        now=_selectSet.getNow();
                    }
                    if (_writeBlocked && timeoutMs>0 && now>=end)
                        return false;
                }
            }
            finally
            {
                _writeBlocked=false;
                setCheckForIdle(check);
            }
        }
        return true;
    }

    /* ------------------------------------------------------------ */
    /** Set the interruptable mode of the endpoint.
     * If set to false (default), then interrupts are assumed to be spurious 
     * and blocking operations continue unless the endpoint has been closed.
     * If true, then interrupts of blocking operations result in InterruptedIOExceptions
     * being thrown.
     * @param interupable
     */
    public void setInterruptable(boolean interupable)
    {
        synchronized (this)
        {
            _interruptable=interupable;
        }
    }

    /* ------------------------------------------------------------ */
    public boolean isInterruptable()
    {
        return _interruptable;
    }
    
    /* ------------------------------------------------------------ */
    /**
     * @see org.eclipse.jetty.io.AsyncEndPoint#scheduleWrite()
     */
    public void scheduleWrite()
    {
        if (_writable)
            LOG.debug("Required scheduleWrite {}",this);

        _writable=false;
        updateKey();
    }

    /* ------------------------------------------------------------ */
    public boolean isWritable()
    {
        return _writable;
    }

    /* ------------------------------------------------------------ */
    public boolean hasProgressed()
    {
        return false;
    }

    /* ------------------------------------------------------------ */
    /**
     * Updates selection key. Adds operations types to the selection key as needed. No operations
     * are removed as this is only done during dispatch. This method records the new key and
     * schedules a call to doUpdateKey to do the keyChange
     */
    private void updateKey()
    {
        final boolean changed;
        synchronized (this)
        {
            int current_ops=-1;
            if (getChannel().isOpen())
            {
                boolean read_interest = _readBlocked || (_state<STATE_DISPATCHED && !_connection.isSuspended());
                boolean write_interest= _writeBlocked || (_state<STATE_DISPATCHED && !_writable);

                _interestOps =
                    ((!_socket.isInputShutdown() && read_interest ) ? SelectionKey.OP_READ  : 0)
                |   ((!_socket.isOutputShutdown()&& write_interest) ? SelectionKey.OP_WRITE : 0);
                try
                {
                    current_ops = ((_key!=null && _key.isValid())?_key.interestOps():-1);
                }
                catch(Exception e)
                {
                    _key=null;
                    LOG.ignore(e);
                }
            }
            changed=_interestOps!=current_ops;
        }

        if(changed)
        {
            _selectSet.addChange(this);
            _selectSet.wakeup();
        }
    }


    /* ------------------------------------------------------------ */
    /**
     * Synchronize the interestOps with the actual key. Call is scheduled by a call to updateKey
     */
    void doUpdateKey()
    {
        synchronized (this)
        {
            if (getChannel().isOpen())
            {
                if (_interestOps>0)
                {
                    if (_key==null || !_key.isValid())
                    {
                        SelectableChannel sc = (SelectableChannel)getChannel();
                        if (sc.isRegistered())
                        {
                            updateKey();
                        }
                        else
                        {
                            try
                            {
                                _key=((SelectableChannel)getChannel()).register(_selectSet.getSelector(),_interestOps,this);
                            }
                            catch (Exception e)
                            {
                                LOG.ignore(e);
                                if (_key!=null && _key.isValid())
                                {
                                    _key.cancel();
                                }

                                if (_open)
                                {
                                    _selectSet.destroyEndPoint(this);
                                }
                                _open=false;
                                _key = null;
                            }
                        }
                    }
                    else
                    {
                        _key.interestOps(_interestOps);
                    }
                }
                else
                {
                    if (_key!=null && _key.isValid())
                        _key.interestOps(0);
                    else
                        _key=null;
                }
            }
            else
            {
                if (_key!=null && _key.isValid())
                    _key.cancel();

                if (_open)
                {
                    _open=false;
                    _selectSet.destroyEndPoint(this);
                }
                _key = null;
            }
        }
    }

    /* ------------------------------------------------------------ */
    /*
     */
    protected void handle()
    {
        boolean dispatched=true;
        try
        {
            while(dispatched)
            {
                try
                {
                    while(true)
                    {
                        final AsyncConnection next = (AsyncConnection)_connection.handle();
                        if (next!=_connection)
                        {
                            LOG.debug("{} replaced {}",next,_connection);
                            Connection old=_connection;
                            _connection=next;
                            _manager.endPointUpgraded(this,old);
                            continue;
                        }
                        break;
                    }
                }
                catch (ClosedChannelException e)
                {
                    LOG.ignore(e);
                }
                catch (EofException e)
                {
                    LOG.debug("EOF", e);
                    try{close();}
                    catch(IOException e2){LOG.ignore(e2);}
                }
                catch (IOException e)
                {
                    LOG.warn(e.toString());
                    try{close();}
                    catch(IOException e2){LOG.ignore(e2);}
                }
                catch (Throwable e)
                {
                    LOG.warn("handle failed", e);
                    try{close();}
                    catch(IOException e2){LOG.ignore(e2);}
                }
                finally
                {
                    if (!_ishut && isInputShutdown() && isOpen())
                    {
                        _ishut=true;
                        try
                        {
                            _connection.onInputShutdown();
                        }
                        catch(Throwable x)
                        {
                            LOG.warn("onInputShutdown failed", x);
                            try{close();}
                            catch(IOException e2){LOG.ignore(e2);}
                        }
                        finally
                        {
                            updateKey();
                        }
                    }
                    dispatched=!undispatch();
                }
            }
        }
        finally
        {
            if (dispatched)
            {
                dispatched=!undispatch();
                while (dispatched)
                {
                    LOG.warn("SCEP.run() finally DISPATCHED");
                    dispatched=!undispatch();
                }
            }
        }
    }

    /* ------------------------------------------------------------ */
    /*
     * @see org.eclipse.io.nio.ChannelEndPoint#close()
     */
    @Override
    public void close() throws IOException
    {
        // On unix systems there is a JVM issue that if you cancel before closing, it can 
        // cause the selector to block waiting for a channel to close and that channel can 
        // block waiting for the remote end.  But on windows, if you don't cancel before a 
        // close, then the selector can block anyway!
        // https://bugs.eclipse.org/bugs/show_bug.cgi?id=357318
        if (WORK_AROUND_JVM_BUG_6346658)
        {
            try
            {
                SelectionKey key = _key;
                if (key!=null)
                    key.cancel();
            }
            catch (Throwable e)
            {
                LOG.ignore(e);
            }
        }

        try
        {
            super.close();
        }
        catch (IOException e)
        {
            LOG.ignore(e);
        }
        finally
        {
            updateKey();
        }
    }

    /* ------------------------------------------------------------ */
    @Override
    public String toString()
    {
        // Do NOT use synchronized (this)
        // because it's very easy to deadlock when debugging is enabled.
        // We do a best effort to print the right toString() and that's it.
        SelectionKey key = _key;
        String keyString = "";
        if (key != null)
        {
            if (key.isValid())
            {
                if (key.isReadable())
                    keyString += "r";
                if (key.isWritable())
                    keyString += "w";
            }
            else
            {
                keyString += "!";
            }
        }
        else
        {
            keyString += "-";
        }
        return String.format("SCEP@%x{l(%s)<->r(%s),s=%d,open=%b,ishut=%b,oshut=%b,rb=%b,wb=%b,w=%b,i=%d%s}-{%s}",
                hashCode(),
                _socket.getRemoteSocketAddress(),
                _socket.getLocalSocketAddress(),
                _state,
                isOpen(),
                isInputShutdown(),
                isOutputShutdown(),
                _readBlocked,
                _writeBlocked,
                _writable,
                _interestOps,
                keyString,
                _connection);
    }

    /* ------------------------------------------------------------ */
    public SelectSet getSelectSet()
    {
        return _selectSet;
    }

    /* ------------------------------------------------------------ */
    /**
     * Don't set the SoTimeout
     * @see org.eclipse.jetty.io.nio.ChannelEndPoint#setMaxIdleTime(int)
     */
    @Override
    public void setMaxIdleTime(int timeMs) throws IOException
    {
        _maxIdleTime=timeMs;
    }

}