pepeercontrolimpl.java

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

/*
 * Created by Olivier Chalouhi
 * Modified Apr 13, 2004 by Alon Rohter
 * Heavily modified Sep 2005 by Joseph Bridgewater
 * Copyright (C) 2004, 2005, 2006 Aelitis, All Rights Reserved.
 * 
 * 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, or (at your option) any later version.
 * 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.
 * 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.
 *
 * AELITIS, SAS au capital de 46,603.30 euros
 * 8 Allee Lenotre, La Grille Royale, 78600 Le Mesnil le Roi, France.
 */

package org.gudy.azureus2.core3.peer.impl.control;


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.ipfilter.*;
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.torrent.TOTorrentException;
import org.gudy.azureus2.core3.tracker.client.*;
import org.gudy.azureus2.core3.util.*;
import org.gudy.azureus2.plugins.download.DownloadAnnounceResultPeer;
import org.gudy.azureus2.plugins.peers.PeerDescriptor;

import com.aelitis.azureus.core.networkmanager.LimitedRateGroup;
import com.aelitis.azureus.core.networkmanager.impl.tcp.ConnectDisconnectManager;
import com.aelitis.azureus.core.networkmanager.impl.tcp.TCPNetworkManager;
import com.aelitis.azureus.core.networkmanager.impl.udp.UDPNetworkManager;
import com.aelitis.azureus.core.peermanager.control.*;
import com.aelitis.azureus.core.peermanager.nat.PeerNATInitiator;
import com.aelitis.azureus.core.peermanager.nat.PeerNATTraversalAdapter;
import com.aelitis.azureus.core.peermanager.nat.PeerNATTraverser;
import com.aelitis.azureus.core.peermanager.peerdb.*;
import com.aelitis.azureus.core.peermanager.piecepicker.*;
import com.aelitis.azureus.core.peermanager.unchoker.*;
import com.aelitis.azureus.core.peermanager.uploadslots.UploadHelper;
import com.aelitis.azureus.core.peermanager.uploadslots.UploadSlotManager;

/**
 * manages all peer transports for a torrent
 * 
 * @author MjrTom
 *			2005/Oct/08: Numerous changes for new piece-picking. Also
 *						a few optimizations and multi-thread cleanups
 *			2006/Jan/02: refactoring piece picking related code
 */


public class 
PEPeerControlImpl
	extends LogRelation
	implements 	PEPeerControl, ParameterListener, DiskManagerWriteRequestListener, PeerControlInstance, PeerNATInitiator,
		DiskManagerCheckRequestListener, IPFilterListener
{
	private static final LogIDs LOGID = LogIDs.PEER;
  
	private static final int	WARNINGS_LIMIT = 2;
	
	private static final int	CHECK_REASON_DOWNLOADED			= 1;
	private static final int	CHECK_REASON_COMPLETE			= 2;
	private static final int	CHECK_REASON_SCAN				= 3;
	private static final int	CHECK_REASON_SEEDING_CHECK		= 4;

	private static final int	SEED_CHECK_WAIT_MARKER	= 65526;

	private static boolean disconnect_seeds_when_seeding = COConfigurationManager.getBooleanParameter("Disconnect Seed");
	private static boolean enable_seeding_piece_rechecks = COConfigurationManager.getBooleanParameter("Seeding Piece Check Recheck Enable");
	
	private static IpFilter ip_filter = IpFilterManagerFactory.getSingleton().getIPFilter();
    
    private volatile boolean    is_running = false;

    private volatile ArrayList  peer_transports_cow = new ArrayList();  // Copy on write!
	private final AEMonitor     peer_transports_mon	= new AEMonitor( "PEPeerControl:PT");

    protected final PEPeerManagerAdapter	adapter;
    private final DiskManager           disk_mgr;
    private final DiskManagerPiece[]    dm_pieces;

    private final PiecePicker	piecePicker;
    private long            	lastNeededUndonePieceChange;
    
    /** literally seeding as in 100% torrent complete */
	private boolean 		seeding_mode;
	private boolean			restart_initiated;

    private final int       _nbPieces;     //how many pieces in the torrent
    private PEPieceImpl[]	pePieces;      //pieces that are currently in progress
    private int				nbPiecesActive;	// how many pieces are currently in progress
    
    private int				nbPeersSnubbed;

    private PeerIdentityDataID	_hash;
    private final byte[]        _myPeerId;
    private PEPeerManagerStats        _stats;
	//private final TRTrackerAnnouncer _tracker;
	//  private int _maxUploads;
	private int		_seeds, _peers,_remotes;
    private long    last_remote_time;
	private long	_timeStarted;
	private long	_timeStartedSeeding = -1;
	private long	_timeFinished;
	private Average	_averageReceptionSpeed;

	private long mainloop_loop_count;

    private static final int MAINLOOP_ONE_SECOND_INTERVAL = 1000 / PeerControlScheduler.SCHEDULE_PERIOD_MILLIS;
    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 static final int MAINLOOP_SIXTY_SECOND_INTERVAL = MAINLOOP_ONE_SECOND_INTERVAL * 60;
    private static final int MAINLOOP_TEN_MINUTE_INTERVAL = MAINLOOP_SIXTY_SECOND_INTERVAL * 10;
    

    private volatile ArrayList  peer_manager_listeners_cow = new ArrayList();  //copy on write


    private final List		piece_check_result_list     = new ArrayList();
    private final AEMonitor	piece_check_result_list_mon  = new AEMonitor( "PEPeerControl:PCRL");
    
	private boolean 			superSeedMode;
	private int 				superSeedModeCurrentPiece;
	private int 				superSeedModeNumberOfAnnounces;
	private SuperSeedPiece[]	superSeedPieces;
	
    private final AEMonitor     this_mon = new AEMonitor( "PEPeerControl");

	private long		ip_filter_last_update_time;

    private Map                 user_data;    
    
    private Unchoker unchoker;
  
    private static boolean fast_unchoke_new_peers;
    
    static{
    	COConfigurationManager.addAndFireParameterListener(
    		"Peer.Fast.Initial.Unchoke.Enabled",
    		new ParameterListener()
    		{
    			public void 
    			parameterChanged(
    				String name ) 
    			{
    				fast_unchoke_new_peers = COConfigurationManager.getBooleanParameter( name );
    			}
    		});
    }
    
	private final UploadHelper upload_helper = new UploadHelper() {		
		public int getPriority() {			
			return UploadHelper.PRIORITY_NORMAL;  //TODO also must call UploadSlotManager.getSingleton().updateHelper( upload_helper ); on priority change
		}
		
		public ArrayList getAllPeers() {
			ArrayList peer_transports = peer_transports_cow;
			return peer_transports;
		}		
	
		public boolean isSeeding() {
			return seeding_mode;
		}		
	};
	
	
	private PeerDatabase	peer_database = PeerDatabaseFactory.createPeerDatabase();
	
	private int				next_rescan_piece		= -1;
	private long			rescan_piece_time		= -1;
	
	private long			last_eta;
	private long			last_eta_calculation;
	
	private static final int UDP_FALLBACK_MAX			= 32;
	private static final int MAX_UDP_TRAVERSAL_COUNT	= 3;
	private static final int MAX_UDP_CONNECTIONS		= 10;
	
	private Map	udp_fallbacks = 
		new LinkedHashMap(UDP_FALLBACK_MAX,0.75f,true)
		{
			protected boolean 
			removeEldestEntry(
		   		Map.Entry eldest) 
			{
				return size() > UDP_FALLBACK_MAX;
			}
		};	
		
	private int udp_traversal_count;
		
		
	private final LimitedRateGroup upload_limited_rate_group = new LimitedRateGroup() {
		public int getRateLimitBytesPerSecond() {
      		return adapter.getUploadRateLimitBytesPerSecond();
		}
	};
	
	private final LimitedRateGroup download_limited_rate_group = new LimitedRateGroup() {
		public int getRateLimitBytesPerSecond() {
      		return adapter.getDownloadRateLimitBytesPerSecond();
		}
	};
	
	public 
	PEPeerControlImpl(
		byte[]					_peer_id,
		PEPeerManagerAdapter 	_adapter,
		DiskManager 			diskManager) 
	{
		_myPeerId		= _peer_id;
		adapter 		= _adapter;
  
		disk_mgr = diskManager;
		_nbPieces =disk_mgr.getNbPieces();
        dm_pieces =disk_mgr.getPieces();

        pePieces =new PEPieceImpl[_nbPieces];

		piecePicker = PiecePickerFactory.create( this );

		COConfigurationManager.addParameterListener("Ip Filter Enabled", this);
		COConfigurationManager.addParameterListener( "Disconnect Seed", this );

		ip_filter.addListener( this );

	}
 

	public void
	start()
	{
		// This torrent Hash
		try
		{

			_hash =PeerIdentityManager.createDataID(disk_mgr.getTorrent().getHash());

		} catch (TOTorrentException e)
		{

			// this should never happen
			Debug.printStackTrace(e);

			_hash =PeerIdentityManager.createDataID(new byte[20]);
		}

        // the recovered active pieces
        for (int i =0; i <_nbPieces; i++ )
        {
            final DiskManagerPiece dmPiece =dm_pieces[i];
            if (!dmPiece.isDone() &&dmPiece.getNbWritten() >0)
            {
                addPiece(new PEPieceImpl(this, dmPiece, 0), i, true );
            }
        }

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

		// 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);

        // the stats
        _stats =new PEPeerManagerStatsImpl(this);

		superSeedMode =(COConfigurationManager.getBooleanParameter("Use Super Seeding") &&this.getRemaining() ==0);

		superSeedModeCurrentPiece =0;

		if (superSeedMode)
		{
			initialiseSuperSeedMode();
		}

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

		// Moved out of mainLoop() so that it runs immediately, possibly changing
		// the state to seeding.

		checkFinished(true);

		UploadSlotManager.getSingleton().registerHelper( upload_helper );
		
		lastNeededUndonePieceChange =Long.MIN_VALUE;
		_timeStarted =SystemTime.getCurrentTime();

		is_running = true;
		
			// activate after marked as running as we may synchronously add connections here due to pending activations
		
		adapter.getPeerManagerRegistration().activate( this );
		
		PeerNATTraverser.getSingleton().register( this );
		
		PeerControlSchedulerFactory.getSingleton().register(this);
	}

	public void stopAll()
	{
		is_running = false;

		UploadSlotManager.getSingleton().deregisterHelper( upload_helper );
		
		PeerControlSchedulerFactory.getSingleton().unregister(this);

		PeerNATTraverser.getSingleton().unregister( this );

			// remove legacy controller activation
		
		adapter.getPeerManagerRegistration().deactivate();
		
		closeAndRemoveAllPeers("download stopped", false);

		// clear pieces
		for (int i =0; i <_nbPieces; i++ )
		{
			if (pePieces[i] !=null)
				removePiece(pePieces[i], i);
		}

		// 5. Remove listeners
		COConfigurationManager.removeParameterListener("Ip Filter Enabled", this);
		COConfigurationManager.removeParameterListener("Disconnect Seed", this);

		ip_filter.removeListener(this);
		
		final ArrayList peer_manager_listeners = peer_manager_listeners_cow;

		for( int i=0; i < peer_manager_listeners.size(); i++ ) {
      		((PEPeerManagerListener)peer_manager_listeners.get(i)).destroyed();
		}
	}

	public DiskManager getDiskManager() {  return disk_mgr;   }
	public PiecePicker getPiecePicker()
	{
		return piecePicker;
	}
	
  public PEPeerManagerAdapter	getAdapter(){ return( adapter ); }
	
  public String getDisplayName(){ return( adapter.getDisplayName()); }
  
  public void
  schedule()
  {
      try {
			updateTrackerAnnounceInterval();
			doConnectionChecks();
			processPieceChecks();
			
				// note that seeding_mode -> torrent totally downloaded, not just non-dnd files
				// complete, so there is no change of a new piece appearing done by a means such as
				// background periodic file rescans
			
			if ( !seeding_mode ){
				
				checkCompletedPieces();		//check to see if we've completed anything else
			}
			
			updateStats();
			
            checkInterested();      // see if need to recheck Interested on all peers
			
			piecePicker.updateAvailability();
						
			checkCompletionState();	// pick up changes in completion caused by dnd file changes
			
			if ( seeding_mode ){
				
				checkSeeds();

			}else{
					// if we're not finished
				
				checkRequests();
				
				piecePicker.allocateRequests();
               
				checkRescan();
				
				checkSpeedAndReserved();
				
				check99PercentBug();
			}
			
				
			updatePeersInSuperSeedMode();
			
			doUnchokes();
			
      }catch (Throwable e) {
   	
        Debug.printStackTrace( e );
      }

      mainloop_loop_count++;
  }

  
  
	/**
	 * A private method that does analysis of the result sent by the tracker.
	 * It will mainly open new connections with peers provided
	 * and set the timeToWait variable according to the tracker response.
	 * @param tracker_response
	 */
	
	private void 
	analyseTrackerResponse(
		TRTrackerAnnouncerResponse	tracker_response )
	{
		// tracker_response.print();
		final TRTrackerAnnouncerResponsePeer[]	peers = tracker_response.getPeers();
		
		if ( peers != null ){
			addPeersFromTracker( tracker_response.getPeers());  
		}
		
		final Map extensions = tracker_response.getExtensions();
		
		if (extensions != null ){
			addExtendedPeersFromTracker( extensions );
		}