package modules.gossip;

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

/* 
   Gossip, like broadcast, fills an area with information. Gossip,
   however, is centerless. Rather, a Gossip item merges data from
   all sources is the region, spreading it omnidirectionally through
   neighbors.
*/

public class GossipManager extends Observable implements ProcessorModule {
  PortedProcessor pp; RegionManager rm; NeighborSet ns;

  public Symbol getName() { return Symbol.GetSymbol("GossipManager"); }
  public void link(PortedProcessor p) {
    pp = p;
    rm = (RegionManager)pp.getModule("RegionManager");
    ns = (NeighborSet)pp.getModule("NeighborSet");
    pp.addClockListener(this);
  }
  
  public void init() {}
  
  static final Symbol gtype = Symbol.GetSymbol("gossip");
  public boolean signalEvent(Symbol name,Object data) {
    boolean changed = false;
    if(name==PortedProcessor.kClock) { // run updates
      // merge neighbor information for each gossip item
      Iterator i = ns.getNeighbors();
      while(i.hasNext()) {
        Pair p = (Pair)ns.getData(gtype,i.next());
        if(p==null) continue;
        Hashtable gt = (Hashtable)((Pair)p).car();
        Hashtable st = (Hashtable)((Pair)p).cdr();
        Iterator j = gt.keySet().iterator();
        while(j.hasNext()) {
          Object gid = j.next();
          Tuple gdata = (Tuple)gt.get(gid);
          Tuple sdata = (Tuple)st.get(gid);
          // integrate neighbors from the right region
          if(rm.inRegion(gdata.third())) {
            if(sdata.first()==DEAD) { // dead pass contagion
              if(contains(gid)) kill(gid);
            } else if(sdata.first()==LIVE) { // live pass information
              if(contains(gid)) 
                getGossip(gid).integrate((Gossip)gdata.second(),pp);
              else {
                addGossip(gid,gdata);
                setChanged(); notifyObservers(new Pair(gid,getGossip(gid)));
              }
            } else {
              // do nothing with the dying
            }
          }
        }
      }

      // now the local update...
      Vector deletes=new Vector();
      i=gossip.keySet().iterator();
      while(i.hasNext()) {
        Object gid = i.next();
        Tuple gt = (Tuple)gossip.get(gid);
        Gossip g = (Gossip)gt.second();
        Object rid = gt.third();
        Tuple stat = (Tuple)status.get(gid);
        // if not in region, prune
        if(!rm.inRegion(rid)) { deletes.add(gid); }
        else if(stat.first()==LIVE) { // update live gossips with local info
          g.update(pp);
        } else if(stat.first()==DYING) { // advance dying progress to dead
          Counter c = (Counter)stat.second();
          c.tick();
          if(c.expired()) status.put(gid,new Tuple(DEAD,new Counter(timeout)));
        } else if(stat.first()==DEAD) { // advance dead to delete
          Counter c = (Counter)stat.second();
          c.tick();
          if(c.expired()) deletes.add(gid);
        }
      }
      // delete dead ones
      for(int n=0;n<deletes.size();n++) { delete(deletes.get(n)); }
      // and export the state
      Hashtable h = new Hashtable();
      i=gossip.keySet().iterator();
      while(i.hasNext()) { 
        Object gid = i.next();
        Tuple t = (Tuple)gossip.get(gid);
        h.put(gid,new Tuple(t.first(),((Gossip)t.second()).clone(),t.third()));
      }
      ns.setData(gtype, new Pair(h,status.clone()));
    }
    return changed;
  }

  public String toString() { 
    String s = "Gossip Comm: "+gossip+" Status:"+status;
    return s;
  }

  Hashtable gossip = new Hashtable();
  Hashtable types = new Hashtable();
  Hashtable status = new Hashtable();
  public boolean contains(Object gid) { return gossip.containsKey(gid); }
  public Set getGossip() { return gossip.keySet(); }
  public Gossip getGossip(Object gid) {
    Tuple t = (Tuple)gossip.get(gid);
    if(t==null) return null; else return (Gossip)t.second(); 
  }
  public Set getType(Object type)
  { return (Set)types.get(type); }
  public Object typeOf(Object gid) {
    Tuple t = (Tuple)gossip.get(gid);
    if(t==null) return null; else return t.first(); 
  }
  public void addGossip(Object gid, Object type, Gossip g)
  { addGossip(gid,type,g,RegionManager.rootID); }
  public void addGossip(Object gid, Object type, Gossip g, Object region) {
    status.put(gid,new Tuple(LIVE));
    if(contains(gid)) return; // don't duplicate
    gossip.put(gid,new Tuple(type,g,region));
    Set s = (Set)types.get(type);
    if(s==null) { s=new HashSet(); types.put(type,s); }
    s.add(gid);
  }
  // for adding gossip learned about via gossip
  private void addGossip(Object gid, Tuple gd) {
    addGossip(gid,gd.first(),(Gossip)((Gossip)gd.second()).clone(),gd.third());
  }
  public void removeGossip(Object gid) {
    Tuple t = (Tuple)status.get(gid);
    if(t!=null && t.first()==LIVE) kill(gid);
  }
  private void kill(Object gid) {
    status.put(gid,new Tuple(DYING, new Counter(timeout)));
  }
  private void delete(Object gid) {
    Set typeset = getType(typeOf(gid));
    typeset.remove(gid);
    if(typeset.isEmpty()) types.remove(typeOf(gid));
    gossip.remove(gid);
    status.remove(gid);
  }
  
  static final Symbol LIVE = Symbol.GetSymbol("LIVE");
  static final Symbol DYING = Symbol.GetSymbol("DYING");
  static final Symbol DEAD = Symbol.GetSymbol("DEAD");
  static final int timeout = 3; // time to transition between stages
}


