org/sonews/daemon/ChannelReader.java
changeset 1 6fceb66e1ad7
child 3 2fdc9cc89502
     1.1 --- /dev/null	Thu Jan 01 00:00:00 1970 +0000
     1.2 +++ b/org/sonews/daemon/ChannelReader.java	Fri Jun 26 16:48:50 2009 +0200
     1.3 @@ -0,0 +1,202 @@
     1.4 +/*
     1.5 + *   SONEWS News Server
     1.6 + *   see AUTHORS for the list of contributors
     1.7 + *
     1.8 + *   This program is free software: you can redistribute it and/or modify
     1.9 + *   it under the terms of the GNU General Public License as published by
    1.10 + *   the Free Software Foundation, either version 3 of the License, or
    1.11 + *   (at your option) any later version.
    1.12 + *
    1.13 + *   This program is distributed in the hope that it will be useful,
    1.14 + *   but WITHOUT ANY WARRANTY; without even the implied warranty of
    1.15 + *   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
    1.16 + *   GNU General Public License for more details.
    1.17 + *
    1.18 + *   You should have received a copy of the GNU General Public License
    1.19 + *   along with this program.  If not, see <http://www.gnu.org/licenses/>.
    1.20 + */
    1.21 +
    1.22 +package org.sonews.daemon;
    1.23 +
    1.24 +import org.sonews.util.Log;
    1.25 +import java.io.IOException;
    1.26 +import java.nio.ByteBuffer;
    1.27 +import java.nio.channels.CancelledKeyException;
    1.28 +import java.nio.channels.SelectionKey;
    1.29 +import java.nio.channels.Selector;
    1.30 +import java.nio.channels.SocketChannel;
    1.31 +import java.util.Iterator;
    1.32 +import java.util.Set;
    1.33 +
    1.34 +/**
    1.35 + * A Thread task listening for OP_READ events from SocketChannels.
    1.36 + * @author Christian Lins
    1.37 + * @since sonews/0.5.0
    1.38 + */
    1.39 +class ChannelReader extends AbstractDaemon
    1.40 +{
    1.41 +
    1.42 +  private static ChannelReader instance = new ChannelReader();
    1.43 +
    1.44 +  /**
    1.45 +   * @return Active ChannelReader instance.
    1.46 +   */
    1.47 +  public static ChannelReader getInstance()
    1.48 +  {
    1.49 +    return instance;
    1.50 +  }
    1.51 +  
    1.52 +  private Selector selector = null;
    1.53 +  
    1.54 +  protected ChannelReader()
    1.55 +  {
    1.56 +  }
    1.57 +  
    1.58 +  /**
    1.59 +   * Sets the selector which is used by this reader to determine the channel
    1.60 +   * to read from.
    1.61 +   * @param selector
    1.62 +   */
    1.63 +  public void setSelector(final Selector selector)
    1.64 +  {
    1.65 +    this.selector = selector;
    1.66 +  }
    1.67 +  
    1.68 +  /**
    1.69 +   * Run loop. Blocks until some data is available in a channel.
    1.70 +   */
    1.71 +  @Override
    1.72 +  public void run()
    1.73 +  {
    1.74 +    assert selector != null;
    1.75 +
    1.76 +    while(isRunning())
    1.77 +    {
    1.78 +      try
    1.79 +      {
    1.80 +        // select() blocks until some SelectableChannels are ready for
    1.81 +        // processing. There is no need to lock the selector as we have only
    1.82 +        // one thread per selector.
    1.83 +        selector.select();
    1.84 +
    1.85 +        // Get list of selection keys with pending events.
    1.86 +        // Note: the selected key set is not thread-safe
    1.87 +        SocketChannel channel = null;
    1.88 +        NNTPConnection conn = null;
    1.89 +        final Set<SelectionKey> selKeys = selector.selectedKeys();
    1.90 +        SelectionKey selKey = null;
    1.91 +
    1.92 +        synchronized (selKeys)
    1.93 +        {
    1.94 +          Iterator it = selKeys.iterator();
    1.95 +
    1.96 +          // Process the first pending event
    1.97 +          while (it.hasNext())
    1.98 +          {
    1.99 +            selKey = (SelectionKey) it.next();
   1.100 +            channel = (SocketChannel) selKey.channel();
   1.101 +            conn = Connections.getInstance().get(channel);
   1.102 +
   1.103 +            // Because we cannot lock the selKey as that would cause a deadlock
   1.104 +            // we lock the connection. To preserve the order of the received
   1.105 +            // byte blocks a selection key for a connection that has pending
   1.106 +            // read events is skipped.
   1.107 +            if (conn == null || conn.tryReadLock())
   1.108 +            {
   1.109 +              // Remove from set to indicate that it's being processed
   1.110 +              it.remove();
   1.111 +              if (conn != null)
   1.112 +              {
   1.113 +                break; // End while loop
   1.114 +              }
   1.115 +            }
   1.116 +            else
   1.117 +            {
   1.118 +              selKey = null;
   1.119 +              channel = null;
   1.120 +              conn = null;
   1.121 +            }
   1.122 +          }
   1.123 +        }
   1.124 +
   1.125 +        // Do not lock the selKeys while processing because this causes
   1.126 +        // a deadlock in sun.nio.ch.SelectorImpl.lockAndDoSelect()
   1.127 +        if (selKey != null && channel != null && conn != null)
   1.128 +        {
   1.129 +          processSelectionKey(conn, channel, selKey);
   1.130 +          conn.unlockReadLock();
   1.131 +        }
   1.132 +
   1.133 +      }
   1.134 +      catch(CancelledKeyException ex)
   1.135 +      {
   1.136 +        Log.msg("ChannelReader.run(): " + ex, false);
   1.137 +        if(Log.isDebug())
   1.138 +        {
   1.139 +          ex.printStackTrace();
   1.140 +        }
   1.141 +      }
   1.142 +      catch(Exception ex)
   1.143 +      {
   1.144 +        ex.printStackTrace();
   1.145 +      }
   1.146 +      
   1.147 +      // Eventually wait for a register operation
   1.148 +      synchronized (NNTPDaemon.RegisterGate)
   1.149 +      {
   1.150 +      // Do nothing; FindBugs may warn about an empty synchronized 
   1.151 +      // statement, but we cannot use a wait()/notify() mechanism here.
   1.152 +      // If we used something like RegisterGate.wait() we block here
   1.153 +      // until the NNTPDaemon calls notify(). But the daemon only
   1.154 +      // calls notify() if itself is NOT blocked in the listening socket.
   1.155 +      }
   1.156 +    } // while(isRunning())
   1.157 +  }
   1.158 +  
   1.159 +  private void processSelectionKey(final NNTPConnection connection,
   1.160 +    final SocketChannel socketChannel, final SelectionKey selKey)
   1.161 +    throws InterruptedException, IOException
   1.162 +  {
   1.163 +    assert selKey != null;
   1.164 +    assert selKey.isReadable();
   1.165 +    
   1.166 +    // Some bytes are available for reading
   1.167 +    if(selKey.isValid())
   1.168 +    {      
   1.169 +      // Lock the channel
   1.170 +      //synchronized(socketChannel)
   1.171 +      {
   1.172 +        // Read the data into the appropriate buffer
   1.173 +        ByteBuffer buf = connection.getInputBuffer();
   1.174 +        int read = -1;
   1.175 +        try 
   1.176 +        {
   1.177 +          read = socketChannel.read(buf);
   1.178 +        } 
   1.179 +        catch(Exception ex) 
   1.180 +        {
   1.181 +          Log.msg("ChannelReader.processSelectionKey(): " + ex, false);
   1.182 +          if(Log.isDebug())
   1.183 +          {
   1.184 +            ex.printStackTrace();
   1.185 +          }
   1.186 +        }
   1.187 +        
   1.188 +        if(read == -1) // End of stream
   1.189 +        {
   1.190 +          selKey.cancel();
   1.191 +        }
   1.192 +        else if(read > 0) // If some data was read
   1.193 +        {
   1.194 +          ConnectionWorker.addChannel(socketChannel);
   1.195 +        }
   1.196 +      }
   1.197 +    }
   1.198 +    else
   1.199 +    {
   1.200 +      // Should not happen
   1.201 +      Log.msg(selKey, false);
   1.202 +    }
   1.203 +  }
   1.204 +  
   1.205 +}