pepeertransportprotocol.java

这是一个基于java编写的torrent的P2P源码

/*
 * 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.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.util.*;

import org.gudy.azureus2.core3.config.*;
import org.gudy.azureus2.core3.disk.*;
import org.gudy.azureus2.core3.logging.*;
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.util.*;
import org.gudy.azureus2.plugins.network.Connection;
import org.gudy.azureus2.pluginsimpl.local.network.ConnectionImpl;

import com.aelitis.azureus.core.networkmanager.*;
import com.aelitis.azureus.core.networkmanager.impl.tcp.ProtocolEndpointTCP;
import com.aelitis.azureus.core.networkmanager.impl.tcp.TCPNetworkManager;
import com.aelitis.azureus.core.networkmanager.impl.udp.ProtocolEndpointUDP;
import com.aelitis.azureus.core.networkmanager.impl.udp.UDPNetworkManager;
import com.aelitis.azureus.core.peermanager.messaging.*;
import com.aelitis.azureus.core.peermanager.messaging.azureus.*;
import com.aelitis.azureus.core.peermanager.messaging.bittorrent.*;
import com.aelitis.azureus.core.peermanager.peerdb.*;
import com.aelitis.azureus.core.peermanager.piecepicker.PiecePicker;
import com.aelitis.azureus.core.peermanager.piecepicker.util.BitFlags;
import com.aelitis.azureus.core.peermanager.utils.*;



public class 
PEPeerTransportProtocol
	extends LogRelation
	implements PEPeerTransport
{
	protected final static LogIDs LOGID = LogIDs.PEER;
  
	private volatile int	_lastPiece =-1;		//last piece that was requested from this peer (mostly to try to request from same one)
	
	protected final PEPeerControl 	manager;
	protected final DiskManager		diskManager;
	protected final PiecePicker		piecePicker;
	
	protected final int		nbPieces;
	
	private final String			peer_source;
	private byte[] peer_id;
	private final String ip;
	protected String ip_resolved;
	private IPToHostNameResolverRequest	ip_resolver_request;
	
	private int port;
  
  private PeerItem peer_item_identity;
  private int tcp_listen_port = 0;
  private int udp_listen_port = 0;
  private int udp_non_data_port = 0;
	
  private byte	crypto_level;
  
  protected final PEPeerStats peer_stats;
  
  private final ArrayList requested = new ArrayList();
  private final AEMonitor	requested_mon = new AEMonitor( "PEPeerTransportProtocol:Req" );

  private HashMap data;
  	
	private long lastNeededUndonePieceChange;
	
  protected boolean choked_by_other_peer = true;
  /** total time the other peer has unchoked us while not snubbed */
  protected long unchokedTimeTotal;
  /** the time at which the other peer last unchoked us when not snubbed */
  protected long unchokedTime;
  protected boolean choking_other_peer = true;
  private boolean interested_in_other_peer = false;
  private boolean other_peer_interested_in_me = false;
  private long snubbed =0;

  /** lazy allocation; null until needed */
  private volatile BitFlags	peerHavePieces =null; 
  private volatile boolean	availabilityAdded =false;

  private boolean seed_set_by_accessor = false;
 
  private final boolean incoming;
  
  protected volatile boolean closing = false;
  private volatile int current_peer_state;
  
  protected NetworkConnection connection;
  private OutgoingBTPieceMessageHandler outgoing_piece_message_handler;
  private OutgoingBTHaveMessageAggregator outgoing_have_message_aggregator;
  private Connection	plugin_connection;
  
  private boolean identityAdded = false;  //needed so we don't remove id's in closeAll() on duplicate connection attempts

  protected int connection_state = PEPeerTransport.CONNECTION_PENDING;

  //The client name identification	
	private String client = "";
    	
	//When superSeeding, number of unique piece announced
	private int uniquePiece = -1;
  
  //When downloading a piece in exclusivity mode the piece number being downloaded
  private int reservedPiece = -1;
	
	//Spread time (0 secs , fake default)
	private int spreadTimeHint = 0 * 1000;

	protected long last_message_sent_time = 0;
	protected long last_message_received_time = 0;
	protected long last_data_message_received_time = -1;
	protected long last_good_data_time =-1;			// time data written to disk was recieved
	protected long last_data_message_sent_time = -1;
  
  private long connection_established_time = 0;

  private int consecutive_no_request_count;
  
  private boolean az_messaging_mode = false;
  private Message[] supported_messages = null;
  
  
  private final AEMonitor	closing_mon	= new AEMonitor( "PEPeerTransportProtocol:closing" );
  private final AEMonitor data_mon  = new AEMonitor( "PEPeerTransportProtocol:data" );
  

  private LinkedHashMap recent_outgoing_requests;
  private AEMonitor	recent_outgoing_requests_mon;
  
  private static final boolean SHOW_DISCARD_RATE_STATS;
  static {
	  final 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_discarded_endgame = 0;
  private static int requests_recovered = 0;
  private static int requests_completed = 0;

  private List peer_listeners_cow;
  private final AEMonitor	peer_listeners_mon = new AEMonitor( "PEPeerTransportProtocol:PL" );

  
  //certain Optimum Online networks block peer seeding via "complete" bitfield message filtering
  //lazy mode makes sure we never send a complete (seed) bitfield
  protected static boolean ENABLE_LAZY_BITFIELD;
  
  static {
    
    COConfigurationManager.addAndFireParameterListeners(
    		new String[]{ "Use Lazy Bitfield" },
    		new ParameterListener()
    		{
    			 public final void 
    			 parameterChanged(
    				String ignore )
    			 {
    				 final String  prop = System.getProperty( "azureus.lazy.bitfield" );
    				 
    				 ENABLE_LAZY_BITFIELD = prop != null && prop.equals( "1" );
 
    				 ENABLE_LAZY_BITFIELD |= COConfigurationManager.getBooleanParameter( "Use Lazy Bitfield" );
    			 }
    		});
  }
  
  
  private boolean is_optimistic_unchoke = false;

  private PeerExchangerItem peer_exchange_item = null;
  private boolean peer_exchange_supported = false;
    
  protected PeerMessageLimiter message_limiter;
  
  
  
  
  
  //INCOMING
  public 
  PEPeerTransportProtocol( PEPeerControl _manager, String _peer_source, NetworkConnection _connection ) {
    manager 		= _manager;
    peer_source		= _peer_source;
    connection 		= _connection;

    incoming = true;
    
    diskManager =manager.getDiskManager();
    piecePicker =manager.getPiecePicker();
    nbPieces =diskManager.getNbPieces();
    
    
    InetSocketAddress notional_address = _connection.getEndpoint().getNotionalAddress();
    
    ip    = notional_address.getAddress().getHostAddress();
    port  = notional_address.getPort();
    
    peer_item_identity = PeerItemFactory.createPeerItem( ip, port, PeerItem.convertSourceID( _peer_source ), PeerItemFactory.HANDSHAKE_TYPE_PLAIN, 0, PeerItemFactory.CRYPTO_LEVEL_1, 0 );  //this will be recreated upon az handshake decode
    
    plugin_connection = new ConnectionImpl(connection);
    
    peer_stats = manager.createPeerStats( this );

    changePeerState( PEPeer.CONNECTING );
  }
  
  public void
  start()
  {
	  	// split out connection initiation from constructor so we can get access to the peer transport
	  	// before message processing starts
	  
	if ( incoming ){
		
			//"fake" a connect request to register our listener
		
	    connection.connect( new NetworkConnection.ConnectionListener() {
	      public final void connectStarted() {
	        connection_state = PEPeerTransport.CONNECTION_CONNECTING;
	      }
	      
	      public final void connectSuccess( ByteBuffer remaining_initial_data ) {  //will be called immediately
	      	if (Logger.isEnabled())
						Logger.log(new LogEvent(PEPeerTransportProtocol.this, LOGID,
								"In: Established incoming connection"));
	        initializeConnection();
	        
	        /*
	         * Waiting until we've received the initiating-end's full handshake, before sending back our own,
	         * really should be the "proper" behavior.  However, classic BT trackers running NAT checking will
	         * only send the first 48 bytes (up to infohash) of the peer handshake, skipping peerid, which means
	         * we'll never get their complete handshake, and thus never reply, which causes the NAT check to fail.
	         * So, we need to send our handshake earlier, after we've verified the infohash.
	         * NOTE:
	         * This code makes the assumption that the inbound infohash has already been validated,
	         * as we don't check their handshake fully before sending our own.
	         */
	        sendBTHandshake();
	      }
	      
	      public final void connectFailure( Throwable failure_msg ) {  //should never happen
	        Debug.out( "ERROR: incoming connect failure: ", failure_msg );
	        closeConnectionInternally( "ERROR: incoming connect failure [" + PEPeerTransportProtocol.this + "] : " + failure_msg.getMessage());
	      }
	      
	      public final void exceptionThrown( Throwable error ) {
	        if( error.getMessage() == null ) {
	          Debug.out( error );
	        }
	        
	        closeConnectionInternally( "connection exception: " + error.getMessage());
	      }
	    });
	}else{
		// not pulled out startup from outbound connections yet...
	}
  }
  
  
  
  	//OUTGOING
  
  public 
  PEPeerTransportProtocol( 
		PEPeerControl 	_manager, 
		String 			_peer_source, 
		String 			_ip, 
		int 			_tcp_port, 
		int 			_udp_port,
		boolean			_use_tcp,
		boolean 		_require_crypto_handshake,
		byte			_crypto_level ) 
  {
    manager = _manager;
    diskManager =manager.getDiskManager();
    piecePicker =manager.getPiecePicker();
    nbPieces =diskManager.getNbPieces();
    lastNeededUndonePieceChange =Long.MIN_VALUE;

    peer_source	= _peer_source;
    ip    = _ip;
    port  = _tcp_port;
    tcp_listen_port = _tcp_port;
    udp_listen_port	= _udp_port;
    crypto_level	= _crypto_level;
    
    udp_non_data_port = UDPNetworkManager.getSingleton().getUDPNonDataListeningPortNumber();
    	
    peer_item_identity = PeerItemFactory.createPeerItem( ip, tcp_listen_port, PeerItem.convertSourceID( _peer_source ), PeerItemFactory.HANDSHAKE_TYPE_PLAIN, _udp_port, crypto_level, 0 );  //this will be recreated upon az handshake decode
    
    incoming = false;
    
    peer_stats = manager.createPeerStats( this );
    
    if( port < 0 || port > 65535 ) {
      closeConnectionInternally( "given remote port is invalid: " + port );
      return;
    }

    boolean use_crypto = _require_crypto_handshake || NetworkManager.REQUIRE_CRYPTO_HANDSHAKE;  //either peer specific or global pref
    
    if( isLANLocal() )  use_crypto = false;  //dont bother with PHE for lan peers
    
    InetSocketAddress	endpoint_address;
    ProtocolEndpoint	pe;
    
    if ( _use_tcp ){
    	
    	endpoint_address = new InetSocketAddress( ip, tcp_listen_port );
    	   
    	pe = new ProtocolEndpointTCP( endpoint_address );
    	
    }else{
    	
      	endpoint_address = new InetSocketAddress( ip, udp_listen_port );
 	   
    	pe = new ProtocolEndpointUDP( endpoint_address );
    }
	
    ConnectionEndpoint connection_endpoint	= new ConnectionEndpoint( endpoint_address );
    
    connection_endpoint.addProtocol( pe );
       
    connection = 
    	NetworkManager.getSingleton().createConnection(
    			connection_endpoint, 
    			new BTMessageEncoder(), 
    			new BTMessageDecoder(), 
    			use_crypto, 
    			!_require_crypto_handshake, 
    			manager.getSecrets( _crypto_level ));
    
    plugin_connection = new ConnectionImpl(connection);
    
    changePeerState( PEPeer.CONNECTING );
    
    connection.connect( new NetworkConnection.ConnectionListener() {
      public final void connectStarted() {
        connection_state = PEPeerTransport.CONNECTION_CONNECTING;
      }
 
      public final void connectSuccess( ByteBuffer remaining_initial_data ) {
        if( closing ) {
          //Debug.out( "PEPeerTransportProtocol::connectSuccess() called when closing." );
          return;
        }
        
        if (Logger.isEnabled())
					Logger.log(new LogEvent(PEPeerTransportProtocol.this, LOGID,
							"Out: Established outgoing connection"));
        initializeConnection();
        sendBTHandshake();
      }
        
      public final void connectFailure( Throwable failure_msg ) {
        closeConnectionInternally( "failed to establish outgoing connection: " + failure_msg.getMessage(), true );
      }
        
      public final void exceptionThrown( Throwable error ) {
        if( error.getMessage() == null ) {
          Debug.out( "error.getMessage() == null", error );
        }
        
        closeConnectionInternally( "connection exception: " + error.getMessage(), true );
      }
    });
      
    if (Logger.isEnabled())
			Logger.log(new LogEvent(this, LOGID,
					"Out: Creating outgoing connection"));
  }
  
  

  
  protected void initializeConnection() {
    if( closing )  return;

    recent_outgoing_requests = new LinkedHashMap( 16, .75F, true ) {
      public final boolean removeEldestEntry(Map.Entry eldest) {
        return size() > 16;
      }
    };
    recent_outgoing_requests_mon  = new AEMonitor( "PEPeerTransportProtocol:ROR" );