dhtcontrolimpl.java

一个基于JAVA的多torrent下载程序

		
				Map.Entry	entry = (Map.Entry)it.next();
				
				HashWrapper	key		= (HashWrapper)entry.getKey();
				
				List		values	= (List)entry.getValue();
		
				keys[index] 		= key.getHash();
				value_sets[index]	= new DHTTransportValue[values.size()];
				
				
				for (int i=0;i<values.size();i++){
					
					value_sets[index][i] = ((DHTDBValue)values.get(i)).getValueForRelay( local_contact );
				}
				
				index++;
			}
			
				// move to anti-spoof for cache forwards. we gotta do a findNode to update the
				// contact's latest random id
			
			t_contact.sendFindNode(
					new DHTTransportReplyHandlerAdapter()
					{
						public void
						findNodeReply(
							DHTTransportContact 	contact,
							DHTTransportContact[]	contacts )
						{	
							// System.out.println( "nodeAdded: pre-store findNode OK" );
							
							t_contact.sendStore( 
									new DHTTransportReplyHandlerAdapter()
									{
										public void
										storeReply(
											DHTTransportContact _contact,
											byte[]				_diversifications )
										{
											// System.out.println( "nodeAdded: store OK" );

												// don't consider diversifications for node additions as they're not interested
												// in getting values from us, they need to get them from nodes 'near' to the 
												// diversification targets or the originator
											
											DHTLog.log( "add store ok" );
											
											router.contactAlive( _contact.getID(), new DHTControlContactImpl(_contact));
										}	
										
										public void
										failed(
											DHTTransportContact 	_contact,
											Throwable				_error )
										{
											// System.out.println( "nodeAdded: store Failed" );

											DHTLog.log( "add store failed " + DHTLog.getString( _contact ) + " -> failed: " + _error.getMessage());
																					
											router.contactDead( _contact.getID(), false);
										}
									},
									keys, 
									value_sets );
						}
						
						public void
						failed(
							DHTTransportContact 	_contact,
							Throwable				_error )
						{
							// System.out.println( "nodeAdded: pre-store findNode Failed" );

							DHTLog.log( "pre-store findNode failed " + DHTLog.getString( _contact ) + " -> failed: " + _error.getMessage());
																	
							router.contactDead( _contact.getID(), false);
						}
					},
					t_contact.getProtocolVersion() >= DHTTransportUDP.PROTOCOL_VERSION_ANTI_SPOOF2?new byte[0]:new byte[20] );
						
		}else{
			
			if ( !new_contact.hasBeenAlive()){
				
				requestPing( new_contact );
			}
		}
	}
	
	protected Set
	getClosestContactsSet(
		byte[]		id,
		boolean		live_only )
	{
		List	l = router.findClosestContacts( id, live_only );
		
		Set	sorted_set	= new sortedTransportContactSet( id, true ).getSet(); 

		for (int i=0;i<l.size();i++){
			
			sorted_set.add(((DHTControlContactImpl)((DHTRouterContact)l.get(i)).getAttachment()).getTransportContact());
		}
		
		return( sorted_set );
	}
	
	public List
	getClosestKContactsList(
		byte[]		id,
		boolean		live_only )
	{
		Set	sorted_set	= getClosestContactsSet( id, live_only );
					
		List	res = new ArrayList(K);
		
		Iterator	it = sorted_set.iterator();
		
		while( it.hasNext() && res.size() < K ){
			
			res.add( it.next());
		}
		
		return( res );
	}
	
	protected byte[]
	encodeKey(
		byte[]		key )
	{
		byte[]	temp = new SHA1Simple().calculateHash( key );
		
		byte[]	result =  new byte[node_id_byte_count];
		
		System.arraycopy( temp, 0, result, 0, node_id_byte_count );
		
		return( result );
	}
	
	public int
	computeAndCompareDistances(
		byte[]		t1,
		byte[]		t2,
		byte[]		pivot )
	{
		return( computeAndCompareDistances2( t1, t2, pivot ));
	}
	
	protected static int
	computeAndCompareDistances2(
		byte[]		t1,
		byte[]		t2,
		byte[]		pivot )
	{
		for (int i=0;i<t1.length;i++){

			byte d1 = (byte)( t1[i] ^ pivot[i] );
			byte d2 = (byte)( t2[i] ^ pivot[i] );

			int diff = (d1&0xff) - (d2&0xff);
			
			if ( diff != 0 ){
				
				return( diff );
			}
		}
		
		return( 0 );
	}
	
	public byte[]
	computeDistance(
		byte[]		n1,
		byte[]		n2 )
	{
		return( computeDistance2( n1, n2 ));
	}
	
	protected static byte[]
	computeDistance2(
		byte[]		n1,
		byte[]		n2 )
	{
		byte[]	res = new byte[n1.length];
		
		for (int i=0;i<res.length;i++){
			
			res[i] = (byte)( n1[i] ^ n2[i] );
		}
		
		return( res );
	}
	
		/**
		 * -ve -> n1 < n2
		 * @param n1
		 * @param n2
		 * @return
		 */
	
	public int
	compareDistances(
		byte[]		n1,
		byte[]		n2 )
	{
		return( compareDistances2( n1,n2 ));
	}
	
	protected static int
	compareDistances2(
		byte[]		n1,
		byte[]		n2 )
	{
		for (int i=0;i<n1.length;i++){
			
			int diff = (n1[i]&0xff) - (n2[i]&0xff);
			
			if ( diff != 0 ){
				
				return( diff );
			}
		}
		
		return( 0 );
	}
	
	public void
	addListener(
		DHTControlListener	l )
	{
		try{
			activity_mon.enter();
			
			listeners.addListener( l );
			
			for (int i=0;i<activities.size();i++){
				
				listeners.dispatch( DHTControlListener.CT_ADDED, activities.get(i));
			}
			
		}finally{
			
			activity_mon.exit();
		}
	}
	
	public void
	removeListener(
		DHTControlListener	l )
	{
		listeners.removeListener( l );	
	}
		
	public DHTControlActivity[]
	getActivities()
	{
		List	res;
		
		try{
			
			activity_mon.enter();
			
			res = new ArrayList( activities );
			
		}finally{
		
			activity_mon.exit();
		}
		
		DHTControlActivity[]	x = new DHTControlActivity[res.size()];
		
		res.toArray( x );
		
		return( x );
	}
	
	public void
	setTransportEstimatedDHTSize(
		int						size )
	{
		if ( size > 0 ){
			
			try{
				estimate_mon.enter();
				
				remote_estimate_values.add( new Integer( size ));
				
				if ( remote_estimate_values.size() > REMOTE_ESTIMATE_HISTORY ){
					
					remote_estimate_values.remove(0);
				}
			}finally{
				
				estimate_mon.exit();
			}
		}
		// System.out.println( "estimated dht size: " + size );
	}
	
	public int
	getTransportEstimatedDHTSize()
	{
		return((int)local_dht_estimate );
	}
	
	public int
	getEstimatedDHTSize()
	{
			// public method, trigger actual computation periodically
		
		long	now = SystemTime.getCurrentTime();
		
		long	diff = now - last_dht_estimate_time;
		
		if ( diff < 0 || diff > 60*1000 ){

			estimateDHTSize( router.getID(), null, router.getK());
		}
		
		return((int)combined_dht_estimate );
	}
	
	protected void
	estimateDHTSize(
		byte[]	id,
		Map		contacts,
		int		contacts_to_use )
	{
			// if called with contacts then this is in internal estimation based on lookup values
		
		long	now = SystemTime.getCurrentTime();
		
		long	diff = now - last_dht_estimate_time;
			
			// 5 second limiter here
		
		if ( diff < 0 || diff > 5*1000 ){

			try{
				estimate_mon.enter();
	
				last_dht_estimate_time	= now;
				
				List	l;
				
				if ( contacts == null ){
					
					l = getClosestKContactsList( id, false );
					
				}else{
					
					Set	sorted_set	= new sortedTransportContactSet( id, true ).getSet(); 
		
					sorted_set.addAll( contacts.values());
					
					l = new ArrayList( sorted_set );
					
					if ( l.size() > 0 ){
				
							// algorithm works relative to a starting point in the ID space so we grab
							// the first here rather than using the initial lookup target
						
						id = ((DHTTransportContact)l.get(0)).getID();
					}
					
					/*
					String	str = "";
					for (int i=0;i<l.size();i++){
						str += (i==0?"":",") + DHTLog.getString2( ((DHTTransportContact)l.get(i)).getID());
					}
					System.out.println( "trace: " + str );
					*/
				}
				
					// can't estimate with less than 2
				
				if ( l.size() > 2 ){
					
				
					/*
					<Gudy> if you call N0 yourself, N1 the nearest peer, N2 the 2nd nearest peer ... Np the pth nearest peer that you know (for example, N1 .. N20)
					<Gudy> and if you call D1 the Kad distance between you and N1, D2 between you and N2 ...
					<Gudy> then you have to compute :
					<Gudy> Dc = sum(i * Di) / sum( i * i)
					<Gudy> and then :
					<Gudy> NbPeers = 2^160 / Dc
					*/
					
					BigInteger	sum1 = new BigInteger("0");
					BigInteger	sum2 = new BigInteger("0");
					
						// first entry should be us
							
					for (int i=1;i<Math.min( l.size(), contacts_to_use );i++){
						
						DHTTransportContact	node = (DHTTransportContact)l.get(i);
						
						byte[]	dist = computeDistance( id, node.getID());
						
						BigInteger b_dist = IDToBigInteger( dist );
						
						BigInteger	b_i = new BigInteger(""+i);
						
						sum1 = sum1.add( b_i.multiply(b_dist));
						
						sum2 = sum2.add( b_i.multiply( b_i ));
					}
					
					byte[]	max = new byte[id.length+1];
					
					max[0] = 0x01;
					
					long this_estimate;
					
					if ( sum1.compareTo( new BigInteger("0")) == 0 ){
						
						this_estimate = 0;
						
					}else{
						
						this_estimate = IDToBigInteger(max).multiply( sum2 ).divide( sum1 ).longValue();
					}
					
						// there's always us!!!!
					
					if ( this_estimate < 1 ){
						
						this_estimate	= 1;
					}
					
					local_estimate_values.put( new HashWrapper( id ), new Long( this_estimate ));
					
					long	new_estimate	= 0;
					
					Iterator	it = local_estimate_values.values().iterator();
					
					String	sizes = "";
						
					while( it.hasNext()){
						
						long	estimate = ((Long)it.next()).longValue();
						
						sizes += (sizes.length()==0?"":",") + estimate;
						
						new_estimate += estimate;
					}
					
					local_dht_estimate = new_estimate/local_estimate_values.size();
					
					// System.out.println( "getEstimatedDHTSize: " + sizes + "->" + dht_estimate + " (id=" + DHTLog.getString2(id) + ",cont=" + (contacts==null?"null":(""+contacts.size())) + ",use=" + contacts_to_use );
				}
				
				List rems = new ArrayList(new TreeSet( remote_estimate_values ));
				
				// ignore largest and smallest few values
				
				long	rem_average = local_dht_estimate;
				int		rem_vals	= 1;
				
				for (int i=3;i<rems.size()-3;i++){
				
					rem_average += ((Integer)rems.get(i)).intValue();
					
					rem_vals++;
				}
				
				combined_dht_estimate = rem_average / rem_vals;
				
				// System.out.println( "estimateDHTSize: loc =" + local_dht_estimate + ", comb = " + combined_dht_estimate + " [" + remote_estimate_values.size() + "]");
			}finally{
				
				estimate_mon.exit();
			}
		}
	}
	
	protected BigInteger
	IDToBigInteger(
		byte[]		data )
	{
		String	str_key = "";
		
		for (int i=0;i<data.length;i++){
			
			String	hex = Integer.toHexString( data[i]&0xff );
			
			while( hex.length() < 2 ){
				
				hex = "0" + hex;
			}
				
			str_key += hex;
		}
				
		BigInteger	res		= new BigInteger( str_key, 16 );	
		
		return( res );
	}
	
	protected int
	generateSpoofID(
		DHTTransportContact	contact )
	{
		if ( spoof_cipher == null  ){
			
			return( 0 );
		}
		
		try{
			spoof_mon.enter();
			
			spoof_cipher.init(Cipher.ENCRYPT_MODE, spoof_key ); 
		
			byte[]	address = contact.getAddress().getAddress().getAddress();
					
			byte[]	data_out = spoof_cipher.doFinal( add