Debug synchronization + add return code

6 files changed, 76 insertions, 52 deletions

diff --git a/esds/__init__.py b/esds/__init__.py
index af3fa98..10c6881 100644
--- a/esds/__init__.py
+++ b/esds/__init__.py
@@ -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
 
diff --git a/esds/node.py b/esds/node.py
index 4f16f46..12e7a13 100644
--- a/esds/node.py
+++ b/esds/node.py
@@ -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):
diff --git a/esds/plugins/node_plugin.py b/esds/plugins/node_plugin.py
index 1b62d60..2e6c8e4 100644
--- a/esds/plugins/node_plugin.py
+++ b/esds/plugins/node_plugin.py
@@ -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)
diff --git a/esds/plugins/power_states.py b/esds/plugins/power_states.py
index e5a0e19..3c97c4f 100644
--- a/esds/plugins/power_states.py
+++ b/esds/plugins/power_states.py
@@ -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):
+    def on_turn_on(self):
+        for interface in self.power.keys():
+            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 sync_idle(self):
         clock=self.api.read("clock")
-        idle=0
         for interface in self.power.keys():
-            idle+=(clock-self.idle_clock)*self.power[interface]["idle"]
-        return idle
-
-    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
+            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")
diff --git a/esds/rcode.py b/esds/rcode.py
new file mode 100644
index 0000000..f4ca775
--- /dev/null
+++ b/esds/rcode.py
@@ -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
+
diff --git a/esds/simulator.py b/esds/simulator.py
index b774468..44c45c4 100644
--- a/esds/simulator.py
+++ b/esds/simulator.py
@@ -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)