vivaldiclient.java

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

	synchronized public void reset() {
		sys_coord.reset();
		app_coord.reset();
		error = MAX_ERROR;
		rs_map.clear();
		start_coords.clear();
		current_coords.clear();
		nearest_neighbor = null;		
	}

	/**
	 * Returns whether the application level coordinates have been updated at
	 * least once.
	 * 
	 * @return <code>true</code> if updated, <code>false</code> otherwise
	 */
  synchronized public boolean updatedYet() {
		return updated_app_coord_at_least_once;
	}

	/**
	 * Returns the dimension of the Euclidian space coordinates are embedded in.
	 * 
	 * @return the coordinate space dimension
	 */
  synchronized public int getNumDimensions() {
		return num_dims;
	}

	/**
	 * Returns the application-level Vivaldi coordinates.
	 * 
	 * @return the application-level coordinates
	 */
  synchronized public Coordinate getApplicationCoords() {
		return new Coordinate(sys_coord);
	}

	/**
	 * Returns the system-level Vivaldi coordinates. These coordinates change
	 * more frequently than the application-level coordinates.
	 * 
	 * @return the system-level coordinates
	 */
  synchronized public Coordinate getSystemCoords() {
		return new Coordinate(sys_coord);
	}

  /**
   * Returns the system-level error, which denotes the accuracy of the
   * system-level coordinates.
   * 
   * @return the system-level error
   */
  synchronized public double getSystemError() {
    return error;
  }

  /**
   * Returns the age of our coordinate
   * Note that this does not require clock-synchronization
   * because it is relative to our coordinate
   * 
   * @return relative age of our coordinate since we last updated it
   */
  synchronized public long getAge(long curr_time) {
    if (curr_time < time_of_last_sys_update) return 0;
    return curr_time - time_of_last_sys_update;
  }

  
	/**
	 * Returns the list of observers, to which observers for the
	 * application-level coordinate can be added, removed, and so forth.
	 * 
	 * @return the list of observers for the application-level coordinate
	 */
  synchronized public ObserverList getObserverList() {
		return obs_list;
	}

	/**
	 * Notifies this <code>VivaldiClient</code> object that a host that
	 * supports Vivaldi has joined the system. State associated with the new
	 * host is created. This method succeeds and returns <code>true</code>
	 * only if the host is not already registered with this
	 * <code>VivaldiClient</code> object.
	 * 
	 * @param addr
	 * the address of the joining host
	 * @return <code>true</code> if <code>addr</code> is registered and its
	 * associated state created, <code>false</code> otherwise
	 */
  synchronized public boolean addHost(T addr) {
		if (rs_map.containsKey(addr)) {
			return false;
		}

		RemoteState<T> rs = new RemoteState<T>(addr);
		rs_map.put(addr, rs);
		return true;
	}

	/**
	 * Notifies this <code>VivaldiClient</code> object that a host that
	 * supports Vivaldi and has the provided coordinates and error has joined
	 * the system. State associated with the new host is created. This method
	 * succeeds and returns <code>true</code> only if the host is not already
	 * registered with this <code>VivaldiClient</code> object.
	 * 
	 * @param addr
	 * the address of the joining host
	 * @param _r_coord
	 * the app-level coordinates of the remote host
	 * @param r_error
	 * the system-level error of the remote host
	 * @param sample_rtt
	 * the RTT sample to the remote host
	 * @param curr_time
	 * the current time, in milliseconds
	 * @param can_update
	 * <code>true</code> if this method can update a host already present
	 * @return <code>true</code> if <code>addr</code> is registered and its
	 * associated state created, <code>false</code> otherwise
	 */

  synchronized public boolean addHost(T addr, Coordinate _r_coord, double r_error,
			long curr_time, boolean can_update) {
		RemoteState<T> rs = null;
		if (rs_map.containsKey(addr)) {
			if (!can_update) {
				return false;
			}
			rs = rs_map.get(addr);
		}
		else {
			rs = new RemoteState<T>(addr);
			rs_map.put(addr, rs);
		}

		Coordinate r_coord = _r_coord.makeCopy();
		rs.assign(r_coord, r_error, curr_time);
		return true;
	}

	/**
	 * Notifies this <code>VivaldiClient</code> object that a host that
	 * supports Vivaldi has left the system. 
   * However, the state (i.e. short list of RTT values) is kept because
   * it will be useful if and when the node returns into the system
	 * @param addr
	 * the address of the departing host
	 * @return <code>true</code> if <code>addr</code> was a known node
	 */

  synchronized public boolean removeHost(T addr) {
    if (rs_map.containsKey(addr)) {
      return true;
    }
    return false;
  }

	/**
	 * Returns whether the given host has been registered with this
	 * <code>VivaldiClient</code> object.
	 * 
	 * @param addr
	 * the address to query as registered
	 * @return <code>true</code> if registered, <code>false</code> otherwise
	 */
  synchronized public boolean containsHost(T addr) {
		return rs_map.containsKey(addr);
	}

	/**
	 * Returns all hosts that support Vivaldi and have been registered with this
	 * <code>VivaldiClient</code> object. The returned set is backed by the
	 * true set of registered hosts, but cannot be modified.
	 * 
	 * @return the set of registered Vivaldi-supporting hosts
	 */
  synchronized public Set<T> getHosts() {
		return hosts;
	}

	/**
	 * This method is invoked when a new RTT sample is made to a host that
	 * supports Vivaldi. This method succeeds and returns <code>true</code>
	 * only if the host is already registered with this
	 * <code>VivaldiClient</code> object, and the RTT sample is valid.
	 * 
	 * @param addr
	 * the address of the host
	 * @param _r_coord
	 * the system-level coordinates of the remote host
	 * @param r_error
	 * the system-level error of the remote host
	 * @param sample_rtt
	 * the RTT sample to the remote host
	 * @param curr_time
	 * the current time, in milliseconds
	 * @param can_add
	 * <code>true</code> if this method can add a host not already present
	 * @return <code>true</code> if <code>addr</code> is registered and the
	 * sample is processed, <code>false</code> otherwise
	 */

  synchronized public boolean processSample(T addr, Coordinate _r_coord, double r_error,
			double sample_rtt, long sample_age, long curr_time, boolean can_add) {

    //if (debugCrawler) crawler_log.info("sample addr="+addr+" rtt="+sample_rtt);
    int id = getIdFromAddr (addr);
    if (debugCrawler && debugGood) crawler_log.info(id+" START");
    
    assert (_r_coord != sys_coord);
		assert (_r_coord != null);
		assert (sys_coord != null);
   
    if (!sys_coord.isCompatible(_r_coord)) {
      if (debugCrawler && debug) crawler_log.info("INVALID "+id+" s "+sample_rtt+" NOT_COMPAT "+_r_coord.getVersion());
      return false;
    }
    
		// There is a major problem with the coord.
		// However, if this is happening, it will probably
		// happen again and again.
		// Note that error is checked and fixed in updateError()
		if (!sys_coord.isValid() || Double.isNaN(error)) {
			//System.err.println("Warning: resetting Vivaldi coordinate");
      if (debugCrawler || SIMULATION) crawler_log.info(id + " RESET, USE_HEIGHT="+USE_HEIGHT);
			reset();
		}
   
    if (r_error <= 0. || r_error > MAX_ERROR || Double.isNaN(r_error) || !_r_coord.isValid()) {
      if (debugCrawler && debug) crawler_log.info(id+" BUSTED his coord is busted: r_error "+r_error+" r_coord "+_r_coord);  
      return false;
    }

		if (sample_rtt > OUTRAGEOUSLY_LARGE_RTT) {
			if (debug)
			  System.err.println("Warning: skipping huge RTT of "
						+ nf.format(sample_rtt) + " from " + addr);
      if (debugCrawler) crawler_log.info(id+ " HUGE "+sample_rtt);
			return false;
		}

		RemoteState<T> addr_rs = rs_map.get(addr);
		if (addr_rs == null) {
			if (!can_add) {
        if (debugCrawler) crawler_log.info(id+ " NO_ADD");  
				return false;
			}
			addHost(addr);
			addr_rs = rs_map.get(addr);
		}
		Coordinate r_coord = _r_coord.makeCopy();
    
		// add sample to history, then get smoothed rtt based on percentile
		addr_rs.addSample(sample_rtt, sample_age, r_coord, r_error, curr_time);

    // even if we aren't going to use him this time around, we remember this RTT
    
    if (sys_coord.atOrigin()) {
	sys_coord.bump();
    }
    
    boolean didUpdate = false;
    int sample_size = addr_rs.getSampleSize();
    double smoothed_rtt = addr_rs.getSample();
    
    if (addr_rs.isValid(curr_time)) {
      addNeighbor(addr_rs);
      // first update our error
      updateError(addr, r_coord, r_error, smoothed_rtt, sample_rtt, sample_age, sample_size, curr_time);
      // next, update our system-level coordinate
      updateSystemCoordinate(curr_time);  
      // last, try to update our application-level coordinate
      tryUpdateAppCoordinate(curr_time);
      didUpdate = true;

    } else {
      if (debugCrawler && debug) {
        String reason;
        if (addr_rs.getSampleSize() < RemoteState.MIN_SAMPLE_SIZE) {
          reason = "TOO_FEW";
        } else if (addr_rs.getSample() <= 0) {
          reason = "sample is "+addr_rs.getSample();
        } else if (addr_rs.getLastError() <= 0.) {
          reason = "error is "+addr_rs.getLastError();
        } else if (addr_rs.getLastUpdateTime() <= 0) {
          reason = "last update "+addr_rs.getLastUpdateTime();
        } else if (addr_rs.getLastCoordinate().atOrigin()) {
          reason = "AT_ORIGIN";
        } else {
          reason = "UNKNOWN";
        }
        crawler_log.info("INVALID "+id+" s "+sample_rtt+" ss "+smoothed_rtt+" c "+sample_size+
          " "+reason);

      } 
    }

    //System.out.println ("maint?");
    if (lastMaintenanceStamp < curr_time - MAINTENANCE_PERIOD) {
	performMaintenance (curr_time);
	lastMaintenanceStamp = curr_time;
    }
    
		return didUpdate;
	}
  
  private Map<T,Integer> addr2id = new HashMap<T,Integer>();
  private int idCounter = 0;

  // If remote nodes are already represented by ints, just use them
  // otherwise, translate into more easily read-able ID
  private int getIdFromAddr (T addr) {
	if ( debugCrawler || SIMULATION ){
		
	    if (addr instanceof Integer) {
	      return ((Integer)addr).intValue();
	    }
	      
	    if (!addr2id.containsKey(addr)) {
	      addr2id.put(addr,idCounter);  
	      idCounter++;
	    }
	    return addr2id.get(addr);
	}
	
	return(0);
  }
  
  protected void updateError(T addr, Coordinate r_coord, double r_error,
    double smoothed_rtt, double sample_rtt, long sample_age, int sample_size, long curr_time) {
		// get the coordinate distance
    double sys_distance = sys_coord.distanceTo(r_coord);
    double app_distance = 0;

    if (app_coord != null)
      app_distance = app_coord.distanceTo(r_coord);
    if (app_distance == 0. || sys_distance == 0.) {
      if (debugCrawler) crawler_log.info("bad distance sys "+sys_distance+" app "+app_distance);      
      return;
    }

    
    // get sample error in terms of coordinate distance and sample rtt
    // Note that smoothed_rtt must be greater than zero
    // or we wouldn't have entered this function

    // Note that app_sample_error is only giving us a limited amount of info
    // because his app coord is not going over the wire
    
    assert (smoothed_rtt > 0.); 
    double sys_sample_error = Math.abs(sys_distance - smoothed_rtt) / smoothed_rtt;
    double app_sample_error = Math.abs(app_distance - smoothed_rtt) / smoothed_rtt;


    if (debugCrawler) {
        int remote_id = getIdFromAddr (addr);
        String info = 
        //"lID "+local_addr+" rID "+
        "UPDATE "+remote_id+
        " re "+nf.format(sys_sample_error)+
        " rtt "+nf.format(smoothed_rtt)+
        " raw "+nf.format(sample_rtt)+
        " age "+sample_age+
        " dist "+nf.format(sys_distance)+
        " ssize "+sample_size+
        " lE "+nf.format(error)+
        " rE "+nf.format(r_error)+
        " rV "+r_coord.getVersion()+
        " lc "+sys_coord+
        " rc "+r_coord;
        
      crawler_log.info(info);
    }