[ovs-dev] [gc 11/13] ovsdb: Provide a way for for_each_txn_row() callback to delete any row.

[Ben Pfaff]

for_each_txn_row() restricts the txn_rows that its callback may delete.
Until now, this has meant that its callback could not delete any rows
that were created within the transaction being processed.  These rows have
txn_rows with null 'old' and nonnull 'new', so to delete them requires
either removing the txn_row entirely (forbidden by for_each_txn_row()) or
clearing its 'new' to null.  The latter is forbidden because a txn_row
is not allowed to have both 'old' and 'new' null.

Until now, this has not been a significant restriction, because none of
the processing at transaction commit time required deleting arbitrary rows.
Implementing garbage collection, however, does require this ability, so
this commit makes it possible by eliminating the requirement that at least
'old' or 'new' be nonnull.
---
 ovsdb/transaction.c |   46 ++++++++++++++++++++++++++++------------------
 1 files changed, 28 insertions(+), 18 deletions(-)

diff --git a/ovsdb/transaction.c b/ovsdb/transaction.c
index b7c57e5..f67018b 100644
--- a/ovsdb/transaction.c
+++ b/ovsdb/transaction.c
@@ -57,9 +57,11 @@ struct ovsdb_txn_table {
  *
  *      - A row modified by a transaction will have non-null 'old' and 'new'.
  *
- *      - 'old' and 'new' both null is invalid.  It would indicate that a row
- *        was added then deleted within a single transaction, but we instead
- *        handle that case by deleting the txn_row entirely.
+ *      - 'old' and 'new' both null indicates that a row was added then deleted
+ *        within a single transaction.  Most of the time we instead delete the
+ *        ovsdb_txn_row entirely, but inside a for_each_txn_row() callback
+ *        there are restrictions that sometimes mean we have to leave the
+ *        ovsdb_txn_row in place.
  */
 struct ovsdb_txn_row {
     struct hmap_node hmap_node; /* In ovsdb_txn_table's txn_rows hmap. */
@@ -67,6 +69,12 @@ struct ovsdb_txn_row {
     struct ovsdb_row *new;      /* The new row. */
     size_t n_refs;              /* Number of remaining references. */
 
+    /* These members are the same as the corresponding members of 'old' or
+     * 'new'.  They are present here for convenience and because occasionally
+     * there can be an ovsdb_txn_row where both 'old' and 'new' are NULL. */
+    struct uuid uuid;
+    struct ovsdb_table *table;
+
     /* Used by for_each_txn_row(). */
     unsigned int serial;        /* Serial number of in-progress commit. */
 
@@ -110,7 +118,9 @@ ovsdb_txn_row_abort(struct ovsdb_txn *txn OVS_UNUSED,
 
     ovsdb_txn_row_prefree(txn_row);
     if (!old) {
-        hmap_remove(&new->table->rows, &new->hmap_node);
+        if (new) {
+            hmap_remove(&new->table->rows, &new->hmap_node);
+        }
     } else if (!new) {
         hmap_insert(&old->table->rows, &old->hmap_node, ovsdb_row_hash(old));
     } else {
@@ -140,10 +150,7 @@ find_txn_row(const struct ovsdb_table *table, const struct uuid *uuid)
 
     HMAP_FOR_EACH_WITH_HASH (txn_row, hmap_node,
                              uuid_hash(uuid), &table->txn_table->txn_rows) {
-        const struct ovsdb_row *row;
-
-        row = txn_row->old ? txn_row->old : txn_row->new;
-        if (uuid_equals(uuid, ovsdb_row_get_uuid(row))) {
+        if (uuid_equals(uuid, &txn_row->uuid)) {
             return txn_row;
         }
     }
@@ -208,7 +215,7 @@ ovsdb_txn_adjust_row_refs(struct ovsdb_txn *txn, const struct ovsdb_row *r,
 static struct ovsdb_error * WARN_UNUSED_RESULT
 update_row_ref_count(struct ovsdb_txn *txn, struct ovsdb_txn_row *r)
 {
-    struct ovsdb_table *table = r->old ? r->old->table : r->new->table;
+    struct ovsdb_table *table = r->table;
     struct shash_node *node;
 
     SHASH_FOR_EACH (node, &table->schema->columns) {
@@ -241,8 +248,7 @@ check_ref_count(struct ovsdb_txn *txn OVS_UNUSED, struct ovsdb_txn_row *r)
         return ovsdb_error("referential integrity violation",
                            "cannot delete %s row "UUID_FMT" because "
                            "of %zu remaining reference(s)",
-                           r->old->table->schema->name,
-                           UUID_ARGS(ovsdb_row_get_uuid(r->old)),
+                           r->table->schema->name, UUID_ARGS(&r->uuid),
                            r->n_refs);
     }
 }
@@ -328,7 +334,7 @@ assess_weak_refs(struct ovsdb_txn *txn, struct ovsdb_txn_row *txn_row)
         return NULL;
     }
 
-    table = txn_row->new->table;
+    table = txn_row->table;
     SHASH_FOR_EACH (node, &table->schema->columns) {
         const struct ovsdb_column *column = node->data;
         struct ovsdb_datum *datum = &txn_row->new->fields[column->index];
@@ -412,9 +418,8 @@ assess_weak_refs(struct ovsdb_txn *txn, struct ovsdb_txn_row *txn_row)
 static struct ovsdb_error * WARN_UNUSED_RESULT
 determine_changes(struct ovsdb_txn *txn, struct ovsdb_txn_row *txn_row)
 {
-    struct ovsdb_table *table;
+    struct ovsdb_table *table = txn_row->table;
 
-    table = (txn_row->old ? txn_row->old : txn_row->new)->table;
     if (txn_row->old && txn_row->new) {
         struct shash_node *node;
         bool changed = false;
@@ -534,7 +539,7 @@ ovsdb_txn_for_each_change(const struct ovsdb_txn *txn,
 
     LIST_FOR_EACH (t, node, &txn->txn_tables) {
         HMAP_FOR_EACH (r, hmap_node, &t->txn_rows) {
-            if (!cb(r->old, r->new, r->changed, aux)) {
+            if ((r->old || r->new) && !cb(r->old, r->new, r->changed, aux)) {
                 break;
             }
         }
@@ -560,6 +565,7 @@ static struct ovsdb_txn_row *
 ovsdb_txn_row_create(struct ovsdb_txn *txn, struct ovsdb_table *table,
                      const struct ovsdb_row *old_, struct ovsdb_row *new)
 {
+    const struct ovsdb_row *row = old_ ? old_ : new;
     struct ovsdb_row *old = (struct ovsdb_row *) old_;
     size_t n_columns = shash_count(&table->schema->columns);
     struct ovsdb_txn_table *txn_table;
@@ -567,7 +573,9 @@ ovsdb_txn_row_create(struct ovsdb_txn *txn, struct ovsdb_table *table,
 
     txn_row = xzalloc(offsetof(struct ovsdb_txn_row, changed)
                       + bitmap_n_bytes(n_columns));
-    txn_row->old = (struct ovsdb_row *) old;
+    txn_row->uuid = *ovsdb_row_get_uuid(row);
+    txn_row->table = row->table;
+    txn_row->old = old;
     txn_row->new = new;
     txn_row->n_refs = old ? old->n_refs : 0;
     txn_row->serial = serial - 1;
@@ -663,8 +671,7 @@ ovsdb_txn_get_comment(const struct ovsdb_txn *txn)
 static void
 ovsdb_txn_row_prefree(struct ovsdb_txn_row *txn_row)
 {
-    struct ovsdb_row *row = txn_row->old ? txn_row->old : txn_row->new;
-    struct ovsdb_txn_table *txn_table = row->table->txn_table;
+    struct ovsdb_txn_table *txn_table = txn_row->table->txn_table;
 
     txn_table->n_processed--;
     hmap_remove(&txn_table->txn_rows, &txn_row->hmap_node);
@@ -697,6 +704,9 @@ ovsdb_txn_table_destroy(struct ovsdb_txn_table *txn_table)
  * in within the same txn_table.  It may *not* delete any txn_tables.  As long
  * as these rules are followed, 'cb' will be called exactly once for each
  * txn_row in 'txn', even those added by 'cb'.
+ *
+ * (Even though 'cb' is not allowed to delete some txn_rows, it can still
+ * delete any actual row by clearing a txn_row's 'new' member.)
  */
 static struct ovsdb_error * WARN_UNUSED_RESULT
 for_each_txn_row(struct ovsdb_txn *txn,
-- 
1.7.1