ovsdb: Weak references performance fix

Prevents the cloning of rows with outgoing or incoming weak references when
those rows aren't being modified.

It improves the OVSDB Server performance when many rows with weak references
are involved in a transaction.

In the original code (dst_refs is created from scratch):

old->dst_refs = all the rows that weak referenced old

new->dst_refs = all the rows that weak referenced old and are still weak
+referencing new + rows in the transaction that weak referenced new

In the patch (dst_refs incrementally built):
Old->dst_refs = all the rows that weak referenced old

Ideally, but expansive to calculate:
New->dst_refs = old->dst_refs - "weak references removed within this TXN" +
+"weak references created within this TXN"

What this patch implements:
New->dst_refs = old->dst_refs - "weak references in old rows in TXN" + "weak
+references in new rows in TXN"

The resulting sets should be equal in both cases.

We do some more optimizations:

- If we know that the transactions must be successful at some point then,
  instead of cloning dst_refs we could just move the elements between
  the lists.

- At that point we lost the rollback feature, but we aren't going to need
  it anyway (note that we didn't really touch the src_refs part).

- The references in dst_refs must point to new instead than old.
  Previously we iterated over all the weak references in dst_refs
  to change that pointer, but using an UUID is easier, and prevents
  that iteration completely.

For some more commentary, see:
http://openvswitch.org/pipermail/dev/2016-July/074840.html

Signed-off-by: Esteban Rodriguez Betancourt <estebarb@hpe.com>
Signed-off-by: Ben Pfaff <blp@ovn.org>

include/openvswitch/list.h

@@ -288,4 +288,11 @@ ovs_list_is_short(const struct ovs_list *list)
     return list->next == list->prev;
 }
 
+/* Transplant a list into another, and resets the origin list */
+static inline void
+ovs_list_push_back_all(struct ovs_list *dst, struct ovs_list *src)
+{
+    ovs_list_splice(dst, src->next, src);
+}
+
 #endif /* list.h */

ovsdb/row.h

@@ -36,9 +36,11 @@ struct ovsdb_column_set;
  * ovsdb_weak_ref" structures are created for them.
  */
 struct ovsdb_weak_ref {
-    struct ovs_list src_node;   /* In src->src_refs list. */
-    struct ovs_list dst_node;   /* In destination row's dst_refs list. */
-    struct ovsdb_row *src;      /* Source row. */
+    struct ovs_list src_node;      /* In src->src_refs list. */
+    struct ovs_list dst_node;      /* In destination row's dst_refs list. */
+    struct ovsdb_row *src;         /* Source row. */
+    struct ovsdb_table *dst_table; /* Destination table. */
+    struct uuid dst;               /* Destination row uuid. */
 };
 
 /* A row in a database table. */

ovsdb/transaction.c

@@ -436,9 +436,48 @@ ovsdb_txn_row_commit(struct ovsdb_txn *txn OVS_UNUSED,
     return NULL;
 }
 
+static struct ovsdb_error *
+ovsdb_txn_update_weak_refs(struct ovsdb_txn *txn OVS_UNUSED,
+                           struct ovsdb_txn_row *txn_row)
+{
+    struct ovsdb_weak_ref *weak, *next;
+
+    /* Remove the weak references originating in the old version of the row. */
+    if (txn_row->old) {
+        LIST_FOR_EACH_SAFE (weak, next, src_node, &txn_row->old->src_refs) {
+            ovs_list_remove(&weak->src_node);
+            ovs_list_remove(&weak->dst_node);
+            free(weak);
+        }
+    }
+
+    /* Although the originating rows have the responsibility of updating the
+     * weak references in the dst, it is possible that some source rows aren't
+     * part of the transaction.  In that situation this row needs to move the
+     * list of incoming weak references from the old row into the new one.
+     */
+    if (txn_row->old && txn_row->new) {
+        /* Move the incoming weak references from old to new. */
+        ovs_list_push_back_all(&txn_row->new->dst_refs,
+                               &txn_row->old->dst_refs);
+    }
+
+    /* Insert the weak references originating in the new version of the row. */
+    struct ovsdb_row *dst_row;
+    if (txn_row->new) {
+        LIST_FOR_EACH (weak, src_node, &txn_row->new->src_refs) {
+            /* dst_row MUST exist. */
+            dst_row = CONST_CAST(struct ovsdb_row *,
+                    ovsdb_table_get_row(weak->dst_table, &weak->dst));
+            ovs_list_insert(&dst_row->dst_refs, &weak->dst_node);
+        }
+    }
+
+    return NULL;
+}
+
 static void
-add_weak_ref(struct ovsdb_txn *txn,
-             const struct ovsdb_row *src_, const struct ovsdb_row *dst_)
+add_weak_ref(const struct ovsdb_row *src_, const struct ovsdb_row *dst_)
 {
     struct ovsdb_row *src = CONST_CAST(struct ovsdb_row *, src_);
     struct ovsdb_row *dst = CONST_CAST(struct ovsdb_row *, dst_);
@@ -448,8 +487,6 @@ add_weak_ref(struct ovsdb_txn *txn,
         return;
     }
 
-    dst = ovsdb_txn_row_modify(txn, dst);
-
     if (!ovs_list_is_empty(&dst->dst_refs)) {
         /* Omit duplicates. */
         weak = CONTAINER_OF(ovs_list_back(&dst->dst_refs),
@@ -461,7 +498,10 @@ add_weak_ref(struct ovsdb_txn *txn,
 
     weak = xmalloc(sizeof *weak);
     weak->src = src;
-    ovs_list_push_back(&dst->dst_refs, &weak->dst_node);
+    weak->dst_table = dst->table;
+    weak->dst = *ovsdb_row_get_uuid(dst);
+    /* The dst_refs list is updated at commit time. */
+    ovs_list_init(&weak->dst_node);
     ovs_list_push_back(&src->src_refs, &weak->src_node);
 }
 
@@ -471,7 +511,7 @@ assess_weak_refs(struct ovsdb_txn *txn, struct ovsdb_txn_row *txn_row)
     struct ovsdb_table *table;
     struct shash_node *node;
 
-    if (txn_row->old) {
+    if (txn_row->old && !txn_row->new) {
         /* Mark rows that have weak references to 'txn_row' as modified, so
          * that their weak references will get reassessed. */
         struct ovsdb_weak_ref *weak, *next;
@@ -506,7 +546,7 @@ assess_weak_refs(struct ovsdb_txn *txn, struct ovsdb_txn_row *txn_row)
                 row = ovsdb_table_get_row(column->type.key.u.uuid.refTable,
                                           &datum->keys[i].uuid);
                 if (row) {
-                    add_weak_ref(txn, txn_row->new, row);
+                    add_weak_ref(txn_row->new, row);
                     i++;
                 } else {
                     if (uuid_is_zero(&datum->keys[i].uuid)) {
@@ -524,7 +564,7 @@ assess_weak_refs(struct ovsdb_txn *txn, struct ovsdb_txn_row *txn_row)
                 row = ovsdb_table_get_row(column->type.value.u.uuid.refTable,
                                           &datum->values[i].uuid);
                 if (row) {
-                    add_weak_ref(txn, txn_row->new, row);
+                    add_weak_ref(txn_row->new, row);
                     i++;
                 } else {
                     if (uuid_is_zero(&datum->values[i].uuid)) {
@@ -838,6 +878,7 @@ ovsdb_txn_commit_(struct ovsdb_txn *txn, bool durable)
 
     /* Finalize commit. */
     txn->db->run_triggers = true;
+    ovsdb_error_assert(for_each_txn_row(txn, ovsdb_txn_update_weak_refs));
     ovsdb_error_assert(for_each_txn_row(txn, ovsdb_txn_row_commit));
     ovsdb_txn_free(txn);
 

tests/library.at

@@ -43,8 +43,8 @@ AT_SETUP([atomic operations])
 AT_CHECK([ovstest test-atomic])
 AT_CLEANUP
 
-AT_SETUP([linked lists])
-AT_CHECK([ovstest test-list], [0], [...
+AT_SETUP([test linked lists])
+AT_CHECK([ovstest test-list], [0], [....
 ])
 AT_CLEANUP
 

tests/test-list.c

@@ -185,6 +185,46 @@ test_list_for_each_pop(void)
     }
 }
 
+/* Tests the transplant of one list into another  */
+static void
+test_list_push_back_all(void)
+{
+    struct ovs_list list_a, list_b;
+    struct element a, b, c, d;
+
+    a.value = 0;
+    b.value = 1;
+    c.value = 2;
+    d.value = 3;
+
+    ovs_list_init(&list_a);
+    ovs_list_init(&list_b);
+
+    ovs_list_insert(&list_a, &a.node);
+    ovs_list_insert(&list_a, &b.node);
+    ovs_list_insert(&list_b, &c.node);
+    ovs_list_insert(&list_b, &d.node);
+
+    /* Check test preconditions */
+    assert(2 == ovs_list_size(&list_a));
+    assert(2 == ovs_list_size(&list_b));
+
+    /* Perform transplant */
+    ovs_list_push_back_all(&list_a, &list_b);
+
+    /* Check expected result */
+    assert(4 == ovs_list_size(&list_a));
+    assert(0 == ovs_list_size(&list_b));
+
+    struct element *node;
+    int n = 0;
+    LIST_FOR_EACH(node, node, &list_a) {
+        assert(n == node->value);
+        n++;
+    }
+    assert(n == 4);
+}
+
 static void
 run_test(void (*function)(void))
 {
@@ -198,6 +238,7 @@ test_list_main(int argc OVS_UNUSED, char *argv[] OVS_UNUSED)
     run_test(test_list_construction);
     run_test(test_list_for_each_safe);
     run_test(test_list_for_each_pop);
+    run_test(test_list_push_back_all);
     printf("\n");
 }