basicconnectiontable.java

来自「JGRoups源码」· Java 代码 · 共 783 行 · 第 1/2 页

                   tmp.join(MAX_JOIN_TIMEOUT);
               }
               catch(InterruptedException e) {
               }
               if(tmp.isAlive()) {
                   if(log.isWarnEnabled())
                   log.warn("stopped receiver thread, but thread (" + tmp + ") is still alive !");
               }
           }
       }


       /**
        *
        * @param data Guaranteed to be non null
        * @param offset
        * @param length
        */
       void send(byte[] data, int offset, int length) {
           if(!is_running) {
               if(log.isWarnEnabled())
                   log.warn("Connection is not running, discarding message");
               return;
           }
           if(use_send_queues) {
               try {
                   // we need to copy the byte[] buffer here because the original buffer might get changed meanwhile
                   byte[] tmp=new byte[length];
                   System.arraycopy(data, offset, tmp, 0, length);
                   send_queue.add(tmp);
                   if(!sender.isRunning())
                       sender.start();
               }
               catch(QueueClosedException e) {
                   log.error("failed adding message to send_queue", e);
               }
           }
           else
               _send(data, offset, length);
       }


       private void _send(byte[] data, int offset, int length) {
           synchronized(send_mutex) {
               try {
                   doSend(data, offset, length);
                   updateLastAccessed();
               }
               catch(IOException io_ex) {
                   if(log.isWarnEnabled())
                       log.warn("peer closed connection, trying to re-send msg");
                   try {
                       doSend(data, offset, length);
                       updateLastAccessed();
                   }
                   catch(IOException io_ex2) {
                       if(log.isErrorEnabled()) log.error("2nd attempt to send data failed too");
                   }
                   catch(Exception ex2) {
                       if(log.isErrorEnabled()) log.error("exception is " + ex2);
                   }
               }
               catch(InterruptedException iex) {}
               catch(Throwable ex) {
                   if(log.isErrorEnabled()) log.error("exception is " + ex);
               }
           }
       }


       void doSend(byte[] data, int offset, int length) throws Exception {
           try {
               // we're using 'double-writes', sending the buffer to the destination in 2 pieces. this would
               // ensure that, if the peer closed the connection while we were idle, we would get an exception.
               // this won't happen if we use a single write (see Stevens, ch. 5.13).
               if(out != null) {
                   out.writeInt(length); // write the length of the data buffer first
                   Util.doubleWrite(data, offset, length, out);
                   out.flush();  // may not be very efficient (but safe)
               }
           }
           catch(Exception ex) {
               remove(peer_addr);
               throw ex;
           }
       }


       /**
        * Reads the peer's address. First a cookie has to be sent which has to match my own cookie, otherwise
        * the connection will be refused
        */
       Address readPeerAddress(Socket client_sock) throws Exception {
           Address     client_peer_addr=null;
           byte[]      input_cookie=new byte[cookie.length];
           int         client_port=client_sock != null? client_sock.getPort() : 0;
           short       version;
           InetAddress client_addr=client_sock != null? client_sock.getInetAddress() : null;

           if(in != null) {
               initCookie(input_cookie);

               // read the cookie first
               in.read(input_cookie, 0, input_cookie.length);
               if(!matchCookie(input_cookie))
                   throw new SocketException("ConnectionTable.Connection.readPeerAddress(): cookie sent by " +
                                             client_peer_addr + " does not match own cookie; terminating connection");
               // then read the version
               version=in.readShort();

               if(Version.compareTo(version) == false) {
                   if(log.isWarnEnabled())
                       log.warn(new StringBuffer("packet from ").append(client_addr).append(':').append(client_port).
                              append(" has different version (").append(version).append(") from ours (").
                                append(Version.version).append("). This may cause problems"));
               }
               client_peer_addr=new IpAddress();
               client_peer_addr.readFrom(in);

               updateLastAccessed();
           }
           return client_peer_addr;
       }


       /**
        * Send the cookie first, then the our port number. If the cookie doesn't match the receiver's cookie,
        * the receiver will reject the connection and close it.
        */
       void sendLocalAddress(Address local_addr) {
           if(local_addr == null) {
               if(log.isWarnEnabled()) log.warn("local_addr is null");
               return;
           }
           if(out != null) {
               try {
                   // write the cookie
                   out.write(cookie, 0, cookie.length);

                   // write the version
                   out.writeShort(Version.version);
                   local_addr.writeTo(out);
                   out.flush(); // needed ?
                   updateLastAccessed();
               }
               catch(Throwable t) {
                   if(log.isErrorEnabled()) log.error("exception is " + t);
               }
           }
       }


       void initCookie(byte[] c) {
           if(c != null)
               for(int i=0; i < c.length; i++)
                   c[i]=0;
       }

       boolean matchCookie(byte[] input) {
           if(input == null || input.length < cookie.length) return false;
           for(int i=0; i < cookie.length; i++)
               if(cookie[i] != input[i]) return false;
           return true;
       }


       String printCookie(byte[] c) {
           if(c == null) return "";
           return new String(c);
       }


       public void run() {
           byte[] buf=new byte[256]; // start with 256, increase as we go
           int len=0;

           while(receiverThread != null && receiverThread.equals(Thread.currentThread()) && is_running) {
               try {
                   if(in == null) {
                       if(log.isErrorEnabled()) log.error("input stream is null !");
                       break;
                   }
                   len=in.readInt();
                   if(len > buf.length)
                       buf=new byte[len];
                   in.readFully(buf, 0, len);
                   updateLastAccessed();
                   receive(peer_addr, buf, 0, len); // calls receiver.receive(msg)
               }
               catch(OutOfMemoryError mem_ex) {
                   if(log.isWarnEnabled()) log.warn("dropped invalid message, closing connection");
                   break; // continue;
               }
               catch(EOFException eof_ex) {  // peer closed connection
                   if(log.isTraceEnabled()) log.trace("exception is " + eof_ex);
                   notifyConnectionClosed(peer_addr);
                   break;
               }
               catch(IOException io_ex) {
                   if(log.isTraceEnabled()) log.trace("exception is " + io_ex);
                   notifyConnectionClosed(peer_addr);
                   break;
               }
               catch(Throwable e) {
                   if(log.isWarnEnabled()) log.warn("exception is " + e);
               }
           }
           if(log.isTraceEnabled())
               log.trace("ConnectionTable.Connection.Receiver terminated");
           receiverThread=null;
           closeSocket();
           // remove(peer_addr);
       }


       public String toString() {
           StringBuffer ret=new StringBuffer();
           InetAddress local=null, remote=null;
           String local_str, remote_str;

           if(sock == null)
               ret.append("<null socket>");
           else {
               //since the sock variable gets set to null we want to make
               //make sure we make it through here without a nullpointer exception
               Socket tmp_sock=sock;
               local=tmp_sock.getLocalAddress();
               remote=tmp_sock.getInetAddress();
               local_str=local != null ? Util.shortName(local) : "<null>";
               remote_str=remote != null ? Util.shortName(remote) : "<null>";
               ret.append('<' + local_str + ':' + tmp_sock.getLocalPort() +
                          " --> " + remote_str + ':' + tmp_sock.getPort() + "> (" +
                          ((System.currentTimeMillis() - last_access) / 1000) + " secs old)");
               tmp_sock=null;
           }

           return ret.toString();
       }


       void closeSocket() {
           Util.close(sock); // should actually close in/out (so we don't need to close them explicitly)
           sock=null;
           Util.close(out);  // flushes data
           // removed 4/22/2003 (request by Roland Kurmann)
           // out=null;
           Util.close(in);
       }


       class Sender implements Runnable {
           Thread senderThread;
           private boolean is_it_running=false;

           void start() {
               if(senderThread == null || !senderThread.isAlive()) {
                   senderThread=new Thread(thread_group, this, "ConnectionTable.Connection.Sender [" + getSockAddress() + "]");
                   senderThread.setDaemon(true);
                   senderThread.start();
                   is_it_running=true;
                   if(log.isTraceEnabled())
                       log.trace("ConnectionTable.Connection.Sender thread started");
               }
           }

           void stop() {
               is_it_running=false;
               if(send_queue != null)
                   send_queue.close(false);
               if(senderThread != null) {
                   Thread tmp=senderThread;
                   senderThread=null;
                   tmp.interrupt();
                   try {
                       tmp.join(MAX_JOIN_TIMEOUT);
                   }
                   catch(InterruptedException e) {
                   }
                   if(tmp.isAlive()) {
                       if(log.isWarnEnabled())
                           log.warn("sender thread was interrupted, but is still alive: " + tmp);
                   }
               }
           }

           boolean isRunning() {
               return is_it_running && senderThread != null;
           }

           public void run() {
               byte[] data;
               while(senderThread != null && senderThread.equals(Thread.currentThread()) && is_it_running) {
                   try {
                       data=(byte[])send_queue.remove();
                       if(data == null)
                           continue;
                       _send(data, 0, data.length);
                   }
                   catch(QueueClosedException e) {
                       break;
                   }
               }
               is_it_running=false;
               if(log.isTraceEnabled())
                   log.trace("ConnectionTable.Connection.Sender thread terminated");
           }
       }


   }

   class Reaper implements Runnable {
       Thread t=null;

       Reaper() {
           ;
       }

       public void start() {
           if(conns.size() == 0)
               return;
           if(t != null && !t.isAlive())
               t=null;
           if(t == null) {
               //RKU 7.4.2003, put in threadgroup
               t=new Thread(thread_group, this, "ConnectionTable.ReaperThread");
               t.setDaemon(true); // will allow us to terminate if all remaining threads are daemons
               t.start();
           }
       }

       public void stop() {
           Thread tmp=t;
           if(t != null)
               t=null;
           if(tmp != null) {
               tmp.interrupt(); // interrupts the sleep()
               try {
                   tmp.join(MAX_JOIN_TIMEOUT);
               }
               catch(InterruptedException e) {
               }
               if(tmp.isAlive()) {
                   if(log.isWarnEnabled())
                       log.warn("reaper thread was interrupted, but is still alive: " + tmp);
               }
           }
       }


       public boolean isRunning() {
           return t != null;
       }

       public void run() {
           Connection value;
           Map.Entry entry;
           long curr_time;

           if(log.isInfoEnabled()) log.info("connection reaper thread was started. Number of connections=" +
                                            conns.size() + ", reaper_interval=" + reaper_interval + ", conn_expire_time=" +
                                            conn_expire_time);

           while(conns.size() > 0 && t != null && t.equals(Thread.currentThread())) {
               Util.sleep(reaper_interval);
               if(t == null || !Thread.currentThread().equals(t))
                   break;
               synchronized(conns) {
                   curr_time=System.currentTimeMillis();
                   for(Iterator it=conns.entrySet().iterator(); it.hasNext();) {
                       entry=(Map.Entry)it.next();
                       value=(Connection)entry.getValue();
                       if(log.isInfoEnabled()) log.info("connection is " +
                                                        ((curr_time - value.last_access) / 1000) + " seconds old (curr-time=" +
                                                        curr_time + ", last_access=" + value.last_access + ')');
                       if(value.last_access + conn_expire_time < curr_time) {
                           if(log.isInfoEnabled()) log.info("connection " + value +
                                                            " has been idle for too long (conn_expire_time=" + conn_expire_time +
                                                            "), will be removed");
                           value.destroy();
                           it.remove();
                       }
                   }
               }
           }
           if(log.isInfoEnabled()) log.info("reaper terminated");
           t=null;
       }
   }
}