pepeercontrolimpl.java

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

                                pePiece.setSpeed((int)(calcSpeed >0 ?calcSpeed :0));
                        }
                    } else if (timeSinceActivity >(120 *1000))
					{
						// has reserved piece gone stagnant?
						final String reservingPeer =pePiece.getReservedBy();
						if (reservingPeer !=null)
						{
							if (needsMD5CheckOnCompletion(i))
								badPeerDetected(reservingPeer);
							else
							{
								final PEPeerTransport pt =getTransportFromAddress(reservingPeer);
								if (pt !=null)
									closeAndRemovePeer(pt, "Reserved piece data timeout; 120 seconds", true);
							}
                            pePiece.setReservedBy(null);
						}
//						if (!piecePicker.isInEndGameMode())
//							pePiece.checkRequests();
                        checkEmptyPiece(i);
					}
				}
			}
		}
	}

	private void
	check99PercentBug()
	{
			// there's a bug whereby pieces are left downloaded but never written. might have been fixed by
			// changes to the "write result" logic, however as a stop gap I'm adding code to scan for such
			// stuck pieces and reset them
		
		if ( mainloop_loop_count % MAINLOOP_SIXTY_SECOND_INTERVAL == 0 ) {
			
			long	now = SystemTime.getCurrentTime();
			
			for ( int i=0;i<pePieces.length;i++){
				
				PEPiece	pe_piece = pePieces[ i ];

				if ( pe_piece != null ){
									
					DiskManagerPiece	dm_piece = dm_pieces[i];
				
					if ( !dm_piece.isDone()){
					
						if ( pe_piece.isDownloaded()){
						
							if ( now - pe_piece.getLastDownloadTime(now) > 60*1000 ){
							
									// people with *very* slow disk writes can trigger this (I've been talking to a user
									// with a SAN that has .5 second write latencies when checking a file at the same time
									// this means that when dowloading > 32K/sec things start backing up). Eventually the
									// write controller will start blocking the network thread to prevent unlimited
									// queueing but until that time we need to handle this situation slightly better)
								
									// if there are any outstanding writes for this piece then leave it alone
								
								if ( !disk_mgr.hasOutstandingWriteRequestForPiece( i )){
									
									Debug.out( "Fully downloaded piece stalled pending write, resetting p_piece " + i );
															
									pe_piece.reset();
								}
							}
						}
					}
				}
			}
		}
	}
	
    private void checkInterested()
    {
        if ( (mainloop_loop_count %MAINLOOP_ONE_SECOND_INTERVAL) != 0 ){
            return;
        }
        
        if (lastNeededUndonePieceChange >=piecePicker.getNeededUndonePieceChange())
            return;
        
        lastNeededUndonePieceChange =piecePicker.getNeededUndonePieceChange();

		final ArrayList peer_transports = peer_transports_cow;
        int cntPeersSnubbed =0;	// recount # snubbed peers while we're at it
        for (int i =0; i <peer_transports.size(); i++)
        {
            final PEPeerTransport peer =(PEPeerTransport)peer_transports.get(i);
            peer.checkInterested();
            if (peer.isSnubbed())
            	cntPeersSnubbed++;
        }
        setNbPeersSnubbed(cntPeersSnubbed);
    }


	/**
	 * Private method to process the results given by DiskManager's
	 * piece checking thread via asyncPieceChecked(..)
	 */
	private void 
	processPieceChecks() 
	{
		if ( piece_check_result_list.size() > 0 ){
			
			final List pieces;
			
			// process complete piece results
			
			try{
				piece_check_result_list_mon.enter();
				
				pieces = new ArrayList( piece_check_result_list );
				
				piece_check_result_list.clear();
				
			}finally{
				
				piece_check_result_list_mon.exit();
			}
			
			final Iterator it = pieces.iterator();
			
			while (it.hasNext()) {
				
				final Object[]	data = (Object[])it.next();
				
	    		processPieceCheckResult((DiskManagerCheckRequest)data[0],((Integer)data[1]).intValue());
	    			
			}
		}
	}

	private void
	checkRescan()
	{
		if ( rescan_piece_time == 0 ){
			
				// pending a piece completion
			
			return;
		}
		
		if ( next_rescan_piece == -1 ){
			
			if ( mainloop_loop_count % MAINLOOP_FIVE_SECOND_INTERVAL == 0 ){
			
				if ( adapter.isPeriodicRescanEnabled()){
					
					next_rescan_piece	= 0;
				}
			}
		}else{
			
			if ( mainloop_loop_count % MAINLOOP_TEN_MINUTE_INTERVAL == 0 ){

				if ( !adapter.isPeriodicRescanEnabled()){
					
					next_rescan_piece	= -1;
				}
			}
		}
		
		if ( next_rescan_piece == -1 ){
			
			return;
		}
		
			// delay as required
		
		final long	now = SystemTime.getCurrentTime();
		
		if ( rescan_piece_time > now ){
			
			rescan_piece_time	= now;
		}
		
			// 250K/sec limit
		
		final long	piece_size = disk_mgr.getPieceLength();
		
		final long	millis_per_piece = piece_size / 250;
		
		if ( now - rescan_piece_time < millis_per_piece ){
			
			return;
		}
		
		while( next_rescan_piece != -1 ){
			
			int	this_piece = next_rescan_piece;
			
			next_rescan_piece++;
			
			if ( next_rescan_piece == _nbPieces ){
				
				next_rescan_piece	= -1;
			}
			
			if ( pePieces[this_piece] == null && !dm_pieces[this_piece].isDone()){
				
				DiskManagerCheckRequest	req = 
					disk_mgr.createCheckRequest(
						this_piece, 
						new Integer( CHECK_REASON_SCAN ));	
				
				req.setLowPriority( true );
				
			   	if ( Logger.isEnabled()){
			   		
					Logger.log(
							new LogEvent(
								disk_mgr.getTorrent(), LOGID,
								"Rescanning piece " + this_piece ));
							
			   	}
			   	
				rescan_piece_time	= 0;	// mark as check piece in process
				
				try{
					disk_mgr.enqueueCheckRequest( req, this );
					
				}catch( Throwable e ){
					
					rescan_piece_time	= now;
					
					Debug.printStackTrace(e);
				}
				
				break;
			}
		}
	}
	
	/**
	 * This method checks if the downloading process is finished.
	 * 
	 */
	private void 
	checkFinished(
		boolean start_of_day )
	{
		final boolean all_pieces_done =disk_mgr.getRemainingExcludingDND() ==0;

		if (all_pieces_done)
		{
			seeding_mode =true;
			piecePicker.clearEndGameChunks();

			if (!start_of_day)
				adapter.setStateFinishing();

			_timeFinished =SystemTime.getCurrentTime();
			final ArrayList peer_transports = peer_transports_cow;

			// remove previous snubbing
			for (int i =0; i <peer_transports.size(); i++ )
			{
				final PEPeerTransport pc =(PEPeerTransport) peer_transports.get(i);
				pc.setSnubbed(false);
			}
			setNbPeersSnubbed(0);
			
			final boolean checkPieces =COConfigurationManager.getBooleanParameter("Check Pieces on Completion", true);

			// re-check all pieces to make sure they are not corrupt, but only if we weren't already complete
			if (checkPieces &&!start_of_day)
			{
				final DiskManagerCheckRequest req =disk_mgr.createCheckRequest(-1, new Integer(CHECK_REASON_COMPLETE));
				disk_mgr.enqueueCompleteRecheckRequest(req, this);
			}

			disk_mgr.downloadEnded();
			_timeStartedSeeding =SystemTime.getCurrentTime();
			adapter.setStateSeeding(start_of_day);
		}
	}
	
	protected void
	checkCompletionState()
	{
		if ( mainloop_loop_count % MAINLOOP_ONE_SECOND_INTERVAL != 0 ){
	  		
	  		return;
	  	}
		
		boolean dm_done = disk_mgr.getRemainingExcludingDND() == 0;
		
		if ( seeding_mode ){
			
			if ( !dm_done ){
				
				seeding_mode = false;
				
				_timeStartedSeeding = -1;
				_timeFinished		= 0;
				
				Logger.log(
						new LogEvent(	disk_mgr.getTorrent(), LOGID,
										"Turning off seeding mode for PEPeerManager"));
			}
		}else{
			
			if ( dm_done ){
				
				checkFinished( false );
				
				if ( seeding_mode ){
					
					Logger.log(
							new LogEvent(	disk_mgr.getTorrent(), LOGID,
											"Turning on seeding mode for PEPeerManager"));
				}
			}
		}
	}
	
	/**
	 * This method will locate expired requests on peers, will cancel them,
	 * and mark the peer as snubbed if we haven't received usefull data from
	 * them within the last 60 seconds
	 */
	private void checkRequests()
	{
			// to be honest I don't see why this can't be 5 seconds, but I'm trying 1 second
			// now as the existing 0.1 second is crazy given we're checking for events that occur
			// at 60+ second intervals
		
	 	if ( mainloop_loop_count % MAINLOOP_ONE_SECOND_INTERVAL != 0 ){
	  		
	  		return;
	  	}
	 	
		final long now =SystemTime.getCurrentTime();

		//for every connection
		final ArrayList peer_transports = peer_transports_cow;
		for (int i =peer_transports.size() -1; i >=0 ; i--)
		{
			final PEPeerTransport pc =(PEPeerTransport)peer_transports.get(i);
			if (pc.getPeerState() ==PEPeer.TRANSFERING)
			{
				final List expired =pc.getExpiredRequests();
				if (expired !=null &&expired.size() >0)
				{   // now we know there's a request that's > 60 seconds old
                    final boolean isSeed =pc.isSeed();
                    // snub peers that haven't sent any good data for a minute
                    final long timeSinceGoodData =pc.getTimeSinceGoodDataReceived();
                    if (timeSinceGoodData <0 ||timeSinceGoodData >60 *1000)
                        pc.setSnubbed(true);
                    
                    final long timeSinceData =pc.getTimeSinceLastDataMessageReceived();
                    final long timeSinceOldestRequest =now -((DiskManagerReadRequest) expired.get(0)).getTimeCreated(now);
                    
                    for (int j =0; j <expired.size(); j++)
                    {
                        //for every expired request                              
                        //get the request object
                        final DiskManagerReadRequest request =(DiskManagerReadRequest) expired.get(j);
                        //Only cancel first request if more than 2 mins have passed
                        if (j >0 ||(timeSinceOldestRequest >120 *1000
                            &&(timeSinceData <0 ||timeSinceData >(isSeed ?120 :60) *1000)))
                        {
                            pc.sendCancel(request);             //cancel the request object
                            //get the piece number
                            final int pieceNumber = request.getPieceNumber();
                            PEPiece	pe_piece = pePieces[pieceNumber];
                            //unmark the request on the block
                            if ( pe_piece != null )
                            	pe_piece.clearRequested(request.getOffset() /DiskManager.BLOCK_SIZE);
                             // remove piece if empty so peers can choose something else, except in end game
                            if (!piecePicker.isInEndGameMode())
                                checkEmptyPiece(pieceNumber);
                        }
                    }
				}
			}
		}
	}
	
	
  private void 
  updateTrackerAnnounceInterval() 
  {
  	if ( mainloop_loop_count % MAINLOOP_FIVE_SECOND_INTERVAL != 0 ){
  		
  		return;
  	}
  	
  	final int WANT_LIMIT = 100;
      
	  int num_wanted = getMaxNewConnectionsAllowed();
	  
	  final boolean has_remote = adapter.isNATHealthy();
	  if( has_remote ) {
	    //is not firewalled, so can accept incoming connections,
	    //which means no need to continually keep asking the tracker for peers
	    num_wanted = (int)(num_wanted / 1.5);
	  }
	  
	  if ( num_wanted < 0 || num_wanted > WANT_LIMIT ) {
	    num_wanted = WANT_LIMIT;
	  }
	  
	  int current_connection_count = PeerIdentityManager.getIdentityCount( _hash );
	  
	  final TRTrackerScraperResponse tsr = adapter.getTrackerScrapeResponse();
	  
	  if( tsr != null && tsr.isValid() ) {  //we've got valid scrape info
	    final int num_seeds = tsr.getSeeds();   
	    final int num_peers = tsr.getPeers();
	
	    final int swarm_size;
	    
	    if( seeding_mode ) {
	      //Only use peer count when seeding, as other seeds are unconnectable.
	      //Since trackers return peers randomly (some of which will be seeds),
	      //backoff by the seed2peer ratio since we're given only that many peers
	      //on average each announce.
	      final float ratio = (float)num_peers / (num_seeds + num_peers);
	      swarm_size = (int)(num_peers * ratio);
	    }
	    else {
	      swarm_size = num_peers + num_seeds;
	    }
	    
	    if( swarm_size < num_wanted ) {  //lower limit to swarm size if necessary
	      num_wanted = swarm_size;
	    }
	  }
	  
	  if( num_wanted < 1 ) {  //we dont need any more connections
	    adapter.setTrackerRefreshDelayOverrides( 100 );  //use normal announce interval
	    return;
	  }
	  
	  if( current_connection_count == 0 )  current_connection_count = 1;  //fudge it :)
	  
	  final int current_percent = (current_connection_count * 100) / (current_connection_count + num_wanted);
	  
	  adapter.setTrackerRefreshDelayOverrides( current_percent );  //set dynamic interval override
  }

  	public boolean
  	hasDownloadablePiece()
  	{
  		return( piecePicker.hasDownloadablePiece());