Switch to deferred call

evio/broker/broker.py

@@ -60,7 +60,7 @@ from .controller_module import ControllerModule
 from .nexus import Nexus
 from .process_proxy import ProcessProxy, ProxyMsg
 from .subscription import Subscription
-from .timed_transactions import TimedTransactions, Transaction
+from .timed_transactions import TimedTransactions
 
 
 class Broker:
@@ -453,18 +453,13 @@ class Broker:
         if cbt.is_response:
             recipient = cbt.response.recipient
 
-        def is_cmplt(x: CBT):
-            return x.is_completed or x.is_expired
-
         with self._nexus_lock:
             nexus = self._nexus_map[recipient]
             nexus.work_queue.put(cbt)
             if cbt.is_pending:
                 initiator = cbt.request.initiator
                 owner = self._nexus_map[initiator]
-                self._timers.register(
-                    Transaction(cbt, is_cmplt, owner.on_cbt_expired, cbt.lifespan)
-                )
+                self._timers.register_dpc(cbt.lifespan, owner.on_cbt_expired, (cbt,))
 
     # Caller is the subscription source
     def publish_subscription(self, publisher_name, subscription_name, publisher):
@@ -518,9 +513,6 @@ class Broker:
         if sub is not None:
             sub.remove_subscriber(sink)
 
-    def register_timed_transaction(self, entry: Transaction):
-        self._timers.register(entry)
-
     def register_dpc(self, delay, call, params=()):
         self._timers.register_dpc(delay, call, params)
 

evio/broker/controller_module.py

@@ -252,12 +252,9 @@ class ControllerModule:
     def end_subscription(self, publisher_name, subscription_name):
         self._nexus.end_subscription(publisher_name, subscription_name)
 
-    def register_timed_transaction(self, obj, is_completed, on_expired, lifespan):
-        if is_completed(obj):
-            raise ValueError(f"Object already marked as completed {obj}")
-        self._nexus.register_timed_transaction(obj, is_completed, on_expired, lifespan)
-
-    def register_deferred_call(self, delay, call, params=()):
+    def register_deferred_call(self, delay, call, params=None):
+        if params is None:
+            params = ()
         self._nexus.register_deferred_call(delay, call, params)
 
     @abstractmethod

evio/broker/nexus.py

@@ -26,7 +26,6 @@ import time
 from . import CM_TIMER_EVENT_INTERVAL, TIMER_EVENT_PERIOD
 from .cbt import CBT
 from .process_proxy import ProxyMsg
-from .timed_transactions import Transaction
 
 
 class Nexus:
@@ -151,16 +150,18 @@ class Nexus:
             finally:
                 self._cm_queue.task_done()
 
-    def on_cbt_expired(self, cbt: CBT, time_expired: float):
+    def on_cbt_expired(self, cbt: CBT):
         """Callback from the TimedTransaction to indicate a CBT has expired.
         The CBT must be in a pending state, ie., it has not already been expired or completed
         """
+        if cbt.is_completed or cbt.is_expired:
+            return
         if cbt.request.initiator != self._controller.name:
             raise RuntimeWarning(
                 f"Invalid Operation: Attemptng to expire a CBT that is not own by this controller {self._controller.name} {cbt}"
             )
         if cbt.is_pending:
-            cbt.time_expired = time_expired
+            cbt.time_expired = time.time()
             self.work_queue.put(cbt)
         else:
             self._controller.logger.info(
@@ -206,16 +207,6 @@ class Nexus:
             publisher_name, subscription_name, self._controller
         )
 
-    def register_timed_transaction(self, obj, is_completed, on_expired, lifespan):
-        self._broker.register_timed_transaction(
-            Transaction(
-                item=obj,
-                is_completed=is_completed,
-                on_expired=on_expired,
-                lifespan=lifespan,
-            )
-        )
-
     def register_deferred_call(self, delay, call, params):
         self._broker.register_dpc(delay, call, params)
 

evio/broker/timed_transactions.py

@@ -1,23 +1,10 @@
 import logging
 import sched
 import threading
-import time
 
 from . import TIMER_EVENT_PERIOD
 
 
-class Transaction:
-    def __init__(self, item, is_completed, on_expired, lifespan) -> None:
-        self.item = item
-        self._is_completed = is_completed
-        self.lifespan = lifespan
-        self.on_expired = on_expired
-        self.priority: int = 10
-
-    def is_completed(self):
-        return self._is_completed(self.item)
-
-
 class TimedTransactions:
     def __init__(self) -> None:
         self._exit_ev = threading.Event()
@@ -28,20 +15,11 @@ class TimedTransactions:
         self._chk_interval = float(TIMER_EVENT_PERIOD)
         self._sched = sched.scheduler()
 
-    def register(self, entry: Transaction):
-        if self._exit_ev.is_set():
-            return
-        self._sched.enter(entry.lifespan, entry.priority, self._get_expired, (entry,))
-
     def register_dpc(self, delay: float, call, params: tuple):
         if self._exit_ev.is_set():
             return
         self._sched.enter(delay, 15, call, params)
 
-    def _get_expired(self, entry):
-        if not entry.is_completed():
-            entry.on_expired(entry.item, time.time())
-
     def _run(self):
         while not self._exit_ev.wait(self._chk_interval):
             try:

evio/controllers/link_manager.py

@@ -164,11 +164,10 @@ class LinkManager(ControllerModule):
                 dataplane=DATAPLANE_TYPES.Tincan,
             )
             self._tunnels[tnlid] = tnl
-            self.register_timed_transaction(
-                tnl,
-                self.is_tnl_online,
-                self.on_tnl_timeout,
+            self.register_deferred_call(
                 LINK_SETUP_TIMEOUT,
+                self.on_tnl_timeout,
+                (tnl,),
             )
             self.logger.debug(
                 "Tunnel %s authorized for peer %s.", tnlid[:7], peer_id[:7]
@@ -663,7 +662,9 @@ class LinkManager(ControllerModule):
                 )
         self.free_cbt(cbt)
 
-    def on_tnl_timeout(self, tnl: Tunnel, timeout: float):
+    def on_tnl_timeout(self, tnl: Tunnel):
+        if tnl.is_tnl_online():
+            return
         self._rollback_link_creation_changes(tnl.tnlid)
 
     def _register_abort_handlers(self):
@@ -725,9 +726,6 @@ class LinkManager(ControllerModule):
         ign_netinf |= self._ignored_net_interfaces[overlay_id]
         return ign_netinf
 
-    def is_tnl_online(self, tnl: Tunnel) -> bool:
-        return tnl.is_tnl_online()
-
     def _remove_link_from_tunnel(self, tnlid):
         tnl = self._tunnels.get(tnlid)
         if tnl:
@@ -860,11 +858,10 @@ class LinkManager(ControllerModule):
         )
         tnl.fpr = None
         if not tnl.is_tnl_online():
-            self.register_timed_transaction(
-                tnl,
-                self.is_tnl_online,
-                self.on_tnl_timeout,
+            self.register_deferred_call(
                 LINK_SETUP_TIMEOUT,
+                self.on_tnl_timeout,
+                (tnl,),
             )
 
     def _complete_link_endpt_request(self, cbt: CBT):

evio/controllers/tincan_tunnel.py

@@ -30,7 +30,7 @@ from copy import deepcopy
 from threading import Event
 
 import broker
-from broker import TC_PRCS_CHK_INTERVAL
+from broker import TC_PRCS_CHK_INTERVAL, TC_REQUEST_TIMEOUT
 from broker.cbt import CBT
 from broker.controller_module import ControllerModule
 from broker.process_proxy import ProxyMsg
@@ -148,11 +148,10 @@ class TincanTunnel(ControllerModule):
             req["IgnoredNetInterfaces"] = msg.get("IgnoredNetInterfaces")
             tc_proc = self._tc_proc_tbl[tnlid]
             self._tnl_cbts[cbt.tag] = cbt
-            self.register_timed_transaction(
-                cbt.tag,
-                self.is_tc_req_cmpl,
+            self.register_deferred_call(
+                TC_REQUEST_TIMEOUT,
                 self.on_tc_req_expire,
-                60,
+                (cbt.tag,),
             )
             self.send_control(tc_proc.ipc_id, json.dumps(ctl))
         except Exception:
@@ -198,11 +197,10 @@ class TincanTunnel(ControllerModule):
             req["IgnoredNetInterfaces"] = msg.get("IgnoredNetInterfaces")
             tc_proc = self._tc_proc_tbl[tnlid]
             self._tnl_cbts[cbt.tag] = cbt
-            self.register_timed_transaction(
-                cbt.tag,
-                self.is_tc_req_cmpl,
+            self.register_deferred_call(
+                TC_REQUEST_TIMEOUT,
                 self.on_tc_req_expire,
-                60,
+                (cbt.tag,),
             )
             self.send_control(tc_proc.ipc_id, json.dumps(ctl))
         except Exception:
@@ -223,11 +221,10 @@ class TincanTunnel(ControllerModule):
             ctl["Request"]["TunnelId"] = tnlid
             tc_proc = self._tc_proc_tbl[tnlid]
             self._tnl_cbts[cbt.tag] = cbt
-            self.register_timed_transaction(
-                cbt.tag,
-                self.is_tc_req_cmpl,
+            self.register_deferred_call(
+                TC_REQUEST_TIMEOUT,
                 self.on_tc_req_expire,
-                60,
+                (cbt.tag,),
             )
             self.send_control(tc_proc.ipc_id, json.dumps(ctl))
         except Exception:
@@ -248,11 +245,10 @@ class TincanTunnel(ControllerModule):
             ctl["TransactionId"] = cbt.tag
             ctl["Request"]["TunnelId"] = tnlid
             self._tnl_cbts[cbt.tag] = cbt
-            self.register_timed_transaction(
-                cbt.tag,
-                self.is_tc_req_cmpl,
+            self.register_deferred_call(
+                TC_REQUEST_TIMEOUT,
                 self.on_tc_req_expire,
-                60,
+                (cbt.tag,),
             )
             self.send_control(tc_proc.ipc_id, json.dumps(ctl))
         except Exception as excep:
@@ -372,7 +368,9 @@ class TincanTunnel(ControllerModule):
         cbt.set_response(rmv, True)
         self.complete_cbt(cbt)
 
-    def on_tc_req_expire(self, tag: int, timeout: float):
+    def on_tc_req_expire(self, tag: int, timeout: float = None):
+        if self.is_tc_req_cmpl(tag):
+            return
         self.logger.info("Tincan request expired %s", tag)
         cbt: CBT = self._tnl_cbts.pop(tag, None)
         if cbt and cbt.is_pending: