On 12/1/21 18:51, Terry Wilson wrote: > Add support for monitor_cond_since / update3 to python-ovs to > allow more efficient reconnections when connecting to clustered > OVSDB servers. > > Signed-off-by: Terry Wilson <[email protected]> > --- > python/ovs/db/idl.py | 245 ++++++++++++++++++++++++++++++++++++------- > tests/ovsdb-idl.at | 2 +- > 2 files changed, 211 insertions(+), 36 deletions(-) > > diff --git a/python/ovs/db/idl.py b/python/ovs/db/idl.py > index 60e58b03e..0d5e00208 100644 > --- a/python/ovs/db/idl.py > +++ b/python/ovs/db/idl.py > @@ -13,6 +13,7 @@ > # limitations under the License. > > import collections > +import enum > import functools > import uuid > > @@ -36,6 +37,7 @@ ROW_DELETE = "delete" > > OVSDB_UPDATE = 0 > OVSDB_UPDATE2 = 1 > +OVSDB_UPDATE3 = 2 > > CLUSTERED = "clustered" > RELAY = "relay" > @@ -75,6 +77,65 @@ class ColumnDefaultDict(dict): > return item in self.keys() > > > +class Monitor(enum.IntEnum): > + monitor = OVSDB_UPDATE > + monitor_cond = OVSDB_UPDATE2 > + monitor_cond_since = OVSDB_UPDATE3 > + > + > +class ConditionState(object): > + def __init__(self): > + self._ack_cond = None > + self._req_cond = None > + self._new_cond = [True] > + > + def __iter__(self): > + return iter([self._new_cond, self._req_cond, self._ack_cond]) > + > + @property > + def new(self): > + """The latest freshly initialized condition change""" > + return self._new_cond > + > + @property > + def acked(self): > + """The last condition change that has been accepted by the server""" > + return self._ack_cond > + > + @property > + def requested(self): > + """A condition that's been requested, but not acked by the server""" > + return self._req_cond > + > + @property > + def latest(self): > + """The most recent condition change""" > + return next(cond for cond in self if cond is not None) > + > + @staticmethod > + def is_true(condition): > + return condition == [True] > + > + def init(self, cond): > + """Signal that a condition change is being initiated""" > + self._new_cond = cond > + > + def ack(self): > + """Signal that a condition change has been acked""" > + if self._req_cond is not None: > + self._ack_cond, self._req_cond = (self._req_cond, None) > + > + def request(self): > + """Signal that a condition change has been requested""" > + if self._new_cond is not None: > + self._req_cond, self._new_cond = (self._new_cond, None) > + > + def reset(self): > + """Reset a requested condition change back to new""" > + if self._req_cond is not None and self._new_cond is None: > + self._new_cond, self._req_cond = (self._req_cond, None) > + > + > class Idl(object): > """Open vSwitch Database Interface Definition Language (OVSDB IDL). > > @@ -132,7 +193,13 @@ class Idl(object): > IDL_S_SERVER_MONITOR_REQUESTED = 2 > IDL_S_DATA_MONITOR_REQUESTED = 3 > IDL_S_DATA_MONITOR_COND_REQUESTED = 4 > - IDL_S_MONITORING = 5 > + IDL_S_DATA_MONITOR_COND_SINCE_REQUESTED = 5 > + IDL_S_MONITORING = 6 > + > + monitor_map = { > + Monitor.monitor: IDL_S_SERVER_MONITOR_REQUESTED, > + Monitor.monitor_cond: IDL_S_DATA_MONITOR_COND_REQUESTED, > + Monitor.monitor_cond_since: IDL_S_DATA_MONITOR_COND_SINCE_REQUESTED} > > def __init__(self, remote, schema_helper, probe_interval=None, > leader_only=True): > @@ -176,10 +243,12 @@ class Idl(object): > remotes = self._parse_remotes(remote) > self._session = ovs.jsonrpc.Session.open_multiple(remotes, > probe_interval=probe_interval) > + self._request_id = None > self._monitor_request_id = None > self._last_seqno = None > self.change_seqno = 0 > self.uuid = uuid.uuid1() > + self.last_id = str(uuid.UUID(int=0)) > > # Server monitor. > self._server_schema_request_id = None > @@ -206,6 +275,9 @@ class Idl(object): > self.txn = None > self._outstanding_txns = {} > > + self.cond_changed = False > + self.cond_seqno = 0 > + > for table in schema.tables.values(): > for column in table.columns.values(): > if not hasattr(column, 'alert'): > @@ -213,8 +285,7 @@ class Idl(object): > table.need_table = False > table.rows = custom_index.IndexedRows(table) > table.idl = self > - table.condition = [True] > - table.cond_changed = False > + table.condition = ConditionState() > > def _parse_remotes(self, remote): > # If remote is - > @@ -252,6 +323,38 @@ class Idl(object): > update.""" > self._session.close() > > + def ack_conditions(self): > + """Mark all requested table conditions as acked""" > + for table in self.tables.values(): > + table.condition.ack() > + > + def sync_conditions(self): > + """Synchronize condition state when the FSM is restarted > + > + If a non-zero last_id is available for the DB, then upon reconnect > + the IDL should first request acked conditions to avoid missing > updates > + about records that were added before the transaction with > + txn-id == last_id. If there were requested condition changes in > flight > + and the IDL client didn't set new conditions, then reset the > requested > + conditions to new to trigger a follow-up monitor_cond_change request > + """ > + ack_all = self.last_id == str(uuid.UUID(int=0)) > + for table in self.tables.values(): > + if ack_all: > + table.condition.request() > + table.condition.ack() > + else: > + table.condition.reset() > + self.cond_changed = True > + > + def restart_fsm(self): > + # Resync data DB table conditions to avoid missing updated due to > + # conditions that were in flight or changed locally while the > + # connection was down. > + self.sync_conditions() > + self.__send_server_schema_request() > + self.state = self.IDL_S_SERVER_SCHEMA_REQUESTED > + > def run(self): > """Processes a batch of messages from the database server. Returns > True if the database as seen through the IDL changed, False if it did > @@ -286,7 +389,7 @@ class Idl(object): > if seqno != self._last_seqno: > self._last_seqno = seqno > self.__txn_abort_all() > - self.__send_server_schema_request() > + self.restart_fsm() > if self.lock_name: > self.__send_lock_request() > break > @@ -294,8 +397,20 @@ class Idl(object): > msg = self._session.recv() > if msg is None: > break > + is_response = msg.type in (ovs.jsonrpc.Message.T_REPLY, > + ovs.jsonrpc.Message.T_ERROR) > + > + if is_response and self._request_id and self._request_id == > msg.id: > + self._request_id = None > + # process_response follows > > if (msg.type == ovs.jsonrpc.Message.T_NOTIFY > + and msg.method == "update3" > + and len(msg.params) == 3): > + # Database contents changed. > + self.__parse_update(msg.params[2], OVSDB_UPDATE3) > + self.last_id = msg.params[1] > + elif (msg.type == ovs.jsonrpc.Message.T_NOTIFY > and msg.method == "update2" > and len(msg.params) == 2): > # Database contents changed. > @@ -320,11 +435,18 @@ class Idl(object): > try: > self.change_seqno += 1 > self._monitor_request_id = None > - self.__clear() > - if self.state == self.IDL_S_DATA_MONITOR_COND_REQUESTED: > + if (self.state == > + self.IDL_S_DATA_MONITOR_COND_SINCE_REQUESTED): > + # If 'found' is false, clear table rows for new dump > + if not msg.result[0]: > + self.__clear() > + self.__parse_update(msg.result[2], OVSDB_UPDATE3) > + elif self.state == > self.IDL_S_DATA_MONITOR_COND_REQUESTED: > + self.__clear() > self.__parse_update(msg.result, OVSDB_UPDATE2) > else: > assert self.state == > self.IDL_S_DATA_MONITOR_REQUESTED > + self.__clear() > self.__parse_update(msg.result, OVSDB_UPDATE) > self.state = self.IDL_S_MONITORING > > @@ -398,11 +520,17 @@ class Idl(object): > elif msg.type == ovs.jsonrpc.Message.T_NOTIFY and msg.id == > "echo": > # Reply to our echo request. Ignore it. > pass > + elif (msg.type == ovs.jsonrpc.Message.T_ERROR and > + self.state == ( > + self.IDL_S_DATA_MONITOR_COND_SINCE_REQUESTED) and > + self._monitor_request_id == msg.id): > + if msg.error == "unknown method": > + self.__send_monitor_request(Monitor.monitor_cond) > elif (msg.type == ovs.jsonrpc.Message.T_ERROR and > self.state == self.IDL_S_DATA_MONITOR_COND_REQUESTED and > self._monitor_request_id == msg.id): > if msg.error == "unknown method": > - self.__send_monitor_request() > + self.__send_monitor_request(Monitor.monitor) > elif (msg.type == ovs.jsonrpc.Message.T_ERROR and > self._server_schema_request_id is not None and > self._server_schema_request_id == msg.id): > @@ -418,6 +546,13 @@ class Idl(object): > and self.__txn_process_reply(msg)): > # __txn_process_reply() did everything needed. > pass > + elif (msg.type == ovs.jsonrpc.Message.T_REPLY and > + self.state == self.IDL_S_MONITORING): > + # Mark the last requested conditions as acked and if further > + # condition changes were pending, send them now. > + self.ack_conditions() > + self.send_cond_change() > + self.cond_seqno += 1 > else: > # This can happen if a transaction is destroyed before we > # receive the reply, so keep the log level low. > @@ -427,14 +562,36 @@ class Idl(object): > > return initial_change_seqno != self.change_seqno > > - def send_cond_change(self): > - if not self._session.is_connected(): > + def compose_cond_change(self): > + if not self.cond_changed: > return > > + change_requests = {} > for table in self.tables.values(): > - if table.cond_changed: > - self.__send_cond_change(table, table.condition) > - table.cond_changed = False > + # Always use the most recent conditions set by the IDL client > when > + # requesting monitor_cond_change > + if table.condition.new is not None: > + change_requests[table.name] = [ > + {"where": table.condition.new}] > + table.condition.request() > + > + if not change_requests: > + return > + > + self.cond_changed = False > + old_uuid = str(self.uuid) > + self.uuid = uuid.uuid1() > + params = [old_uuid, str(self.uuid), change_requests] > + return ovs.jsonrpc.Message.create_request( > + "monitor_cond_change", params) > + > + def send_cond_change(self): > + if not self._session.is_connected() or self._request_id is not None: > + return > + > + msg = self.compose_cond_change() > + if msg: > + self.send_request(msg) > > def cond_change(self, table_name, cond): > """Sets the condition for 'table_name' to 'cond', which should be a > @@ -450,13 +607,28 @@ class Idl(object): > > if cond == []: > cond = [False] > - if table.condition != cond: > - table.condition = cond > - table.cond_changed = True > + > + # Compare the new condition to the last known condition > + if table.condition.latest != cond: > + table.condition.init(cond) > + self.cond_changed = True > + > + # New condition will be sent out after all already requested ones > + # are acked. > + if table.condition.new: > + any_reqs = any(t.condition.request for t in self.tables.values()) > + return self.cond_seqno + int(any_reqs) + 1 > + > + # Already requested conditions should be up to date at > + # self.cond_seqno + 1 while acked conditions are already up to date > + return self.cond_seqno + int(bool(table.condition.requested)) > > def wait(self, poller): > """Arranges for poller.block() to wake up when self.run() has > something > to do or when activity occurs on a transaction on 'self'.""" > + if self.cond_changed: > + poller.immediate_wake() > + return
Nit: In theory, self._session.wait()/recv_wait() should never do more than register to be awakened when the appropriate events occur, so skipping that should be fine (and seems to be with the current code). But what if that changes in the future? Ilya, what do you think? In any case, this works for me today, with or without that "return" there: Acked-by: Dumitru Ceara <[email protected]> Thanks! Dumitru _______________________________________________ dev mailing list [email protected] https://mail.openvswitch.org/mailman/listinfo/ovs-dev
