loic/esds
1
0
Fork 0
mirror of https://gitlab.com/manzerbredes/esds.git synced 2025-04-07 02:26:28 +02:00
Code Issues Projects Releases Packages Wiki Activity Actions

Improve plugin callbacks

Browse source
This commit is contained in:
Loic Guegan 2022-09-07 07:04:40 +02:00
parent 8252ecb385
commit 02aac0ef99
2 changed files with 12 additions and 6 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

7
esds/node.py
View file

@ -13,7 +13,7 @@ class Node:
self.rargs=None # Store the requests arguments self.rargs=None # Store the requests arguments
self.plugins=list() # Contains all registered node plugins self.plugins=list() # Contains all registered node plugins
self.rqueue=queue.Queue() # Receive simulator acknowledgments self.rqueue=queue.Queue() # Receive simulator acknowledgments
self.chest={"state":"running", "turned_on":True, "request": None, "interfaces":dict(), "interfaces_queue_size":dict()} self.chest={"state":"running", "turned_on":True, "request": None, "interfaces":dict(), "interfaces_queue_size":dict(), "pending_plugin_notify":0}
for interface in interfaces: for interface in interfaces:
self.chest["interfaces"][interface]=queue.Queue() self.chest["interfaces"][interface]=queue.Queue()
self.chest["interfaces_queue_size"][interface]=0 self.chest["interfaces_queue_size"][interface]=0
@ -200,7 +200,10 @@ class Node:
ack=None ack=None
while True: while True:
ack=self.rqueue.get() # Wait for simulator acknowledgments ack=self.rqueue.get() # Wait for simulator acknowledgments
if ack[0] not in ack_types: if ack[0] == "plugin_notify":
self.plugin_notify(ack[1],ack[2])
self["pending_plugin_notify"]-=1
elif ack[0] not in ack_types:
ack_buffer.append(ack) ack_buffer.append(ack)
else: else:
break break

11
esds/simulator.py
View file

@ -124,9 +124,6 @@ class Simulator:
sorted_indexes=np.lexsort((self.events[:,3],self.events[:,1])) sorted_indexes=np.lexsort((self.events[:,3],self.events[:,1]))
self.events=self.events[sorted_indexes] self.events=self.events[sorted_indexes]
def notify_node_plugin(self,node,function,args):
self.nodes[node]["plugin_notify"][function]=args
def sync_node_non_blocking(self,node, timeout_remove_only=False): def sync_node_non_blocking(self,node, timeout_remove_only=False):
""" """
Process all call request and wait for Node.sync() to return Process all call request and wait for Node.sync() to return
@ -362,7 +359,13 @@ class Simulator:
selector = self.netmat[interface]["bandwidth"][node.node_id,] > 0 selector = self.netmat[interface]["bandwidth"][node.node_id,] > 0
return np.arange(0,selector.shape[0])[selector] return np.arange(0,selector.shape[0])[selector]
def notify_node_plugins(self,node,callback,args):
node[pending_plugin_notify]+=1
self.nodes[int(event[2][1])].rqueue.put(("plugin_notify",callback,args))
# Now ensure that all callbacks are called before continuing
while node[pending_plugin_notify] > 0:
pass
def add_event(self,event_type,event_ts,event,priority=2): def add_event(self,event_type,event_ts,event,priority=2):
""" """
Call this function with sort=True the least amount of time possible Call this function with sort=True the least amount of time possible