import java.io.*;
import java.util.*;
import java.lang.*;
import java.util.concurrent.PriorityBlockingQueue;


/** This class represents the physical processors involved in the time warp 
 * simulation engine.
  * @author Yin Xiong
  * @version 1.7 (07/2006)
  */
public class PE implements Runnable{
  
  
  /**The id of this PE.*/
  int id;

  /**Flag for end of sim*/
  Flag eosFlag;
  
  /**The most recent GVT*/
  int GVT;

  /**Sleep time*/
  int sleepTime;

  /**The message pool where all messages are stored*/
  MessagePool msgPool;
  
  /**The LocalNameServer in charge of routing messages.*/
  LocalNameServer localNameServer;
  
  /**The GlobalNameServer*/
  GlobalNameServer gns;
  

  /**The queue that holds messages that have been cancelled. When a processor
    * wishes to cancel a message, it enqueues the message being cancelled into
    * this queue. Logically, each message enqueued in this queue can be viewed
    * as an anti-message, however, it is the message itself rather than an
    * explicit anti-message that is enqueued. Access this queue is synchronized
    * with locks. 
    */
  PriorityBlockingQueue<Message> cancelQ;

  
 /**The queue that holds the outgoing messages. Messages are placed into this
    *by the Scheduler and the sender(main thread) will come here to get the
    *outgoing message and send it out. A copy is kept in this queue, but its sign
    *is changed from positive to negative.
    */
  PriorityBlockingQueue<Message> sendQ;
    
/**The queue that holds all unprocessed events for all LPs mapped to this PE.  
 * A single queue is used here in order to eliminates the need for a
 * separate "scheduling queue" to enumerate the executable LPs.
  */
  //Queue calendarQ;
  

  /**The queue that holds the processed and unprocessed events for LPs mapped
   * to this PE. Each processed event has pointers to messages scheduled by  
   * the computation associated with this event and pointer to state vector  
   * information to allow event computation to roll back.
   */
  PriorityBlockingQueue<Message> eventQ;
  
  /**The queue that holds the events whose memeory is free to reclaim.*/
  PriorityBlockingQueue<Message> freeQ;

  /**The scheduler that schedules events for this PE.*/
  SchedulingLoop schedulingLoop;
  

 /**set the sleep time*/
public void setSleepTime(int st)
{
   sleepTime=st;
}

  /**Constructs a new PE instance. Sets the id of the PE to peid.
    *@param peid int the id of the PE, i.e. the id of this process in MPI.comm.
	*/
  public PE(int peid, MessagePool msgp, Flag ef, Flag gf, int gc)
  {
     id=peid;
     GVT=0;
     msgPool=msgp;
     eosFlag=ef;
     localNameServer=new LocalNameServer(id);
     cancelQ=new PriorityBlockingQueue<Message>();
     eventQ=new PriorityBlockingQueue<Message>();
     sendQ=new PriorityBlockingQueue<Message>();
     schedulingLoop=new SchedulingLoop(id);
     schedulingLoop.setFlags(ef,gf);
     schedulingLoop.setGVTClock(gc);

  }//end constructor
  

/**Runs the PE(processing events).
   *This is the communication thread that send and receive messages.
   */
    
public void run()
   {
    System.out.println("PE " + id + " is running ...\n");
    initLPs();
           
    Thread scheduler=new Thread(schedulingLoop);
    scheduler.start();
    
    boolean endOfSim=false;
  
  while(!endOfSim){
   try{
    
    //receive messages
    receiveMsg();

    //send messages stored in sednQ
    sendMessages();
   
    //process the queues
    processQs();  

    if(eosFlag.get()<0){
     //System.out.println("PE " + id + " gets flag -1");
       endOfSim=true;
      }    
        
    Thread thread=Thread.currentThread();
    thread.sleep(sleepTime);
  }
  catch(Exception e){
    e.printStackTrace();
  }
     
 }//end while

    finalizeLPs();
    
    finalizePE();
    
}//end run
    
void finalizePE()
{
    System.out.println("PE " + id + " finished running ");
  
}//end finalizePE

/**Receives messages sent to this PE.
  */
public void receiveMsg()
{
    Vector msgsForMe=msgPool.getMsgsForMe(id);
        for(int i=0;i<msgsForMe.size();i++){
            Message msg=(Message)msgsForMe.get(i);
    
       schedulingLoop.incoming.add(msg);
  }//end for
                                
}//end receiveMsg
   
/**Checks the sendQ and sends message if any.
  */
         
public void sendMessages()
{   
  //System.out.println("PE " + id + " has " + scheduler.sendQ.size() + " msgs to send");
  for(int i=0;i<sendQ.size();i++){
      Message msg=(Message)sendQ.poll();
      msgPool.addMsgToPool(msg);
  }//end for
    
}//end sendMessages
        


/**Processes the queues.
  */
       
void processQs()
{
    int cqsize,eqsize;
    cqsize=eqsize=0;
  
    //process events
    eqsize=schedulingLoop.outgoing.size();
    if(eqsize>0)
       processEvents(); 
   
    cqsize=cancelQ.size();
    if(cqsize>0)
      processCanQ();
    
}//end processQs

/**Processes events. It takes out each message in SchedulingLoop's outgoing
  *queue, finds out the routing information by application-specified id of the
  *destination or the LP id of the destination, consults the local name server
  *or the global name server (in case local name server does not have the info)
  *for the routing information. It then puts the message into the sendQ for
  *sending out.
  */
void processEvents()
{
  
  int size=schedulingLoop.outgoing.size();

//  System.out.println("PE " + id + "has " + size + " events to process ");         
   for(int i=0;i<size;i++){
      Message msg=(Message)schedulingLoop.outgoing.poll(); 
      if(msg!=null){
         String sname=msg.getSenderAPPid();
         String rname=msg.getReceiverAPPid();
        
         RoutingInfo rinfo=null;
         boolean byName=false;
         int lid=-1;
           
       if(rname!=null){
            //check lns for receiver lp
          //System.out.println("PE " + id + " send msg to APP "  +rname);
          rinfo=localNameServer.getRoutingInfoByAid(rname);
             byName=true;
           }
        else{
            lid=msg.getReceiverLP();
            //System.out.println("PE " + id + " send msg to LP "  + lid);
            rinfo=localNameServer.getRoutingInfoByLid(lid);
        }
 
       if(rinfo==null){ 
        //if lns doesn't have the receiver info, check gns
            if(byName==true)
                rinfo=gns.getRoutingInfoByAid(rname);
            else
                rinfo=gns.getRoutingInfoByLid(lid);
          
          //store rinfo in local name server for future use
          localNameServer.addRoutingInfo(rinfo);
       }//end if rinfo=null
         
       if(rinfo!=null){ 
          msg.setReceiverPE(rinfo.PEid);
          msg.setReceiverAPPid(rinfo.APPid);
          msg.setReceiverLP(rinfo.LPid);
            
         //if the receiver is not on this machine, put the msg in sendQ
         if(rinfo.PEid!=id){
           sendQ.add(msg);
         }//end if receiver is not on this machine
        
          else
             schedulingLoop.incoming.offer(msg);
      }//end if rinfo is not null
     }//end if msg!=null
  }//end for
}//end ProcessEvents

/**Places AntiMessages in CanQ.  These are events that have been scheduled by other events.
  *They are removed and deleted.
  */
public void processCanQ()
{
     //System.out.println("PE " + id + " is processing can queue...");
      
}//end ProcessCanQ
        

/**Create LPs in the specified subclass.
    *@param pid int the id of the PE where the LP resides.
    *@param lid int the LP id assigned by the system. 
    *@param appid String the application-assigned id of the instance.
    *@param appName String the application-assigned name of the instance.
    *@param className String the name of the instance.
    *@param dat String the configuration data provided by application.
  */

public void createLPs(int pid, int lid, 
				String appid, String appName,
				String className, String dat)
{
   schedulingLoop.createLP(pid,lid,appid,appName,className,dat);

}//end createLPs


/**Gets the LP by its id.
  *@param lpid int the LP id.
  */
public LogicalProcess getLPByID(int lpid)
{
//System.out.println("in getLPByID,size:" + LPs.size());

  for(int i=0;i<schedulingLoop.LPs.size();i++){
    LogicalProcess lp=(LogicalProcess)schedulingLoop.LPs.get(i);
    if(lp.LPid==lpid)
      return lp;
  }  
  return null;
}//end getLPByID  


/**Sets the global name server for this PE. 
  */
public void setGNS(GlobalNameServer g)
{
    gns=g;
}//end setGNS

/**Sets the local name server for this PE. 
  */
public void setLNS(LocalNameServer lns)
{
   localNameServer=lns;
}//end setLNS

/**Display the contents of the routing table.*/
public void dumpLNS()
{
   System.out.println("\nin PE: " + id + " dumping local routing table...");
   localNameServer.dump();

   //System.out.println("\nin PE: " + id + " dumping global routing table...");
   //gns.dump();

}//end dumpRoutingTable

/**Initialize LPs by invoking initializeLP method provided by subclasses 
  *of LogicalProcess.
  */
public void initLPs()
{


   for(int i=0;i<schedulingLoop.LPs.size();i++){
     LogicalProcess sublp=(LogicalProcess)schedulingLoop.LPs.get(i);
     sublp.initializeLP();
  }//end for     
      
}//end initLPs

/**Finalizes LPs by invoking finalizeLP method provided by subclasses of
  *LogicalProcess.
  */
public void finalizeLPs()
{

   for(int i=0;i<schedulingLoop.LPs.size();i++){
     LogicalProcess sublp=(LogicalProcess)schedulingLoop.LPs.get(i);
     sublp.finalizeLP();
  }//end for     
  
}//end finalizeLPs

/**Returns the routing table.
  */
public Vector getRoutingTable()
{
  return localNameServer.routingTable;

}//end getRoutingTable


  /*Displays LP's publishing and subscribing attributes.*/
  /*
  public void displayInterests()
  {
   System.out.println("\nPE: " + id);
    for(int i=0;i<LPs.size();i++){
	   LogicalProcess lp=(LogicalProcess)LPs.get(i);
	   lp.displayInterests();
	}//end for
  }//end displayInterests
  */
  
  /**Adds a LP into LPs.*/
  public void addLP(LogicalProcess lp)
  {
      schedulingLoop.LPs.add(lp);
  }//end addLP
  
  
  /**Returns the id of the PE.*/
  public int getID()
  {
      return id;
  }
  
  /**Sets the id for the PE (dummy method).
    *@param theID int representing the value that the PE's id will be set to.
	*/
  public void setID(int theID)
  {
      id=theID;  
  }
 



}//end class PE