package modules.partition;

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

/*
  A partition object takes a region and partitions it into a set
  of subregions
*/

public class WeightedPartition implements ProcessorModule {
  PortedProcessor pp; NeighborSet ns; Broadcast bc;

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

  public void init() {
    root=new Region(null,rootID,rootID,new UnlimitedRegion());
    regions.put(rootID,root);
    bc.openBroadcastPort(this,type);
  }

  static final Symbol type = Symbol.GetSymbol("Regions");
  public static final Symbol rootID = Symbol.GetSymbol("root");

  public boolean signalEvent(Symbol name,Object data) {
    boolean changed = false;
    if(name==PortedProcessor.kClock) {
      integrateNeighbors();
      pruneRegions();
      exportData();
    } else if(name==type) { // kill bcast
      RegionKill rk = (RegionKill)data;
      Region r = (Region)regions.get(rk.id);
      if(r!=null) r.dead=true; // mark as dead
    } else if(name==PortedProcessor.kTransmit) { // ack of a kill bcast
      Object id = ((Pair)((Pair)data).cdr()).cdr();
      deleteRegion(id); // actually delete
    }
    return changed;
  }

  private void integrateNeighbors() {
    Iterator i = ns.getNeighbors();
    while(i.hasNext()) {
      Hashtable nr = (Hashtable)ns.getData(type,i.next());
      if(nr==null) continue;
      // walk the neighbor's tree, adding to our own where possible
      Queue q = new Queue(); 
      if(regions.containsKey(rootID)) q.push(rootID);
      while(!q.isEmpty()) {
        Region r = (Region)nr.get(q.pop());
        boolean walk=false;
        if(regions.containsKey(r.id)) { // if already got it, get its children
          if(r.dead) ((Region)regions.get(r.id)).dead=true;
          walk=true;
        } else if(!r.dead && r.rmt.isMember(pp)) { // else see if we want it...
          addRegion(r.id,r.type,r.parent,r.rmt);
          walk=true;
        }
        if(walk) { // recurse on children
          Iterator j = r.children.iterator();
          while(j.hasNext()) { q.push(j.next()); }
        }
      }
    }
  }
  // delete all regions for which isMember is false
  private void pruneRegions() {
    Queue q = new Queue();
    Iterator i = regions.keySet().iterator();
    while(i.hasNext()) {
      Region r = (Region)regions.get(i.next());
      if(!r.rmt.isMember(pp)) q.push(r.id); // enqueue deadbeats
    }
    while(!q.isEmpty()) { deleteRegion(q.pop()); } // now delete them
  }
  // garbage collector:
  private void exportData() {
    Hashtable rcopy = new Hashtable(regions.size());
    Iterator i = regions.values().iterator();
    while(i.hasNext()) { 
      Region r = (Region)i.next();
      rcopy.put(r.id,r.commit());
    }
    ns.setData(type,rcopy);
  }
  private void gc() {
    HashSet gct = new HashSet(regions.keySet());
    // phase 1: walk the region tree, unmarking live members
    Queue q=new Queue(); gct.remove(root); q.push(root);
    while(!q.isEmpty()) {
      Region r = (Region)q.pop();
      Iterator i = r.children.iterator();
      while(i.hasNext()) {
        Region c = (Region)regions.get(i.next());
        if(c==null) continue;
        if(gct.contains(c.id)) {
          gct.remove(c.id); q.push(c);
        }
      }
    }
    // phase 2: prune all still-marked regions
    Iterator i=gct.iterator();
    while(i.hasNext()) {
      deleteRegion(i.next());
    }
  }
  // deletes and unlinks a region
  private void deleteRegion(Object id) {
    Region r = (Region)regions.get(id);
    if(r==null) return;
    Region p = (Region)regions.get(r.parent);
    if(p!=null) p.children.remove(id); // unlink
    regions.remove(r.id);
    // finally, delete all the children
    Iterator i = r.children.iterator();
    while(i.hasNext()) { deleteRegion(i.next()); }
  }

  public String toString() { 
    return "RegionManager: ("+regions.toString()+")"; 
  }


  // public interface
  Region root;
  Hashtable regions = new Hashtable(2);

  public boolean inRegion(Object id) { return regions.containsKey(id); }
  public void addRegion(Object id, Object type, RegionMembershipTest rmt)
  { addRegion(id,type,rootID,rmt); } 
  public void addRegion(Object id, Object type, Object parent,
                        RegionMembershipTest rmt) { 
    if(regions.containsKey(id) || !regions.containsKey(parent)) return;
    Region r = new Region((Region)regions.get(parent),id,type,rmt);
    regions.put(id,r);
  }
  // killing is much more serious than adding, and requires a convergecast
  public void killRegion(Object id) {
    if(!regions.containsKey(id)) return;
    Region r = (Region)regions.get(id);
    r.dead=true;
    bc.convergecast(type,new RegionKill(id),id);
  }
  public int numRegions() { return regions.size(); }
  public Set getRegions() { return regions.keySet(); }
}

class RegionKill implements ConvergeCast {
  Object id;
  RegionKill(Object id) { this.id=id; }

  public Object deliverLeaf(PortedProcessor pp) { return id; }
  public Object deliverBranch(PortedProcessor pp) { return id; }
  public Object deliverRoot(PortedProcessor pp) { return id; }

  public Object getLeafValue(PortedProcessor pp) { return null; }
  public void putBranchValue(PortedProcessor pp, Object input) {}
  public Object getBranchValue(PortedProcessor pp) { return null; }
  public void putRootValue(PortedProcessor pp, Object input) {}

  //static final Symbol fail = Symbol.GetSymbol("RegionKill failure");
  public Object fail(PortedProcessor pp) { return id; }

  public ConvergeCast getChild() { return new RegionKill(id); }
}

/*
  A weighted-region has a class of competitors.
  Processors on the border compare the pressure of the WR they're
  part of with that of its competitors, and join the highest pressure
  WR.
*/

/*
class WeightCompetition implements RegionMembershipTest {
  public boolean isMember(PortedProcessor pp) { return false; }
}
*/

//Tomorrow:
//1. broadcast gets regionalized
//2. test region marking
