pepeercontrolimpl.java

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

/*
 * Created on Oct 15, 2003
 * Created by Olivier Chalouhi
 * Modified Apr 13, 2004 by Alon Rohter
 * Copyright (C) 2004 Aelitis, All Rights Reserved.
 * 
 */
 
package org.gudy.azureus2.core3.peer.impl.control;


import java.nio.ByteBuffer;
import java.util.*;


import org.gudy.azureus2.core3.torrent.*;
import org.gudy.azureus2.core3.tracker.client.*;
import org.gudy.azureus2.core3.util.*;
import org.gudy.azureus2.core3.config.*;
import org.gudy.azureus2.core3.disk.*;
import org.gudy.azureus2.core3.download.DownloadManager;
import org.gudy.azureus2.core3.ipfilter.*;
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.impl.transport.base.DataReaderOwner;
import org.gudy.azureus2.core3.peer.impl.transport.base.DataReaderSpeedLimiter;
import org.gudy.azureus2.core3.peer.util.*;

import com.aelitis.azureus.core.networkmanager.ConnectDisconnectManager;
import com.aelitis.azureus.core.peermanager.LimitedRateGroup;
import com.aelitis.azureus.core.peermanager.utils.PeerConnectInfoStorage;


public class 
PEPeerControlImpl
	implements 	PEPeerControl, ParameterListener, DiskManagerWriteRequestListener, DiskManagerCheckRequestListener
{
  
  //Min number of requests sent to a peer
  private static final int MIN_REQUESTS = 2;
  //Default number of requests sent to a peer
  //(for each X B/s a new request will be used)
  private static final int SLOPE_REQUESTS = 2 * 1024;
  //Max number of request sent to a peer
  private static final int MAX_REQUESTS = 64;
  
  private static final int BAD_CHUNKS_LIMIT = 3;
  private static final int WARNINGS_LIMIT = 3;
  
  private static boolean oldPolling = COConfigurationManager.getBooleanParameter("Old.Socket.Polling.Style", false);
  private static boolean disconnect_seeds_when_seeding = COConfigurationManager.getBooleanParameter("Disconnect Seed", true);
  
    
  private static IpFilter ip_filter = IpFilterManagerFactory.getSingleton().getIPFilter();
  
  private int peer_manager_state = PS_INITIALISED;
  
  private int[] 	availability_cow;
  
  private boolean _bContinue;    
  
  private volatile ArrayList peer_transports_cow = new ArrayList();	// Copy on write!
  
  private AEMonitor	peer_transports_mon	= new AEMonitor( "PEPeerControl:PT");
  
  private DiskManager 			_diskManager;
  private DiskManagerPiece[]	dm_pieces;
  
  private boolean[] _downloading;
  private boolean 	seeding_mode;
  private boolean	restart_initiated;
  
  private boolean[] 			_piecesRarest;
  private PeerIdentityDataID 	_hash;
  private byte[] _myPeerId;
  private int _nbPieces;
  private PEPieceImpl[] 				_pieces;
  private PEPeerServerHelper 			_server;
  private PEPeerManagerStatsImpl 		_stats;
  private TRTrackerClient _tracker;
   //  private int _maxUploads;
  private int _seeds, _peers,_remotes;
  private long _timeStarted;
  private long _timeStartedSeeding = -1;
  private long _timeFinished;
  private Average _averageReceptionSpeed;
  private PEPeerTransport currentOptimisticUnchoke;
  
  
  private static final long	END_GAME_MODE_SIZE_TRIGGER	= 20*1024*1024;
  private static final long	END_GAME_MODE_TIMEOUT		= 10*60*1000;
  
  	//A flag to indicate when we're in endgame mode
  private boolean endGameMode;
  private boolean endGameModeAbandoned;
  private long	  timeEndGameModeEntered;
  
    //The list of chunks needing to be downloaded (the mechanism change when entering end-game mode)
  private List 		endGameModeChunks;
  private AEMonitor	endGameModeChunks_mon	= new AEMonitor( "PEPeerControl:EG");

  
  private DownloadManager _downloadManager;
  private PeerUpdater peerUpdater;
  
  private int nbHashFails;
  

  private static final int PEER_UPDATER_INTERVAL = 50;
  
  private long mainloop_loop_count;
  private static final int MAINLOOP_INTERVAL   = 100;
  private static final int MAINLOOP_ONE_SECOND_INTERVAL = 1000 / MAINLOOP_INTERVAL;
  private static final int MAINLOOP_FIVE_SECOND_INTERVAL = MAINLOOP_ONE_SECOND_INTERVAL * 5;
  private static final int MAINLOOP_TEN_SECOND_INTERVAL = MAINLOOP_ONE_SECOND_INTERVAL * 10;
  private static final int MAINLOOP_THIRTY_SECOND_INTERVAL = MAINLOOP_ONE_SECOND_INTERVAL * 30;
  
  

  private List	peer_manager_listeners 		= new ArrayList();
  //private List	peer_transport_listeners 	= new ArrayList();
  
  private List 		piece_check_result_list 	= new ArrayList();
  private AEMonitor	piece_check_result_list_mon	= new AEMonitor( "PEPeerControl:PCRL");

  private boolean superSeedMode;
  private int superSeedModeCurrentPiece;
  private int superSeedModeNumberOfAnnounces;
  private SuperSeedPiece[] superSeedPieces;
  
  private final PeerConnectInfoStorage peer_info_storage = new PeerConnectInfoStorage( 200 );  //size will be updated later on
  
  private final HashMap 	reconnect_counts 		= new HashMap();
  private final AEMonitor	reconnect_counts_mon	= new AEMonitor( "PEPeerControl:RC");

  private AEMonitor	this_mon	= new AEMonitor( "PEPeerControl");
  
  private long		ip_filter_last_update_time;
  
  private Map		user_data;
  
  private PEPeerTransportDataReader		download_speed_limiter;
  
  private final LimitedRateGroup upload_limited_rate_group = new LimitedRateGroup() {
    public int getRateLimitBytesPerSecond() {
      return _downloadManager.getStats().getUploadRateLimitBytesPerSecond();
    }
  };
  

  public 
  PEPeerControlImpl
  (
    DownloadManager 	manager,
    PEPeerServerHelper 	server,
	TRTrackerClient 	tracker,
    DiskManager 		diskManager) 
  {
  	  _server = server;
  	  this._downloadManager = manager;
  	  _tracker = tracker;
  	  this._diskManager = diskManager;
  	  COConfigurationManager.addParameterListener("Old.Socket.Polling.Style", this);
      COConfigurationManager.addParameterListener("Ip Filter Enabled", this);
      COConfigurationManager.addParameterListener( "Disconnect Seed", this );
      
 }
  
	public DownloadManager
	getDownloadManager()
	{
		return( _downloadManager );
	}
 
	public int
	getState()
	{
		return( peer_manager_state );
	}
	
  public void
  start()
  {  
    endGameMode = false;
    //This torrent Hash
    
    try{
    
    	_hash = PeerIdentityManager.createDataID( _tracker.getTorrent().getHash());
    	
    }catch( TOTorrentException e ){
    	
    		// this should never happen
    	Debug.printStackTrace( e );
    	
    	_hash = PeerIdentityManager.createDataID( new byte[20] ); 
    }
    
    this.nbHashFails = 0;

    //The connection to the tracker
    _tracker.setAnnounceDataProvider(
    		new TrackerClientAnnounceDataProvider()
    		{
    			public String
				getName()
    			{
    				return( getDownloadManager().getDisplayName());
    			}
    			
    			public long
    			getTotalSent()
    			{
    				return(getStats().getTotalSent());
    			}
    			public long
    			getTotalReceived()
    			{
    				return(getStats().getTotalReceived());
    			}
    			
    			public long
    			getRemaining()
    			{
    				return( PEPeerControlImpl.this.getRemaining());
    			}
    		});
    
    _myPeerId = _tracker.getPeerId();

    //The peer connections
    peer_transports_cow = new ArrayList();

    //The Server that handle incoming connections
     _server.setServerAdapter(this);

    //BtManager is threaded, this variable represents the
    // current loop iteration. It's used by some components only called
    // at some specific times.
    mainloop_loop_count = 0;

    //The current tracker state
    //this could be start or update

    _averageReceptionSpeed = Average.getInstance(1000, 30);

	download_speed_limiter = 
				DataReaderSpeedLimiter.getSingleton().getDataReader(
						new DataReaderOwner()
					{
							public int
							getMaximumBytesPerSecond()
							{
								return( _downloadManager.getStats().getMaxDownloadKBSpeed() * 1024 );
							}
					});
    
    setDiskManager(_diskManager);
    
    superSeedMode = (COConfigurationManager.getBooleanParameter("Use Super Seeding") && this.getRemaining() == 0);
    
    superSeedModeCurrentPiece = 0;
    
    if ( superSeedMode ){
    	initialiseSuperSeedMode();
    }
    
    peerUpdater = new PeerUpdater();
    peerUpdater.start();
    
    
    new AEThread( "Peer Manager"){
      public void
      runSupport()
      {
        mainLoop();
      }
    }.start();
  }

  private class 
  PeerUpdater 
  extends AEThread 
  {
    private boolean bContinue = true;

    public PeerUpdater() {
      super("Peer Updater"); //$NON-NLS-1$
      setPriority(Thread.NORM_PRIORITY - 1);
    }

    public void runSupport() {
      while (bContinue) {
        
        long start_time = SystemTime.getCurrentTime();
        
        try{
        	List	peer_transports = peer_transports_cow;
                       	
	          for (int i=0; i < peer_transports.size(); i++) {
	            PEPeerTransport ps = (PEPeerTransport) peer_transports.get(i);
	
	            if (SystemTime.isErrorLast5sec() || oldPolling || (SystemTime.getCurrentTime() > (ps.getLastReadTime() + ps.getReadSleepTime()))) {
	              ps.setReadSleepTime( ps.processRead() );
	              if ( !oldPolling ) ps.setLastReadTime( SystemTime.getCurrentTime() );
	            }
	          }
        }catch( Throwable e ){
        	
        	Debug.printStackTrace( e );
        }
         
        long loop_time = SystemTime.getCurrentTime() - start_time;
        
        //TODO : BOTTLENECK for download speed HERE (100 : max 500kB/s from BitTornado, 50 : 1MB/s, 25 : 2MB/s, 10 : 3MB/s
        
        if( loop_time < PEER_UPDATER_INTERVAL && loop_time >= 0 ) {
          try {  Thread.sleep( PEER_UPDATER_INTERVAL - loop_time );  } catch(Exception e) {}
        }

      }
    }

    public void stopIt() {
      bContinue = false;
    }
  }
  
  
  
  private void mainLoop() {
    _bContinue = true;

    _downloadManager.setState( DownloadManager.STATE_DOWNLOADING );

    _timeStarted = SystemTime.getCurrentTime();

    // initial check on finished state - future checks are driven by piece check results

    checkFinished( true );

    while( _bContinue ) { //loop until stopAll() kills us

      try {
        long timeStart = SystemTime.getCurrentTime();

        updateTrackerAnnounceInterval();
        
        doConnectionChecks();

        processPieceChecks();

        checkCompletedPieces(); //check to see if we've completed anything else

        computeAvailability(); //compute the availablity

        updateStats();

        checkFastPieces();

        if( !seeding_mode ) { //if we're not finished

          _diskManager.computePriorityIndicator();

          checkRequests(); //check the requests

          checkDLPossible(); //look for downloadable pieces
        }

        checkSeeds( false );

        updatePeersInSuperSeedMode();

        doUnchokes();
        

        long loop_time = SystemTime.getCurrentTime() - timeStart;
        if( loop_time < MAINLOOP_INTERVAL && loop_time >= 0 ) {
          try {  Thread.sleep( MAINLOOP_INTERVAL - loop_time );  } catch(Exception e) {}
        }
        
      }
      catch (Throwable e) {

        Debug.printStackTrace( e );
      }

      mainloop_loop_count++;
    }
  }

  
  
  public void 
  stopAll() 
  {
  		// send stopped event
  	
    _tracker.stop();
    
    	//  Stop the server
    
    _server.stopServer();
    
    _server.clearServerAdapter();
    
    	// Close all clients
    
    try{
    	peer_transports_mon.enter();
      
        	//  Stop itself
    	
        _bContinue = false;

        while (peer_transports_cow.size() != 0) {
        	
          removeFromPeerTransports((PEPeerTransport)peer_transports_cow.get(0), "Closing all Connections");
        }
    }finally{
      	
      	peer_transports_mon.exit();
    }
  
    
    // Stop the peer updater
    peerUpdater.stopIt();

    //clear pieces
    for (int i = 0; i < _pieces.length; i++) {
      if (_pieces[i] != null)
        pieceRemoved(_pieces[i]);
    }