On Fri, Dec 3, 2021 at 6:06 AM Dumitru Ceara <[email protected]> wrote: > > 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?
The C code does a poll_immediate_wake() if cond_changed in ovsdb_cs_db_set_condition() which is cond_change() here. From what I could tell, the C poll code has a poller per thread, whereas the pattern in the Python IDL is that we create a Poller() object and pass it around. Since the Transaction code is the code creating the Poller object, and calling idl.wait(), this seemed to be the only way to wake up. I put the return in because it seemed that if you are immediately waking upon the next block() call, that the other wakeups would be redundant. > 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
