ovsdb-idl: fix index row setting with references.

IDL index should be able to be used without having to be in a
transaction. However, current implementation leads to crash if
a reference type column is being set in an index row for querying
purpose when it is not in a transaction. It is because of the
uninitialized arcs and unnecessary updates of the arcs. This patch
fixes it by identifying index rows by a magic uuid, so that when
parsing index row, the arcs are not updated. A new test case is
added to cover this scenario.

Signed-off-by: Han Zhou <zhouhan@gmail.com>
Signed-off-by: Ben Pfaff <blp@ovn.org>

lib/ovsdb-idl.c

@@ -2298,6 +2298,20 @@ ovsdb_idl_index_write_(struct ovsdb_idl_row *const_row,
     (column->parse)(row, &row->new_datum[column_idx]);
 }
 
+/* Magic UUID for index rows */
+static const struct uuid index_row_uuid = {
+        .parts = {0xdeadbeef,
+                  0xdeadbeef,
+                  0xdeadbeef,
+                  0xdeadbeef}};
+
+/* Check if a row is an index row */
+static bool
+is_index_row(struct ovsdb_idl_row *row)
+{
+    return uuid_equals(&row->uuid, &index_row_uuid);
+}
+
 /* Initializes a row for use in an indexed query.
  * Not intended for direct usage.
  */
@@ -2307,9 +2321,13 @@ ovsdb_idl_index_init_row(struct ovsdb_idl * idl,
 {
     struct ovsdb_idl_row *row = xzalloc(class->allocation_size);
     class->row_init(row);
+    row->uuid = index_row_uuid;
     row->new_datum = xmalloc(class->n_columns * sizeof *row->new_datum);
     row->written = bitmap_allocate(class->n_columns);
     row->table = ovsdb_idl_table_from_class(idl, class);
+    /* arcs are not used for index row, but it doesn't harm to initialize */
+    ovs_list_init(&row->src_arcs);
+    ovs_list_init(&row->dst_arcs);
     return row;
 }
 
@@ -2325,6 +2343,8 @@ ovsdb_idl_index_destroy_row__(const struct ovsdb_idl_row *row_)
     const struct ovsdb_idl_column *c;
     size_t i;
 
+    ovs_assert(ovs_list_is_empty(&row_->src_arcs));
+    ovs_assert(ovs_list_is_empty(&row_->dst_arcs));
     BITMAP_FOR_EACH_1 (i, class->n_columns, row->written) {
         c = &class->columns[i];
         (c->unparse) (row);
@@ -2726,13 +2746,17 @@ ovsdb_idl_get_row_arc(struct ovsdb_idl_row *src,
 
     dst_table = ovsdb_idl_table_from_class(idl, dst_table_class);
     dst = ovsdb_idl_get_row(dst_table, dst_uuid);
-    if (idl->txn) {
+    if (idl->txn || is_index_row(src)) {
-        /* We're being called from ovsdb_idl_txn_write().  We must not update
+        /* There are two cases we should not update any arcs:
+         *
+         * 1. We're being called from ovsdb_idl_txn_write(). We must not update
          * any arcs, because the transaction will be backed out at commit or
          * abort time and we don't want our graph screwed up.
          *
-         * Just return the destination row, if there is one and it has not been
+         * 2. The row is used as an index for querying purpose only.
-         * deleted. */
+         *
+         * In these cases, just return the destination row, if there is one and
+         * it has not been deleted. */
         if (dst && (hmap_node_is_null(&dst->txn_node) || dst->new_datum)) {
             return dst;
         }

tests/idltest.ovsschema

@@ -34,7 +34,8 @@
             "min": 0
           }
         }
-      }
+      },
+      "isRoot" : true
     },
     "link2": {
       "columns": {
@@ -50,7 +51,8 @@
             "min": 0
           }
         }
-      }
+      },
+      "isRoot" : true
     },
     "simple": {
       "columns": {
@@ -104,7 +106,8 @@
             "min": 0
           }
         }
-      }
+      },
+      "isRoot" : true
     },
     "simple2" : {
       "columns" : {
@@ -133,7 +136,8 @@
             "max": "unlimited"
           }
         }
-      }
+      },
+      "isRoot" : true
     },
     "simple3" : {
       "columns" : {
@@ -156,14 +160,16 @@
              "max": "unlimited"
           }
         }
-      }
+      },
+      "isRoot" : true
     },
     "simple4" : {
       "columns" : {
         "name" : {
           "type": "string"
         }
-      }
+      },
+      "isRoot" : false
     }
   }
 }

tests/ovsdb-idl.at

@@ -1596,3 +1596,29 @@ OVSDB_CHECK_IDL_COMPOUND_INDEX_DOUBLE_COLUMN_C([Compound_index, double column te
 006: i=4 s=List001 b=True r=130.000000
 006: i=5 s=List005 b=True r=130.000000
 ])
+
+m4_define([OVSDB_CHECK_IDL_COMPOUND_INDEX_WITH_REF],
+  [AT_SETUP([$1 - C])
+   AT_KEYWORDS([ovsdb server idl compound_index compound_index_with_ref positive $5])
+   AT_CHECK([ovsdb_start_idltest])
+   m4_if([$2], [], [],
+     [AT_CHECK([ovsdb-client transact unix:socket $2], [0], [ignore], [ignore])])
+   AT_CHECK([test-ovsdb '-vPATTERN:console:test-ovsdb|%c|%m' -vjsonrpc -t10 -c idl-compound-index-with-ref unix:socket $3],
+            [0], [stdout], [ignore])
+   AT_CHECK([sort stdout | ${PERL} $srcdir/uuidfilt.pl]m4_if([$6],,, [[| $6]]),
+            [0], [$4])
+   OVSDB_SERVER_SHUTDOWN
+   AT_CLEANUP])
+
+OVSDB_CHECK_IDL_COMPOUND_INDEX_WITH_REF([set, simple3 idl-compound-index-with-ref, initially populated],
+[],
+[],
+[[000: After add to other table + set of strong ref
+001: name= uset=[] uref=[[<0>]]
+002: check simple4: not empty
+003: Query using index with reference
+004: name= uset=[] uref=[[<0>]]
+005: After delete
+007: check simple4: empty
+008: End test
+]])

tests/test-ovsdb.c

@@ -2661,6 +2661,87 @@ do_idl_partial_update_set_column(struct ovs_cmdl_context *ctx)
     return;
 }
 
+static void
+do_idl_compound_index_with_ref(struct ovs_cmdl_context *ctx)
+{
+    struct ovsdb_idl *idl;
+    struct ovsdb_idl_txn *myTxn;
+    const struct idltest_simple3 *myRow;
+    struct idltest_simple4 *myRow2;
+    const struct ovsdb_datum *uset OVS_UNUSED;
+    const struct ovsdb_datum *uref OVS_UNUSED;
+    struct ovsdb_idl_index_cursor cursor;
+    int step = 0;
+
+    idl = ovsdb_idl_create(ctx->argv[1], &idltest_idl_class, false, true);
+    ovsdb_idl_add_table(idl, &idltest_table_simple3);
+    ovsdb_idl_add_column(idl, &idltest_simple3_col_name);
+    ovsdb_idl_add_column(idl, &idltest_simple3_col_uset);
+    ovsdb_idl_add_column(idl, &idltest_simple3_col_uref);
+    ovsdb_idl_add_table(idl, &idltest_table_simple4);
+    ovsdb_idl_add_column(idl, &idltest_simple4_col_name);
+
+    struct ovsdb_idl_index *index;
+    index = ovsdb_idl_create_index(idl, &idltest_table_simple3, "uref");
+    ovsdb_idl_index_add_column(index, &idltest_simple3_col_uref,
+                               OVSDB_INDEX_ASC, NULL);
+
+    ovsdb_idl_get_initial_snapshot(idl);
+
+    ovsdb_idl_initialize_cursor(idl, &idltest_table_simple3, "uref",
+                                &cursor);
+
+    setvbuf(stdout, NULL, _IONBF, 0);
+    ovsdb_idl_run(idl);
+
+    /* Adds to a table and update a strong reference in another table. */
+    myTxn = ovsdb_idl_txn_create(idl);
+    myRow = idltest_simple3_insert(myTxn);
+    myRow2 = idltest_simple4_insert(myTxn);
+    idltest_simple4_set_name(myRow2, "test");
+    idltest_simple3_update_uref_addvalue(myRow, myRow2);
+    ovsdb_idl_txn_commit_block(myTxn);
+    ovsdb_idl_txn_destroy(myTxn);
+    ovsdb_idl_get_initial_snapshot(idl);
+    printf("%03d: After add to other table + set of strong ref\n", step++);
+    dump_simple3(idl, myRow, step++);
+
+    myRow2 = (struct idltest_simple4 *) idltest_simple4_first(idl);
+    printf("%03d: check simple4: %s\n", step++,
+           myRow2 ? "not empty" : "empty");
+
+    /* Use index to query the row with reference */
+
+    struct idltest_simple3 *equal;
+    equal = idltest_simple3_index_init_row(idl, &idltest_table_simple3);
+    myRow2 = (struct idltest_simple4 *) idltest_simple4_first(idl);
+    idltest_simple3_index_set_uref(equal, &myRow2, 1);
+    printf("%03d: Query using index with reference\n", step++);
+    IDLTEST_SIMPLE3_FOR_EACH_EQUAL (myRow, &cursor, equal) {
+        print_idl_row_simple3(myRow, step++);
+    }
+    idltest_simple3_index_destroy_row(equal);
+
+    /* Delete the row with reference */
+    myTxn = ovsdb_idl_txn_create(idl);
+    myRow = idltest_simple3_first(idl);
+    idltest_simple3_delete(myRow);
+    ovsdb_idl_txn_commit_block(myTxn);
+    ovsdb_idl_txn_destroy(myTxn);
+    ovsdb_idl_get_initial_snapshot(idl);
+    printf("%03d: After delete\n", step++);
+    dump_simple3(idl, myRow, step++);
+
+    myRow2 = (struct idltest_simple4 *) idltest_simple4_first(idl);
+    printf("%03d: check simple4: %s\n", step++,
+           myRow2 ? "not empty" : "empty");
+
+    ovsdb_idl_destroy(idl);
+    printf("%03d: End test\n", step);
+    return;
+}
+
+
 static int
 test_idl_compound_index_single_column(struct ovsdb_idl *idl,
         struct ovsdb_idl_index_cursor *sCursor,
@@ -2962,6 +3043,8 @@ static struct ovs_cmdl_command all_commands[] = {
     { "trigger", NULL, 2, INT_MAX, do_trigger, OVS_RO },
     { "idl", NULL, 1, INT_MAX, do_idl, OVS_RO },
     { "idl-compound-index", NULL, 2, 2, do_idl_compound_index, OVS_RW },
+    { "idl-compound-index-with-ref", NULL, 1, INT_MAX,
+        do_idl_compound_index_with_ref, OVS_RO },
     { "idl-partial-update-map-column", NULL, 1, INT_MAX,
         do_idl_partial_update_map_column, OVS_RO },
     { "idl-partial-update-set-column", NULL, 1, INT_MAX,