#!/usr/bin/env python import numpy as np import threading,importlib,queue,sys,time class Node: available_node_id=0 def __init__(self,src,interfaces): """ """ self.node_id=Node.available_node_id Node.available_node_id+=1 # Refresh node id self.src=src # Store the node source code self.args=None # Store the node arguments (passed through Simulator.create_node() self.rargs=None # Store the requests arguments self.plugins=list() # Contains all registered node plugins self.rqueue=queue.Queue() # Receive simulator acknowledgments self.chest={"state":"running", "turned_on":True, "request": None, "interfaces":dict(), "interfaces_queue_size":dict()} for interface in interfaces: self.chest["interfaces"][interface]=queue.Queue() self.chest["interfaces_queue_size"][interface]=0 self.chest_lock=threading.Lock() # To access/modify self.chest def plugin_register(self,plugin): self.plugins.append(plugin) def plugin_notify(self,reason,args): """ This function strives to avoid using Python specific features """ for p in self.plugins: if reason == "receive_return" or reason == "receivet_return": p.on_receive_return(args[0],args[1],args[2],args[3]) if reason == "send_call": p.on_send_call(args[0],args[1],args[2],args[3]) if reason == "send_return": p.on_send_return(args[0],args[1],args[2],args[3],args[4]) if reason == "terminated": p.on_terminated() def __getitem__(self,key): self.chest_lock.acquire() value=self.chest[key] self.chest_lock.release() return value def __setitem__(self,key,value): self.chest_lock.acquire() value=self.chest[key]=value self.chest_lock.release() def log(self,msg): self.rargs=msg self["request"]="log" self["state"]="call" self.wait_ack(["log"]) def read(self, register): self["request"]="read" self.rargs=register self["state"]="call" ack=self.wait_ack(["read"]) return ack[1] def wait(self,duration): self.rargs=duration self["request"]="timeout_add" self["state"]="call" self.wait_ack(["timeout_add"]) self["state"]="pending" self.wait_ack(["timeout"]) def wait_end(self): self["request"]="wait_end" self["state"]="request" self.wait_ack(["wait_end"]) self.wait_ack(["sim_end"]) def turn_off(self): self["turned_on"]=False self["request"]="turn_off" self["state"]="call" self.wait_ack(["turn_off"]) def turn_on(self): self["turned_on"]=True self["request"]="turn_on" self["state"]="call" self.wait_ack(["turn_on"]) def send(self, interface, data, datasize, dst): self.plugin_notify("send_call",(interface,data,datasize,dst)) self.rargs=(interface, data, datasize, dst) self["request"]="send" self["state"]="request" ack=self.wait_ack(["send","send_cancel"]) self.plugin_notify("send_return",(interface,data,datasize,dst,ack[1])) return ack[1] def receive(self,interface): self["request"]="receive" self.rargs=interface self["state"]="request" self.wait_ack(["receive"]) data,start_at,end_at=self["interfaces"][interface].get() self.plugin_notify("receive_return",(interface,data,start_at,end_at)) return (0,data) def sendt(self, interface, data, datasize, dst, timeout): self.rargs=timeout self["request"]="timeout_add" self["state"]="call" self.wait_ack(["timeout_add"]) self.rargs=(interface, data, datasize, dst) self["request"]="send" self["state"]="request" ack=self.wait_ack(["send","timeout","send_cancel"]) if ack[0] == "timeout": self["request"]="send_cancel" self["state"]="call" self.wait_ack(["send_cancel"]) return -1 self["request"]="timeout_remove" self["state"]="call" self.wait_ack(["timeout_remove"]) return ack[1] def receivet(self,interface, timeout): self.rargs=timeout self["request"]="timeout_add" self["state"]="call" self.wait_ack(["timeout_add"]) self["request"]="receive" self.rargs=interface self["state"]="request" ack=self.wait_ack(["receive","timeout"]) if ack[0] == "timeout": return (-1,None) self["request"]="timeout_remove" self["state"]="call" self.wait_ack(["timeout_remove"]) data,start_at,end_at=self["interfaces"][interface].get() self.plugin_notify("receivet_return",(interface,data,start_at,end_at)) return (0,data) def wait_ack(self, ack_types): """ Wait for specific acks from the request queue (rqueue) """ ack_buffer=list() # To filter ack ack=None while True: ack=self.rqueue.get() # Wait for simulator acknowledgments if ack[0] not in ack_types: ack_buffer.append(ack) else: break # Push back the filtered ack for cur_ack in ack_buffer: self.rqueue.put(cur_ack) return(ack) def sync(self): """ Wait until node stop running """ while self["state"] == "running": pass def run(self,args): """ Load and run the user program """ self.node=importlib.import_module(self.src) self.args=args # Allow access to arguments self.node.execute(self) self["state"]="terminated" class Simulator: """ Flow-Level Discrete Event Simulator for Cyber-Physical Systems The general format for an event is (type,timestamp,event,priority) Event types: - 0 send (0,timestamp,(src,dst,interface,data,datasize,duration,datasize_remaining), 1) - 1 timeout (1,timestamp,node_id,4) - 2 breakpoint_manual (3,timestamp,0,0) - 3 breakpoint_auto (4,timestamp,0,0) Very important: when the simulator wakes up a node (changing is state to running) data that should be received by that node on the current simulated time SHOULD be in the queue! Thus, the send event must be handle before the other event (priority equals to 1). Otherwise plugings such as the power states one may not gives accurate results because of missing entries in the nodes received queues. """ def __init__(self,netmat): """ Format of netmat: { "interface": {"bandwidth": numpy_matrix, "latency": numpy_matrix, "is_wired":bool}} For wireless interfaces the diagonals of the bandwidth and latency matrices are very important. They determine the duration of the tranmission for THE SENDER. It allows to have a different tx duration per node and per interface. Thus, at each wireless communication, an addionnal event is created for the sender that corresponds to a send to himself (diagonals of the matrices) used to unlock him from the api.send() call. Consequently, the duration of the transmission (by the sender) can be different from the time at which the receivers actually receive the data (non-diagonal entries of the matrices). """ self.netmat=netmat self.nodes=list() self.sharing=dict() for interface in netmat.keys(): if netmat[interface]["is_wired"]: self.sharing[interface]=np.zeros(len(netmat[interface]["bandwidth"])) self.events=np.empty((0,4),dtype=object) self.events_dirty=True # For optimization reasons self.startat=-1 self.time=0 self.debug_file_path="./esds.debug" self.precision=".3f" self.interferences=True self.wait_end_nodes=list() # Keep track of nodes that wait for the end of the simulation self.time_truncated=format(self.time,self.precision) # Truncated version is used in log print def update_network(self,netmat): for event in self.events: if int(event[0]) == 0: cur_event=event[2] ts=float(event[1]) src_id,dst_id,interface, data, datasize,duration, datasize_remaining,start_at=cur_event new_bw=netmat[interface]["bandwidth"][int(src_id),int(dst_id)] old_bw=self.netmat[interface]["bandwidth"][int(src_id),int(dst_id)] new_lat=netmat[interface]["latency"][int(src_id),int(dst_id)] old_lat=self.netmat[interface]["latency"][int(src_id),int(dst_id)] if new_bw != old_bw or new_lat != old_lat: new_datasize_remaining=float(datasize_remaining)*((ts-self.time)/float(duration)) if new_datasize_remaining > 0: latency_factor=new_datasize_remaining/float(datasize) if self.netmat[interface]["is_wired"]: new_duration=new_datasize_remaining*8/(new_bw/self.sharing[interface][int(dst_id)])+new_lat*latency_factor else: new_duration=new_datasize_remaining*8/new_bw+new_lat*latency_factor event[1]=self.time+new_duration event[2][6]=new_datasize_remaining event[2][5]=new_duration self.netmat=netmat def debug(self): """ Log all the informations for debugging """ stdout_save = sys.stdout with open(self.debug_file_path, "a") as debug_file: sys.stdout = debug_file print("-----------------------------------------------") print("Started since {}s".format(round(time.time()-self.startat,2))) print("Simulated time {}s (or more precisely {}s)".format(self.time_truncated,self.time)) states=dict() timeout_mode=list() sharing=dict() for node in self.nodes: s=node["state"] states[s]=states[s]+1 if s in states else 1 node_key="n"+str(node.node_id) for interface in self.sharing.keys(): if self.sharing[interface][node.node_id] > 0: if node_key not in sharing: sharing[node_key] = "" sharing[node_key]+=str(int(self.sharing[interface][node.node_id])) print("Node number per state: ",end="") for key in states: print(key+"="+str(states[key]), end=" ") print("\nNode sharing: ",end="") for node_id in sharing: print(node_id+"="+sharing[node_id], end=" ") print("\nIds of node in timeout mode: ", end="") for n in timeout_mode: print(n,end=" ") print("\nSorted events list:") print(self.events) sys.stdout = stdout_save def create_node(self, src, args=None): """ Create a node thread and run it """ node=Node(src, self.netmat.keys()) self.nodes.append(node) thread=threading.Thread(target=node.run, daemon=False,args=[args]) thread.start() def log(self,msg,node=None): src = "esds" if node is None else "n"+str(node) print("[t="+str(self.time_truncated)+",src="+src+"] "+msg) def sort_events(self): """ Sort the events by timestamp and priorities """ sorted_indexes=np.lexsort((self.events[:,3],self.events[:,1])) self.events=self.events[sorted_indexes] def sync_node(self,node): """ Process all call request and wait for Node.sync() to return """ node.sync() while node["state"] == "call": if node["request"] == "log": self.log(node.rargs,node=node.node_id) node["state"]="running" node.rqueue.put(("log",0)) elif node["request"] == "timeout_add": self.add_event(1,self.time+node.rargs,node.node_id,priority=3) node["state"]="running" node.rqueue.put(("timeout_add",0)) elif node["request"] == "timeout_remove": selector=list() for event in self.events: if event[0] == 1 and event[2]==node.node_id: selector.append(True) else: selector.append(False) self.events=self.events[~np.array(selector)] node["state"]="running" node.rqueue.put(("timeout_remove",0)) elif node["request"] == "read": node["state"]="running" if node.rargs == "clock": node.rqueue.put(("read",self.time)) elif node.rargs[0:5] == "ncom_": # ncom_ register interface=node.rargs[5:] count=0 # Count number of communication on interface for event in self.events: if event[0] == 0 and event[2][1] == node.node_id and event[2][2] == interface: count+=1 node.rqueue.put(("read",count)) else: node.rqueue.put(("read",0)) # Always return 0 if register is unknown elif node["request"] == "turn_on": node["state"]="running" node.rqueue.put(("turn_on",0)) self.log("Turned on",node=node.node_id) elif node["request"] == "turn_off": # Create communications selectors (True/False arrays) selector_wireless=list() # Select all wireless events where node is involved selector_wired=list() # Select all wired events where node is involved for event in self.events: if event[0]==0 and int(event[2][1])==node.node_id: if self.netmat[event[2][2]]["is_wired"]: selector_wireless.append(False) selector_wired.append(True) else: selector_wireless.append(True) selector_wired.append(False) else: selector_wireless.append(False) selector_wired.append(False) # Informed senders of wired events to cancel send for event in self.events[selector_wired]: sender=self.nodes[int(event[2][0])] sender["state"]="running" sender.rqueue.put(("send_cancel",2)) # Remove communications from the event list if(len(self.events) != 0): self.events=self.events[~(np.array(selector_wireless)|np.array(selector_wired))] # Refresh wired sharing for interface in self.sharing.keys(): self.sharing[interface][node.node_id]=0 # Sharing goes back to zero # Update node state after turning off node["state"]="running" node.rqueue.put(("turn_off",0)) self.log("Turned off",node=node.node_id) elif node["request"] == "send_cancel": selector=list() for event in self.events: if event[0]==0 and int(event[2][0]) == node.node_id: selector.append(True) if self.netmat[event[2][2]]["is_wired"]: self.update_sharing(int(event[2][1]),-1,event[2][2]) else: selector.append(False) self.events=self.events[~np.array(selector)] node["state"]="running" node.rqueue.put(("send_cancel",0)) node.sync() def update_sharing(self, dst, amount,interface): """ Manage bandwidth sharing on wired interfaces """ sharing=self.sharing[interface][dst] new_sharing=sharing+amount for event in self.events: if event[0] == 0 and self.netmat[event[2][2]]["is_wired"] and int(event[2][1]) == dst: remaining=event[1]-self.time if remaining > 0: remaining=remaining/sharing if sharing>1 else remaining # First restore sharing remaining=remaining*new_sharing if new_sharing > 1 else remaining # Then apply new sharing event[2][5]=remaining # Update duration event[1]=self.time+remaining # Update timestamp self.sharing[interface][dst]=new_sharing self.sort_events() def handle_interferences(self,sender,receiver, interface): """ Interferences are detected by looking for conflicts between new events and existing events. """ status=False selector=list() notify=set() for event in self.events: event_type=event[0] com=event[2] if event_type==0 and com[2] == interface: com_sender=int(com[0]) com_receiver=int(com[1]) select=False if receiver==com_sender: status=True notify.add(receiver) elif receiver==com_receiver: status=True select=True notify.add(receiver) if sender==com_receiver and com_sender != com_receiver: select=True notify.add(sender) selector.append(select) else: selector.append(False) if len(selector) != 0: self.events=self.events[~np.array(selector)] for node in notify: self.log("Interferences on "+interface,node=node) return status def sync_event(self, node): """ Collect events from the nodes """ if node["state"] == "request": if node["request"] == "send": node["state"]="pending" interface, data, datasize, dst=node.rargs self.communicate(interface, node.node_id, dst, data, datasize) elif node["request"] == "receive": interface=node.rargs if node["interfaces_queue_size"][interface] > 0: node["interfaces_queue_size"][interface]-=1 node.rqueue.put(("receive",0)) node["state"]="running" # Do not forget to collect the next event. This is the only request which is processed here self.sync_node(node) self.sync_event(node) elif node["request"] == "wait_end": node["state"]="pending" node.rqueue.put(("wait_end",0)) self.wait_end_nodes.append(node.node_id) def communicate(self, interface, src, dst, data, datasize): """ Create communication event between src and dst """ nsrc=self.nodes[src] if self.netmat[interface]["is_wired"]: if self.nodes[dst]["turned_on"]: self.log("Send "+str(datasize)+" bytes to n"+str(dst)+" on "+interface,node=src) self.update_sharing(dst,1,interface) # Update sharing first # Note that in the following we send more data than expected to handle bandwidth sharing (datasize*8*sharing): duration=datasize*8/(self.netmat[interface]["bandwidth"][src,dst]/self.sharing[interface][dst])+self.netmat[interface]["latency"][src,dst] self.add_event(0,duration+self.time,(src,dst,interface,data,datasize,duration,datasize,self.time)) else: nsrc["state"]="request" # Try later when node is on else: self.log("Send "+str(datasize)+" bytes on "+interface,node=src) for dst in self.list_receivers(nsrc,interface): if self.nodes[dst]["turned_on"]: duration=datasize*8/self.netmat[interface]["bandwidth"][src,dst]+self.netmat[interface]["latency"][src,dst] if src == dst: # This event (where src == dst) is used to notify the sender when data is received! # Correspond to the diagonal of the network matrices (bandwidth and latency) self.add_event(0,duration+self.time,(src,dst,interface,data,datasize,duration,datasize,self.time)) elif not self.interferences: self.add_event(0,duration+self.time,(src,dst,interface,data,datasize,duration,datasize,self.time)) elif not self.handle_interferences(src,dst, interface): self.add_event(0,duration+self.time,(src,dst,interface,data,datasize,duration,datasize,self.time)) def list_receivers(self,node,interface): """ Deduce reachable receivers from the bandwidth matrix """ selector = self.netmat[interface]["bandwidth"][node.node_id,] > 0 return np.arange(0,selector.shape[0])[selector] def add_event(self,event_type,event_ts,event,priority=1): """ Call this function with sort=True the least amount of time possible """ self.events=np.concatenate([self.events,[np.array([event_type,event_ts,np.array(event,dtype=object),priority],dtype=object)]]) # Add new events self.sort_events() def run(self, breakpoints=[],breakpoint_callback=lambda s:None,breakpoints_every=None,debug=False,interferences=True): """ Run the simulation with the created nodes """ ##### Setup simulation self.startat=time.time() self.interferences=interferences for bp in breakpoints: self.add_event(2,bp,0,0) if breakpoints_every != None: self.add_event(3,breakpoints_every,0,0) if debug: with open(self.debug_file_path, "w") as f: f.write("Python version {}\n".format(sys.version)) f.write("Simulation started at {}\n".format(self.startat)) f.write("Number of nodes is "+str(len(self.nodes))+"\n") f.write("Manual breakpoints list: "+str(breakpoints)+"\n") f.write("Breakpoints every "+str(breakpoints_every)+"s\n") ##### Simulation loop while True: # Synchronize every nodes for node in self.nodes: self.sync_node(node) # Manage events for node in self.nodes: self.sync_event(node) # Generate debug logs if debug: self.debug() # Simulation end if len(self.events) <= 0 or len(self.events) == 1 and self.events[0,0] == 3: # Notify nodes that wait for the end of the simulation # Note that we do not allow them to create new events (even if they try, they will not be processed) for node_id in self.wait_end_nodes: self.nodes[node_id].rqueue.put(("sim_end",0)) self.nodes[node_id]["state"]="running" self.sync_node(self.nodes[node_id]) # Allow them for make call requests (printing logs for example) break # End the event processing loop # Update simulation time self.time=self.events[0,1] self.time_truncated=format(self.time,self.precision) # refresh truncated time # Process events while len(self.events) > 0 and self.events[0,1] == self.time: event_type=int(self.events[0,0]) ts=self.events[0,1] event=self.events[0,2] self.events=np.delete(self.events,0,0) # Consume events NOW! not at the end of the loop (event list may change in between) if event_type == 0: src_id,dst_id,interface, data, datasize,duration,datasize_remaining,start_at=event src=self.nodes[int(src_id)] dst=self.nodes[int(dst_id)] if self.netmat[interface]["is_wired"]: dst["interfaces"][interface].put((data,start_at,self.time)) dst["interfaces_queue_size"][interface]+=1 self.update_sharing(dst.node_id,-1,interface) self.log("Receive "+str(datasize)+" bytes on "+interface,node=int(dst_id)) # If node is receiving makes it consume (this way if there is a timeout, it will be removed!) if dst["state"] == "request" and dst["request"] == "receive": dst["interfaces_queue_size"][interface]-=1 dst.rqueue.put(("receive",0)) dst["state"]="running" self.sync_node(dst) src["state"]="running" src.rqueue.put(("send",0)) else: if src.node_id != dst.node_id: dst["interfaces"][interface].put((data,start_at,self.time)) dst["interfaces_queue_size"][interface]+=1 self.log("Receive "+str(datasize)+" bytes on "+interface,node=int(dst_id)) # If node is receiving makes it consume (this way if there is a timeout, it will be removed!) if dst["state"] == "request" and dst["request"] == "receive": dst["interfaces_queue_size"][interface]-=1 dst.rqueue.put(("receive",0)) dst["state"]="running" self.sync_node(dst) else: src["state"]="running" src.rqueue.put(("send",0)) elif event_type == 1: node=self.nodes[int(event)] node["state"]="running" node.rqueue.put(("timeout",0)) self.sync_node(node) elif event_type == 2 or event_type == 3: breakpoint_callback(self) if event_type == 3: self.add_event(3,self.time+breakpoints_every,0,0) ##### Simulation ends self.log("Simulation ends")