package modules.broadcast;

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

/* 
   The difficulty that needs to be addressed in making a broadcast
   work is cleanup. In order to provide for message termination, the
   broadcast must maintain a space-filling structure until the message
   is acknowledged. When a processor acks, it must not propagate the
   message any more.

   There are 7 states:
   send: processor is spreading the message
   build: processor is building multi-tree
   recv_ack: a reply has been send to the processor
   send_ack: processor has sent a reply
   gc-ack: processor is deleting on command of sender
   gc-noack: processor is deleting q. process failure [3x slow, send inhibits]
   dead: it's deletion time
   
   The real problem with a broadcast is the expense:
   for a correctly operating broadcast, a bfs tree must be maintained for
   O(d) rounds, costing O(nd) messages.

   The broadcast system makes a "best effort" delivery: failures
   may cut off part or all of the network. It is also possible for
   a message to get delivered multiple times, if certain failure
   patterns occur.
*/

/**
   TODO: convergecast, region-limited broadcast
*/

public class Broadcast implements ProcessorModule {
  PortedProcessor pp; RegionManager rm;

  public Symbol getName() { return Symbol.GetSymbol("Broadcast"); }
  public void link(PortedProcessor p) {
    pp = p;
    rm = (RegionManager)pp.getModule("RegionManager");
    pp.addClockListener(this);
    pp.openNetworkPort(this,port);
  }

  static final Symbol port = Symbol.GetSymbol("broadcast");
  public static final Symbol ack = Symbol.GetSymbol("ack");
  
  public void init() {}
  
  int period=2;
  public boolean signalEvent(Symbol name,Object data) {
    boolean changed = false;
    if(name==PortedProcessor.kClock) { // run updates
      int t = ((Integer)data).intValue();
      if(t%period==0) {
        Vector deliveries = new Vector();
        Iterator i=broadcasts.keySet().iterator();
        while(i.hasNext()) {
          Object k = i.next();
          BCExecution bce = (BCExecution)broadcasts.get(k);
          bce.update();
          if(bce.delivervalue!=null && !acks.contains(k)) { 
            deliveries.add(new Pair(bce.port,new Pair(k,bce.delivervalue)));
            acks.add(k);
          }
          if(bce.state==BCExecution.DEAD) {
            i.remove(); acks.remove(k);
          }
        }
        // make internal delivery of messages
        i=deliveries.iterator();
        while(i.hasNext()) portSend(PortedProcessor.kTransmit,i.next());
        // export our state
        if(!broadcasts.isEmpty())
          pp.sendMessage(port,broadcasts);
      }
    } else if(name==port) { // integrate neighbor information
      Hashtable nbr = (Hashtable)data;
      Iterator i = nbr.keySet().iterator();
      while(i.hasNext()) {
        Object msgid = i.next();
        BCExecution nbrbce = (BCExecution)nbr.get(msgid);
        BCExecution bce = (BCExecution)broadcasts.get(msgid);
        if(bce==null && nbrbce.state==BCExecution.SEND && 
           rm.inRegion(nbrbce.region)) {
          bce = new BCExecution(pp,nbrbce.port,nbrbce.message,
                                new Double(pp.getUID()),nbrbce.region,
                                nbrbce.hop+1);
          broadcasts.put(msgid,bce);
          portSend(bce.port,nbrbce.message);
        }
        if(bce!=null)
          bce.integrateNeighbor(nbrbce);
      }
    }
    return changed;
  }

  private void portSend(Symbol port,Object msg) {
    Symbol modulename = null;
    if(port==PortedProcessor.kTransmit && msg instanceof Pair)
      modulename = (Symbol)ports.get(((Pair)msg).car());
    else
      modulename = (Symbol)ports.get(port);
    if(modulename==null) return;
    ProcessorModule pm = pp.getModule(modulename);
    if(pm==null) return;
    pm.signalEvent(port,msg);
  }

  public String toString() { 
    String s = "Broadcast System: ports="+ports;
    s+= " broadcasts="+broadcasts;
    return s;
  }

  // KLUDGE FOR DEBUGGING!
  public int debugState() { /* @deprecated */
    BCExecution a=null;
    Iterator i = broadcasts.values().iterator();
    if(i.hasNext()) a=(BCExecution)i.next();
    if(a==null) return -1; else return a.state;
  }

  // port interface: just like PortedProcessor
  // communicates via events of type=port and data=message
  HashSet acks = new HashSet(2);
  Hashtable ports = new Hashtable(2);
  Hashtable broadcasts = new Hashtable(2);
  public void openBroadcastPort(ProcessorModule m, Symbol port) {
    ports.put(port,m.getName());
  }
  // sends an ack (possibly with an answer) when it completes
  // returns the message UID so the ack can be identified
  public Object broadcast(Symbol port, Object message)
  { return broadcast(port,message,RegionManager.rootID); }
  public Object broadcast(Symbol port, Object message, Object region) {
    if(ports.get(port)==null) return null; // don't send to non-existent ports
    BCExecution bce = new BCExecution(pp,port,message,
                                      new Double(pp.getUID()),region);
    Object key = new Double(Math.random());
    broadcasts.put(key,bce);
    return key;
  }
  // convergecasts are data-gathering queries
  // returns the message UID so the ack can be identified
  public Object convergecast(Symbol port, ConvergeCast message) 
  { return convergecast(port,message,RegionManager.rootID); }
  public Object convergecast(Symbol port, ConvergeCast message, Object region){
    if(ports.get(port)==null) return null; // don't send to non-existent ports
    BCExecution bce = new BCExecution(pp,port,message,
                                      new Double(pp.getUID()),region);
    Object key = new Double(Math.random());
    broadcasts.put(key,bce);
    return key;
  }
}


// a execution of the broadcast process
// Originally, the messageID lived in this, but that's not actually
// necessary: it can live in the time in the broadcast module
class BCExecution {
  public static final int SEND = 0;
  public static final int BUILD = 1;
  public static final int RECV_ACK = 2;
  public static final int SEND_ACK = 3;
  public static final int GC_ACK = 4;
  public static final int GC_NOACK = 5;
  public static final int DEAD = 6;
  static final int timeout = 4;
  static final int decaytimeout = 6;
  static final int fasttimeout = 3;

  PortedProcessor pp; RegionManager rm;
  Hashtable parents=new Hashtable(2);
  Hashtable children=new Hashtable(2);
  Object uid; // UID for this processor
  Object region; // the region the bcast is confined to
  Object message; // ack behaves differently if it's a convergecast
  Symbol port; // destination of the message
  int state = SEND;
  int hop;
  Counter timer = new Counter(timeout);
  Object ackparent=null; Object ackvalue=null; Object delivervalue=null;
  // used by convergecast
  
  public BCExecution(PortedProcessor pp, Symbol port, Object msg, Object uid,
                     Object region) 
  { init(pp,port,msg,uid,region,0); }
  public BCExecution(PortedProcessor pp, Symbol port, Object msg, 
                     Object uid, Object region, int h) 
  { init(pp,port,msg,uid,region,h); }
  
  private void init(PortedProcessor pp, Symbol port, Object msg, 
                    Object uid, Object region, int h) {
    this.pp=pp; this.uid=uid; hop=h; this.port=port; this.region=region;
    rm = (RegionManager)pp.getModule("RegionManager");
    if(msg instanceof ConvergeCast) 
      message=((ConvergeCast)msg).getChild();
    else message=msg;
  }

  public String toString() {
    return "("+message+", "+port+", "+state+", "+hop+", "+timer+", "+
      ackparent+", "+parents+", "+children+")";
  }

  private static Random r = new Random(); // random used by everybody
  public void update() {
    timer.tick();
    ageTree();
    switch(state) {
    case SEND: // expired timer means stop sending
      if(timer.expired()) { timer.reset(); state=BUILD; }
      break;
    case BUILD: // expired timer means leaf, go to SEND_ACK
      if(timer.expired()) { sendAck(true); }
    case SEND_ACK:
      if(hop==0 && state==SEND_ACK)
        { state=GC_ACK; timer.setAlarm(fasttimeout); timer.reset(); } 
    case RECV_ACK: // no parents means disruption: go to GC_NOACK
      if(hop!=0 && parents.isEmpty()) { 
        timer.reset(); state=GC_NOACK; 
        if(message instanceof ConvergeCast)
          delivervalue = ((ConvergeCast)message).fail(pp);
      }
      // if no children left, return ack
      if(state==RECV_ACK && children.isEmpty()) { sendAck(false); }
      break;
    case GC_NOACK: // expired timer means we should delete
    case GC_ACK:
      if(timer.expired()) { state=DEAD; }
      break;
    }
    // finally, check if we've fallen from the region: if so, we're dead
    if(!rm.inRegion(region)) state=DEAD;
  }
  public void integrateNeighbor(BCExecution nbrex) {
    improveHopcount(nbrex); // improve hopcount
    maintainParents(nbrex); // maintain tree up
    maintainChildren(nbrex); // maintain tree down
    collectAcks(nbrex); // collect acks from down the tree
    runGC(nbrex); // GC fast on ack
    inhibitGC(nbrex); // don't GC with senders
  }

  private void improveHopcount(BCExecution nbr) {
    if(!(state==SEND || state==BUILD)) return; // construction states only
    if(nbr.state==SEND && nbr.hop<(hop-1)) {
      hop=nbr.hop+1; timer.reset(); state=SEND;
      parents=new Hashtable(2); parents.put(nbr.uid,new Counter(decaytimeout));
      children=new Hashtable(2);
    }
  }
  private void maintainParents(BCExecution nbr) {
    if(!(state==SEND || state==BUILD || state==RECV_ACK || state==SEND_ACK))
      return; // GC & dead don't care
    if((nbr.state==SEND || nbr.state==BUILD || nbr.state==RECV_ACK || 
        nbr.state==SEND_ACK)
       && nbr.hop<hop) {
      Counter c = (Counter)parents.get(nbr.uid);
      if(c==null) parents.put(nbr.uid,new Counter(decaytimeout));
      else c.reset();
    }
  }
  private void maintainChildren(BCExecution nbr) {
    if(!(state==SEND || state==BUILD || state==RECV_ACK)) 
      return; // construction & wait states only
    if((nbr.state==SEND || nbr.state==BUILD || nbr.state==RECV_ACK) 
       && nbr.hop==(hop+1)) {
      if(state==BUILD) timer.reset(); // not a leaf
      Counter c = (Counter)children.get(nbr.uid);
      if(c==null)
        children.put(nbr.uid,new Counter(decaytimeout));
      else c.reset();
    }
  }
  private void collectAcks(BCExecution nbr) {
    if(!(state==BUILD || state==RECV_ACK)) return; // ack collectors only
    if(nbr.state==SEND_ACK && 
       (children.containsKey(nbr.uid) && uid.equals(nbr.ackparent))) {
      if(uid.equals(nbr.ackparent)) { // if we're the ones, convergecast
        if(message instanceof ConvergeCast) {
          if(hop==0) ((ConvergeCast)message).putRootValue(pp,nbr.ackvalue);
          else ((ConvergeCast)message).putBranchValue(pp,nbr.ackvalue);
        }
      }
      children.remove(nbr.uid);
      state=RECV_ACK;
    }
  }
  private void runGC(BCExecution nbr) {
    if(nbr.state==GC_ACK && state!=GC_ACK) { // spreads unidirectionally
      state=GC_ACK; timer.setAlarm(fasttimeout); timer.reset();
    }
  }
  // prevents sends from flowing into newly deleted spaces
  private void inhibitGC(BCExecution nbr) {
    if(!(state==GC_NOACK || state==GC_ACK)) return; // GC states only
    if(nbr.state==SEND) timer.reset(); // don't GC while there's a sender
  }


  private void sendAck(boolean leaf) {
    if(hop==0) { // case: root
      if(message instanceof ConvergeCast) 
        delivervalue=((ConvergeCast)message).deliverRoot(pp);
      if(delivervalue==null) delivervalue=Broadcast.ack;
    } else if(leaf) {
      if(message instanceof ConvergeCast) {
        ackvalue = ((ConvergeCast)message).getLeafValue(pp);
        delivervalue = ((ConvergeCast)message).deliverLeaf(pp);
      }
    } else {
      if(message instanceof ConvergeCast) {
        ackvalue = ((ConvergeCast)message).getBranchValue(pp);
        delivervalue = ((ConvergeCast)message).deliverBranch(pp);
      }
    }
    timer.reset(); state=SEND_ACK; 
    Iterator i = parents.keySet().iterator();
    if(i.hasNext()) ackparent=i.next(); // choose arbitrary parent for ack
  }
  private void ageTree() {
    Iterator i = parents.values().iterator();
    while(i.hasNext()) { 
      Counter c = (Counter)i.next();
      c.tick(); if(c.expired()) i.remove();
    }
    i = children.values().iterator();
    while(i.hasNext()) { 
      Counter c = (Counter)i.next();
      c.tick(); if(c.expired()) i.remove();
    }
  }
}

