From a529ced50861408cb4e0439c8eda41ae6a61b73d Mon Sep 17 00:00:00 2001 From: Ken Date: Thu, 25 Jan 2024 20:28:46 +0000 Subject: [PATCH] Switch to deferred call --- evio/broker/broker.py | 12 ++--------- evio/broker/controller_module.py | 9 +++----- evio/broker/nexus.py | 17 ++++------------ evio/broker/timed_transactions.py | 22 -------------------- evio/controllers/link_manager.py | 21 ++++++++----------- evio/controllers/tincan_tunnel.py | 34 +++++++++++++++---------------- 6 files changed, 34 insertions(+), 81 deletions(-) diff --git a/evio/broker/broker.py b/evio/broker/broker.py index 202a5e0..1a0c719 100644 --- a/evio/broker/broker.py +++ b/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) diff --git a/evio/broker/controller_module.py b/evio/broker/controller_module.py index a29c034..a326b8c 100644 --- a/evio/broker/controller_module.py +++ b/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 diff --git a/evio/broker/nexus.py b/evio/broker/nexus.py index 9a672b5..cae0670 100644 --- a/evio/broker/nexus.py +++ b/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) diff --git a/evio/broker/timed_transactions.py b/evio/broker/timed_transactions.py index 1616d63..cbb9ded 100644 --- a/evio/broker/timed_transactions.py +++ b/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: diff --git a/evio/controllers/link_manager.py b/evio/controllers/link_manager.py index 9513015..6d3cd7c 100644 --- a/evio/controllers/link_manager.py +++ b/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): diff --git a/evio/controllers/tincan_tunnel.py b/evio/controllers/tincan_tunnel.py index 1e7e7c7..e739e33 100644 --- a/evio/controllers/tincan_tunnel.py +++ b/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: