ovsdb-idl: Add loop_run until timeout or completion.

This change makes the ovsdb_idl_loop_run_completion() function from the
northd code in OVN usable for other parts of the code and adapts the
client synchronisation layer functions accordingly.

It allows users to continue processing idl messages until some timeout
arrives or no more messages exist.

Signed-off-by: Martin Morgenstern <martin.morgenstern@cloudandheat.com>
Signed-off-by: Felix Huettner <felix.huettner@stackit.cloud>
(cherry picked from commit d00299c)

Author: felixhuettner

lib/ovsdb-cs.c

@@ -611,12 +611,12 @@ ovsdb_cs_db_add_event(struct ovsdb_cs_db *db, enum ovsdb_cs_event_type type)
  *
  * Initializes 'events' with a list of events that occurred on 'cs'.  The
  * caller must process and destroy all of the events. */
-void
+int
 ovsdb_cs_run(struct ovsdb_cs *cs, struct ovs_list *events)
 {
     ovs_list_init(events);
     if (!cs->session) {
-        return;
+        return EINVAL;
     }
 
     ovsdb_cs_send_cond_change(cs);
@@ -660,6 +660,12 @@ ovsdb_cs_run(struct ovsdb_cs *cs, struct ovs_list *events)
         }
     }
     ovs_list_push_back_all(events, &cs->data.events);
+
+    if (ret == EAGAIN) {
+        return EAGAIN;
+    } else {
+        return 0;
+    }
 }
 
 /* Arranges for poll_block() to wake up when ovsdb_cs_run() has something to

lib/ovsdb-cs.h

@@ -118,7 +118,7 @@ struct ovsdb_cs *ovsdb_cs_create(const char *database, int max_version,
                                  void *ops_aux);
 void ovsdb_cs_destroy(struct ovsdb_cs *);
 
-void ovsdb_cs_run(struct ovsdb_cs *, struct ovs_list *events);
+int ovsdb_cs_run(struct ovsdb_cs *, struct ovs_list *events);
 void ovsdb_cs_wait(struct ovsdb_cs *);
 
 /* Network connection. */

lib/ovsdb-idl.c

@@ -45,6 +45,7 @@
 #include "simap.h"
 #include "sset.h"
 #include "svec.h"
+#include "timeval.h"
 #include "util.h"
 #include "uuid.h"
 #include "openvswitch/vlog.h"
@@ -439,13 +440,14 @@ ovsdb_idl_clear(struct ovsdb_idl *db)
 /* Processes a batch of messages from the database server on 'idl'.  This may
  * cause the IDL's contents to change.  The client may check for that with
  * ovsdb_idl_get_seqno(). */
-void
+int
 ovsdb_idl_run(struct ovsdb_idl *idl)
 {
     ovs_assert(!idl->txn);
 
     struct ovs_list events;
-    ovsdb_cs_run(idl->cs, &events);
+    int ret;
+    ret = ovsdb_cs_run(idl->cs, &events);
 
     struct ovsdb_cs_event *event;
     LIST_FOR_EACH_POP (event, list_node, &events) {
@@ -479,6 +481,8 @@ ovsdb_idl_run(struct ovsdb_idl *idl)
     ovsdb_idl_reparse_refs_to_inserted(idl);
     ovsdb_idl_reparse_deleted(idl);
     ovsdb_idl_row_destroy_postprocess(idl);
+
+    return ret;
 }
 
 /* Arranges for poll_block() to wake up when ovsdb_idl_run() has something to
@@ -4391,6 +4395,44 @@ ovsdb_idl_loop_run(struct ovsdb_idl_loop *loop)
     return loop->open_txn;
 }
 
+/* Run the ovsdb_idl_loop until there is no more data to be received.
+ * A timeout in milliseconds can be provided. The actual duration of the
+ * function is returned in the duration parameter. */
+struct ovsdb_idl_txn *
+ovsdb_idl_loop_run_completion(struct ovsdb_idl_loop *idl_loop,
+                              unsigned long long timeout,
+                              uint64_t *idl_duration)
+{
+    unsigned long long duration, start = time_msec();
+    struct ovsdb_idl_txn *txn;
+    int n = 0;
+
+    /* Accumulate database changes as long as there are some,
+     * but no longer than the given timeout. */
+    while (time_msec() - start < timeout) {
+        if (ovsdb_idl_run(idl_loop->idl) == EAGAIN) {
+            break;
+        }
+        n++;
+    }
+
+    txn = ovsdb_idl_loop_run(idl_loop);
+
+    duration = time_msec() - start;
+    if (idl_duration) {
+        *idl_duration = duration;
+    }
+    /* ovsdb_idl_run() is called at least 2 times.  Once directly and
+     * once in the ovsdb_idl_loop_run().  n > 2 means that we received
+     * data on at least 2 subsequent calls. */
+    if (n > 2 || duration > 100) {
+        VLOG(duration > timeout ? VLL_INFO : VLL_DBG,
+             "%d iterations in %lld ms", n + 1, duration);
+    }
+
+    return txn;
+}
+
 /* Attempts to commit the current transaction, if one is open.
  *
  * If a transaction was open, in this or a previous iteration of the main loop,

lib/ovsdb-idl.h

@@ -73,7 +73,7 @@ void ovsdb_idl_destroy(struct ovsdb_idl *);
 
 void ovsdb_idl_set_leader_only(struct ovsdb_idl *, bool leader_only);
 
-void ovsdb_idl_run(struct ovsdb_idl *);
+int ovsdb_idl_run(struct ovsdb_idl *);
 void ovsdb_idl_wait(struct ovsdb_idl *);
 
 void ovsdb_idl_get_memory_usage(struct ovsdb_idl *, struct simap *usage);
@@ -409,6 +409,9 @@ struct ovsdb_idl_loop {
 
 void ovsdb_idl_loop_destroy(struct ovsdb_idl_loop *);
 struct ovsdb_idl_txn *ovsdb_idl_loop_run(struct ovsdb_idl_loop *);
+struct ovsdb_idl_txn *ovsdb_idl_loop_run_completion(
+    struct ovsdb_idl_loop *idl_loop, unsigned long long timeout,
+    uint64_t *idl_duration);
 int ovsdb_idl_loop_commit_and_wait(struct ovsdb_idl_loop *);
 
 /* Conditional Replication