package modules.consensus;

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

// uses Paxos consensus
  //**KEY PROBLEM: WHEN NODES LEAVE, THEY AREN'T REMOVED FROM THE SUMMARY

public class Consensus implements ProcessorModule {
  PortedProcessor pp; NeighborSet ns; RegionManager rm;

  public Symbol getName() { return Symbol.GetSymbol("Consensus"); }
  public void link(PortedProcessor p) { 
    pp = p;
    ns = (NeighborSet)p.getModule("NeighborSet");
    rm = (RegionManager)pp.getModule("RegionManager");
    pp.addClockListener(this);
  }

  public void init() {
  }

  Hashtable paxos = new Hashtable();
  public static final Symbol type = Symbol.GetSymbol("consensus");

  public boolean signalEvent(Symbol name,Object data) {
    boolean changed = false;
    if(name==PortedProcessor.kClock) {
      // run the update
      Iterator i = paxos.keySet().iterator();
      while(i.hasNext()) {
        Object k = i.next();
        Paxos p = (Paxos)paxos.get(k);
        p.update();
      }
      join();
      prune();
    }
    return changed;
  }

  public Set getConsensusProcesses() { return paxos.keySet(); }
  public boolean contains(Object pid) { return paxos.containsKey(pid); }
  public boolean containsFull(Object pid) 
  { return paxos.containsKey(pid) && ((Paxos)paxos.get(pid)).getFull(); }
  public boolean isConverged(Object pid) {
    Paxos p = (Paxos)paxos.get(pid);
    if(p==null) return false; else return p.isConverged();
  }
  public VersionedData getConsensus(Object pid) {
    Paxos p = (Paxos)paxos.get(pid);
    if(p==null) return null; else return p.consensus();
  }
  public void propose(Object pid, Object v) {
    Paxos p = (Paxos)paxos.get(pid);
    if(p!=null) p.propose(v);
  }
  public void startConsensus(Object pid, int majority) {
    startConsensus(pid,majority,new ConsensusRegion(pid));
  }
  public void startConsensus(Object pid, int majority, 
                             RegionMembershipTest rmt) {
    rm.addRegion(RID(pid,majority),type,rmt);
    Paxos p = (Paxos)paxos.get(pid);
    if(p==null) paxos.put(pid,new Paxos(pid,majority,pp));
    else {
      p.majority=majority;
      p.setFull(true);
    }
  }
  //**KEY PROBLEM: WHEN NODES LEAVE THIS, THEY AREN'T REMOVED FROM THE SUMMARY
  public void leaveConsensus(Object pid) {
    Paxos p = (Paxos)paxos.get(pid);
    if(p!=null) {
      p.setFull(false); 
    }
  }
  public void terminateConsensus(Object pid) {
    Paxos p = (Paxos)paxos.get(pid);
    if(p!=null) {
      p.setFull(false); 
      rm.killRegion(RID(p.name,p.majority));
    }
  }
  // keep only live ones
  private void prune() {
    Iterator i = paxos.keySet().iterator();
    while(i.hasNext()) {
      Object pid = i.next();
      Paxos p = (Paxos)paxos.get(pid);
      if(!p.getFull() && !rm.inRegion(RID(p.name,p.majority))) {
        i.remove(); ns.setData(new Pair(Paxos.paxos,pid),null);
      }
    }
  }
  // join if in region
  private void join() {
    Iterator i = rm.getRegions(type).iterator();
    while(i.hasNext()) {
      Tuple rid = (Tuple)i.next();
      Object pid = rid.first();
      if(!paxos.containsKey(pid)) paxos.put(pid,new Paxos(rid,pp));
    }
  }
  Tuple RID(Object name, int majority) 
  { return new Tuple(name,new Integer(majority)); }

  public String toString() {
    return "Consensus Processes: "+paxos.toString();
  }

}

// a Paxos instance is a consensus agency
// Functionality:
// 1. Proposers create request and accept flows
// 2. Acceptors mark accepted values
// 3. Learners watch for consensus and update consensus version
// Every Paxos process is an acceptor.
// 

class Paxos {
  // the summary needs to know the propose/accept state of all procs
  // and the highest accepted v
  class Summary {
    VersionedData chosen=new VersionedData(); // most recent chosen value
    HashMap prepared=new HashMap();
    HashMap accepted=new HashMap();

    public String toString()
    { return "("+chosen.toString(false)+
        ", "+prepared.size()+", "+accepted.size()+")"; }
    public Object clone() {
      Summary s = new Summary(); 
      s.chosen=(VersionedData)chosen.clone(); 
      s.prepared=new HashMap();
      Iterator i = prepared.keySet().iterator();
      while(i.hasNext()) {
        Object k = i.next(); VersionedData vd = (VersionedData)prepared.get(k);
        s.prepared.put(k,vd.clone()); 
      }
      s.accepted=new HashMap();
      i = accepted.keySet().iterator();
      while(i.hasNext()) {
        Object k = i.next(); VersionedData vd = (VersionedData)accepted.get(k);
        s.accepted.put(k,vd.clone()); 
      }
      return s;
    }
    // walk through and update the values stored
    public void integrate(Summary s) {
      chosen.merge(s.chosen);
      Iterator i = s.prepared.keySet().iterator();
      while(i.hasNext()) {
        Object k = i.next();
        VersionedData nbrv = (VersionedData)s.prepared.get(k);
        VersionedData v = (VersionedData)prepared.get(k);
        if(v==null || v.compareTo(nbrv)==-1) prepared.put(k,nbrv);
      }
      i = s.accepted.keySet().iterator();
      while(i.hasNext()) {
        Object k = i.next();
        VersionedData nbrv = (VersionedData)s.accepted.get(k);
        VersionedData v = (VersionedData)accepted.get(k);
        if(v==null || v.compareTo(nbrv)==-1) accepted.put(k,nbrv);
      }
    }
  }

  // at every time step:
  // 1. check for consensus
  // 2. look for proposals that can be pushed to the next state
  // 3. forward changes to neighbors
  static final Symbol paxos = Symbol.GetSymbol("Paxos");

  VersionedData prepared=new VersionedData(); // highest prepared value
  VersionedData accepted=new VersionedData(); // highest accepted value
  VersionedData chosen=new VersionedData(); // highest chosen value
  VersionedData candidate=new VersionedData(); // candidate value
  Object name; // name of this paxos instance
  int majority; // minimum size for a majority
  Summary summary=new Summary();
  PortedProcessor pp; NeighborSet ns;

  boolean full_participant; // if false, means it's just part of the comm

  public Paxos(Object name, int majority,PortedProcessor pp)
  { init(name,majority,pp,true); }
  public Paxos(Tuple data,PortedProcessor pp)
  { init(data.first(),((Integer)data.second()).intValue(),pp,false); }

  private void init(Object name, int majority,PortedProcessor pp,boolean full){
    this.name=name; this.majority=majority; this.pp=pp;
    ns = (NeighborSet)pp.getModule("NeighborSet");
    summary.chosen=chosen;
    full_participant=full;
    if(full) {
      summary.prepared.put(pp.getUIDd(),prepared);
      summary.accepted.put(pp.getUIDd(),accepted);
    }
  }
  void setFull(boolean full) { full_participant=full; }
  boolean getFull() { return full_participant; }

  public void update() {
    /**** INTEGRATION ****/
    // first, integrate the neighbor data
    Iterator i = ns.getNeighbors();
    while(i.hasNext()) {
      Summary s = (Summary)ns.getData(new Pair(paxos,name),i.next());
      if(s!=null) summary.integrate(s);
    }
    // next, raise the prepared promise to the highest known value
    i = summary.prepared.keySet().iterator(); 
    while(i.hasNext()) {
      Object k = i.next();
      prepared.merge((VersionedData)summary.prepared.get(k));
    }
    // third, raise the acceptances likewise
    i = summary.accepted.keySet().iterator(); 
    while(i.hasNext()) {
      Object k = i.next();
      VersionedData vd = (VersionedData)summary.accepted.get(k);
      if(vd.compareTo(prepared)!=-1) // only accept if at least prep-value
        accepted.merge(vd);
    }

    /**** PROGRESS ****/
    // try to migrate prepare to accept
    int count = 0; VersionedData best = chosen;
    i = summary.prepared.keySet().iterator(); 
    while(i.hasNext()) {
      Object k = i.next();
      VersionedData vd = (VersionedData)summary.prepared.get(k);
      // count only acceptors with prepare>=candidate
      if(vd.compareTo(candidate)!=-1) {
        VersionedData ad = (VersionedData)summary.accepted.get(k);
        best.merge(ad);
        count++;
      }
    }
    // if no accepteds yet, use the candidate. Otherwise use the best
    if(count>=majority) {
      if(best.equals(chosen)) {
        accepted.merge(candidate);
      } else {
        accepted.merge(best);
      }
    }

    // try to migrate accept to choose
    Hashtable props = new Hashtable();
    i = summary.accepted.keySet().iterator();
    while(i.hasNext()) {
      Object k = i.next();
      VersionedData vd = (VersionedData)summary.accepted.get(k);
      if(vd.get()!=null && vd.getVersion()>chosen.getVersion()) {
        Counter c = (Counter)props.get(vd.get());
        if(c==null) { c=new Counter(majority); props.put(vd.get(),c); }
        c.tick();
        if(c.expired()) { // if a majority has been achieved
          //System.out.println("hubba");
          chosen.merge(vd); chosen.setQuality(0);
        }
      }
    }

    /**** PROPAGATION ****/
    // finally, put our values into the summary 
    if(full_participant) {
      summary.prepared.put(new Double(pp.getUID()),prepared);
      summary.accepted.put(new Double(pp.getUID()),accepted);
    }
    summary.chosen.merge(chosen);
    ns.setData(new Pair(paxos,name),summary.clone());
  }

  // propose simply picks a random number and adds it to the current chosen
  // this will spread the proposal around
  public void propose(Object c) {
    if(!full_participant) return; // connective tissue doesn't interface
    candidate = new VersionedData(c,chosen.getVersion()+1,
                                  prepared.getQuality()+Math.random());
    prepared.merge(candidate);
  }

  public VersionedData consensus() { 
    if(!full_participant) return null; // connective tissue doesn't interface
    return chosen; 
  }
  public boolean isConverged() {
    if(!full_participant) return false; // connective tissue doesn't interface
    return chosen.getVersion()>=prepared.getVersion(); 
  }

  public static boolean verbose = false;
  public String toString() {
    String s = "(["+name+", "+majority+", "+chosen.toString(verbose);
    if(!full_participant) s+= ", OBS";
    s += "] "+prepared.toString(verbose)+", "+ accepted.toString(verbose)+
      ", "+candidate.toString(verbose)+")";
    return s;
  }
}

