package testprocs;

import simulator.*;
import scripting.*;
import modules.color.*;
import modules.hierarchy.*;
import modules.neighborset.*;
import modules.gradient.*;
import modules.node.*;
import modules.consensus.*;
import utils.*;
import java.awt.*;
import java.util.*;
import java.io.*;



public class RNAtomExp implements ProcessorModule, Observer {
  static final String script = "modules.color.ColorModule\nmodules.gradient.ActiveGradient\nmodules.neighborset.NeighborSet\nmodules.consensus.Consensus\nmodules.broadcast.Broadcast\nmodules.region.RegionManager\nmodules.node.RamboNode\ntestprocs.RNAtomExp";

  
  public static double killrate = 0.001; //prob. of killing particle each round
  public static int maxrunlength = 100; // length to run sim
  public static int nodesize = 3; // size of the test node
  public static double oprate = 0.005; // frequency of operation initiation
  public static double readbias = 0.5; // probability op is a read
  public static String filestem = "rnatomtest";

  public static void main (String argv[]) {

    for(int i=0;i<argv.length;i++) {
      if(argv[i].equals("-k")) 
        { i++; killrate=(new Double(argv[i])).doubleValue(); }
      else if(argv[i].equals("-l")) 
        { i++; maxrunlength=(new Integer(argv[i])).intValue(); }
      else if(argv[i].equals("-n")) 
        { i++; nodesize=(new Integer(argv[i])).intValue(); }
      else if(argv[i].equals("-r")) 
        { i++; oprate=(new Double(argv[i])).doubleValue(); }
      else if(argv[i].equals("-b")) 
        { i++; readbias=(new Double(argv[i])).doubleValue(); }
      else if(argv[i].equals("-f"))
        { i++; filestem=argv[i]; }
    }

    try {
      out = new PrintWriter(new FileWriter(filestem+".out"),true);
      out.println("Atomicity log for RN");
      outdb = new PrintWriter(new FileWriter(filestem+".db"),true);
      rawdb = new PrintWriter(new FileWriter(filestem+".raw"),true);
      //outdb.println("Operations Database for RN");
    } catch(Exception e) {
      System.out.println("failure: "+e);
    }

    out.println("Parameters:");
    out.println("Kill Rate = "+killrate);
    out.println("Run Length = "+maxrunlength);
    out.println("Node Size = "+nodesize);
    out.println("Op Rate = "+oprate);
    out.println("Read Bias = "+readbias);
    
    CommunicationsModel comm = new PerfectCommunicationsModel();
    ProcessorFactory pf = new ScriptProcFactory(script);
    sim = new Simulator(comm,pf,2000,0.04);

    sim.addObserver(new RNAtomExpCritic(sim));
        
    // and... run!
    sim.setPaused(false);
    startTime=System.currentTimeMillis();
    sim.run();

    System.out.println("All Systems Go");
  }
  static Simulator sim;

  // this attempts to produce an atomic trace of operations, then ends
  static long startTime=0;
  public static void endexperiment() {
    sim.setPaused(true);

    // add in some notes on event-processing rates:
    int n_evt = sim.numEvents();
    double t_elapsed = System.currentTimeMillis()-startTime;
    System.out.println
      ("\nSimulation processed "+n_evt+
       " events over the course of "+t_elapsed/1000+
       " seconds at a rate of "+(1000.*n_evt/t_elapsed)+" events per second.");

    // first, dump the raw op data 
    System.out.println("Dumping raw data");
    rawdb.println("RW, start, end, serial, tag_v, tag_q, value, proc, duration, success");
    Iterator i=opList.keySet().iterator();
    while(i.hasNext()) {
      Object key = i.next(); 
      Operation op = (Operation)opList.get(key);
      String s = "";
      if(op.isRead) s+="1, "; else s+="0, ";
      s += op.startTime+", "+op.endTime+", "+op.serializedTime+", ";
      if(op.versioning!=null)
        s += op.versioning.getVersion()+", "+op.versioning.getQuality()+", ";
      else s+=", , ";
      if(op.value!=null) s += op.value+", "; else s+= "null, ";
      s += op.procid+", ";
      s += op.endTime-op.startTime+", ";
      if(op.successful) s+="1"; else s+="0";
      rawdb.println(s);
    }


    System.out.println("Completing unfinished operations");
    // collect all pending ops
    int counter=0;
    i=sim.getNet().getNodeSet().iterator();
    while(i.hasNext()) {
      PortedProcessor p = (PortedProcessor)i.next();
      RNAtomExp mm = (RNAtomExp)p.getModule("RNAtomExp");
      RamboNode rn = (RamboNode)p.getModule("RamboNode");
      if(mm.curOp!=null) {
        mm.curOp.versioning=rn.debugit2(nid,mm.curOp.isRead);
        if(mm.curOp.versioning==null) {
          mm.curOp.versioning=(VersionedData)((VersionedData)((Tuple)opList.lastKey()).first()).clone();
          mm.curOp.versioning.incrementVersion();
          mm.curOp.versioning.setQuality(-1);
        }
        opList.put(mm.poRank(mm.curOp),mm.curOp);
        counter++;
      }
    }
    System.out.println(counter+" unfinished operations collected.");

    System.out.print("\nAnalyzing "+opList.size()+" operations");

    boolean good=true;
    // walk the list, attempting to assign serialized times
    i=opList.keySet().iterator();
    double timeFinger = 0; double epsilon=1e-8;
    int count=0;
    while(i.hasNext()) {
      count++; if(count%100==1) System.out.print(".");
      //System.out.println("a");
      // weed out failed reads and writes
      // the serialization must be between start and end times
      Object key = i.next(); 
      Operation op = (Operation)opList.get(key);
      if(op==null) System.out.println(key+" "+opList);
      boolean usable=op.successful;
      if(!usable && !op.isRead && i.hasNext()) { //write maybe usable anyway...
        //System.out.println("Dead Write: "+op.startTime);
        SortedMap tail = opList.tailMap(key);
        Iterator j = tail.keySet().iterator(); j.next();
        Tuple nextkey = (Tuple)j.next(); // get key following this one
        if(nextkey==null) 
          System.out.println("Op unusable because no next key");
        if(nextkey!=null && nextkey.first()!=null && 
           ((Tuple)key).first()!=null &&
           nextkey.first().equals(((Tuple)key).first())) usable=true;
      }
      if(usable) {
        op.serializedTime = Math.max(op.startTime,timeFinger+epsilon);
        timeFinger=op.serializedTime;
        sortList.put(new Double(op.serializedTime),op);
      }
    }
    // certify that all reads return the preceding write & all serialization 
    // is between start and end
    i=sortList.keySet().iterator();
    Object value=null;
    while(i.hasNext()) {
      Double stime = (Double)i.next();
      Operation op = (Operation)sortList.get(stime);
      if(op.isRead) {
        if(!((value==null && op.value==null) || value.equals(op.value))) {
          good=false;
          out.println("FAIL: Read/Write mismatch at "+op.serializedTime);
        }
      } else {
        value=op.value;
      }
      if(op.serializedTime<op.startTime || op.serializedTime>op.endTime) {
        good=false;
        out.println("FAIL: Out of bounds at "+op.serializedTime);
      }
    }

    out.println("\n\n");
    out.println("Number of operations: "+opList.size());
    // announce whether it succeeded
    if(good) out.println("SUCCESS: Sequence is atomic.");
    else out.println("FAILURE: Sequence is not atomic.");
    
    // print the serialized event list
    out.println("Serialized\tOpType\tValue\tStart\tEnd");
    i=sortList.keySet().iterator();
    while(i.hasNext()) {
      Double stime = (Double)i.next();
      Operation op = (Operation)sortList.get(stime);
      String ops="Write"; if(op.isRead) ops="Read";
      out.println(stime+"\t"+ops+"\t"+op.value+"\t"+op.startTime+
                  "\t"+op.endTime);
    }
    

    // last of all, print the operations database for post-run analysis
    outdb.println("RW, start, end, serial, tag_v, tag_q, value, proc, duration, success");
    i=opList.keySet().iterator();
    while(i.hasNext()) {
      Object key = i.next(); 
      Operation op = (Operation)opList.get(key);
      String s = "";
      if(op.isRead) s+="1, "; else s+="0, ";
      s += op.startTime+", "+op.endTime+", "+op.serializedTime+", ";
      if(op.versioning!=null)
        s += op.versioning.getVersion()+", "+op.versioning.getQuality()+", ";
      else s+=", , ";
      s += op.value+", ";
      s += op.procid+", ";
      s += op.endTime-op.startTime+", ";
      if(op.successful) s+="1"; else s+="0";
      outdb.println(s);
    }

    System.exit(0);
  }
  static TreeMap opList = new TreeMap();
  static TreeMap sortList = new TreeMap();

  // records for the operations carried out by the PNs
  class Operation {
    double procid;
    boolean isRead;
    double startTime;
    double endTime=Double.NaN;
    double serializedTime=Double.NaN;
    VersionedData versioning=null;
    Object value=null;
    boolean successful=false;
    Operation(boolean r,double s) { isRead=r;startTime=s; procid=pp.getUID();}
    public String toString() {
      String vname="null"; if(value!=null) vname=value.toString();
      return "("+isRead+", "+startTime+", "+endTime+", "+serializedTime+", "+
        versioning+", "+vname+", "+successful+")";
    }
  }

  public Symbol getName() { return Symbol.GetSymbol("RNAtomExp"); }
  public void link(PortedProcessor p) {
    pp = p;
    pp.addClockListener(this);
    rn = (RamboNode)p.getModule("RamboNode");
    cs = (Consensus)p.getModule("Consensus");
  }

  PortedProcessor pp; RamboNode rn; Consensus cs;
  static Symbol nid = Symbol.GetSymbol("test");

  static double initiator = Double.NaN;
  // create the node
  public void init() {
    //System.out.print("i");
    if(Double.isNaN(initiator)) {
      initiator=pp.getUID(); System.out.println("Starting...");
      out.println("Time\tUID\tEvent");
      rn.initiateNode(nid,nodesize);
    }
  }
  static PrintWriter out;
  static PrintWriter outdb;
  static PrintWriter rawdb;

  
  public void update(Observable o, Object arg) {
    if(curOp.isRead) {
      curOp.value=arg;
      out.println(sim.getTime()+"\t"+pp.getUID()+"\t end read "+
                  curOp.value);
    } else {
      out.println(sim.getTime()+"\t"+pp.getUID()+"\t end write");
    }
    curOp.versioning=rn.debugit2(nid,curOp.isRead);
    curOp.successful=true;
    curOp.endTime = sim.getTime();
    if(curOp.endTime-curOp.startTime>100)
      System.out.println("Long op! "+pp.getUID());
    if(true) { // online data dump 
      String s = ""; Operation op=curOp;
      if(op.isRead) s+="1, "; else s+="0, ";
      s += op.startTime+", "+op.endTime+", "+op.serializedTime+", ";
      if(op.versioning!=null)
        s += op.versioning.getVersion()+", "+op.versioning.getQuality()+", ";
      else s+=", , ";
      if(op.value!=null) s += op.value+", "; else s+= "null, ";
      s += op.procid+", ";
      s += op.endTime-op.startTime+", ";
      if(op.successful) s+="1"; else s+="0";
      rawdb.println(s);
    }

    opList.put(poRank(curOp),curOp);
    curOp=null;
  }

  static int topcs = 0;
  Object CID() { return new Tuple(nid,Symbol.GetSymbol("RN:CS")); }
  // tasks for this process:
  // 1. occasionally initiate operations
  // 2. if an op is completed, log the result
  Operation curOp = null; boolean dead=false;
  synchronized public boolean signalEvent(Symbol name,Object data) {
    if(dead) return false;
    //System.out.print("u");
    boolean dirty = false;
    if(name==PortedProcessor.kClock) {
      int t = ((Integer)data).intValue();

      VersionedData vd = cs.getConsensus(CID());
      if(vd!=null && vd.getVersion()>topcs) {
        topcs=vd.getVersion();
        out.println(sim.getTime()+"\tConsensus Update\t"+topcs);
        rawdb.println(sim.getTime()+"\tConsensus Update\t"+topcs+"\t"+((Map)vd.get()).size());
      }

      //if(pp.getUID()==initiator) System.out.print("i");
      //if(rn.inNode(nid)) System.out.print("a");
      if(curOp==null) {
        if(rn.inNode(nid) && Math.random()<oprate) {
          boolean read = (Math.random()<readbias);
          curOp = new Operation(read,sim.getTime());
          if(read) {
            DelayedReturn dr = new DelayedReturn.Object();
            dr.addObserver(this);
            rn.getData(nid,dr);
            out.println(sim.getTime()+"\t"+pp.getUID()+"\t start read");
            //if(curOp.value==null) System.out.println("Null read: "+curOp+"\n"+pp);
          } else { 
            curOp.value=new Double(Math.random()); 
            DelayedReturn dr = new DelayedReturn.Object();
            dr.addObserver(this);
            rn.setData(nid,curOp.value,dr);
            out.println(sim.getTime()+"\t"+pp.getUID()+"\t start write "+
                        curOp.value);
          }
        }
      }
      //if(curOp!=null && !rn.inNode(nid) && !tick) {
      //  tick=true;
      //  System.out.println("going long... "+sim.getTime()+"\t"+pp.getUID());
      //}
    }
    return dirty;
  }
  //boolean tick=false;
  Tuple poRank(Operation op) {
    int RW = 0; if(op.isRead) RW=1;
    return new Tuple(op.versioning,new Integer(RW),new Double(op.startTime));
  }
  
  
  // when a proc is killed, this is notified to let it terminate its op
  synchronized public void kill() {
    dead=true;
    if(curOp!=null) { 
      curOp.successful=false;
      curOp.endTime = sim.getTime();
      if(curOp.isRead) {
        out.println(sim.getTime()+"\t"+pp.getUID()+"\t kill read");
      } else {
        out.println(sim.getTime()+"\t"+pp.getUID()+"\t kill write");
      }
      opList.put(poRank(curOp),curOp);
    }
  }
}

// this has two tasks:
// 1. reboot particles which are not in the middle of an operation
// 2. count number of reboots performed
// 3. terminate the sim when it's run long enough
class RNAtomExpCritic implements Observer {
  
  int kills = 0;
  Simulator sim;

  public RNAtomExpCritic(Simulator sim) {
    this.sim = sim;
  }
  
  // the standard update
  static Random rnd = new Random();
  double lastT=-100;
  public void update(Observable o, Object arg) {
    double t = sim.getTime();
    
    if(t-lastT>1) { lastT=t; System.out.print("."); }
    if(t>=RNAtomExp.maxrunlength) {
      RNAtomExp.endexperiment(); // terminate sim
    } else { // kill, as needed
      int killtarget = (int)(t*2000*RNAtomExp.killrate);
      while(kills<killtarget) {
        //System.out.print("k");
        kills++;
        Vector procs = new Vector(sim.getNet().getNodeSet());
        if(procs.size()!=2000) 
          System.out.println("Bad processor list size: "+procs.size());
        PortedProcessor p = (PortedProcessor)procs.get(rnd.nextInt(2000));
        RNAtomExp ae = (RNAtomExp)p.getModule("RNAtomExp");
        ae.kill(); // notify experiment module that it's being killed
        sim.resetProcessor(p);
      }
    }
    
  }
}



