package modules.node;

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


public class RamboNode implements ProcessorModule {
  PortedProcessor pp; NeighborSet ns; Consensus cs; Broadcast bc;

  public Symbol getName() { return Symbol.GetSymbol("RamboNode"); }
  public void link(PortedProcessor p) { 
    pp = p;
    ns = (NeighborSet)p.getModule("NeighborSet");
    cs = (Consensus)pp.getModule("Consensus");
    bc = (Broadcast)pp.getModule("Broadcast");
    bc.openBroadcastPort(this,rambo);
    pp.openNetworkPort(this,annoint);
    pp.addClockListener(this);
  }

  public void init() {}

  public static final Symbol rambo = Symbol.GetSymbol("rambo");
  static final Symbol annoint = Symbol.GetSymbol("rambo:A");

  Hashtable nodes = new Hashtable(); // full node data structures
  HashSet activenodes = new HashSet();
  Hashtable configs = new Hashtable(); // node sketches
  private Object isRNCS(Object id) {
    if(!(id instanceof Tuple)) return null;
    Tuple t = (Tuple)id;
    if(t.size()!=2 || !Symbol.GetSymbol("RN:CS").equals(t.second())) 
      return null;
    return t.first();
  }
  Object NDID(Object id) { return new Tuple(id,Symbol.GetSymbol("RN:NS")); }
  Object DSID(Object id) { return new Tuple(id,Symbol.GetSymbol("RN:DS")); }
  private void deleteNode(Object id) {
    nodes.remove(id); 
    configs.remove(id); 
    activenodes.remove(id);
    ns.setData(NDID(id),null);
    ns.setData(DSID(id),null);
  }
  private RNConfig getCfgFromNbrs(Object id) {
    Iterator i = ns.getNeighbors();
    while(i.hasNext()) {
      Object nbrid=i.next();
      Hashtable nbrnodes = (Hashtable)ns.getData(rambo,nbrid);
      if(nbrnodes!=null && nbrnodes.containsKey(id))
        return (RNConfig)nbrnodes.get(id);
    }
    return null;
  }

  public boolean signalEvent(Symbol name,Object data) {
    boolean changed = false;
    Set deletes = new HashSet(nodes.keySet());
    Vector adds = new Vector();
    if(name==PortedProcessor.kClock) {
      // we're in a node if we're in it's consensus fringe
      Iterator i = cs.getConsensusProcesses().iterator();
      while(i.hasNext()) {
        Object cid = i.next();
        Object key = isRNCS(cid); if(key==null) continue;  // is RN consensus?
        if(nodes.containsKey(key)) {
          deletes.remove(key);
        } else {
          RNConfig cfg = getCfgFromNbrs(((Tuple)cid).first()); 
          if(cfg==null) continue;
          adds.add(cfg);
        }
      }
      i = adds.iterator();
      while(i.hasNext()) { addNode((RNConfig)i.next(),false); }
      i = deletes.iterator();
      while(i.hasNext()) { deleteNode(i.next()); }
      
      // update all of my nodes
      i=nodes.keySet().iterator();
      while(i.hasNext()) {
        Object key = i.next();
        RNode node = (RNode)nodes.get(key);
        node.update();
        // update activity list
        if(cs.containsFull(node.CID())) activenodes.add(node.cfg.nid);
        else activenodes.remove(node.cfg.nid);
      }
      
      // push data for neighbors
      ns.setData(rambo,new Hashtable(configs));
    } else if(name==annoint) { // contents: (name, destID, cfgid)
      Tuple t = (Tuple)data;
      if(t.second()!=null && pp.getUID()==((Double)t.second()).doubleValue()) {
        RNode node = (RNode)nodes.get(t.first());
        Integer cfgid = (Integer)t.third();
        if(node!=null && node.state.max_cfg==cfgid.intValue()) 
          node.center.becomeCandidate();
      }
    }
    return changed;
  }

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

  // Interface functions:
  // start a persistent node:
  int defaultRadius = 3;
  ProcessorLoadEstimator defaultEstimator = new RNodeCountEstimator();
  public void initiateNode(Object nid) 
  { initiateNode(nid,defaultRadius); }
  public void initiateNode(Object nid, int r)
  { initiateNode(nid,r,defaultEstimator); }
  public void initiateNode(Object nid, ProcessorLoadEstimator l)
  { initiateNode(nid,defaultRadius,l); }
  public void initiateNode(Object nid, int r, ProcessorLoadEstimator l) {
    if(nodes.contains(nid)) return; // no duplication
    addNode(new RNConfig(nid,r,l),true);
  }
  void addNode(RNConfig cfg, boolean coldboot) {
    configs.put(cfg.nid,cfg);
    nodes.put(cfg.nid,new RNode(pp,cfg,coldboot));
  }

  // end a persistent node
  public void killNode(Object nid) {
    // NEED TO WRITE THIS FUNCTION!
  }

  // debugging function to watch node progressions
  public void center(Object nid) {
    RNode node = (RNode)nodes.get(nid);
    if(node==null) return;
    node.center.becomeCandidate();
  }
  // get
  public boolean getData(Object nid, DelayedReturn ret)
    throws ConcurrentModificationException {
    RNode node = (RNode)nodes.get(nid);
    if(node==null) return false;
    node.getData(ret); 
    return true;
  }
  // 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 {
    RNode node = (RNode)nodes.get(nid);
    if(node==null) return false;
    node.setData(data,ret); 
    return true;
  }
  // list off nodes
  public Set getNodes() { // list only active nodes
    return activenodes;
  }
  public Set getNodeUmbras() { // list everything
    return nodes.keySet();
  }

  public boolean inNode(Object nid) {
    return activenodes.contains(nid);
  }

  // illegal accessor used to get data on nodes
  public int debugit(Object nid) {
    RNode n = (RNode)nodes.get(nid);
    if(n==null) return 0;
    return n.state.cfgs.size();
  }

  // illegal accessor used for pulling versioning data out for the atom test
  public VersionedData debugit2(Object nid,boolean read) { /* @deprecated */
    RNode n = (RNode)nodes.get(nid);
    if(n==null) return null;
    return n.data.op.queryFixedPoint;
  }
}






class RNConfig {
  final Object nid;
  final int r;
  final ProcessorLoadEstimator l;
  public RNConfig(Object nid, int r, ProcessorLoadEstimator l) 
  { this.nid=nid; this.r=r; this.l=l; }
  public String toString() { return "("+nid+", "+r+", "+l+")"; }
}

class RNState {
  Vector cfgs=new Vector(); // list of configurations
  Vector cfgdone=new Vector(); // list of config expirations
  int max_cfg=0; // Version ID of most recent config
  double v = 0.0;
  public Object clone() {
    RNState rns = new RNState();
    rns.cfgs = (Vector)cfgs.clone(); 
    Iterator i = cfgdone.iterator();
    while(i.hasNext()) { 
      Tuple t = (Tuple)i.next();
      rns.cfgdone.add(new Tuple(t.first(),((HashSet)t.second()).clone(),
                                ((Counter)t.third()).clone())); 
    }
    rns.max_cfg=max_cfg;
    rns.v = v;
    return rns;
  }
  public String toString() {
    String dstr="Dones=";
    Iterator i=cfgdone.iterator();
    while(i.hasNext()) {
      Tuple t = (Tuple)i.next(); 
      dstr+="["+t.first()+", "+((Set)t.second()).size()+","+t.third()+"]"; 
    }
    return "("+max_cfg+", #CFGs="+cfgs.size()+", "+dstr+", "+v+")";
  }
}

// this structure holds state for center-gathering
// and in fact, most of the relevant center-process...
class RNCenterProcess {
  Counter stable = new Counter(RNode.ageThreshold);
  Object successor = null; // chosen successor ID
  boolean candidate=false;
  boolean running=false;
  Object proposal = null;
  Object gid = null;
  void becomeCandidate() { candidate=true; }
  public String toString() {
    return "("+stable+", "+successor+", "+candidate+", "+running+", "+
      (proposal!=null)+", "+gid+")";
  }
}

class RNData {
  VersionedData value=new VersionedData(); // (v,t)
  int count = 0;
  int vers = 0;
  Hashtable extCounts = new Hashtable();
  Op op = new Op();
  class Op {
    static final int IDLE=0;
    static final int QUERY=1;
    static final int PROP=2;
    static final int DONE=3;
    boolean write;
    int phase = IDLE;
    int count;
    Set accepted;
    Object value;
    VersionedData queryFixedPoint=new VersionedData();
    DelayedReturn ret;
    public String toString() { 
      String s = "null"; if(value!=null) s=value.toString();
      int a = 0; if(accepted!=null) a=accepted.size();
      String p = "N/A";
      switch(phase) { 
      case IDLE: {p="IDLE";break;}
      case QUERY: {p="QUERY";break;}
      case PROP: {p="PROP";break;}
      case DONE: {p="DONE";break;}
      }
      return "("+write+", "+p+", "+count+", A="+a+", "+s+")";
    }
  }
  public String toString() { return "("+value+", "+count+", "+op+")"; }
}

class RNode {
  RNState state = new RNState();
  RNCenterProcess center = new RNCenterProcess();
  RNConfig cfg;
  RNData data = new RNData();

  static final int gc_count = 10;

  public RNode(PortedProcessor pp, RNConfig cfg, boolean coldboot) {
    this.cfg=cfg;
    this.pp = pp;
    ns = (NeighborSet)pp.getModule("NeighborSet");
    cs = (Consensus)pp.getModule("Consensus");
    ag = (ActiveGradient)pp.getModule("ActiveGradient");
    rm = (RegionManager)pp.getModule("RegionManager");
    bc = (Broadcast)pp.getModule("Broadcast");
    if(coldboot) {
      cs.startConsensus(CID(),1,RMT());
      Hashtable metable = new Hashtable(); 
      metable.put(pp.getUIDd(),new RNGather.DS(0,true,1));
      metable.put(RNPC(),new Integer(1));
      state.cfgs.add(new Tuple(new Integer(0),metable));
      state.cfgdone.add(new Tuple(new Integer(state.max_cfg),new HashSet(), 
                                  new Counter(gc_count)));
      center.becomeCandidate();
    }
  }

  PortedProcessor pp; NeighborSet ns; Consensus cs; ActiveGradient ag;
  RegionManager rm; Broadcast bc;

  public String toString() {
    return "("+cfg+", "+state+", "+center+", "+data+")";
  }

  static Object RNPC() { return Symbol.GetSymbol("RN:Participants"); }
  Object CID() { return new Tuple(cfg.nid,Symbol.GetSymbol("RN:CS")); }
  Object NDID() { return new Tuple(cfg.nid,Symbol.GetSymbol("RN:NS")); }
  RegionMembershipTest RMT() { return new ConsensusFringe(CID(),cfg.r+1,pp); }
  Object makeCGID() { 
    return new Tuple(cfg.nid,Symbol.GetSymbol("RN:CG"),
                     new Double(Math.random())); 
  }
  static final int ageThreshold = 10; // change-time bef. annointing successor
  static final double p_kill = 0.2;
  
  void update() {
    config_update();
    center_update();
    data_update();
    
    // last but not least, push state to neighbors
    ns.setData(NDID(),state.clone());
  }

  static final int maxdelta = 50000; // how long old configs are kept
  void config_update() {
    // if we're in the consensus process, check for a step forward
    if(cs.containsFull(CID())) { 
      VersionedData cfg = cs.getConsensus(CID());
      if(cfg.getVersion()>state.max_cfg) {
        state.max_cfg=cfg.getVersion();
        state.cfgs.add(new Tuple(new Integer(cfg.getVersion()),
                                 ((Hashtable)cfg.get()).clone())); 
        // mark all previous dones
        Iterator j=state.cfgdone.iterator();
        while(j.hasNext()) { 
          Tuple t = (Tuple)j.next();
          ((HashSet)t.second()).add(pp.getUIDd());
        }
        state.cfgdone.add(new Tuple(new Integer(state.max_cfg),new HashSet(),new Counter(gc_count)));
        center = new RNCenterProcess();
      }
    }
    
    // integrate config data from neighbors
    Iterator i = ns.getNeighbors();
    while(i.hasNext()) {
      Object nbr = i.next();
      RNState nbrstate = (RNState)ns.getData(NDID(),nbr);
      if(nbrstate!=null) {
        if(nbrstate.max_cfg>state.max_cfg) {
          state.max_cfg=nbrstate.max_cfg;
          // just copy the top one
          state.cfgs.add(((Tuple)nbrstate.cfgs.lastElement()).clone());
          state.cfgdone.add(new Tuple(new Integer(state.max_cfg),
                                      new HashSet(),new Counter(gc_count)));
          center = new RNCenterProcess();
        }
        // integrate done-ness measures
        Iterator j = nbrstate.cfgdone.iterator();
        while(j.hasNext()) {
          Tuple nbrdone = (Tuple)j.next();
          Iterator k = state.cfgdone.iterator();
          while(k.hasNext()) {
            Tuple mydone = (Tuple)k.next();
            if(nbrdone.first().equals(mydone.first())) {
              ((HashSet)mydone.second()).addAll((HashSet)nbrdone.second());
            }
          }
        }
      }
    }
    
    // garbage collect old configs
    i = state.cfgs.iterator();
    Iterator dit = state.cfgdone.iterator();
    while(i.hasNext()) {
      if(!dit.hasNext()) System.out.println("FAILURE: donelist!=cfglist");
      Tuple cfg = (Tuple)i.next();
      Tuple mydone = (Tuple)dit.next();
      if((state.max_cfg-((Integer)cfg.first()).intValue())>maxdelta) {
        System.out.println("Unhappy GC");
        i.remove(); dit.remove(); continue;
      }
      Integer csize = (Integer)((Hashtable)cfg.second()).get(RNPC());
      int majority = (csize.intValue()/2)+1;
      if(((HashSet)mydone.second()).size()>=majority) {
        //System.out.println("GC");
        Counter c = (Counter)mydone.third();
        if(c.expired()) {
          if(state.cfgs.size()>2) { System.out.println("possible failstate\t"+pp.getUID()+"\t"+cfg.first()+"\t"+state.cfgs.size()+"\t"+(data.op.phase==data.op.IDLE)); }
          i.remove();
          dit.remove();
        } else {
          c.tick();
        }
      }
      
    }

    // mark ourselves as active if we're in the top config
    Hashtable topcfg = null;
    if(!state.cfgs.isEmpty()) {
      topcfg=(Hashtable)((Tuple)state.cfgs.lastElement()).second();
    }
    if(topcfg!=null && topcfg.containsKey(pp.getUIDd())
       && ((RNGather.DS)topcfg.get(pp.getUIDd())).in_node) {
      VersionedData cfg = cs.getConsensus(CID());
      cs.startConsensus(CID(),(((Integer)topcfg.get(RNPC())).intValue()/2)+1,
                        RMT());
    } else {
      cs.leaveConsensus(CID());
      //center.candidate=false;
    }
  }
  
  // this is used for deciding to become the new center
  void center_update() {
    // calculate heuristic
    if(!state.cfgs.isEmpty()) {
      Hashtable topcfg=(Hashtable)((Tuple)state.cfgs.lastElement()).second();
      RNGather.DS myds = (RNGather.DS)topcfg.get(pp.getUIDd());
      if(myds!=null) {
        double v = 0;
        boolean lowerNbr = false;
        Iterator i = ns.getNeighbors();
        while(i.hasNext()) {
          Object nbrid = i.next();
          RNState s = (RNState)ns.getData(NDID(),nbrid);
          RNGather.DS nbrds = (RNGather.DS)topcfg.get(nbrid);
          if(s!=null && nbrds!=null) {
            if(nbrds.hop==myds.hop+1) 
              v+=s.v;
            if(nbrds.hop<myds.hop)
              lowerNbr=true;
          }
        }
        if(myds.loadEst<0) { v=(v-1)/(-myds.loadEst); } 
        else { v=(v+1)/myds.loadEst; } 
        if(v!=state.v) {
          state.v=v; // set the state variable
          center.stable.reset(); // change means reset timer
        } else if(lowerNbr) {
          center.stable.reset(); // low nbr means reset timer
        } else {
          center.stable.tick();
        }
      }
    }

    // annoint a successor if we're stable, in charge, and not in the running
    if(center.stable.expired()) { // old: && !center.candidate
      annointsuccessor();
    }

    // manage the center-finding gradients
    if(center.candidate) { // send the gradient
      if(center.gid==null) center.gid=makeCGID();
      ag.sendGradient(center.gid,RamboNode.rambo,null,
                      new DistancePropagation(2*cfg.r));
      if(!center.running) { // start the query...
        //System.out.println("start!");
        rm.addRegion(center.gid,RamboNode.rambo,
                     new GradientRegion(center.gid));
        bc.convergecast(RamboNode.rambo,
                        new RNGather(cfg,center.gid),center.gid);
        center.running = true;
      }
    }

    // if query is complete, propose the new config & quiesce
    if(center.proposal!=null) {
      //System.out.println("complete!"+center.proposal);
      // only propose if we're still a candidate
      if(center.candidate) {
        //System.out.println("propose!");
        cs.propose(CID(),center.proposal);
      }
      // wipe the data, whether or not we succeeded
      center = new RNCenterProcess();
    }
  }

  void annointsuccessor() {
    // check if the old successor is still valid
    if(center.successor != null) {
      RNState s = (RNState)ns.getData(NDID(),center.successor);
      if(s==null) center.successor=null;
    }
    // attempt to fill in a blank
    if(center.successor==null) {
      SortedSet candidates = new TreeSet();
      Iterator i = ns.getNeighbors();
      while(i.hasNext()) {
        Object nbr = i.next();
        RNState s = (RNState)ns.getData(NDID(),nbr);
        // add to successor counts
        if(s!=null) {
          candidates.add(new VersionedData(nbr,0,s.v));
        }
      }
      // choose the nth best, chosen randomly
      while(Math.random()<p_kill && candidates.size()>1) {
        candidates.remove(candidates.last());
      }
      if(!candidates.isEmpty()) { // if got candidate, annoint successor
        // chose the successor
        center.successor = ((VersionedData)candidates.last()).get();
        //System.out.println(pp.getUID()+" annointing "+center.successor);
      }
    }
    // do the actual annointment
    if(center.successor != null) {
      //center.stable.reset(); // make sure we don't reannoint for a while
      Tuple t=new Tuple(cfg.nid,center.successor,new Integer(state.max_cfg));
      pp.sendMessage(RamboNode.annoint,t);
    }
  }

  void reportProposal(Object proposal, RNConfig cfg, Object gid) {
    if(center.proposal==null)
      center.proposal=proposal;
  }


  // now we get to the reason why RAMBO doesn't scale:
  // the gossip net we're about to create is required to have
  // O(n^2) size messages!
  // Here in Java, we'll alleviate some of this suckage by 
  // having many messages be passed as pointers, but in the real
  // world we're screwed.
  void getData(DelayedReturn ret)
    throws ConcurrentModificationException {
    if(data.op.phase!=RNData.Op.IDLE) 
      throw new ConcurrentModificationException();
    data.count++;
    data.op.write = false;
    data.op.phase = RNData.Op.QUERY;
    data.op.ret = ret;
    data.op.count = data.count;
    data.op.accepted = new HashSet();
    data.op.value = null;
  }
  void setData(Object d, DelayedReturn ret) 
    throws ConcurrentModificationException {
    if(data.op.phase!=RNData.Op.IDLE) 
      throw new ConcurrentModificationException();
    data.count++;
    data.op.write = true;
    data.op.phase = RNData.Op.QUERY;
    data.op.ret = ret;
    data.op.count = data.count;
    data.op.accepted = new HashSet();
    data.op.value = d;
  }
  // Data transferred between neighbors is just value & extCounts
  // format: Tuple of (data, extCounts) where each extCount entry
  // is a hashtable containing its own value too
  // phases:
  // 1. merge value with neighbor values & merge extCounts
  // 1a. prune extcounts for nodes no longer in the config
  // 2. increase the "accepted" set for the current op
  // 3. move completed query forward to propagate
  // 4. return ack for completed propagate
  // 5. put data for neighbors
  // make sure we only play if we're in the current config
  Object DSID() { return new Tuple(cfg.nid,Symbol.GetSymbol("RN:DS")); }
  static final Symbol vers = Symbol.GetSymbol("version");
  boolean in_node(Object uid, Hashtable cfg) { 
    return (cfg.containsKey(pp.getUIDd()) &&
            ((RNGather.DS)cfg.get(pp.getUIDd())).in_node);
  }
  void data_update() {
    // 0. only work with data if we're in the current config
    Hashtable topcfg = null;
    if(!state.cfgs.isEmpty()) 
      topcfg = (Hashtable)((Tuple)state.cfgs.lastElement()).second();
    if(topcfg==null || !topcfg.containsKey(pp.getUIDd())) {
      data = new RNData();
      ns.setData(DSID(),null);
      return; // nuke data and return
    }

    // 0a. ensure that myht exists, if nothing else...
    boolean participant = in_node(pp.getUIDd(),topcfg);
    Hashtable myht = (Hashtable)data.extCounts.get(pp.getUIDd());
    if(myht==null && participant) {
      myht = new Hashtable();
      myht.put(pp.getUIDd(),new Integer(data.count));
      myht.put(vers,new Integer(data.vers));
      data.extCounts.put(pp.getUIDd(),myht);
    }

    // 1. merge value with neighbor values & merge extCounts
    Iterator i = ns.getNeighbors();
    while(i.hasNext()) {
      Object nbr = i.next();
      Tuple t = (Tuple)ns.getData(DSID(),nbr);
      if(t==null) continue;
      data.value.merge((VersionedData)t.first()); // integrate value
      Hashtable nbrCounts = (Hashtable)t.second();
      Iterator j = nbrCounts.keySet().iterator();
      while(j.hasNext()) {
        Object id = j.next();
        Hashtable nbrtbl = (Hashtable)nbrCounts.get(id);
        Hashtable mytbl = (Hashtable)data.extCounts.get(id);
        boolean failon=false;
        if(mytbl!=null && mytbl.get(vers)==null) {
          failon=true; System.out.println("FAILURE IMMINENT!");
        }
        if(mytbl==null || 
           ((Integer)mytbl.get(vers)).compareTo(nbrtbl.get(vers))==-1) {
          data.extCounts.put(id,nbrtbl);
          if(participant) myht.put(id,nbrtbl.get(id));
        }
        if(failon) System.out.println("I guess I cried wolf.");
      }
    }
    
    // 1a. prune extcounts for nodes no longer in the config
    i = data.extCounts.keySet().iterator();
    while(i.hasNext()) { 
      Object key = i.next();
      if(!in_node(key,topcfg)) {
        if(participant) myht.remove(key);
        i.remove(); 
      }
    }

    // 2. increase the "accepted" set for the current op
    if(data.op.phase == RNData.Op.QUERY || data.op.phase == RNData.Op.PROP) {
      i = data.extCounts.keySet().iterator();
      while(i.hasNext()) {
        Object key = i.next();
        Hashtable tbl = (Hashtable)data.extCounts.get(key);
        Integer pnr = (Integer)tbl.get(pp.getUIDd());
        if(pnr!=null && pnr.intValue()>=data.op.count) 
          data.op.accepted.add(key);
      }
    }

    // 3. move completed query forward to propagate
    if(data.op.phase==RNData.Op.QUERY) {
      // check ALL live configs
      boolean ok = true;
      Iterator cit = state.cfgs.iterator();
      while(cit.hasNext()) {
        Hashtable curcfg = (Hashtable)((Tuple)cit.next()).second();
        int majority = (((Integer)curcfg.get(RNPC())).intValue()/2)+1;
        int tags = 0;
        i = data.op.accepted.iterator();
        while(i.hasNext()) { if(in_node(i.next(),curcfg)) tags++; }
        if(tags<majority) { ok=false; break; }
      }
      if(ok) {
        if(data.op.write) {
          data.value.set(data.op.value); 
          data.value.incrementVersion(); data.value.setQuality(pp.getUID());
        } else {
          data.op.value = data.value.get();
        }
        data.op.queryFixedPoint = (VersionedData)data.value.clone();
        data.op.count++; data.count++; 
        data.op.phase = RNData.Op.PROP;
        data.op.accepted = new HashSet();
      }
    }

    // 4. return ack for completed propagate
    if(data.op.phase==RNData.Op.PROP) {
      // check ALL live configs
      boolean ok = true;
      Iterator cit = state.cfgs.iterator();
      while(cit.hasNext()) {
        Hashtable curcfg = (Hashtable)((Tuple)cit.next()).second();
        int majority = (((Integer)curcfg.get(RNPC())).intValue()/2)+1;
        int tags = 0;
        i = data.op.accepted.iterator();
        while(i.hasNext()) { if(in_node(i.next(),curcfg)) tags++; }
        if(tags<majority) { ok=false; break; }
      }
      if(ok) {
        if(data.op.write) {
          if(data.op.ret!=null) {
            data.op.ret.send(Symbol.GetSymbol("ack"));
          }
        } else {
          if(data.op.ret!=null) {
            data.op.ret.send(data.op.value);
          }
        }
        data.op.phase = RNData.Op.IDLE;
      }
    }
    
    // 5. put data for neighbors
    if(participant) {
      data.vers++; 
      myht.put(vers,new Integer(data.vers));
      myht.put(pp.getUIDd(),new Integer(data.count)); // update personal entry
    }
    Hashtable extClone = new Hashtable();
    i=data.extCounts.keySet().iterator();
    while(i.hasNext()) {
      Object key = i.next();
      extClone.put(key,((Hashtable)data.extCounts.get(key)).clone());
    }
    ns.setData(DSID(),new NoPrintTuple(data.value.clone(),extClone));
  }
  static class NoPrintTuple extends Tuple {
    NoPrintTuple(Object a, Object b) { super(a,b); }
    public String toString() { return "(SELF-CENSORED)"; }
  }
}

/*
  OK, new way of doing things, here.
  The gather will also collect hop numbers.
  From there it goes as before...
*/

/*
Recon service:
  cycle of operations: 
    new-config announces a definitive new configuration
    gather marks a possible configuration
    recon proposes a configuration
    paxos chooses the new configuration
  maintains a list of configurations; when stale, old configs are GC'ed

configs: sequence#, config
a configuration maintains itself by gossip-net.
GC is dont by killing off things which are either
  a: unable to make progress quickly enough or
  b: recieve gossip information that leads them to GC all of the
     configurations containing themselves
*/



class RNGather implements ConvergeCast {
  Hashtable h = new Hashtable();
  RNConfig cfg; Object gid;
  public RNGather(RNConfig cfg, Object gid) 
  { this.cfg=cfg; this.gid=gid; h.put(RNode.RNPC(),new Integer(0)); }

  static class DS {
    int hop;
    boolean in_node;
    double loadEst;
    DS(int h, boolean n, double l) {hop=h; in_node=n; loadEst=l;}
  }
  void recordDS(PortedProcessor pp) {
    ActiveGradient ag = (ActiveGradient)pp.getModule("ActiveGradient");
    Gradient g = ag.get(gid);
    if(g!=null) {
      int hop = g.getHopcount();
      double load = cfg.l.estimate(pp);
      h.put(pp.getUIDd(),new DS(hop,hop<cfg.r,load)); 
      if(hop<cfg.r) { // increment the participant count
        Integer pc = (Integer)h.get(RNode.RNPC());
        h.put(RNode.RNPC(),new Integer(pc.intValue()+1));
      }
    } else {
      System.out.println("Why is this null?");
    }
  }
  
  void integrate(Map m) {
    Integer pc = (Integer)h.get(RNode.RNPC());
    Integer mc = (Integer)m.get(RNode.RNPC());
    h.putAll(m); 
    h.put(RNode.RNPC(),new Integer(pc.intValue()+mc.intValue()));
  }

  public Object getLeafValue(PortedProcessor pp) { 
    recordDS(pp);
    //System.out.println("Leaf "+pp.getUIDd()+" reports "+h.get(RNode.RNPC()));
    return h;
  }
  public Object deliverLeaf(PortedProcessor pp) { return null; }
  public Object deliverBranch(PortedProcessor pp) { return null; }
  public Object fail(PortedProcessor pp) { return null; }

  public void putBranchValue(PortedProcessor pp, Object input) 
  { integrate((Map)input); }
  public void putRootValue(PortedProcessor pp, Object input)
  { integrate((Map)input); }
  public Object getBranchValue(PortedProcessor pp) {    
    recordDS(pp);     
    //System.out.println("Mid "+pp.getUIDd()+" reports "+h.get(RNode.RNPC()));
    return h; 
  }
  public Object deliverRoot(PortedProcessor pp) { 
    recordDS(pp);
    //System.out.println("Root "+pp.getUIDd()+" reports "+h.get(RNode.RNPC()));
    // dump stuff directly in the center data field:
    RamboNode rn = (RamboNode)pp.getModule("RamboNode");
    RNode node = (RNode)rn.nodes.get(cfg.nid);
    if(node!=null) {
      node.reportProposal(h,cfg,gid);
    }
    return h; // not that we're going to listen to it.
  }

  
  public ConvergeCast getChild() { return new RNGather(cfg,gid); }
}
