package modules.node;

import utils.*;
import modules.neighborset.*;
import modules.gradient.*;
import modules.region.*;
import simulator.*;
import java.util.*;

// A persistent node is a region assigned to store a piece of data.
// The data stored is held distributed throughout the region, so
// the entire region must die in order to kill the data. The region
// is not tied to a particular set of nodes, but may move and regenerate
// portions which have been destroyed.

// Interface:
// Initiate Node(n): create node n, centered at this processor
// Destroy Node(n): destroy node n if it contains this processor
//   calls a completion when done (and indicates whether it's successful)
// Set Data: atomically store and latch a piece of data
//   calls a completion when done (and indicates whether it's successful)
// Get Data: returns most recently latched data
// Data guarantees: something or other about atomicity.

// Data system:
// read - returns instantaneous value
// write - sends gradient setting new value.
//         if that gradient GCs naturally, then send ack, else wait for ack.
//         ack is unique gradient (*not* overlapping!).
//         when an ack GCs (or somebody else's ack arrives), a write returns
// I believe this'll guarantee atomicity... 

// Some terminology:
// A node of size r is divided into three regions:
// Core: 0<=hop<r   this section is the node "proper"
// Reflector: r<=hop<2r  this section is used in calculating motion of the node
// Umbra: 2r<=hop<kr  this section has awareness of the node (and is thus
//   an exclusion zone for copies) but plays no part in the node's lifecycle

public class PersistentNodeB implements ProcessorModule {
  PortedProcessor pp; ActiveGradient ag; NeighborSet ns;
  RegionManager rm;

  public Symbol getName() { return Symbol.GetSymbol("PersistentNode"); }
  public void link(PortedProcessor p) { 
    pp = p; 
    ag = (ActiveGradient)p.getModule("ActiveGradient");
    ns = (NeighborSet)p.getModule("NeighborSet");
    rm = (RegionManager)p.getModule("RegionManager");
    pp.addClockListener(this);
    pp.openNetworkPort(this,pnXfer);
  }

  public void init() {
  }

  public static Symbol pnType = Symbol.GetSymbol("persistentnode");
  public static Symbol pnXfer = Symbol.GetSymbol("pnXfer");
  Hashtable nodes = new Hashtable();

  // types of events:
  // 1. clock: age & update
  // 2. transfer: move to new center
  // 3. setData:
  // 4. commitData:
  // 5. Kill: latch node into end-state
  public boolean signalEvent(Symbol name,Object data) {
    boolean changed = false;
    if(name==PortedProcessor.kClock) {
      Set glist = ag.getType(pnType);
      Iterator i = glist.iterator();
      while(i.hasNext()) { // 1. add new nodes for unlisted gradients
        Gradient g = ag.get(i.next());
        if(nodes.get(g.gid)==null) nodes.put(g.gid,new Node(g.gid,pp));
      }
      i=nodes.keySet().iterator();
      while(i.hasNext()) { // 2. run node updates, deleting ones w/o gradient
        Object id = i.next();
        if(!((Node)nodes.get(id)).update())
          i.remove();
      }
    } else if(name==pnXfer) { // contents: (name, destID)
      Pair p = (Pair)data;
      if(p.cdr()!=null && pp.getUID()==((Double)p.cdr()).doubleValue()) {
        Node n = (Node)nodes.get(p.car());
        if(n!=null) n.becomeCenter();
      }
    }
    return changed;
  }

  // Interface functions:
  // start a persistent node:
  int defaultRadius = 3;
  double defaultUmbra = 3.0; // k for umbra
  ProcessorLoadEstimator defaultEstimator = new NodeCountEstimator();
  public void initiateNode(Object nid) 
  { initiateNode(nid,defaultRadius,defaultUmbra,RegionManager.rootID); }
  public void initiateNode(Object nid, int r, double u)
  { initiateNode(nid,r,u,defaultEstimator,RegionManager.rootID); }
  public void initiateNode(Object nid, ProcessorLoadEstimator l)
  { initiateNode(nid,defaultRadius,defaultUmbra,l,RegionManager.rootID); }
  public void initiateNode(Object nid, int r, double u, 
                           ProcessorLoadEstimator l)
  { initiateNode(nid,r,u,l,RegionManager.rootID); }
  public void initiateNode(Object nid, Object region) 
  { initiateNode(nid,defaultRadius,defaultUmbra,region); }
  public void initiateNode(Object nid, int r, double u, Object region)
  { initiateNode(nid,r,u,defaultEstimator,region); }
  public void initiateNode(Object nid, ProcessorLoadEstimator l, Object region)
  { initiateNode(nid,defaultRadius,defaultUmbra,l,region); }
  public void initiateNode(Object nid, int r, double u, 
                           ProcessorLoadEstimator l, Object region) {
    nodes.put(nid,new Node(nid,pp,r,u,l,region));
  }
  // end a persistent node
  public void killNode(Object nid) {
    Node n = (Node)nodes.get(nid);
    if(n!=null) n.die();
  }
  // get
  public Object getData(Object nid) {
    Node n = (Node)nodes.get(nid);
    if(n==null) return null;
    else return n.getData();
  }
  // set: returns a boolean to say whether there was a node to set
  public boolean setData(Object nid, Object data) 
    throws ConcurrentModificationException
  { return setData(nid,data,null); }
  public boolean setData(Object nid, Object data, DelayedReturn ret)
    throws ConcurrentModificationException {
    Node n = (Node)nodes.get(nid);
    if(n==null) return false;
    else n.setData(data,ret);
    return true;
  }
  // list off nodes (cores only)
  public Vector getNodes() { 
    Vector nl = new Vector();
    Iterator i = nodes.keySet().iterator();
    while(i.hasNext()) {
      Object id = i.next();
      if(((Node)nodes.get(id)).inNode()) nl.add(id);
    }
    return nl;
  }
  public Vector getNodeReflectors() { // get nodes we're in the reflector of
    Vector nl = new Vector();
    Iterator i = nodes.keySet().iterator();
    while(i.hasNext()) {
      Object id = i.next();
      if(((Node)nodes.get(id)).inReflector()) nl.add(id);
    }
    return nl;
  }
  public Vector getNodeUmbras() { // get nodes we're in the umbra of
    Vector nl = new Vector();
    Iterator i = nodes.keySet().iterator();
    while(i.hasNext()) {
      Object id = i.next();
      if(((Node)nodes.get(id)).inUmbra()) nl.add(id);
    }
    return nl;
  }
  // get nodes we're in an abritrary range of
  public Vector getNodePenumbras(double pu) {
    Vector nl = new Vector();
    Iterator i = nodes.keySet().iterator();
    while(i.hasNext()) {
      Object id = i.next();
      if(((Node)nodes.get(id)).inPenumbra(pu)) nl.add(id);
    }
    return nl;
  }

  public boolean inNode(Object nid) { return inNode(nid,1.0); }
  public boolean inNode(Object nid,double pu) {
    Node n = (Node)nodes.get(nid);
    if(n==null) return false;
    return n.inPenumbra(pu);
  }
  // returns distance to center, or -1 if it's not in the node
  public int centerDistance(Object nid) {
    Node n = (Node)nodes.get(nid);
    if(n==null) return -1;
    return n.centerDistance();
  }

  public String toString() {
    return "Persistent Node: "+nodes.toString();
  }

  // kludge debugging
  public int debugit(Object nid) { /* @deprecated */
    int value=0;
    PNData pnd = (PNData)ns.getData(nid); 
    if(pnd!=null && pnd.data!=null) {
      if(pnd.data.pending!=null) value+=1;
      if(pnd.data.backprop) value+=2;
    }
    return value;
  }
  public VersionedData debugit2(Object nid,boolean read) { /* @deprecated */
    Node n = (Node)nodes.get(nid);
    if(n==null) return null;
    if(read) {
      return (VersionedData)n.data.committed.clone();
    } else {
      return (VersionedData)n.pendingWrite.clone();
    }
  }
}

// since it's being shared by the neighborset, its data should be final
class PNData {
  final NodeState state;
  final NodeData data;
  public PNData(NodeState s,NodeData d) {
    state = (NodeState)s.clone(); 
    if(d!=null) data = (NodeData)d.clone(); else data=null;
  }
  public String toString() { 
    if(data!=null) 
      return "("+state+", "+data+")";
    else
      return "("+state+", null)";
  }
}

class PNGData { // passed around with the gradient
  final int radius;
  final double umbra;
  final ProcessorLoadEstimator load;
  public PNGData(int r, double u, ProcessorLoadEstimator l) 
  { radius=r; umbra=u; load=l; }
  public String toString() { return "("+radius+", "+umbra+", "+load+")"; }
}

//PersistentNode proper method:
// 1. gradient of size kr [r=node, 2r=reflector, kr=passive]
// 2. calculate inward values from neighbors [proceeds from edge]
// 3. declare "I'm the best" and wait
// 4. on timeout, designate successor to send new gradient

class NodeState {
  double lifepulse; int lpID;
  int age; // length of stability
  int hop; // distance from center
  double value; // load estimator value
  boolean dead = false;
  boolean in_node;
  boolean in_reflector;
  public NodeState() { reset(); }
  public void reset() { age=0; hop=0; value=0; }
  public String toString() 
  { return "("+age+", "+hop+", "+value+", "+dead+", "+in_node+", "+in_reflector+", "+lifepulse+", "+lpID+")"; }
  public Object clone() { 
    NodeState c = new NodeState(); 
    c.age=age; c.hop=hop; c.value=value; c.dead=dead; c.in_node=in_node;
    c.in_reflector=in_reflector; c.lifepulse=lifepulse; c.lpID=lpID;
    return c;
  }
  void init_from_neighbors() {
  }
}
class NodeData {
  VersionedData committed;
  VersionedData pending; // strength of a pending=number of agreers?
  int hop=-1;
  boolean backprop; int age=0;
  public NodeData() { committed=new VersionedData(); }
  public String toString() { 
    if(pending!=null)
      return "("+committed+", "+pending+", "+hop+", "+backprop+", "+age+")"; 
    else
      return "("+committed+", null)";
  }
  public Object clone() { 
    NodeData d = new NodeData();
    d.committed = (VersionedData)committed.clone();
    if(pending==null) d.pending=null;
    else d.pending = (VersionedData)pending.clone();
    d.hop=hop; d.backprop=backprop; d.age=age;
    return d;
  }
}


/**********************************/
// The illustrious Node class
/**********************************/
class Node {
  PortedProcessor pp;
  ActiveGradient ag;
  NeighborSet ns;

  Object id;
  PNGData config;
  NodeState state; // group maintainance state
  NodeData data; // storage state
  
  // *** constructors ***
  public Node(Object id,PortedProcessor pp) { // propagation init
    this.id=id; this.pp = pp;
    ag = (ActiveGradient)pp.getModule("ActiveGradient");
    ns = (NeighborSet)pp.getModule("NeighborSet");
    config = (PNGData)ag.get(id).data; // there is guaranteed to be a gradient
    init_state_from_neighbors(); // suck info from neighbors
    commit_state();
  }
  // local init
  public Node(Object id,PortedProcessor pp, int r, double u, 
              ProcessorLoadEstimator l, Object region) {
    if(u<2) u=2; // the umbra cannot be smaller than the reflector
    this.id=id; this.pp=pp;
    ag = (ActiveGradient)pp.getModule("ActiveGradient");
    ns = (NeighborSet)pp.getModule("NeighborSet");
    config = new PNGData(r,u,l);
    VersionedGradient g = 
      new VersionedGradient(id,PersistentNodeB.pnType,config,prop(r,u,region));
    ag.sendGradient(g);
    init_local_creation(); // create data
    commit_state();
  }
  
  private PropagationTest prop(int radius, double umbra, Object region) {
    MultiTypePropagation p = new MultiTypePropagation();
    p.addType(new DistancePropagation((int)(radius*umbra)));
    p.addType(new RegionPropagation(region));
    p.finalize();
    return p;
  }

  public String toString() 
  { return "("+id+", "+config+", "+state+", "+data+", "+pendingWrite+")"; }

  // returns true if it should live, false if it should die
  // (and basically, it lives as long as it's got a gradient to feed on)
  public boolean update() {
    VersionedGradient g = (VersionedGradient)ag.get(id);
    // first, check if we're ready to be deleted
    if(g==null) {
      ns.setData(id,null);
      return false;
    }
    // next, check if we're dead
    if(!state.dead) { // then do updates
      node_update(g);
      data_update(g);
      commit_state();
    }
    return true;
  }

  // *** node maintainance ***
  // Die should be a) idempotent b) capable of propagating to others
  public void die() {
    if(!state.in_node) return; // only works on the node itself.
    state.dead=true;
    state.in_node=false;
    state.in_reflector=false;
    commit_state();
  }
  public void becomeCenter() {
    //System.out.println("be-center:"+state.value);
    VersionedGradient g = (VersionedGradient)ag.get(id);
    g.incrementVersion();
    g.setQuality(state.value);
    ag.sendGradient(g);
    state.reset();
    commit_state();
  }
  // the gradient's data contains the radius of the gradient... only those
  // items within 2x radius are used. The gradient extends to umbra x
  public boolean inNode() { // it's in the node if it's in the inmost radius
    Gradient g = ag.get(id);
    if(g==null || g.isDead()) return false;
    else return (g.getHopcount() < ((PNGData)g.data).radius);
  }
  public boolean inReflector() { // it's in node if it's in the 2nd radius
    Gradient g = ag.get(id);
    if(g==null || g.isDead()) return false;
    else return (g.getHopcount() < 2*((PNGData)g.data).radius);
  }
  public boolean inUmbra() {
    Gradient g = ag.get(id);
    if(g==null || g.isDead()) return false;
    else { 
      PNGData pngd = (PNGData)g.data;
      return (g.getHopcount() < pngd.radius*pngd.umbra);
    }
  }
  public boolean inPenumbra(double pu) {
    Gradient g = ag.get(id);
    if(g==null || g.isDead()) return false;
    else { 
      PNGData pngd = (PNGData)g.data;
      return (g.getHopcount() < pngd.radius*pu);
    }
  }
  // return distance from center
  public int centerDistance() {
    Gradient g = ag.get(id);
    if(g==null || g.isDead()) return -1;
    else { 
      PNGData pngd = (PNGData)g.data;
      return g.getHopcount();
    }
  }

  // *** data maintainance ***
  // only processors in the node acknowledge their data
  public Object getData() { 
    if(inNode()) {
      if(data==null) { System.out.println("READ DATA WAS NULL!\n"+pp.toString()); data=new NodeData(); }
      return data.committed.get(); 
    } else return null; 
  }
  // internal set [in response to local invocation of set]
  // each instance of this is separate, and there can only be one
  // pending at a given time (others will be rejected with an exception)
  VersionedData pendingWrite=null;
  DelayedReturn pendingRet;
  public void setData(Object d, DelayedReturn ret) 
    throws ConcurrentModificationException {
    if(pendingWrite!=null) { throw new ConcurrentModificationException(); }
    pendingRet = ret;
    if(data==null) { System.out.println("WRITE DATA WAS NULL!\n"+pp.toString()); data=new NodeData(); }
    int vers = data.committed.getVersion()+1;
    if(data.pending!=null) vers=data.pending.getVersion()+1;
    pendingWrite=new VersionedData(d,vers,Math.random());
    data.pending = pendingWrite;
    data.hop=0; data.backprop=false; data.age=0;
    commit_state();

    //System.out.println("Send initiated: "+this);
  }
  // external set (propagation of a non-local set)
  // merge the set with that of the current "pending"
  // if the current pending is 
  //public void setData(VersionedData d) {} // external set
  //public void commitData(VersionedData d) {}

  // Tasks of data_update
  // 1. send commits
  // 2. send acks
  // 3. maintain status of "pending"
  // nodes are involved in data-transmission if they're in the reflector
  private void data_update(VersionedGradient g) {
    if(!inReflector()) {
      if(pendingWrite!=null) {
        System.out.println("Write ("+pendingWrite+
                           ") terminated by expulsion from node.");
        if(pendingRet!=null) 
          pendingRet.send(new Boolean(false));
        pendingWrite=null; pendingRet=null;
      }
      data=null;
      return;
    } else {
      if(data==null) {
        data=new NodeData(); // a blank slate to be filled by neighbors
      }
    }

    // for committed, pending, pendingWrite, there are 4 states:
    // 1. P=W=C [p->null, w->null]
    // 2. P>W=C [w->null]
    // 3. P=W>C
    // 4. P>W>C

    // 1. integrate neighbor information
    Iterator i = ns.getNeighbors();
    while(i.hasNext()) {
      PNData d = (PNData)ns.getData(id,i.next());
      if(d!=null && d.data !=null) { 
        // update committed if nbr's better than commit & pending
        if(data.committed.compareTo(d.data.committed)<0 &&
           (data.pending==null || data.pending.compareTo(d.data.committed)<=0))
           data.committed.merge(d.data.committed);
        // update pending if nbr's better 
        if(data.pending!=null && d.data.pending!=null) {
          int compare = data.pending.compareTo(d.data.pending);
          if(compare<0) { // if better data, copy it 
            data.pending.merge(d.data.pending);
            data.hop=d.data.hop+1; // ... and set our hop to nbr+1
            data.backprop=false; data.age=0;
          } else if(compare==0) { // if same data, then minimize hop
            if(data.hop>d.data.hop+1) {
              data.hop=d.data.hop+1; data.backprop=false; data.age=0;
            }
          }
        } else if(d.data.pending!=null){  // if we've no data, copy from nbr 
          data.pending=d.data.pending;
          data.hop=d.data.hop+1;
          data.backprop=false; data.age=0;
        }
      }
    }
    // maintain constraints
    maintain_state(); // takes care of all C/P/W constraints

    // 2. manage state transitions
    // this is really just about dealing with the backprop and starting
    // commit waves
    if(data.pending!=null) {
      // 2a. set backprop value
      boolean got=false;
      i = ns.getNeighbors();
      while(i.hasNext()) {
        PNData d = (PNData)ns.getData(id,i.next());
        if(d!=null && d.data !=null) {
          if(data.pending.equals(d.data.pending)) {
            if(d.data.hop > data.hop && !d.data.backprop)
              { got=true; break; }
          }
        }
      }
      if(got) { data.age=0; /*data.backprop=false;*/ }
      else if(!data.backprop) {
        data.age++;
        if(data.age > backpropthresh) data.backprop=true;
      }
      // 2b. start commit wave
      // WARNING: I have commented out the data.backprop part, removing the
      // pending phase, so that the algorithm closely conforms to the PN paper
      // 7/10/03 JB
      // I think I may have invalidated this already...
      if(data.backprop && data.hop==0) { // if the backprop reaches the center
        data.committed=data.pending; maintain_state();
      }
    }
  }
  private static final int backpropthresh = 3;

  boolean noPending = true;
  private void maintain_state() {
    if(noPending && data.pending!=null) {
      data.committed.merge(data.pending);
      data.pending=null; data.hop=-1; data.backprop=false; data.age=0;
    }

    // Constraint 1: committed<pending
    // if violated, null out pending
    if(data.pending!=null && data.committed.compareTo(data.pending)>=0) {
      data.pending=null; data.hop=-1; data.backprop=false; data.age=0;
    }
    // Constraint 2: committed<pendingWrite
    // if violated, return true on the pending write and null it out
    if(pendingWrite!=null && data.committed.compareTo(pendingWrite)>=0) {
      if(pendingRet!=null) pendingRet.send(new Boolean(true));
      pendingWrite=null; pendingRet=null;
    }
    // Constraint 3: if pending, data.hop>1 || data.hop==0&&pendingWrite!=null
    // if violated, report and error
    if(data.pending!=null &&
       !(data.hop>=1 || (data.hop==0 && data.pending.equals(pendingWrite)))) {
      System.out.println("Data violating hop constraints: "+pp);
      Object a=null; System.out.println(a.toString());
    }
  }

  // *** private functions ***
  private void init_state_from_neighbors() {
    // find best neighbor...
    state = new NodeState();
    data = new NodeData();
    
    int besthop = -1;
    PNData bestdata = null;
    Iterator i = ns.getNeighbors();
    while(i.hasNext()) {
      PNData d = (PNData)ns.getData(id,i.next());
      if(d!=null && (besthop==-1 || d.state.hop<besthop)) {
        besthop = d.state.hop; bestdata = d; // we don't care to break ties
      }
    }
    if(bestdata!=null) {
      VersionedGradient g = (VersionedGradient)ag.get(id);
      state = new NodeState();
      calculate_state(g);
      data_update(g);
    } else {
      System.out.println("Why no nbr state?");
      init_local_creation(); // shouldn't ever happen
    }
  }
  private void init_local_creation() {
    VersionedGradient g = (VersionedGradient)ag.get(id);
    state = new NodeState();
    calculate_state(g);
    data = new NodeData();
    data_update(g);
  }

  // Tasks of node_update:
  // 1. check for death
  // 2. calculate state
  // 3. make gradients happen
  // 4. propagate succession
  private void node_update(VersionedGradient g) {
    renew_node_gradient(g);
    calculate_state(g);
    annoint_successor(g);
  }

  int lpTime = 0; static final int lpTimeout = 10;
  private void renew_node_gradient(VersionedGradient g) {
    // if we're the initiator, renew the gradient
    if(g.getHopcount()==0 && g.limit.propagate(g,pp)) { 
      state.lifepulse = Math.random(); state.lpID++; lpTime=0;
      ag.renewGradient(g); 
    }
    
    // if dead & close in, then become center with value=UID
    //if(g.isDead() && state.in_node && g.limit.propagate(g,pp)) {
    //  state.value = pp.getUID();
    //  ag.renewGradient(g,data.hop);
    //}
    lpupdate();
    if(state.in_node && g.limit.propagate(g,pp)) {
      if(lpTime<(lpTimeout+(2*g.getHopcount()))) {
        ag.renewGradient(g,g.getHopcount());
      } else { 
        state.value = pp.getUID();
        becomeCenter();
      }
    }
  }
  private void lpupdate() {
    lpTime++;
    Iterator i = ns.getNeighbors();
    while(i.hasNext()) {
      PNData d = (PNData)ns.getData(id,i.next());
      if(d==null) continue;
      if(d.state.lpID > state.lpID) {
        state.lpID=d.state.lpID;
        state.lifepulse=d.state.lifepulse;
        lpTime=0;
      }
    }
  }

  private void calculate_state(VersionedGradient g) {
    // check to see if we should become dead
    if(neighbor_dead()) { die(); return; } 

    // determine what areas we are in
    state.in_node = (!g.isDead() && inNode());
    state.in_reflector = (!g.isDead() && inReflector());
    // set our hopcount
    state.hop=g.getHopcount();
    // calculate the value

    // NOTE:it'd be nice to give each one its own randomized factor...
    if(state.in_reflector) { calculate_value(g); }
    else { state.age=0; state.value=Double.NaN; }
  }

  private boolean neighbor_dead() {
    Iterator i = ns.getNeighbors();
    while(i.hasNext()) {
      PNData d = (PNData)ns.getData(id,i.next());
      if(d!=null && d.state.dead)
        return true;
    }
    return false;
  }

  private void calculate_value(VersionedGradient g) {
    double v = 0;
    Iterator i = ns.getNeighbors();
    while(i.hasNext()) {
      PNData d = (PNData)ns.getData(id,i.next());
      if(d!=null && d.state.hop==state.hop+1) 
        if(d.state.in_reflector)
          v+=d.state.value;
    }
    double loadEst = ((PNGData)g.data).load.estimate(pp);
    if(loadEst<0) { v=(v-1)/(-loadEst); } else { v=(v+1)/loadEst; } 
    
    // if the value isn't changing and we've no older neighbor, increment age
    if(v!=state.value) { state.value=v; state.age=0; }
    else { // age remains at zero if there's a lower hopcount neighbor
      boolean found = false;
      i = ns.getNeighbors();
      while(i.hasNext()) {
        PNData d = (PNData)ns.getData(id,i.next());
        if(d!=null && d.state.hop<state.hop) { found=true; break; }
      }
      if(found) state.age=0;
      else state.age++;
    }
  }

  static final int ageThreshold = 10; // change-time bef. annointing successor
  static final double p_kill = 0.2;
  private void annoint_successor(VersionedGradient g) {
    // if age is enough, we're in the node proper, and we've no better
    // neighbor, then annoint a successor.  a "better" neighbor is one
    // with a lower hopcount, or eq hopcount and lower value.
    if(state.age > ageThreshold && state.in_node) {
      boolean center = true;
      SortedSet candidates = new TreeSet();
      Iterator i = ns.getNeighbors();
      while(i.hasNext()) {
        Object nbr = i.next();
        PNData d = (PNData)ns.getData(id,nbr);
        // add to successor counts
        if(d!=null && d.state.in_reflector) {
          candidates.add(new VersionedData(nbr,0,d.state.value));
        }
        // check for better annointer
        if(d!=null && (d.state.hop<state.hop || 
                       (d.state.hop==state.hop && d.state.value>state.value)))
          { center=false; break; }
      }
      // choose the nth best, chosen randomly
      while(Math.random()<p_kill && candidates.size()>1) {
        candidates.remove(candidates.last());
      }
      if(center && !candidates.isEmpty()) { // if survived and got candidate...
        // chose the successor
        VersionedData successor = (VersionedData)candidates.last();
        state.age=0; // make sure we don't reannoint for a while
        if(state.hop != 0) { // this shouldn't happen...
          System.out.println("BHV: "+state+" X "+ageThreshold);
          System.out.println("Big hop!\n"+pp.toString());
        }
        pp.sendMessage(PersistentNodeB.pnXfer,new Pair(id,successor.get()));
      }
    }
  }
  
  private void commit_state() { // put copy of current state into nbr set
    ns.setData(id,new PNData(state,data)); 
  }
}

