pepeertransportprotocol.java

来自「Azureus is a powerful, full-featured, cr」· Java 代码 · 共 1,913 行 · 第 1/5 页

/*
 * File    : PEPeerTransportProtocol.java
 * Created : 22-Oct-2003
 * By      : stuff
 * 
 * Azureus - a Java Bittorrent client
 *
 * This program is free software; you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation; either version 2 of the License.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details ( see the LICENSE file ).
 *
 * You should have received a copy of the GNU General Public License
 * along with this program; if not, write to the Free Software
 * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
 */
package org.gudy.azureus2.core3.peer.impl.transport;


import java.io.IOException;
import java.net.InetAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SocketChannel;
import java.util.*;

import org.gudy.azureus2.core3.util.*;

import org.gudy.azureus2.core3.disk.*;
import org.gudy.azureus2.core3.logging.LGLogger;
import org.gudy.azureus2.core3.peer.*;
import org.gudy.azureus2.core3.peer.impl.*;
import org.gudy.azureus2.core3.peer.util.*;
import org.gudy.azureus2.core3.config.*;

import com.aelitis.azureus.core.networkmanager.*;
import com.aelitis.azureus.core.peermanager.UploadManager;
import com.aelitis.azureus.core.peermanager.messages.ProtocolMessage;
import com.aelitis.azureus.core.peermanager.messages.bittorrent.*;
import com.aelitis.azureus.core.peermanager.utils.*;
import com.aelitis.azureus.core.proxy.AEProxyFactory;



/**
 * @author Olivier
 *
 */
public abstract class 
PEPeerTransportProtocol
	implements PEPeerTransport, ConnectionOwner
{
	//TODO xx
	// these appear in the plugin interface as well so don't renumber without
	// fixing things up
	public final static byte BT_CHOKED 			= 0;
	public final static byte BT_UNCHOKED 		= 1;
	public final static byte BT_INTERESTED 		= 2;
	public final static byte BT_UNINTERESTED 	= 3;
	public final static byte BT_HAVE 			= 4;
	public final static byte BT_BITFIELD 		= 5;
	public final static byte BT_REQUEST 		= 6;
	public final static byte BT_PIECE 			= 7;
	public final static byte BT_CANCEL 			= 8;
	
  
	private PEPeerControl manager;
	private byte[] id;
	private String ip;
	private String ip_resolved;
	private IPToHostNameResolverRequest	ip_resolver_request;
	
	private int port;
	
	private PEPeerStatsImpl stats;
  private List 		requested 		= new ArrayList();
  private AEMonitor	requested_mon	= new AEMonitor( "PEPeerTransportProtocol:Req" );

  private HashMap data;
  
  private boolean choked_by_other_peer = true;
  private boolean choking_other_peer = true;
  private boolean interested_in_other_peer = false;
  private boolean other_peer_interested_in_me = false;
  private boolean snubbed = false;
  private boolean[] other_peer_has_pieces;
  private boolean seed = false;
  
  private boolean connection_registered = false;
  

  private boolean incoming;
  private volatile boolean closing = false;
  private PEPeerTransportProtocolState currentState;
  
  private Connection connection;
  private OutgoingBTPieceMessageHandler outgoing_piece_message_handler;
  private OutgoingBTHaveMessageAggregator outgoing_have_message_aggregator;
  
  private boolean identityAdded = false;  //needed so we don't remove id's in closeAll() on duplicate connection attempts
  
  
  private int connection_state = PEPeerTransport.CONNECTION_PENDING;
  
  
  
  //The client name identification	
	private String client = "";

	//Reader inner loop counter
	private int processLoop;
  
  
	//Flag to indicate if the connection is in a stable enough state to send a request.
	//Used to reduce discarded pieces due to request / choke / unchoke / re-request , and both in fact taken into account.
	private boolean readyToRequest;
  
	//Number of bad chunks received from this peer
	private int nbBadChunks;
	
	//When superSeeding, number of unique piece announced
	private int uniquePiece;
	
	//Spread time (0 secs , fake default)
	private int spreadTimeHint = 0 * 1000;

  //TODO xx
	public final static int componentID = 1;
	public final static int evtProtocol = 0;
  public final static int evtLifeCycle = 1;
  public final static int evtErrors = 2;
  
  private int readSleepTime;
  private long lastReadTime;

  private long last_message_sent_time = 0;
  private long last_message_received_time = 0;
  private long last_data_message_received_time = 0;
  
  private long connection_established_time = 0;
  
  
  protected AEMonitor	this_mon	= new AEMonitor( "PEPeerTransportProtocol" );

  private final Map recent_outgoing_requests = new LinkedHashMap( 100, .75F, true ) {
    public boolean removeEldestEntry(Map.Entry eldest) {
      return size() > 100;
    }
  };
  
  private AEMonitor	recent_outgoing_requests_mon	= new AEMonitor( "PEPeerTransportProtocol:ROR" );
  
  private static final boolean SHOW_DISCARD_RATE_STATS;
  
  static {
  	String	prop = System.getProperty( "show.discard.rate.stats" );
  	
  	SHOW_DISCARD_RATE_STATS = prop != null && prop.equals( "1" );
  }
  
  private static int requests_discarded = 0;
  private static int requests_recovered = 0;
  private static int requests_completed = 0;
  
  
  
  private static final boolean	socks_peer_proxy_enable;
  private static final String	socks_version;
  private static final String	socks_host;
  private static  int			socks_port;
  private static final String	socks_user;
  private static final String	socks_password;

  static{
  	socks_peer_proxy_enable	= 	
  		COConfigurationManager.getBooleanParameter("Proxy.Data.Enable", false);
  	
	socks_version =	COConfigurationManager.getStringParameter("Proxy.Data.SOCKS.version" );
		  	
    boolean	socks_same = COConfigurationManager.getBooleanParameter("Proxy.Data.Same", true);

  	socks_host 				= COConfigurationManager.getStringParameter(socks_same?"Proxy.Host":"Proxy.Data.Host");
  	
  	String socks_port_str 	= COConfigurationManager.getStringParameter(socks_same?"Proxy.Port":"Proxy.Data.Port");
  	
  	if ( socks_peer_proxy_enable ){
  		
	  	try{
	  		socks_port = Integer.parseInt( socks_port_str );
	  		
	  	}catch( Throwable e ){
	  		
	  		Debug.printStackTrace(e);
	  	}
  	}else{
  		
  		socks_port	= 0;
  	}
  	
  	socks_user 		= COConfigurationManager.getStringParameter(socks_same?"Proxy.Username":"Proxy.Data.Username");
  	socks_password 	= COConfigurationManager.getStringParameter(socks_same?"Proxy.Password":"Proxy.Data.Password");

  }

  

  /*
	 * This object constructors will let the PeerConnection partially created,
	 * but hopefully will let us gain some memory for peers not going to be
	 * accepted.
	 */

  /**
   * The Default Contructor for outgoing connections.
   * @param manager the manager that will handle this PeerConnection
   * @param ip the peer Ip Address
   * @param port the peer port
   */
	public 
	PEPeerTransportProtocol(
  		PEPeerControl 	_manager,
  		String 			_ip, 
  		int 			_port,
  		boolean			_incoming_connection,
		SocketChannel 	channel,  //hack for incoming connections. null otherwise
  		final byte[]	data_already_read )
	{		
		manager	= _manager;
		ip 		= _ip;
		port 	= _port;
	 	    
		uniquePiece = -1;
		
		incoming = _incoming_connection;

		stats = (PEPeerStatsImpl)manager.createPeerStats();
		
		if( incoming ) {
      connection = NetworkManager.getSingleton().createNewInboundConnection( this, channel );
      
      //"fake" a connect request to register our listener
      connection.connect( new Connection.ConnectionListener() {
        public void connectStarted() {
          connection_state = PEPeerTransport.CONNECTION_CONNECTING;
        }
        
        public void connectSuccess() {  //will be called immediately
          LGLogger.log(componentID, evtLifeCycle, LGLogger.RECEIVED, "Established incoming connection from " + PEPeerTransportProtocol.this );
          currentState = new StateHandshaking( false, data_already_read );
        }
        
        public void connectFailure( Throwable failure_msg ) {  //should never happen
          Debug.out( "ERROR: incoming connect failure: ", failure_msg );
          closeAll( "ERROR: incoming connect failure [" + PEPeerTransportProtocol.this + "] : " + failure_msg.getMessage(), true, false );
        }
        
        public void exceptionThrown( Throwable error ) {
          closeAll( "Connection [" + PEPeerTransportProtocol.this + "] exception : " + error.getMessage(), true, true );
        }
      });
		}
    else { //outgoing
			currentState = new StateConnecting();
		}
	}



  /**
   * Private method that will finish fields allocation, once the handshaking is ok.
   * Hopefully, that will save some RAM.
   */
  private void allocateAll() {
  	try{
  		this_mon.enter();
  	
  		if ( closing ){
  			
  			return;
  		}
  		
	    other_peer_has_pieces = new boolean[ manager.getPiecesNumber() ];
	  	Arrays.fill( other_peer_has_pieces, false );
	
	    //link in outgoing piece handler
	    outgoing_piece_message_handler = new OutgoingBTPieceMessageHandler( manager.getDiskManager(), connection.getOutgoingMessageQueue() );
	    
	    //link in outgoing have message aggregator
	    outgoing_have_message_aggregator = new OutgoingBTHaveMessageAggregator( connection.getOutgoingMessageQueue() );
	    
	    //register bytes sent listener
	    connection.getOutgoingMessageQueue().registerQueueListener( new OutgoingMessageQueue.MessageQueueListener() {
	      public void messageAdded( ProtocolMessage message ) { /*ignore*/ }
	      public void messageRemoved( ProtocolMessage message ) { /*ignore*/ }
	      
	      public void messageSent( ProtocolMessage message ) {
	        //update keep-alive info
	        last_message_sent_time = SystemTime.getCurrentTime();
	      }
	
	      public void protocolBytesSent( int byte_count ) {
	        //update stats
	        stats.protocol_sent( byte_count );
	        manager.protocol_sent( byte_count );
	      }
	      
	      public void dataBytesSent( int byte_count ) {
	        //update stats
	        stats.sent( byte_count );
	        manager.sent( byte_count );
	      }
	    });
	    
	    //register the new connection with the upload manager so that peer messages get processed
	    UploadManager.getSingleton().registerStandardPeerConnection( connection, manager.getUploadLimitedRateGroup() );
	    
	    connection_registered = true;
	    
  	}finally{
	
  		this_mon.exit();
  	}
  }

   
  
  public void 
  closeAll(
  		String 		reason, 
		boolean 	closedOnError, 
		boolean 	attemptReconnect ) 
  {
  		try{
  			this_mon.enter();
  		
  			if (closing){
  				
  				return;
  			}
  			
  			closing = true;
  			
  		}finally{
  			
  			this_mon.exit();
  		}
  		
	    currentState = new StateClosing();
      
        LGLogger.log( componentID, evtProtocol, closedOnError?LGLogger.ERROR:LGLogger.INFORMATION, reason);
      
	  	//Cancel any pending requests (on the manager side)
	  	cancelRequests();
	  	  		    
	    if( outgoing_piece_message_handler != null ) {
	      outgoing_piece_message_handler.removeAllPieceRequests();
	      outgoing_piece_message_handler.destroy();
	      //outgoing_piece_message_handler = null;
	    }
	    
	    if( outgoing_have_message_aggregator != null ) {
	      outgoing_have_message_aggregator.destroy();
	      //outgoing_have_message_aggregator = null;
	    }
	    	    
	    if( connection != null ) {
	    	if( connection_registered ) {
	    		UploadManager.getSingleton().cancelStandardPeerConnection( connection );
	    	}
	    	connection.close();
	    	//connection = null;
	    }
	    
	    
	    try{
	    	recent_outgoing_requests_mon.enter();
	    
	    	recent_outgoing_requests.clear();
	      
	    }finally{