connectiontablenio.java

来自「JGRoups源码」· Java 代码 · 共 1,494 行 · 第 1/4 页

       if(reaper != null)
           reaper.stop();

      // Stop the main selector
      m_acceptSelector.wakeup();

      // Stop selector threads
      for (int i = 0; i < m_readHandlers.length; i++)
      {
         try
         {
            m_readHandlers[i].add(new Shutdown());
         } catch (InterruptedException e)
         {
            LOG.error("Thread ("+Thread.currentThread().getName() +") was interrupted, failed to shutdown selector", e);
         }
      }
      for (int i = 0; i < m_writeHandlers.length; i++)
      {
         try
         {
            m_writeHandlers[i].QUEUE.put(new Shutdown());
            m_writeHandlers[i].SELECTOR.wakeup();
         } catch (InterruptedException e)
         {
            LOG.error("Thread ("+Thread.currentThread().getName() +") was interrupted, failed to shutdown selector", e);
         }
      }

      // Stop the callback thread pool
      if(m_requestProcessors instanceof PooledExecutor)
         ((PooledExecutor)m_requestProcessors).shutdownNow();

       if(m_requestProcessors instanceof PooledExecutor) {
           try {
               ((PooledExecutor)m_requestProcessors).awaitTerminationAfterShutdown(1000);
           }
           catch(InterruptedException e) {
           }
       }

      // then close the connections
      synchronized(conns) {
          Iterator it=conns.values().iterator();
          while(it.hasNext()) {
              Connection conn=(Connection)it.next();
              conn.destroy();
          }
          conns.clear();
      }

      while(m_backGroundThreads.size() > 0) {
          Thread t = (Thread)m_backGroundThreads.removeFirst();
          try {
            t.join();
          } catch(InterruptedException e) {
            LOG.error("Thread ("+Thread.currentThread().getName() +") was interrupted while waiting on thread " + t.getName() + " to finish.");              
          }
      }
      m_backGroundThreads.clear();

   }

   /**
    * Acceptor thread. Continuously accept new connections and assign readhandler/writehandler
    * to them.
    */
   public void run()
   {
      Connection conn;

      while (m_serverSocketChannel.isOpen() && !serverStopping)
      {
         int num;
         try
         {
            num = m_acceptSelector.select();
         } catch (IOException e)
         {
            if (LOG.isWarnEnabled())
               LOG.warn("Select operation on listening socket failed", e);
            continue;   // Give up this time
         }

         if (num > 0)
         {
            Set readyKeys = m_acceptSelector.selectedKeys();
            for (Iterator i = readyKeys.iterator(); i.hasNext();)
            {
               SelectionKey key = (SelectionKey) i.next();
               i.remove();
               // We only deal with new incoming connections

               ServerSocketChannel readyChannel = (ServerSocketChannel) key.channel();
               SocketChannel client_sock_ch;
               try
               {
                  client_sock_ch = readyChannel.accept();
               } catch (IOException e)
               {
                  if (LOG.isWarnEnabled())
                     LOG.warn("Attempt to accept new connection from listening socket failed" , e);
                  // Give up this connection
                  continue;
               }

               if (LOG.isInfoEnabled())
                  LOG.info("accepted connection, client_sock=" + client_sock_ch.socket());

               try {

                  if (LOG.isTraceEnabled())
                     LOG.trace("About to change new connection send buff size from " + client_sock_ch.socket().getSendBufferSize() + " bytes");
                  client_sock_ch.socket().setSendBufferSize(send_buf_size);
                  if (LOG.isTraceEnabled())
                     LOG.trace("Changed new connection send buff size to " + client_sock_ch.socket().getSendBufferSize() + " bytes");
               }
               catch (IllegalArgumentException ex)
               {
                  if (log.isErrorEnabled()) log.error("exception setting send buffer size to " +
                     send_buf_size + " bytes: " ,ex);
               }
               catch (SocketException e)
               {
                  if (log.isErrorEnabled()) log.error("exception setting send buffer size to " +
                     send_buf_size + " bytes: " , e);
               }

               try
               {
                  if (LOG.isTraceEnabled())
                     LOG.trace("About to change new connection receive buff size from " + client_sock_ch.socket().getReceiveBufferSize() + " bytes");
                  client_sock_ch.socket().setReceiveBufferSize(recv_buf_size);
                  if (LOG.isTraceEnabled())
                     LOG.trace("Changed new connection receive buff size to " + client_sock_ch.socket().getReceiveBufferSize() + " bytes");
               }
               catch (IllegalArgumentException ex)
               {
                  if (log.isErrorEnabled()) log.error("exception setting receive buffer size to " +
                     send_buf_size + " bytes: " , ex);
               }
               catch (SocketException e)
               {
                  if (log.isErrorEnabled()) log.error("exception setting receive buffer size to " +
                     recv_buf_size + " bytes: " , e);
               }

               conn = new Connection(client_sock_ch, null);
               try
               {
                  conn.peer_addr = conn.readPeerAddress(client_sock_ch.socket());

                  synchronized (conns)
                  {
                     if (conns.containsKey(conn.getPeerAddress()))
                     {
                        if (conn.getPeerAddress().equals(getLocalAddress()))
                        {
                           if (LOG.isTraceEnabled())
                              LOG.trace(conn.getPeerAddress() + " is myself, not put it in table twice, but still read from it");
                        } else
                        {
                           if (LOG.isWarnEnabled())
                              LOG.warn(conn.getPeerAddress() + " is already there, will terminate connection");
                           // keep existing connection, close this new one
                           conn.destroy();
                           continue;
                        }
                     } else {
                        addConnection(conn.getPeerAddress(), conn);
                     }
                  }
                  notifyConnectionOpened(conn.getPeerAddress());
                  client_sock_ch.configureBlocking(false);
               }
               catch (IOException e)
               {
                  if (LOG.isWarnEnabled())
                     LOG.warn("Attempt to configure non-blocking mode failed", e);
                  // Give up this connection
                  conn.destroy();
                  continue;
               }
               catch (Exception e)
               {
                  if (LOG.isWarnEnabled())
                     LOG.warn("Attempt to handshake with other peer failed", e);
                  // Give up this connection
                  conn.destroy();
                  continue;
               }

               int idx;
               synchronized (m_lockNextWriteHandler)
               {
                  idx = m_nextWriteHandler = (m_nextWriteHandler + 1) % m_writeHandlers.length;
               }
               conn.setupWriteHandler(m_writeHandlers[idx]);

               try
               {
                  synchronized (m_lockNextReadHandler)
                  {
                     idx = m_nextReadHandler = (m_nextReadHandler + 1) % m_readHandlers.length;
                  }
                  m_readHandlers[idx].add(conn);

               } catch (InterruptedException e)
               {
                  if (LOG.isWarnEnabled())
                     LOG.warn("Attempt to configure read handler for accepted connection failed" , e);
                  // close connection
                  conn.destroy();
               }
            }   // end of iteration
         }   // end of selected key > 0
      }   // end of thread

      if (m_serverSocketChannel.isOpen())
      {
         try
         {
            m_serverSocketChannel.close();
         }
         catch (Exception e)
         {
            log.error("exception closing server listening socket", e);
         }
      }
      if (LOG.isTraceEnabled())
         LOG.trace("acceptor thread terminated");

   }


   /**
    * Finds first available port starting at start_port and returns server socket. Sets srv_port
    */
   protected ServerSocket createServerSocket(int start_port, int end_port) throws Exception
   {
      this.m_acceptSelector = Selector.open();
      m_serverSocketChannel = ServerSocketChannel.open();
      m_serverSocketChannel.configureBlocking(false);
      while (true)
      {
         try
         {
            SocketAddress sockAddr;
            if (bind_addr == null)
            {
               sockAddr=new InetSocketAddress(start_port);
               m_serverSocketChannel.socket().bind(sockAddr);
            }
            else
            {
               sockAddr=new InetSocketAddress(bind_addr, start_port);
               m_serverSocketChannel.socket().bind(sockAddr, backlog);
            }
         }
         catch (BindException bind_ex)
         {
            if (start_port == end_port)
               throw (BindException) ((new BindException("No available port to bind to")).initCause(bind_ex));
            start_port++;
            continue;
         }
         catch (SocketException bind_ex)
         {
            if (start_port == end_port)
               throw (BindException) ((new BindException("No available port to bind to")).initCause(bind_ex));
            start_port++;
            continue;
         }
         catch (IOException io_ex)
         {
            if (LOG.isErrorEnabled()) LOG.error("Attempt to bind serversocket failed, port="+start_port+", bind addr=" + bind_addr ,io_ex);
            throw io_ex;
         }
         srv_port = start_port;
         break;
      }
      m_serverSocketChannel.register(this.m_acceptSelector, SelectionKey.OP_ACCEPT);
      return m_serverSocketChannel.socket();
   }

   protected void runRequest(Address addr, ByteBuffer buf) throws InterruptedException {
      m_requestProcessors.execute(new ExecuteTask(addr, buf));
   }


   // Represents shutdown
   private static class Shutdown {
   }

   // ReadHandler has selector to deal with read, it runs in seperated thread
   private static class ReadHandler implements Runnable {
      private final Selector SELECTOR = initHandler();
      private final LinkedQueue QUEUE = new LinkedQueue();
      private final ConnectionTableNIO connectTable;

      ReadHandler(ConnectionTableNIO ct) {
         connectTable= ct;
      }

      public Selector initHandler()
      {
         // Open the selector
         try
         {
            return Selector.open();
         } catch (IOException e)
         {
            if (LOG.isErrorEnabled()) LOG.error(e);
            throw new IllegalStateException(e.getMessage());
         }

      }

      /**
       * create instances of ReadHandler threads for receiving data.
       *
       * @param workerThreads is the number of threads to create.
       */
      private static ReadHandler[] create(int workerThreads, ConnectionTableNIO ct, ThreadGroup tg, LinkedList backGroundThreads)
      {
         ReadHandler[] handlers = new ReadHandler[workerThreads];
         for (int looper = 0; looper < workerThreads; looper++)
         {
            handlers[looper] = new ReadHandler(ct);

            Thread thread = new Thread(tg, handlers[looper], "nioReadHandlerThread");
            thread.setDaemon(true);
            thread.start();
            backGroundThreads.add(thread);
         }
         return handlers;
      }


      private void add(Object conn) throws InterruptedException
      {
         QUEUE.put(conn);
         wakeup();
      }

      private void wakeup()
      {
         SELECTOR.wakeup();
      }

      public void run()
      {
         while (true)
         {  // m_s can be closed by the management thread
            int events;
            try
            {
               events = SELECTOR.select();
            } catch (IOException e)
            {
               if (LOG.isWarnEnabled())
                  LOG.warn("Select operation on socket failed", e);
               continue;   // Give up this time
            } catch (ClosedSelectorException e)
            {
               if (LOG.isWarnEnabled())
                  LOG.warn("Select operation on socket failed" , e);
               return;     // Selector gets closed, thread stops
            }

            if (events > 0)
            {   // there are read-ready channels
               Set readyKeys = SELECTOR.selectedKeys();
               for (Iterator i = readyKeys.iterator(); i.hasNext();)