dhtrouterimpl.java

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

	findContact(
		byte[]		node_id )
	{
		Object[]	res = findContactSupport( node_id );
				
		return((DHTRouterContact)res[1]);
	}
	
	protected DHTRouterNodeImpl
	findNode(
		byte[]	node_id )
	{
		Object[]	res = findContactSupport( node_id );
				
		return((DHTRouterNodeImpl)res[0]);	
	}
	
	protected Object[]
	findContactSupport(
		byte[]		node_id )
	{
		try{
			this_mon.enter();
		
			DHTRouterNodeImpl	current_node	= root;
			
			for (int i=0;i<node_id.length;i++){
				
				if ( current_node.getBuckets() != null ){
				
					break;
				}
	
				byte	b = node_id[i];
				
				int	j = 7;
				
				while( j >= 0 ){
						
					boolean	bit = ((b>>j)&0x01)==1?true:false;
					
					if ( current_node.getBuckets() != null ){
						
						break;
					}
									
					if ( bit ){
						
						current_node = current_node.getLeft();
						
					}else{
						
						current_node = current_node.getRight();
					}
					
					j--;
				}
			}
			
			List	buckets = current_node.getBuckets();
			
			for (int k=0;k<buckets.size();k++){
				
				DHTRouterContactImpl	contact = (DHTRouterContactImpl)buckets.get(k);
				
				if ( Arrays.equals(node_id, contact.getID())){
	
					return( new Object[]{ current_node, contact });
				}
			}
			
			return( new Object[]{ current_node, null });
			
		}finally{
			
			this_mon.exit();
		}
	}
	
	protected long
	getNodeCount()
	{
		return( getNodeCount( root ));
	}
	
	protected long
	getNodeCount(
		DHTRouterNodeImpl	node )
	{
		if ( node.getBuckets() != null ){
			
			return( 1 );
			
		}else{
			
			return( 1 + getNodeCount( node.getLeft())) + getNodeCount( node.getRight());
		}
	}
	
	protected long
	getContactCount()
	{
		return( getContactCount( root ));
	}
	
	protected long
	getContactCount(
		DHTRouterNodeImpl	node )
	{
		if ( node.getBuckets() != null ){
			
			return( node.getBuckets().size());
			
		}else{
			
			return( getContactCount( node.getLeft())) + getContactCount( node.getRight());
		}
	}
	
	public List
	findBestContacts(
		int		max )
	{
		Set	set = 
			new TreeSet(
					new Comparator()
					{
						public int
						compare(
							Object	o1,
							Object	o2 )
						{
							DHTRouterContactImpl	c1 = (DHTRouterContactImpl)o1;
							DHTRouterContactImpl	c2 = (DHTRouterContactImpl)o2;
							
							return((int)( c2.getTimeAlive() - c1.getTimeAlive()));
						}
					});
		
		
		try{
			this_mon.enter();

			findAllContacts( set, root );
			
		}finally{
			
			this_mon.exit();
		}
		
		List	result = new ArrayList( max );
	
		Iterator	it = set.iterator();
		
		while( it.hasNext() && (max <= 0 || result.size() < max )){
			
			result.add( it.next());
		}
		
		return( result );
	}
	
	public List
	getAllContacts()
	{
		try{
			this_mon.enter();

			List	l = new ArrayList();
			
			findAllContacts( l, root );
			
			return( l );
			
		}finally{
			
			this_mon.exit();
		}
	}
	
	protected void
	findAllContacts(
		Set					set,
		DHTRouterNodeImpl	node )
	{
		List	buckets = node.getBuckets();
		
		if ( buckets == null ){
			
			findAllContacts( set, node.getLeft());
			
			findAllContacts( set, node.getRight());
		}else{
			
			for (int i=0;i<buckets.size();i++){
				
				DHTRouterContactImpl	contact = (DHTRouterContactImpl)buckets.get(i);
								
				set.add( contact );
			}
		}
	}
	
	protected void
	findAllContacts(
		List				list,
		DHTRouterNodeImpl	node )
	{
		List	buckets = node.getBuckets();
		
		if ( buckets == null ){
			
			findAllContacts( list, node.getLeft());
			
			findAllContacts( list, node.getRight());
		}else{
			
			for (int i=0;i<buckets.size();i++){
				
				DHTRouterContactImpl	contact = (DHTRouterContactImpl)buckets.get(i);
								
				list.add( contact );
			}
		}
	}
	
	public void
	seed()
	{
			// refresh all buckets apart from closest neighbour
		
		byte[]	path = new byte[router_node_id.length];
		
		List	ids = new ArrayList();
		
		try{
			this_mon.enter();
			
			refreshNodes( ids, root, path, true, 0 );
			
		}finally{
			
			this_mon.exit();
		}
		
		for (int i=0;i<ids.size();i++){
			
			requestLookup((byte[])ids.get(i), "Seeding DHT" );
		}
	}
	
	protected void
	refreshNodes(
		List				nodes_to_refresh,
		DHTRouterNodeImpl	node,
		byte[]				path,
		boolean				seeding,
		long				max_permitted_idle )	// 0 -> don't check
	{
			// when seeding we don't do the smallest subtree
		
		if ( seeding && node == smallest_subtree ){
			
			return;
		}
		
		if ( max_permitted_idle != 0 ){
			
			if ( node.getTimeSinceLastLookup() <= max_permitted_idle ){
				
				return;
			}
		}
		
		if ( node.getBuckets() != null ){
		
				// and we also don't refresh the bucket containing the router id when seeding
			
			if ( seeding && node.containsRouterNodeID()){
				
				return;
			}
			
			refreshNode( nodes_to_refresh, node, path );		
		}
		
			// synchronous refresh may result in this bucket being split
			// so we retest here to refresh sub-buckets as required
		
		if ( node.getBuckets() == null ){
			
			int	depth = node.getDepth();
			
			byte	mask = (byte)( 0x01<<(7-(depth%8)));
			
			path[depth/8] = (byte)( path[depth/8] | mask );
			
			refreshNodes( nodes_to_refresh, node.getLeft(), path,seeding, max_permitted_idle  );
			
			path[depth/8] = (byte)( path[depth/8] & ~mask );
		
			refreshNodes( nodes_to_refresh, node.getRight(), path,seeding, max_permitted_idle  );
		}
	}
	
	protected void
	refreshNode(
		List				nodes_to_refresh,
		DHTRouterNodeImpl	node,
		byte[]				path )
	{
			// pick a random id in the node's range.
		
		byte[]	id = new byte[router_node_id.length];
		
		random.nextBytes( id );
		
		int	depth = node.getDepth();
		
		for (int i=0;i<depth;i++){
			
			byte	mask = (byte)( 0x01<<(7-(i%8)));
			
			boolean bit = ((path[i/8]>>(7-(i%8)))&0x01 ) == 1;

			if ( bit ){
				
				id[i/8] = (byte)( id[i/8] | mask );

			}else{
				
				id[i/8] = (byte)( id[i/8] & ~mask );	
			}
		}
		
		nodes_to_refresh.add( id );
	}
	
	protected DHTRouterNodeImpl
	getSmallestSubtree()
	{
		return( smallest_subtree );
	}
	
	public void
	recordLookup(
		byte[]	node_id )
	{
		findNode( node_id ).setLastLookupTime();
	}
	
	public void
	refreshIdleLeaves(
		long	idle_max)
	{
		// while we are synchronously refreshing the smallest subtree the tree can mutate underneath us
		// as new contacts are discovered. We NEVER merge things back together
		
		byte[]	path = new byte[router_node_id.length];
		
		List	ids = new ArrayList();
		
		try{
			this_mon.enter();
			
			refreshNodes( ids, root, path, false, idle_max );
			
		}finally{
			
			this_mon.exit();
		}
		
		for (int i=0;i<ids.size();i++){
			
			requestLookup((byte[])ids.get(i), "Idle leaf refresh" );
		}
	}
	
	public boolean
	requestPing(
		byte[]		node_id )
	{
		Object[] res = findContactSupport( node_id );
	
		DHTRouterContactImpl	contact = (DHTRouterContactImpl)res[1];
		
		if ( contact != null ){
			
			adapter.requestPing( contact );
			
			return( true );
		}
		
		return( false );
	}
	
	protected void
	requestPing(
		DHTRouterContactImpl	contact )
	{
			// make sure we don't do the ping when synchronised
		
		DHTLog.log( "DHTRouter: requestPing:" + DHTLog.getString( contact.getID()));
		
		if ( contact == local_contact ){
			
			Debug.out( "pinging local contact" );
		}
		
		try{
			this_mon.enter();
			
			if ( !outstanding_pings.contains( contact )){
			
				outstanding_pings.add( contact );
			}
		}finally{
			
			this_mon.exit();
		}
	}
	
	protected void
	dispatchPings()
	{
		if ( outstanding_pings.size() == 0 ){
			
			return;
		}
		
		List	pings;
		
		try{
			this_mon.enter();
		
			pings	= outstanding_pings;
			
			outstanding_pings = new ArrayList();
			
		}finally{
			
			this_mon.exit();
		}
		
		for (int i=0;i<pings.size();i++){
			
			adapter.requestPing((DHTRouterContactImpl)pings.get(i));
		}
	}
	
	protected void
	requestNodeAdd(
		DHTRouterContactImpl	contact )
	{
			// make sure we don't do the addition when synchronised
		
		DHTLog.log( "DHTRouter: requestNodeAdd:" + DHTLog.getString( contact.getID()));
		
		if ( contact == local_contact ){

			Debug.out( "adding local contact" );
		}
		
		try{
			this_mon.enter();
			
			if ( !outstanding_adds.contains( contact )){
			
				outstanding_adds.add( contact );
			}
		}finally{
			
			this_mon.exit();
		}
	}
	
	protected void
	dispatchNodeAdds()
	{
		if ( outstanding_adds.size() == 0 ){
			
			return;
		}
		
		List	adds;
		
		try{
			this_mon.enter();
		
			adds	= outstanding_adds;
			
			outstanding_adds = new ArrayList();
			
		}finally{
			
			this_mon.exit();
		}
		
		for (int i=0;i<adds.size();i++){
			
			adapter.requestAdd((DHTRouterContactImpl)adds.get(i));
		}
	}

	public byte[]
	refreshRandom()
	{
		byte[]	id = new byte[router_node_id.length];
		
		random.nextBytes( id );
	
		requestLookup( id, "Random Refresh" );
		
		return( id );
	}
	
	protected void
	requestLookup(
		byte[]		id,
		String		description )
	{
		DHTLog.log( "DHTRouter: requestLookup:" + DHTLog.getString( id ));
		
		adapter.requestLookup( id, description );
	}
	
	protected void
	getStatsSupport(
		long[]				stats_array,
		DHTRouterNodeImpl	node )
	{
		stats_array[DHTRouterStats.ST_NODES]++;
		
		List	buckets = node.getBuckets();
		
		if ( buckets == null ){
			
			getStatsSupport( stats_array, node.getLeft());
			
			getStatsSupport( stats_array, node.getRight());
			
		}else{
			
			stats_array[DHTRouterStats.ST_LEAVES]++;
			
			stats_array[DHTRouterStats.ST_CONTACTS] += buckets.size();
			
			for (int i=0;i<buckets.size();i++){
				
				DHTRouterContactImpl	contact = (DHTRouterContactImpl)buckets.get(i);
				
				if ( contact.getFirstFailTime() > 0 ){
					
					stats_array[DHTRouterStats.ST_CONTACTS_DEAD]++;
					
				}else if ( contact.hasBeenAlive()){
					
					stats_array[DHTRouterStats.ST_CONTACTS_LIVE]++;
					
				}else{
					
					stats_array[DHTRouterStats.ST_CONTACTS_UNKNOWN]++;
				}
			}
			
			List	rep = node.getReplacements();
			
			if ( rep != null ){
				
				stats_array[DHTRouterStats.ST_REPLACEMENTS] += rep.size();
			}
		}
	}
	
	protected long[]
	getStatsSupport()
	{
		 /* number of nodes
		 * number of leaves
		 * number of contacts
		 * number of replacements
		 * number of live contacts
		 * number of unknown contacts
		 * number of dying contacts
		 */
		
		try{
			this_mon.enter();
		
			long[]	res = new long[7];
		
			getStatsSupport( res, root );
		
			return( res );
			
		}finally{
			
			this_mon.exit();
		}
	}
	
	protected void
	log(
		String	str )
	{
		logger.log( str );
	}
	
	public void
	print()
	{
		try{
			this_mon.enter();
		
			log( "DHT: " + DHTLog.getString2(router_node_id) + ", node count = " + getNodeCount()+ ", contacts =" + getContactCount());
		
			root.print( "", "" );
			
		}finally{
			
			this_mon.exit();
		}
	}
	
	public void
	destroy()
	{
		notifyDead();
	}
}