[Pyxida-cvs-commits] Pyxida/src/edu/harvard/syrah/pyxida/nc/lib ApplicationObserver.java, 1.3, 1.4 NCClient.java, 1.10, 1.11 RemoteState.java, 1.3, 1.4

[Jonathan L. <le...@us...>]

Update of /cvsroot/pyxida/Pyxida/src/edu/harvard/syrah/pyxida/nc/lib
In directory ddv4jf1.ch3.sourceforge.com:/tmp/cvs-serv30573/edu/harvard/syrah/pyxida/nc/lib

Modified Files:
	ApplicationObserver.java NCClient.java RemoteState.java 
Log Message:
first pass on integrating erics changes which make neighbor set more stable

Index: RemoteState.java
===================================================================
RCS file: /cvsroot/pyxida/Pyxida/src/edu/harvard/syrah/pyxida/nc/lib/RemoteState.java,v
retrieving revision 1.3
retrieving revision 1.4
diff -C2 -d -r1.3 -r1.4
*** RemoteState.java	5 Nov 2008 01:22:49 -0000	1.3
--- RemoteState.java	18 Dec 2008 21:49:51 -0000	1.4
***************
*** 16,161 ****
   * limitations under the License.
   */
! package edu.harvard.syrah.pyxida.nc.lib;
! 
! import java.io.BufferedReader;
! import java.io.File;
! import java.io.FileNotFoundException;
! import java.io.FileReader;
! import java.util.StringTokenizer;
! 
! 
! /**
!  * The state kept of a remote node between samples.
!  * 
!  * @author Michael Parker, Jonathan Ledlie
!  * 
!  * @param <T>
!  * the type of the unique identifier of a host
!  */
! public class RemoteState<T> {
!   // made not final so they can be changed by simulator
! 	protected static double SAMPLE_PERCENTILE = 0.5;
!   // Don't keep more than this many samples
!   public static int MAX_SAMPLE_SIZE = 16;
!   // Don't use a guy unless we have this many samples
!   public static int MIN_SAMPLE_SIZE = 4;
!   
! 	protected final T addr;
! 	protected final WindowStatistic ping_samples;
! 
! 	protected Coordinate last_coords;
! 	protected double last_error;
! 	protected long last_update_time;
! 
! 	public RemoteState(T _addr) {
! 		addr = _addr;
!     ping_samples = new WindowStatistic(MAX_SAMPLE_SIZE);
! 
! 		last_coords = null;
! 		last_error = 0.0;
! 		last_update_time = -1L;
! 	}
! 
!   public T getAddress () {
!     return addr;
!   }
!   
! 	public void assign(Coordinate _last_coords, double _last_error,
! 	    long _curr_time) {
! 	  last_coords = _last_coords;
! 		last_error = _last_error;
! 		last_update_time = _curr_time;
! 	}
! 
!   public void addSample(double sample_rtt, long sample_age, Coordinate r_coord,
!     double r_error, long curr_time) {
! 		ping_samples.add(sample_rtt);
! 		last_coords = r_coord;
! 		last_error = r_error;
!     if (sample_age > 0)
!       last_update_time = curr_time-sample_age;
!     else
!       last_update_time = curr_time;
! 	}
!   
!   public boolean isValid (long curr_time) {
!     if (getLastError() <= 0. || last_update_time <= 0 ||
!         last_coords.atOrigin()) {
!       return false;
!     }
!       
!     if (getSampleSize() >= MIN_SAMPLE_SIZE && getSample() > 0) {
!       return true;
!     }
! 
!     if (getSampleSize() >= 2 && ping_samples.withinVariance(.1)) {
!       return true;
!     }
!     
!     return false;
!   }
!   
! 	public double getSample() {
! 		return ping_samples.getPercentile(SAMPLE_PERCENTILE);
! 	}
! 
!   public int getSampleSize() {
!     return ping_samples.getSize();
!   }
!   /*
!   public boolean isLowVariance () {
!     return ping_samples.isLowVariance();
!   }
!   */
!   
! 	public Coordinate getLastCoordinate() {
! 		return last_coords;
! 	}
! 
! 	public double getLastError() {
! 		return last_error;
! 	}
! /*
! 	public boolean beenSampled() {
! 		return (last_update_time >= 0L);
! 	}
! */
! 	public long getLastUpdateTime() {
! 		return last_update_time;
! 	}
!   
! 	public static void main (String args[]) {
! 	  System.out.println("Testing Remote State Object");
! 	  String sampleFile = args[0];
! 	  RemoteState<String> rs = new RemoteState<String>(sampleFile);
! 	  BufferedReader sampleReader = null;
! 	  try {
! 	    sampleReader = new BufferedReader (new FileReader (new File (sampleFile)));
! 	  }catch (FileNotFoundException ex) {
! 	    System.err.println("Cannot open file "+sampleFile+": "+ex);
! 	    System.exit(-1);
! 	  } 
! 	  
! 	  long sample_age = 0;
! 	  Coordinate r_coord = null;
! 	  double r_error = 0;
! 	  
! 	  try {
! 	    String sampleLine = sampleReader.readLine();
! 	    while (sampleLine != null) {
! 	      // reads in timestamp in ms and raw rtt
! 	      StringTokenizer sampleTokenizer = new StringTokenizer (sampleLine);
! 	      long curr_time = Long.parseLong((String)(sampleTokenizer.nextElement()));
! 	      int rawRTT = Integer.parseInt((String)(sampleTokenizer.nextElement()));
! 	      sampleLine = sampleReader.readLine();
! 	      rs.addSample (rawRTT, sample_age, r_coord, r_error, curr_time);
!         double smoothedRTT = rs.getSample();
!         System.out.println(curr_time+" raw "+rawRTT+" smooth "+smoothedRTT);
! 	    }
! 	  } catch (Exception ex) {
! 	    System.err.println("Problem parsing "+sampleFile+": "+ex);
! 	    System.exit(-1);     
! 	  }
! 	}
!   
! }
--- 16,174 ----
   * limitations under the License.
   */
! 
! package edu.harvard.syrah.pyxida.nc.lib;
! 
! import java.io.BufferedReader;
! import java.io.File;
! import java.io.FileNotFoundException;
! import java.io.FileReader;
! import java.util.StringTokenizer;
! 
! 
! /**
!  * The state kept of a remote node between samples.
!  * 
!  * @author Michael Parker, Jonathan Ledlie
!  * 
!  * @param <T>
!  * the type of the unique identifier of a host
!  */
! public class RemoteState<T> {
!   // made not final so they can be changed by simulator
! 	protected static double SAMPLE_PERCENTILE = 0.5;
!   // Don't keep more than this many samples
!   public static int MAX_SAMPLE_SIZE = 16;
!   // Don't use a guy unless we have this many samples
!   public static int MIN_SAMPLE_SIZE = 4;
!   
! 	protected final T addr;
! 	protected final WindowStatistic ping_samples;
! 
! 	protected Coordinate last_coords;
! 	protected double last_error;
! 	// when we last update our coord vs this node
! 	protected long last_update_time;
! 	// when we last pinged this node
! 	protected long last_ping_time;
! 
! 	public RemoteState(T _addr) {
! 		addr = _addr;
!     ping_samples = new WindowStatistic(MAX_SAMPLE_SIZE);
! 
! 		last_coords = null;
! 		last_error = 0.0;
! 		last_update_time = -1L;
! 		last_ping_time = 0L;
! 	}
! 
!   public T getAddress () {
!     return addr;
!   }
!   
! 	public void assign(Coordinate _last_coords, double _last_error,
! 	    long _curr_time) {
! 	  last_coords = _last_coords;
! 		last_error = _last_error;
! 		last_update_time = _curr_time;
! 	}
! 
!   public void addSample(double sample_rtt, long sample_age, Coordinate r_coord,
!     double r_error, long curr_time) {
! 		ping_samples.add(sample_rtt);
! 		last_coords = r_coord;
! 		last_error = r_error;
!     if (sample_age > 0)
!       last_update_time = curr_time-sample_age;
!     else
!       last_update_time = curr_time;
! 	}
!   
!   public boolean isValid (long curr_time) {
!     if (getLastError() <= 0. || last_update_time <= 0 ||
!         last_coords.atOrigin()) {
!       return false;
!     }
!       
!     if (getSampleSize() >= MIN_SAMPLE_SIZE && getSample() > 0) {
!       return true;
!     }
! 
!     if (getSampleSize() >= 2 && ping_samples.withinVariance(.1)) {
!       return true;
!     }
!     
!     return false;
!   }
!   
! 	public double getSample() {
! 		return ping_samples.getPercentile(SAMPLE_PERCENTILE);
! 	}
! 
!   public int getSampleSize() {
!     return ping_samples.getSize();
!   }
!   /*
!   public boolean isLowVariance () {
!     return ping_samples.isLowVariance();
!   }
!   */
!   
! 	public Coordinate getLastCoordinate() {
! 		return last_coords;
! 	}
! 
! 	public double getLastError() {
! 		return last_error;
! 	}
! /*
! 	public boolean beenSampled() {
! 		return (last_update_time >= 0L);
! 	}
! */
! 	public long getLastUpdateTime() {
! 		return last_update_time;
! 	}
! 	
! 	public long getLastPingTime() {
! 		return last_ping_time;
! 	}
! 	
! 	public void setLastPingTime(long time) {
! 		last_ping_time = time;
! 	}
!   
! 	public static void main (String args[]) {
! 	  System.out.println("Testing Remote State Object");
! 	  String sampleFile = args[0];
! 	  RemoteState<String> rs = new RemoteState<String>(sampleFile);
! 	  BufferedReader sampleReader = null;
! 	  try {
! 	    sampleReader = new BufferedReader (new FileReader (new File (sampleFile)));
! 	  }catch (FileNotFoundException ex) {
! 	    System.err.println("Cannot open file "+sampleFile+": "+ex);
! 	    System.exit(-1);
! 	  } 
! 	  
! 	  long sample_age = 0;
! 	  Coordinate r_coord = null;
! 	  double r_error = 0;
! 	  
! 	  try {
! 	    String sampleLine = sampleReader.readLine();
! 	    while (sampleLine != null) {
! 	      // reads in timestamp in ms and raw rtt
! 	      StringTokenizer sampleTokenizer = new StringTokenizer (sampleLine);
! 	      long curr_time = Long.parseLong((String)(sampleTokenizer.nextElement()));
! 	      int rawRTT = Integer.parseInt((String)(sampleTokenizer.nextElement()));
! 	      sampleLine = sampleReader.readLine();
! 	      rs.addSample (rawRTT, sample_age, r_coord, r_error, curr_time);
!         double smoothedRTT = rs.getSample();
!         System.out.println(curr_time+" raw "+rawRTT+" smooth "+smoothedRTT);
! 	    }
! 	  } catch (Exception ex) {
! 	    System.err.println("Problem parsing "+sampleFile+": "+ex);
! 	    System.exit(-1);     
! 	  }
! 	}
!   
! }

Index: ApplicationObserver.java
===================================================================
RCS file: /cvsroot/pyxida/Pyxida/src/edu/harvard/syrah/pyxida/nc/lib/ApplicationObserver.java,v
retrieving revision 1.3
retrieving revision 1.4
diff -C2 -d -r1.3 -r1.4
*** ApplicationObserver.java	5 Nov 2008 01:22:49 -0000	1.3
--- ApplicationObserver.java	18 Dec 2008 21:49:51 -0000	1.4
***************
*** 16,33 ****
   * limitations under the License.
   */
! package edu.harvard.syrah.pyxida.nc.lib;
! 
! /**
!  * An observer to be notified when the application coordinates change.
!  *
!  * @author Michael Parker, Jonathan Ledlie
!  */
! public interface ApplicationObserver {
! 	/**
! 	 * This method is invoked when the application-level coordinates are
! 	 * updated.
! 	 *
! 	 * @param new_coords the new application-level coordinates
! 	 */
! 	public void coordinatesUpdated(Coordinate new_coords);
! }
--- 16,33 ----
   * limitations under the License.
   */
! package edu.harvard.syrah.pyxida.nc.lib;
! 
! /**
!  * An observer to be notified when the application coordinates change.
!  *
!  * @author Michael Parker, Jonathan Ledlie
!  */
! public interface ApplicationObserver {
! 	/**
! 	 * This method is invoked when the application-level coordinates are
! 	 * updated.
! 	 *
! 	 * @param new_coords the new application-level coordinates
! 	 */
! 	public void coordinatesUpdated(Coordinate new_coords);
! }

Index: NCClient.java
===================================================================
RCS file: /cvsroot/pyxida/Pyxida/src/edu/harvard/syrah/pyxida/nc/lib/NCClient.java,v
retrieving revision 1.10
retrieving revision 1.11
diff -C2 -d -r1.10 -r1.11
*** NCClient.java	5 Nov 2008 01:22:49 -0000	1.10
--- NCClient.java	18 Dec 2008 21:49:51 -0000	1.11
***************
*** 1,3 ****
--- 1,7 ----
+ package edu.harvard.syrah.pyxida.nc.lib;
+ 
  /*
+  * Pyxida - a network coordinate library
+  * 
   * Copyright 2008 Jonathan Ledlie and Peter Pietzuch
   * 
***************
*** 16,20 ****
   * limitations under the License.
   */
- package edu.harvard.syrah.pyxida.nc.lib;
  
  import java.io.DataInputStream;
--- 20,23 ----
***************
*** 43,46 ****
--- 46,54 ----
   * @param <T>
   *          the type of the unique identifier of a host
+  *
+  * A note about time: Try to not use System.currentTimeMillis or
+  * similar references to the system's time in this code.  This code
+  * should be able to run in simulation, where arbitrary times are
+  * passed in as parameters.
   */
  public class NCClient<T> implements NCClientIF {
***************
*** 78,83 ****
  	public static boolean USE_HEIGHT = true;
  
  	// Try to minimize our error between up to MAX_NEIGHBORS guys at once
! 	final public static int MAX_NEIGHBORS = 512;
  
  	final protected static int WINDOW_SIZE = 64;
--- 86,92 ----
  	public static boolean USE_HEIGHT = true;
  
+ 	// Note: for larger installations, use e.g. 512
  	// Try to minimize our error between up to MAX_NEIGHBORS guys at once
! 	final public static int MAX_NEIGHBORS = 32;
  
  	final protected static int WINDOW_SIZE = 64;
***************
*** 90,93 ****
--- 99,104 ----
  	final public static long MAINTENANCE_PERIOD = 10 * 60 * 1000; // ten minutes
  
+ 	final public static long MAX_PING_RESPONSE_TIME = 10 * 60 * 1000; // ten minutes
+ 
  	// target max number of remote states kept
  	// set to be larger than MAX_NEIGHBORS
***************
*** 115,118 ****
--- 126,153 ----
  	// Gravity should keep everybody within this ball
  	public static double MAX_DIST_FROM_ORIGIN = 60000.;
+ 		
+ 	//ERIC START
+ 	/* After STABILIZE_TIME, only ping neighbors from neighbors list
+ 	 * and do not remove first guy from the neighbors list
+ 	 * This is to ensure a stable neighbors list
+ 	 */
+ 	//public static final long STABILIZE_TIME = 1 * 60 * 60 * 1000; //1 hour
+ 	
+ 	/*
+ 	 * The time the NCClient has been created/started.
+ 	 */
+ 	//public long START_TIME;
+ 	
+ 	/*
+ 	 * The minimum time to wait before kicking a neighbor off the list
+ 	 * if that neighbor has not been pinged yet.
+ 	 */
+ 	public static long MIN_UPDATE_TIME_TO_PING = 2 * MAINTENANCE_PERIOD;
+ 	
+ 	/*
+ 	 * The weight to be used in calculating the probability
+ 	 */
+ 	//public static final double WEIGHT_PING = (double)1.5 * (double)MAINTENANCE_PERIOD;
+ 	//ERIC END
  
    final static protected NumberFormat nf = NumberFormat.getInstance();
***************
*** 208,211 ****
--- 243,250 ----
  		current_coords = new LinkedList<Coordinate>();
  		nearest_neighbor = null;
+ 		
+ 		//ERIC START -- added
+ 		//START_TIME = System.currentTimeMillis();
+ 		//ERIC END
  
  		// bootstrapCoordinates ();
***************
*** 656,660 ****
  
  		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,
--- 695,700 ----
  
  		if (addr_rs.isValid(curr_time)) {
! 			if (!stabilized() || neighbors.size() < MAX_NEIGHBORS)
! 				addNeighbor(addr_rs);
  			// first update our error
  			updateError(addr, r_coord, r_error, smoothed_rtt, sample_rtt, sample_age, sample_size,
***************
*** 688,692 ****
  		}
  
! 		// System.out.println ("maint?");
  		if (lastMaintenanceStamp < curr_time - MAINTENANCE_PERIOD) {
  			performMaintenance(curr_time);
--- 728,732 ----
  		}
  
! 
  		if (lastMaintenanceStamp < curr_time - MAINTENANCE_PERIOD) {
  			performMaintenance(curr_time);
***************
*** 787,790 ****
--- 827,904 ----
  	}
  
+ 	
+ 	//ERIC START - updated
+ 	/* If stabilized, then only add to neighbors if list is not full
+ 	 * else do the normal thing, add anyway, and if full kick out the first guy
+ 	 */
+ 	/*
+ 	protected boolean addNeighbor(RemoteState<T> guy) {
+ 		boolean added = false;
+ 		if (stabilized()) {
+ 			crawler_log.info("addNeighbor stabilized");
+ 			if (neighbors.size() < MAX_NEIGHBORS) {
+ 				if (!neighbors.contains(guy)) {
+ 					crawler_log.info("addNeighbor stabilized added");
+ 					neighbors.add(guy);
+ 					added = true;
+ 				}
+ 			}
+ 		} else {
+ 			crawler_log.info("addNeighbor NOT stabilized");
+ 			if (!neighbors.contains(guy)) {
+ 				neighbors.add(guy);
+ 				added = true;
+ 			}
+ 			if (neighbors.size() > MAX_NEIGHBORS) {
+ 				neighbors.remove(0);
+ 			}
+ 		}
+ 		return added;
+ 	}
+ 	*/
+ 	//ERIC END
+ 	
+ 	//ERIC START - added
+ 	/*
+ 	public boolean neighborsFull() {
+ 		if (neighbors.size() < MAX_NEIGHBORS) {
+ 			return false;
+ 		} else {
+ 			return true;
+ 		}
+ 	}
+ 	*/
+ 
+ 	/*
+ 	public boolean stabilized() {
+ 		if (System.currentTimeMillis() >= START_TIME + STABILIZE_TIME) {
+ 			return true;
+ 		} else {
+ 			return false;
+ 		}
+ 	}
+ 	*/
+ 
+ 	public boolean stabilized() {
+ 		// Could also do something more sophisticated with the app level coordinates
+ 		// but this should be a rough approximation.
+ 		// May want to be on the lookout for transitional behavior.
+ 		// That is, we don't want nodes to suddenly act differently based on this value.
+ 		if (error < 0.5) {
+ 			return true;
+ 		}
+ 		return false;
+ 	}
+ 	
+ 	public String printNeighbors() {
+ 		String toPrint = "";
+ 		for (Iterator<RemoteState<T>> i = neighbors.iterator(); i.hasNext();) {
+ 			RemoteState<T> A_rs = i.next();
+ 			toPrint=toPrint+","+A_rs.getAddress();
+ 		}
+ 		return toPrint;
+ 	}
+ 	//ERIC END
+ 
  	protected boolean removeNeighbor(RemoteState<T> guy) {
  		if (neighbors.contains(guy)) {
***************
*** 795,799 ****
  	}
  
! 	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;
--- 909,914 ----
  	}
  
! 	//ERIC START - modified getNeighborToPing to add "weight" probability
! 	/*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;
***************
*** 818,822 ****
  		}
  		return null;
! 	}
  
  	protected void updateSystemCoordinate(long curr_time) {
--- 933,983 ----
  		}
  		return null;
! 	}*/
! 	
! 	/*
! 	 * Pick a "random" neighbor from the neighbors list to send a ping request
! 	 * For each neighbor, calculate the weight (probability that it will be
! 	 * sent a ping)
! 	 * If, on the off chance, that no neighbor was picked, then randomly pick
! 	 * one.
! 	 */
! 
! 	/*
! 	 * JL Note: Eric used a weight here which might be nicer...
! 	 */
! 
! 	synchronized public T getNeighborToPing(long curr_time) {
! 		List<RemoteState<T>> grayingNeighbors = new ArrayList<RemoteState<T>>();
! 
! 		for (RemoteState<T> neighbor : neighbors) {
! 			if (curr_time - neighbor.getLastPingTime() > MIN_UPDATE_TIME_TO_PING) {
! 				grayingNeighbors.add(neighbor);
! 			}
! 		}
! 
! 		if (grayingNeighbors.size() > 0) {
! 			crawler_log.info("picking from grayingNeighbors");
! 			Collections.shuffle(grayingNeighbors);
! 			RemoteState<T> neighbor = grayingNeighbors.get(0);
! 			neighbor.setLastPingTime(curr_time);
! 
! 			// reduce the likely size of graying neighbors
! 			// if it is (relatively) too big, but only if it is (absolutely) too big
! 			if (grayingNeighbors.size() > neighbors.size() / 8 &&
! 					grayingNeighbors.size() > 3) {
! 				MIN_UPDATE_TIME_TO_PING /= 2;
! 				if (MIN_UPDATE_TIME_TO_PING <= 1) MIN_UPDATE_TIME_TO_PING = 1;
! 				crawler_log.info("lowered MIN_UPDATE_TIME_TO_PING "+MIN_UPDATE_TIME_TO_PING);
! 			}
! 
! 			return neighbor.getAddress();
! 		}
! 		crawler_log.info("getNeighborToPing returning null");
! 		// we do want to get some nodes in grayingNeighbors
! 		MIN_UPDATE_TIME_TO_PING *= 2;
! 		crawler_log.info("increased MIN_UPDATE_TIME_TO_PING "+MIN_UPDATE_TIME_TO_PING);
! 		return null;
! 	}	
! 	//ERIC END
  
  	protected void updateSystemCoordinate(long curr_time) {
***************
*** 953,956 ****
--- 1114,1140 ----
  			}
  		}
+ 
+ 		//ERIC START
+ 		//remove neighbors that are "dead" -- the last update time is more than
+ 		//10 minutes (MAINTENANCE_PERIOD) ago
+ 		//Also check that the lastPingTime is not greater than the MIN_UPDATE_TIME_TO_PING
+ 
+ 		// Toss guys we've pinged but who haven't responded.
+ 
+ 		for (Iterator<RemoteState<T>> i = neighbors.iterator(); i.hasNext();) {
+ 			RemoteState<T> rs = i.next();
+ 			if (rs.getLastPingTime() - rs.getLastUpdateTime() > MAX_PING_RESPONSE_TIME) {
+ 				crawler_log.info("performMaintenance removed " + rs.getAddress());
+ 				i.remove();
+ 				//removeNeighbor(A_rs);
+ 			}
+ 		}
+ 
+ 		//for (RemoteState<T> neighbor : neighbors) {
+ 		//	if (neighbor.getLastUpdateTime() < (curr_time - MAINTENANCE_PERIOD)) {
+ 		//		neighbors.remove(neighbor);
+ 		//	}
+ 		//}
+ 		//ERIC END
  	}