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

Debug synchronization + add return code

Browse source
This commit is contained in:
Loic Guegan 2022-09-09 13:16:42 +02:00
parent 57f020e09d
commit cfa677a0c6
21 changed files with 107 additions and 71 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

3
esds/__init__.py
View file

@ -1,4 +1,5 @@
__all__ = ["simulator", "node", "plugins", "helpers"]
__all__ = ["simulator", "node", "plugins", "helpers", "rcode"]
from esds.simulator import Simulator
from esds.rcode import RCode

17
esds/node.py
View file

@ -1,4 +1,5 @@
import threading,importlib,queue
from esds.rcode import RCode
class Node:
available_node_id=0
@ -22,7 +23,7 @@ class Node:
def plugin_register(self,plugin):
self.plugins.append(plugin)
def plugin_notify(self,reason,args,time=None):
def plugin_notify(self,reason,args=None,time=None):
"""
This function strives to avoid using Python specific features
"""
@ -35,6 +36,10 @@ class Node:
p.on_send_return(args[0],args[1],args[2],args[3],args[4])
if reason == "on_communication_end":
p.on_communication_end(time,args)
if reason == "turn_off_return":
p.on_turn_off()
if reason == "turn_on_return":
p.on_turn_on()
if reason == "terminated":
p.on_terminated()
@ -96,12 +101,14 @@ class Node:
self["request"]="turn_off"
self["state"]="call_non_blocking"
self.wait_ack(["turn_off"])
self.plugin_notify("turn_off_return")
def turn_on(self):
self["turned_on"]=True
self["request"]="turn_on"
self["state"]="call_non_blocking"
self.wait_ack(["turn_on"])
self.plugin_notify("turn_on_return")
def send(self, interface, data, datasize, dst, receiver_required=False):
if interface not in self["interfaces"]:
@ -141,7 +148,7 @@ class Node:
self["request"]="send"
self["state"]="call_blocking"
ack=self.wait_ack(["send","timeout","send_cancel"])
status=-1
status=RCode.TIMEOUT_EXPIRE
if ack[0] == "timeout":
self["request"]="send_cancel"
self["state"]="call_non_blocking"
@ -164,7 +171,7 @@ class Node:
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)
return (RCode.SUCCESS,data)
def receivet(self,interface, timeout):
if interface not in self["interfaces"]:
@ -183,14 +190,14 @@ class Node:
self.rargs=interface
self["state"]="call_blocking"
ack=self.wait_ack(["receive","timeout"])
result=(-1,None)
result=(RCode.TIMEOUT_EXPIRE,None)
if ack[0] != "timeout":
self["request"]="timeout_remove"
self["state"]="call_non_blocking"
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))
result=(0,data)
result=(RCode.SUCCESS,data)
return result
def wait_ack(self, ack_types):

6
esds/plugins/node_plugin.py
View file

@ -26,6 +26,12 @@ class NodePlugin:
def on_communication_end(self,time,com_event):
pass
def on_turn_on(self):
pass
def on_turn_off(self):
pass
def log(self,msg):
self.api.log(self.plugin_name+"(NP) "+msg)

45
esds/plugins/power_states.py
View file

@ -110,10 +110,7 @@ class PowerStatesComms(NodePlugin):
def __init__(self,api):
super().__init__("PowerStatesComms",api)
self.energy_dynamic=0.0 # Store the dynamic part of the energy consumption
self.power=dict() # Store the power states
self.tx_clock=0 # Dynamic clock (store the time at which a the last tx starts
self.idle_clock=api.read("clock") # Store the start time (to compute the idle part of the energy consumption)
def on_communication_end(self,time,com_event):
content=com_event[2]
@ -121,39 +118,39 @@ class PowerStatesComms(NodePlugin):
duration=time-content[7]
interface=content[2]
mode= "tx" if content[0] == self.api.node_id else "rx"
self.energy_dynamic+=self.power[interface][mode]*duration
self.power[interface]["consumption_dynamic"]+=self.power[interface][mode]*duration
if self.api.node_id==0:
print("called with +{}J".format(self.power[interface][mode]*duration))
def set_power(self,interface,idle,tx,rx):
self.power[interface]=dict()
self.power[interface]["idle"]=idle
self.power[interface]["rx"]=rx
self.power[interface]["tx"]=tx
self.power[interface]["on_at"]=self.api.read("clock")
self.power[interface]["consumption_idle"]=0
self.power[interface]["consumption_dynamic"]=0
def get_idle(self):
clock=self.api.read("clock")
idle=0
def on_turn_on(self):
for interface in self.power.keys():
idle+=(clock-self.idle_clock)*self.power[interface]["idle"]
return idle
self.power[interface]["on_at"]=self.api.read("clock")
def on_turn_off(self):
self.sync_idle()
self.log("Consumes "+str(self.get_energy()))
def get_receive_queue_energy(self,interface):
"""
Not that call to on_receive_return may not have happened yet (or never).
Thus we should manually compute the energy consumption stored in each queues of the node.
"""
energy=0
# For each interface we should check if there is received data that has not been consumed
for data in list(self.api["interfaces"][interface].queue):
start_at=float(data[1])
end_at=float(data[2])
energy+=(end_at-start_at)*self.power[interface]["rx"]
return energy
def sync_idle(self):
clock=self.api.read("clock")
for interface in self.power.keys():
self.power[interface]["consumption_idle"]+=(clock-self.power[interface]["on_at"])*self.power[interface]["idle"]
self.power[interface]["on_at"]=clock
def get_energy(self):
queue_energy=0
self.sync_idle()
consumption=0
for interface in self.power.keys():
queue_energy+=self.get_receive_queue_energy(interface)
return self.get_idle()+self.energy_dynamic+queue_energy
consumption+=self.power[interface]["consumption_idle"]+self.power[interface]["consumption_dynamic"]
return consumption
def report_energy(self):
self.log("Communications consumed "+str(round(self.get_energy(),2))+"J")

9
esds/rcode.py Normal file
View file

@ -0,0 +1,9 @@
from enum import Enum
class RCode(Enum):
SUCCESS = 0
FAIL = 1
TIMEOUT_EXPIRE = 2
RECEIVER_TURNED_OFF = 3
RECEIVER_UNAVAILABLE = 4

48
esds/simulator.py
View file

@ -1,6 +1,7 @@
import numpy as np
import threading,sys,time
from esds.node import Node
from esds.rcode import RCode
class Simulator:
"""
@ -139,22 +140,22 @@ class Simulator:
selector.append(False)
self.events=self.events[~np.array(selector)]
node["state"]="running"
node.rqueue.put(("timeout_remove",0))
node.rqueue.put(("timeout_remove",RCode.SUCCESS))
elif timeout_remove_only:
break
elif not timeout_remove_only:
if node["request"] == "log":
self.log(node.rargs,node=node.node_id)
node["state"]="running"
node.rqueue.put(("log",0))
node.rqueue.put(("log",RCode.SUCCESS))
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))
node.rqueue.put(("timeout_add",RCode.SUCCESS))
elif node["request"] == "notify_add":
self.add_event(4,self.time+node.rargs,node.node_id,priority=0)
node["state"]="running"
node.rqueue.put(("notify_add",0))
node.rqueue.put(("notify_add",RCode.SUCCESS))
elif node["request"] == "notify_remove":
selector=list()
for event in self.events:
@ -164,7 +165,7 @@ class Simulator:
selector.append(False)
self.events=self.events[~np.array(selector)]
node["state"]="running"
node.rqueue.put(("notify_remove",0))
node.rqueue.put(("notify_remove",RCode.SUCCESS))
elif node["request"] == "abort":
self.log("Simulation aborted: "+node.rargs,node=node.node_id)
exit(1)
@ -184,7 +185,7 @@ class Simulator:
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))
node.rqueue.put(("turn_on",RCode.SUCCESS))
self.log("Turned on",node=node.node_id)
elif node["request"] == "turn_off":
# Create communications selectors (True/False arrays)
@ -209,7 +210,7 @@ class Simulator:
selector_wireless.append(False)
selector_wired.append(False)
# Build the set of senders to notify (only in wired connections,
# indeed IRL, in wireless communications sender would send all its data)
# indeed IRL, in wireless communications, sender would send all its data)
senders_to_notify=set()
for event in self.events[selector_wired]:
senders_to_notify.add(int(event[2][0]))
@ -218,14 +219,14 @@ class Simulator:
self.events=self.events[~(np.array(selector_wireless)|np.array(selector_wired))]
# Update node state after turning off
node["state"]="running"
node.rqueue.put(("turn_off",0))
node.rqueue.put(("turn_off",RCode.SUCCESS))
self.log("Turned off",node=node.node_id)
# Informed senders of wired events that communication ended
for sender_id in senders_to_notify:
# Notify sender (node that wired sharing is updated in the send_cancel request)
sender_node=self.nodes[sender_id]
sender_node["state"]="running"
sender_node.rqueue.put(("send_cancel",2))
sender_node.rqueue.put(("send_cancel",RCode.RECEIVER_TURNED_OFF))
# The node should resume at current self.time. So, sync the sender now:
self.sync_node_non_blocking(sender_node)
self.sync_node_blocking(sender_node)
@ -247,7 +248,7 @@ class Simulator:
for com in sharing_to_update:
self.update_sharing(com[0],-1,com[1])
node["state"]="running"
node.rqueue.put(("send_cancel",0))
node.rqueue.put(("send_cancel",RCode.SUCCESS))
node.sync()
def update_sharing(self, dst, amount,interface):
@ -316,20 +317,23 @@ class Simulator:
exit(1)
if not self.communicate(interface, node.node_id, dst, data, datasize,receiver_required):
node["state"]="running"
node.rqueue.put(("send",4))
self.sync_node_non_blocking(node,timeout_remove_only=True)
node.rqueue.put(("send",RCode.RECEIVER_UNAVAILABLE))
# Do not forget to collect the next event (since current event did not happend)
# Be careful in node implementation to have no infinite loop when receiver_required=True
self.sync_node_non_blocking(node)
self.sync_node_blocking(node)
elif node["request"] == "receive":
interface=node.rargs
if node["interfaces_queue_size"][interface] > 0:
node["interfaces_queue_size"][interface]-=1
node["state"]="running"
node.rqueue.put(("receive",0))
node.rqueue.put(("receive",RCode.SUCCESS))
# Do not forget to collect the next event. This is the only request which is processed here
self.sync_node_non_blocking(node)
self.sync_node_blocking(node)
elif node["request"] == "wait_end":
node["state"]="pending"
node.rqueue.put(("wait_end",0))
node.rqueue.put(("wait_end",RCode.SUCCESS))
self.wait_end_nodes.append(node.node_id)
def communicate(self, interface, src, dst, data, datasize,receiver_required):
@ -413,10 +417,10 @@ class Simulator:
for node in self.nodes:
if node["state"] != "terminated":
node["state"]="running"
node.rqueue.put(("sim_end",0))
node.rqueue.put(("sim_end",RCode.SUCCESS))
self.sync_node_non_blocking(node) # Allow them for make non-blocking call requests (printing logs for example)
else:
node.rqueue.put(("sim_end",0))
node.rqueue.put(("sim_end",RCode.SUCCESS))
break # End the event processing loop
# Update simulation time
@ -442,12 +446,12 @@ class Simulator:
if dst["state"] == "call_blocking" and dst["request"] == "receive":
dst["interfaces_queue_size"][interface]-=1
dst["state"]="running"
dst.rqueue.put(("receive",0))
dst.rqueue.put(("receive",RCode.SUCCESS))
self.sync_node_non_blocking(dst,timeout_remove_only=True)
self.notify_node_plugins(dst, "on_communication_end", event)
self.update_sharing(dst.node_id,-1,interface)
src["state"]="running"
code=0 if perform_delivery else 1
code=RCode.SUCCESS if perform_delivery else RCode.FAIL
src.rqueue.put(("send",code))
self.sync_node_non_blocking(src,timeout_remove_only=True)
self.notify_node_plugins(src, "on_communication_end", event)
@ -461,23 +465,23 @@ class Simulator:
if dst["state"] == "call_blocking" and dst["request"] == "receive":
dst["interfaces_queue_size"][interface]-=1
dst["state"]="running"
dst.rqueue.put(("receive",0))
dst.rqueue.put(("receive",RCode.SUCCESS))
self.sync_node_non_blocking(dst,timeout_remove_only=True)
self.notify_node_plugins(dst, "on_communication_end", event)
else:
src["state"]="running"
src.rqueue.put(("send",0))
src.rqueue.put(("send",RCode.SUCCESS))
self.sync_node_non_blocking(src,timeout_remove_only=True)
self.notify_node_plugins(src, "on_communication_end", event)
elif event_type == 1: # Timeout
node=self.nodes[int(event)]
node["state"]="running"
node.rqueue.put(("timeout",0))
node.rqueue.put(("timeout",RCode.SUCCESS))
self.sync_node_non_blocking(node,timeout_remove_only=True)
elif event_type == 4:
node=self.nodes[int(event)]
node["state"]="running"
node.rqueue.put(("notify",0))
node.rqueue.put(("notify",RCode.SUCCESS))
self.sync_node_non_blocking(node,timeout_remove_only=True)
elif event_type == 2 or event_type == 3:
breakpoint_callback(self)

3
example/receiver.py
View file

@ -1,6 +1,7 @@
#!/usr/bin/env python
import sys, random, time
from esds import RCode
lr=random.Random(6)
@ -17,7 +18,7 @@ def execute(api):
# Receive until uptime seconds if elapsed
while api.read("clock") < wakeuntil:
code, data=api.receivet("wlan0",wakeuntil-api.read("clock"))
if code == 0:
if code == RCode.SUCCESS:
api.log("Receive "+data)
api.log("Was up for {}s".format(api.read("clock")-wakeat))
endoff=3600*(i+1)-api.read("clock")

3
manual/manual.org
View file

@ -136,6 +136,7 @@ called by the orchestrator to execute the code of your node. The api parameter p
#!/usr/bin/env python
import sys, random, time
from esds import RCode
lr=random.Random(6)
@ -152,7 +153,7 @@ called by the orchestrator to execute the code of your node. The api parameter p
# Receive until uptime seconds if elapsed
while api.read("clock") < wakeuntil:
code, data=api.receivet("wlan0",wakeuntil-api.read("clock"))
if code == 0:
if code == RCode.SUCCESS:
api.log("Receive "+data)
api.log("Was up for {}s".format(api.read("clock")-wakeat))
endoff=3600*(i+1)-api.read("clock")

2
tests/api_receivet0_eth0_1s1r/out
View file

@ -1,5 +1,5 @@
[t=0.000,src=n0] Send 1 bytes to n1 on eth0
[t=0.000,src=n1] Receive failed code=-1
[t=0.000,src=n1] Receive failed code=RCode.TIMEOUT_EXPIRE
[t=1.000,src=n1] Receive 1 bytes on eth0
[t=1.000,src=n1] Received: Hello World!
[t=1.000,src=esds] Simulation ends

3
tests/api_receivet0_eth0_1s1r/receiver.py
View file

@ -1,9 +1,10 @@
#!/usr/bin/env python
from esds import RCode
def receivet(node,timeout):
##### Simple receive
code, data=node.receivet("eth0",timeout)
msg="Received: "+data if code == 0 else "Receive failed code="+str(code)
msg="Received: "+data if code == RCode.SUCCESS else "Receive failed code="+str(code)
node.log(msg)
def execute(api):

2
tests/api_receivet_eth0_1s1r/out
View file

@ -2,7 +2,7 @@
[t=1.000,src=n1] Receive 1 bytes on eth0
[t=1.000,src=n1] Received: Hello World!
[t=1.000,src=n0] Send 1 bytes to n1 on eth0
[t=1.500,src=n1] Receive failed code=-1
[t=1.500,src=n1] Receive failed code=RCode.TIMEOUT_EXPIRE
[t=2.000,src=n1] Receive 1 bytes on eth0
[t=2.000,src=n1] Received: Hello World!
[t=2.000,src=esds] Simulation ends

3
tests/api_receivet_eth0_1s1r/receiver.py
View file

@ -1,9 +1,10 @@
#!/usr/bin/env python
from esds import RCode
def receivet(node,timeout):
##### Simple receive
code, data=node.receivet("eth0",timeout)
msg="Received: "+data if code == 0 else "Receive failed code="+str(code)
msg="Received: "+data if code == RCode.SUCCESS else "Receive failed code="+str(code)
node.log(msg)
def execute(api):

5
tests/api_send0_eth0_1s1r/receiver.py
View file

@ -1,9 +1,10 @@
#!/usr/bin/env python
from esds import RCode
def receive(api):
##### Simple receive
code, data=api.receive("eth0")
msg="Received: "+data if code == 0 else "Receive failed code="+str(code)
msg="Received: "+data if code == RCode.SUCCESS else "Receive failed code="+str(code)
api.log(msg)
def execute(api):
@ -16,7 +17,7 @@ def execute(api):
receive(api)
# Simulated time t=3s here
code, data=api.receivet("eth0",0)
msg="Received: "+data if code == 0 else "Receive failed code="+str(code)
msg="Received: "+data if code == RCode.SUCCESS else "Receive failed code="+str(code)
api.log(msg)
# Simulated time t=3s here

2
tests/api_send_eth0_1s1r/out
View file

@ -7,7 +7,7 @@
[t=3.000,src=n1] Received: Hello World!
[t=3.000,src=n1] Turned off
[t=4.000,src=n1] Turned on
[t=5.000,src=n1] Receive failed code=-1
[t=5.000,src=n1] Receive failed code=RCode.TIMEOUT_EXPIRE
[t=18.000,src=n0] End transmission
[t=18.000,src=n0] Send 15 bytes to n1 on eth0
[t=33.000,src=n1] Receive 15 bytes on eth0

7
tests/api_send_eth0_1s1r/receiver.py
View file

@ -1,21 +1,22 @@
#!/usr/bin/env python
from esds import RCode
def execute(api):
##### Simple receive
code, data=api.receive("eth0")
msg="Received: "+data if code == 0 else "Receive failed code="+str(code)
msg="Received: "+data if code == RCode.SUCCESS else "Receive failed code="+str(code)
api.log(msg)
##### Test if we still receive the data when we are not receiving
api.wait(2)
code, data=api.receive("eth0")
msg="Received: "+data if code == 0 else "Receive failed code="+str(code)
msg="Received: "+data if code == RCode.SUCCESS else "Receive failed code="+str(code)
api.log(msg)
##### Ensure data is not receive when turned off but communication must still be ongoing
api.turn_off()
api.wait(1)
api.turn_on()
code, data=api.receivet("eth0",1)
msg="Received: "+data if code == 0 else "Receive failed code="+str(code)
msg="Received: "+data if code == RCode.SUCCESS else "Receive failed code="+str(code)
api.log(msg)
##### Ensure communication get aborted on turned off for the sender
api.wait(28) # Goto t=33s

5
tests/api_send_eth0_2s1r/receiver.py
View file

@ -1,13 +1,14 @@
#!/usr/bin/env python
from esds import RCode
def execute(api):
##### Simple receive from node 0
code, data=api.receive("eth0")
msg="Received: "+data if code == 0 else "Receive failed code="+str(code)
msg="Received: "+data if code == RCode.SUCCESS else "Receive failed code="+str(code)
api.log(msg)
##### Simple receive from node 1
code, data=api.receive("eth0")
msg="Received: "+data if code == 0 else "Receive failed code="+str(code)
msg="Received: "+data if code == RCode.SUCCESS else "Receive failed code="+str(code)
api.log(msg)

3
tests/api_send_eth0_3s1r/receiver.py
View file

@ -1,9 +1,10 @@
#!/usr/bin/env python
from esds import RCode
def receive(node, n):
for i in range(0,n):
code, data=node.receive("eth0")
msg="Received: "+data if code == 0 else "Receive failed code="+str(code)
msg="Received: "+data if code == RCode.SUCCESS else "Receive failed code="+str(code)
node.log(msg)
def execute(api):

3
tests/api_send_wlan0_1s2r/receiver.py
View file

@ -1,9 +1,10 @@
#!/usr/bin/env python
from esds import RCode
def receive(node):
##### Simple receive
code, data=node.receive("wlan0")
msg="Received: "+data if code == 0 else "Receive failed code="+str(code)
msg="Received: "+data if code == RCode.SUCCESS else "Receive failed code="+str(code)
node.log(msg)

3
tests/api_sendt_eth0_1s1r/receiver.py
View file

@ -1,9 +1,10 @@
#!/usr/bin/env python
from esds import RCode
def receive(node):
##### Simple receive
code, data=node.receive("eth0")
msg="Received: "+data if code == 0 else "Receive failed code="+str(code)
msg="Received: "+data if code == RCode.SUCCESS else "Receive failed code="+str(code)
node.log(msg)
def execute(api):

3
tests/api_sendt_eth0_1s1r/sender.py
View file

@ -1,8 +1,9 @@
#!/usr/bin/env python
from esds import RCode
def sendt(node,timeout):
code=node.sendt("eth0","Hello World!",1,1,timeout)
msg="Send worked!" if code == 0 else "Send failed"
msg="Send worked!" if code == RCode.SUCCESS else "Send failed"
node.log(msg)
def execute(api):

3
tests/api_sendt_wlan0_1s2r/receiver.py
View file

@ -1,9 +1,10 @@
#!/usr/bin/env python
from esds import RCode
def receive(node):
##### Simple receive
code, data=node.receive("wlan0")
msg="Received: "+data if code == 0 else "Receive failed code="+str(code)
msg="Received: "+data if code == RCode.SUCCESS else "Receive failed code="+str(code)
node.log(msg)