vivaldiclient.java

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

    if (sys_sample_error < 0) {
      sys_sample_error = 0;
    }
    if (sys_sample_error > MAX_ERROR) {
      sys_sample_error = MAX_ERROR;
    }

    // EWMA on error
    double alpha = error / (error + r_error) * COORD_ERROR;
    error = (sys_sample_error*alpha)+((1-alpha)*error);
    
		if (keepStatistics) {
			running_sys_error.add(sys_sample_error);
			running_app_error.add(app_sample_error);
		}
	}

  

  protected boolean addNeighbor (RemoteState<T> guy) {
    boolean added = false;
    if (!neighbors.contains(guy)) {
      neighbors.add(guy);
      added = true;
    }
    if (neighbors.size() > MAX_NEIGHBORS) {
      neighbors.remove(0);
    }
    return added;
  }

    protected boolean removeNeighbor (RemoteState<T> guy) {
	if (neighbors.contains(guy)) {
	    neighbors.remove(guy);
	    return true;
	}
	return false;
    }

  synchronized public T getNeighborToPing (long curr_time) {
    final long NEIGHBOR_PING_EXPIRE_TIME = 10 * 60 * 1000; // 10 minutes
    long expire_time = curr_time - NEIGHBOR_PING_EXPIRE_TIME;
    List<RemoteState<T>> recentNeighbors = new ArrayList<RemoteState<T>>();
    for (RemoteState<T> neighbor : neighbors) {
      if (neighbor.getLastUpdateTime() > expire_time) {
        recentNeighbors.add(neighbor);
        //if (debugCrawler) {
          //int id = getIdFromAddr (neighbor.getAddress());
          //crawler_log.info ("considering "+id);
        //}
      }
    }
    if (recentNeighbors.size() > 0) {
      Collections.shuffle(recentNeighbors);
      RemoteState<T> neighbor = recentNeighbors.get(0);
      if (debugCrawler && debugGood) {
        int id = getIdFromAddr (neighbor.getAddress());
        crawler_log.info ("pinging neighbor "+id);
      }
      return recentNeighbors.get(0).getAddress();
    }
    return null;
  }

  
  protected void updateSystemCoordinate(long curr_time) {
    
    // figure out the oldest sample we are going to use
    // and recalculate the nearest neighbor
    // as a side effect
    long oldestSample = curr_time;
    double nearest_neighbor_distance = Double.MAX_VALUE;

    Collections.shuffle(neighbors);
    
    for (RemoteState<T> neighbor : neighbors) {
      double distance = sys_coord.distanceTo(neighbor.getLastCoordinate());
      if (distance < nearest_neighbor_distance) { 
        nearest_neighbor_distance = distance;
        nearest_neighbor = neighbor.getLastCoordinate();
      }
      if (oldestSample > neighbor.getLastUpdateTime()) {
        oldestSample = neighbor.getLastUpdateTime();
      }
    } 
    
    double sampleWeightSum = 0.;
    for (RemoteState<T> neighbor : neighbors) {
      double distance = sys_coord.distanceTo(neighbor.getLastCoordinate());
      sampleWeightSum += neighbor.getLastUpdateTime()-oldestSample;
    } 
    assert (sampleWeightSum >= 0.);
    
    Vec force = new Vec (sys_coord.getNumDimensions());

    for (RemoteState<T> neighbor : neighbors) {
      double distance = sys_coord.distanceTo(neighbor.getLastCoordinate());      
      while (distance == 0.) {
        sys_coord.bump();
        distance = sys_coord.distanceTo(neighbor.getLastCoordinate());
      }
      // cannot return null b/c distance is not 0
      Vec unitVector = sys_coord.getDirection(neighbor.getLastCoordinate());
      double latency = neighbor.getSample();
      double weight = error / (neighbor.getLastError()+error);
      if (weight == 0.) continue;
      
      // error of sample
      double sampleError = distance - latency;
      double sampleWeight = 1.;
      if (sampleWeightSum > 0) {
        sampleWeight = (neighbor.getLastUpdateTime()-oldestSample)/sampleWeightSum;
      }
      
      if (debugCrawler && debugGood) {
        int id = getIdFromAddr (neighbor.getAddress());
        crawler_log.info ("f "+id+ " age "+Math.round((curr_time-neighbor.getLastUpdateTime())/1000.)+
            " er "+sampleError+" sw "+sampleWeight+" comb " +
            (sampleError*sampleWeight));
      }
      
      unitVector.scale(sampleError*sampleWeight);
      force.add(unitVector);
    }
    
    if (USE_HEIGHT) {
      force.direction[force.direction.length-1] = -1.*force.direction[force.direction.length-1];
    }
    force.scale(COORD_CONTROL);

    if (debugCrawler && debugGood) {
      crawler_log.info ("t "+force.getLength()+" "+force);
    }
    
    // TODO add in gravity if necessary
    // Might not be with all of the churn
    // Check whether scaling should happen before or after addition of gravity

    /*
    if (GRAVITY_DIAMETER > 0) {
      // include "gravity" to keep coordinates centered on the origin
      Vec gravity = sys_coord.asVectorFromZero(true);
      if (gravity.getLength() > 0) {
        
        // scale gravity s.t. it increases polynomially with distance
        double force_of_gravity = Math.pow(gravity.getLength()
          / GRAVITY_DIAMETER, 2.);
        gravity.makeUnit();
        gravity.scale(force_of_gravity);
        
        // add to total force
        force.subtract(gravity);
        
        if (keepStatistics) {
          running_gravity.add(force_of_gravity);
        }
      }
    }
    */
    
    sys_coord.add(force);
    sys_coord.checkHeight ();
    double distance_delta = force.getLength();

    if (keepStatistics) {
      running_sys_dd.add(distance_delta);
      if (neighbors != null) {
        running_neighbors_used.add(neighbors.size());
      }

      if (time_of_last_sys_update > 0) {
        long since_last_sys_update = curr_time - time_of_last_sys_update;
        running_sys_update_frequency.add(since_last_sys_update);
      }
    }
    time_of_last_sys_update = curr_time;
    
  }

    /*
     * Periodically walk the entire rs_map and toss anybody who has expired.
     * If the map has grown beyond the preferred size (MAX_RS_MAP_SIZE),
     * shrink the max age that a guy can be before he gets kicked out.
    */

    protected void performMaintenance (long curr_time) {
	if (debugCrawler && debugGood) crawler_log.info ("performing maintenance");

	if (rs_map.size() > MAX_RS_MAP_SIZE) {
	    RS_EXPIRATION = (long)(.9 * RS_EXPIRATION);
	    if (debugCrawler && debugGood) crawler_log.info ("lowered RS_EXPIRATION to "+RS_EXPIRATION+ " size "+rs_map.size());
	}

	final long expirationStamp = curr_time - RS_EXPIRATION;
	Set<Map.Entry<T,RemoteState<T>>> states = rs_map.entrySet();

	for (Iterator<Map.Entry<T,RemoteState<T>>> stateIter = states.iterator(); stateIter.hasNext(); ) {
	    Map.Entry<T,RemoteState<T>> entry = stateIter.next();
	    if (entry.getValue().getLastUpdateTime() < expirationStamp) {
		if (debugCrawler && debugGood) crawler_log.info ("tossing "+entry.getValue().getAddress());
		removeNeighbor(entry.getValue());
		stateIter.remove();
	    }
	}
    }

	protected void tryUpdateAppCoordinate(long curr_time) {
		final double scale_factor = 1.0 / ((double) WINDOW_SIZE);

		// Make sure app coord always has a value.

		// calculate centroid of starting coordinates by averaging vectors

		if (start_coords.size() < WINDOW_SIZE) {
			Vec start_vec = new Vec(num_dims);
			start_coords.add(sys_coord);
			for (Iterator<Coordinate> i = start_coords.iterator(); i.hasNext();) {
				Coordinate next_coord = i.next();
				start_vec.add(next_coord.asVectorFromZero(false));
			}
			start_vec.scale(scale_factor);
			start_centroid = start_vec.asCoordinateFromZero(false);
		}

		current_coords.add(sys_coord);
		if (current_coords.size() > WINDOW_SIZE) {
			current_coords.remove(0);
		}

		// calculate centroid of current coordinates by averaging vectors
		Vec curr_vec = new Vec(num_dims);
		for (Iterator<Coordinate> i = current_coords.iterator(); i.hasNext();) {
			Coordinate next_coord = i.next();
			curr_vec.add(next_coord.asVectorFromZero(false));
		}
		curr_vec.scale(scale_factor);

		// create centroids
		Coordinate curr_centroid = curr_vec.asCoordinateFromZero(false);

		// get distances of centroids from nearest neighbor
		double start_dist = start_centroid.distanceTo(nearest_neighbor);
		double curr_dist = curr_centroid.distanceTo(nearest_neighbor);

		// fraction of space moved through, relative to distance to our NN
		double relative_diff = Math.abs((start_dist - curr_dist) / start_dist);

		if (keepStatistics) {
			running_relative_diff.add(relative_diff);
		}

		if (relative_diff > APP_UPDATE_THRESHOLD) {
			// exceed threshold, update application-level coordinate
			updated_app_coord_at_least_once = true;
			// clear coordinate windows
			start_coords.clear();
			current_coords.clear();
		}

		// This will keep updating the observers as we get rolling
		// until we've had one time when the coord windows differ

		boolean did_update = false;
		if (relative_diff > APP_UPDATE_THRESHOLD
				|| !updated_app_coord_at_least_once) {
			if (keepStatistics) {
				double app_dd = app_coord.distanceTo(curr_centroid);
				running_app_dd.add(app_dd);
			}
			app_coord = curr_centroid;
			did_update = true;

			// If we've gotten ourselves into a situation where the coord
			// very accurate, stop always updating the app.
			// Currently can only use this if keep track of statistics

			if (keepStatistics) {
				final double MIN_SYS_ERROR_SIZE = (RUNNING_STAT_HISTORY / 8.);
				if (!updated_app_coord_at_least_once
						&& running_sys_error.getSize() > MIN_SYS_ERROR_SIZE
						&& running_sys_error.getPercentile(.5) < 0.20) {
					updated_app_coord_at_least_once = true;
				}
			}

			// notify observers of new application-level coordinate
			for (Iterator<ApplicationObserver> i = obs_list.iterator(); i
					.hasNext();) {
				ApplicationObserver obs = i.next();
				obs.coordinatesUpdated(app_coord);
			}

			if (keepStatistics) {
				if (time_of_last_app_update > 0) {
					long since_last_app_update = curr_time
							- time_of_last_app_update;
					running_app_update_frequency.add(since_last_app_update);
				}
				time_of_last_app_update = curr_time;
			}
		}

		if (debug && debugCrawler) {
			crawler_log.info("app_coord update: done " + did_update
					+ " rolling " + updated_app_coord_at_least_once + " start "
					+ start_coords.size() + " current " + current_coords.size()
					+ " diff " + nf.format(relative_diff));
		}
	}

  public static void setRandomSeed (long seed) {
    random = new Random (seed);
  }

  public void startUp(DataInputStream is) throws IOException {
    
    if (debugCrawler) crawler_log.info("startUp");
        
    boolean valid = false;

    	// when starting up with no previously stored coord we get zero length input stream
    
    if ( is.available() > 0 ){    	
    
	    int	version = 1;
	    
	    	// migration 2501 when version added...
	
	    if ( is.available() != 25 ){
	    
	    	version = is.readInt();
	     }
	    
	    try {
	      sys_coord = new Coordinate (num_dims, is);
	      error = ((double) is.readFloat());
	      if (sys_coord.isValid() && !(Double.isNaN(error))) {
	         
	          valid = true;
	      }else{
	    	  //System.err.println("Invalid coordinate.");
	      }
	      
	    } catch (IOException ex) {
	    }
    }
    
    if (!valid) {
      //System.err.println("Error deserializing coordinate during startup.  Starting afresh.");
      sys_coord = new Coordinate (num_dims);
      error = MAX_ERROR;
    } else {
     // System.err.println("Deserialized coordinate OK during startup "+sys_coord+ " er "+error);
    }
    
  }
  
  synchronized public void shutDown(DataOutputStream os) throws IOException {

    if (debugCrawler) crawler_log.info("shutDown");
    
    os.writeInt( 1 );	// version
    
    // could also save a number of neighbors
    // but then when we come back into the system, 
    // it would be tricky to know how to treat them (they'd be old
    // so they would get much weight either).
    
    //System.err.println("Saving coordinates during shutdown "+sys_coord+" er="+error);
    
    sys_coord.toSerialized(os);
    os.writeFloat((float) error);
    
    // save app_coord also?
    
  }
  
}