from infinity import * from devs_exceptions import * from solvers import * from DEVS import * from Distributed_Model import * from Pyro.EventService.Clients import Publisher, Subscriber # from server import CHAT_SERVER_NAME from threading import Thread import random import Pyro.core import time from Queue import Queue from Pyro.errors import NamingError, ConnectionClosedError class OptimisticAtomicSolver(): """An optimistic atomic solver object. """ def __init__(self, devs_solver, stateQueue = [] ): # self.parent = parent self.devs_solver = devs_solver self.stateQueue = stateQueue self.gvt = 0 def getTimeNext(self): return self.devs_solver.getTimeNext() def getTimeLast(self): return self.devs_solver.getTimeLast() def setGVT(self, gvt): self.gvt = gvt for item in self.stateQueue: if item[0] < gvt: self.stateQueue.remove(item) def receive(self, msg): #An anti message to roll back is sent if msg[0] == -1: #resotre the state of the simulator to a specific point. t_new = msg[1] tl, tn, model = self.stateQueue.pop() while tl > t_new: (tl,tn,model) = self.stateQueue.pop() self.devs_solver.setModel(model) self.devs_solver.setTimeLast(tl) self.devs_solver.setTimeNext(tn) else: #any other message will be forwarded to the devs_simulator and the return value ( in case of an x- message inducing a y message return value) result = self.devs_solver.receive(msg) model = self.devs_solver.getModel() tl = self.devs_solver.getTimeLast() tn = self.devs_solver.getTimeNext() self.stateQueue.append( (tl,tn,model) ) return result class OptimisticCoupledSolver(): """An optimistic coupled solver object. """ def __init__(self, devs_solver): # self.parent = parent self.devs_solver = devs_solver self.gvt = 0 def getTimeNext(self): return self.devs_solver.getTimeNext() def getTimeLast(self): return self.devs_solver.getTimeLast() def setGVT(self, gvt): assert self.gvt < gvt, "Attempting to set a GVT that is smaller than the older value" self.gvt = gvt self.devs_solver.setGVT() def receive(self, msg): #An anti message to roll back is sent if msg[0] == -1: assert self.gvt < msg[1], "Sending an anti message with time less than GVT" #Loop over all the subsolvers, and for the ones who ran ahead send an anti message for name,solver in self.devs_solver.getSubSolvers().iteritems(): if solver.getTimeLast() > msg[1]: solver.recieve(msg) #then update the tn, tl and current event list. self.devs_solver.resetVariables() else: #any other message will be forwarded to the devs_simulator and the return value ( in case of an x- message inducing a y message return value) result = self.devs_solver.receive(msg) return result """ The Dependencies : 1. External Messages (inputs or anti-inputs) received from other processors 2. Time Last of Other Processors ( events ) 3. Sending Messages to other processors ( events ) 4. Accessing global IO, namely to calculate the minimium event time, request removing some objects """ class OptimisticRootCoordinator(Publisher, Subscriber): """ A Time warp root coordinator""" def __init__(self, child_model, name=""): Publisher.__init__(self) Subscriber.__init__(self) Pyro.core.initClient() #current simulation time self.name = name self.t = 0 self.logger = LogServer(child_model, child_model.getModelName() +"_Simulation_name") assert child_model != None, "The sub DEVS model needs to be specified" # self.child = child if child_model.type() == 'COUPLED': self.child = OptimisticCoupledSolver(CoupledSolver(model=child_model, config=None, logger=self.logger)) elif child_model.type() == 'ATOMIC': self.child = OptimisticAtomicSolver(AtomicSolver(model=child_model, logger=self.logger)) #queue to store input events self.in_queue = [] #queue to store output events self.out_queue = [] #queue to store the set of influncer processors # self.influncers = influncers #queue to store the set of influncees processors # self.influncees = influncees #global queue to store the sent not yet recieved out events # self.io_queue = io_queue self.msgQueue = Queue() # time of the last event within the scope of the root-coordinator self.tl = 0 self.GVT = 0 # maps processor_name -> tl recieved self.tl_processors = {} self.ports_map = {} self.es = Pyro.core.getProxyForURI("PYRONAME://"+Pyro.constants.EVENTSERVER_NAME) def getName(self): return self.name def getTimeLast(self): return self.tl def ConnectPort(self,external_port_name,internal_port_name): #TODO: check that the internal port name exists on the model print "TimeWarp-Coordinator:: " + self.name + " Subscribing to: " + external_port_name # self.es.subscribe(external_port_name) self.subscribe(external_port_name) self.ports_map[external_port_name] = internal_port_name """ Iterate over the asynchronus messages receieved and porcess them """ def handle_msg_queue(self): while not self.msgQueue.empty(): event = self.msgQueue.get(block=True, timeout=5) if event.subject == "TimeLast": sender, t = event.msg self.tl_processors[sender] = t print self.name + " is updating a timeLast message from: " + str(sender) + " with time: " + str(t) elif self.ports_map.has_key(event.subject): msg = event.msg #print self.name + " is handling a message from: " + str(msg) self.external_handle(msg) """ Starts the Main Loop Simulation """ def startSimulation(self): print self.name + " Starting Simulation" self.inputThread=Thread(target=self.events_listener) self.inputThread.start() self.runningThread = Thread(target=self.startEventLoop) self.runningThread.start() """ A method which could be called asynchronusly it Appends the message recieved to the recieved messages queue to be handled later by the main loop""" def events_listener(self): print self.name + 'Input thread is on, Ready for input Messages!' try: self.subscribe("TimeLast") self.listen() except KeyboardInterrupt: print 'Shutting down... (press enter)' self.abort() # need to get new chatbox proxy because we're in a different thread self.abort() print 'Closing the Input Messages thread!' """ Makes sure the simulation keeps running on the current node """ def startEventLoop(self): #Send the initialization method self.child.receive( (0,0) ) self.tl = 0 while True: # time.sleep(2) # To Slow things down a bit tn = self.child.getTimeNext() tl = self.child.getTimeLast() # 1. When the Component is blocked, with time Next infinity we need to skip the loop and wait #check the in_queue for any possible event print self.name + " Has Input Queue: " + str(self.in_queue) self.in_queue.sort() if len(self.in_queue) > 0 and self.in_queue[0][0] <= tn: t = self.in_queue[0][0] #Need to modify x to map the right port x = self.in_queue[0][1] self.in_queue.remove(self.in_queue[0]) assert len(x)==1,"Only one event per port is allowed" Y = {} from_port = x.keys()[0] Y[self.ports_map[from_port]] = x[from_port] print self.name + " Sending the message: " + str((Y,t)) +" to Child" self.child.receive( (Y,t) ) #else execute the * message else: if tn != INFINITY: print self.name + " Sending the star Message with time: " + str(tn) result = self.child.receive( (1,tn) ) self.out_queue.append( (tn,result) ) #Send a request to put the result ( (t,y) message ) into the global IO queue assert len(result)==1,"Only one event per port is allowed" new_result = {} port_name = result.keys()[0].getPortFullName() new_result[port_name] = result[result.keys()[0]] print self.name + " publishing event: " + str((tn,new_result,1)) + " on channel: " + port_name self.es.publish( port_name, (tn,new_result,1) ) self.tl = self.child.devs_solver.getTimeLast() self.es.publish("TimeLast", (self.name,self.tl) ) # Compute GVT the minimum of the time last of all processors and times in IO_queue # 1.Ask all processors for thier time last # gvt = self.tl # # for p, t in self.tl_processors.iteritems(): # gvt = min(gvt,t) # # for p in self.influncees: # gvt = min(gvt, p.getTimeLast()) # for p2 in self.influncers: # gvt = min(gvt, p2.getTimeLast()) # 2.Ask the Global IO_queue for it smallest event time # for t, item in self.io_queue: # gvt = min(gvt,t) # Do fossil collection for states with time t < GVT # self.fossil_collection() #hanlde asychronus events waiting at the ports self.handle_msg_queue() '''Do fossil collection for states with time t < GVT on self and children ''' def fossil_collection(self): self.child.setGVT(gvt) # And inputs, and outputs if len(self.in_queue) > 0: self.in_queue.sort() for item in self.in_queue: if item[0] < gvt: self.in_queue.remove(item) if len(self.out_queue) > 0: self.out_queue.sort() for item in self.out_queue: if item[0] < gvt: self.out_queue.remove(item) '''An Asynch message handler, it places the messages recieved on an event Queue ''' def event(self,event): #print self.name + " Received Message: " + str(event) self.msgQueue.put(event) ''' The message will look something like this : and will be recieved from influcning processor ({},t,1) or ({},t,-1) ''' def external_handle(self, msg): #receive an anti message from an influncing processor print self.name + ":: is Handling the Message: " + str(msg) x = msg[1] new_t = msg[0] if msg[2] == -1: print "\t Message of type Roll-back " tn = self.child.getTimeNext() if new_t < tn : #roll back child print "\t\t Rolling Back child from time: " + str(t) +"to: " + str(new_t) self.child.receive( (-1,new_t) ) #roll-back all outputs events which are not valid anymore print "\t Attempting to cancel already sent outputs" for t_y, y, m_t in self.out_queue: if t_y > new_t: #send anti-message to INFLUNCED processor port_channel = y[0] self.es.publish(port_channel, (y,t_y,-1) ) #delete input event corresponding to this event form Input Queue self.out_queue.remove( (t_y,y) ) # Be careful of clone comparisons caused by pickling self.in_queue.remove( (new_t,x) ) #receive an x message from an influencing processor elif msg[2] == 1: print self.name + " Adding an item: " + str((new_t, x))+ " to the in queue" self.in_queue.append( (new_t, x) ) if new_t < self.child.getTimeLast(): self.child.receive( (-1,new_t) ) #rollback all outputs events which are not valid anymore for t_y, y in self.out_queue: if t_y > new_t: port_channel = y[0].getPortFullName() self.es.publish(port_channel, (y,t_y,-1) ) self.out_queue.remove( (t_y,y) ) else: raise DEVSException("Unrecognized message on Time warp root coordinator" + self.name , 1)