ovsdb/transaction.c

   1 /* Copyright (c) 2009, 2010, 2011, 2012, 2013 Nicira, Inc.
   2  *
   3  * Licensed under the Apache License, Version 2.0 (the "License");
   4  * you may not use this file except in compliance with the License.
   5  * You may obtain a copy of the License at:
   6  *
   7  *     http://www.apache.org/licenses/LICENSE-2.0
   8  *
   9  * Unless required by applicable law or agreed to in writing, software
  10  * distributed under the License is distributed on an "AS IS" BASIS,
  11  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  12  * See the License for the specific language governing permissions and
  13  * limitations under the License.
  14  */
  15
  16 #include <config.h>
  17
  18 #include "transaction.h"
  19
  20 #include "bitmap.h"
  21 #include "dynamic-string.h"
  22 #include "hash.h"
  23 #include "hmap.h"
  24 #include "json.h"
  25 #include "list.h"
  26 #include "ovsdb-error.h"
  27 #include "ovsdb.h"
  28 #include "row.h"
  29 #include "table.h"
  30 #include "uuid.h"
  31
  32 struct ovsdb_txn {
  33     struct ovsdb *db;
  34     struct list txn_tables;     /* Contains "struct ovsdb_txn_table"s. */
  35     struct ds comment;
  36 };
  37
  38 /* A table modified by a transaction. */
  39 struct ovsdb_txn_table {
  40     struct list node;           /* Element in ovsdb_txn's txn_tables list. */
  41     struct ovsdb_table *table;
  42     struct hmap txn_rows;       /* Contains "struct ovsdb_txn_row"s. */
  43
  44     /* This has the same form as the 'indexes' member of struct ovsdb_table,
  45      * but it is only used or updated at transaction commit time, from
  46      * check_index_uniqueness(). */
  47     struct hmap *txn_indexes;
  48
  49     /* Used by for_each_txn_row(). */
  50     unsigned int serial;        /* Serial number of in-progress iteration. */
  51     unsigned int n_processed;   /* Number of rows processed. */
  52 };
  53
  54 /* A row modified by the transaction:
  55  *
  56  *      - A row added by a transaction will have null 'old' and non-null 'new'.
  57  *
  58  *      - A row deleted by a transaction will have non-null 'old' and null
  59  *        'new'.
  60  *
  61  *      - A row modified by a transaction will have non-null 'old' and 'new'.
  62  *
  63  *      - 'old' and 'new' both null indicates that a row was added then deleted
  64  *        within a single transaction.  Most of the time we instead delete the
  65  *        ovsdb_txn_row entirely, but inside a for_each_txn_row() callback
  66  *        there are restrictions that sometimes mean we have to leave the
  67  *        ovsdb_txn_row in place.
  68  */
  69 struct ovsdb_txn_row {
  70     struct hmap_node hmap_node; /* In ovsdb_txn_table's txn_rows hmap. */
  71     struct ovsdb_row *old;      /* The old row. */
  72     struct ovsdb_row *new;      /* The new row. */
  73     size_t n_refs;              /* Number of remaining references. */
  74
  75     /* These members are the same as the corresponding members of 'old' or
  76      * 'new'.  They are present here for convenience and because occasionally
  77      * there can be an ovsdb_txn_row where both 'old' and 'new' are NULL. */
  78     struct uuid uuid;
  79     struct ovsdb_table *table;
  80
  81     /* Used by for_each_txn_row(). */
  82     unsigned int serial;        /* Serial number of in-progress commit. */
  83
  84     unsigned long changed[];    /* Bits set to 1 for columns that changed. */
  85 };
  86
  87 static struct ovsdb_error * WARN_UNUSED_RESULT
  88 delete_garbage_row(struct ovsdb_txn *txn, struct ovsdb_txn_row *r);
  89 static void ovsdb_txn_row_prefree(struct ovsdb_txn_row *);
  90 static struct ovsdb_error * WARN_UNUSED_RESULT
  91 for_each_txn_row(struct ovsdb_txn *txn,
  92                       struct ovsdb_error *(*)(struct ovsdb_txn *,
  93                                               struct ovsdb_txn_row *));
  94
  95 /* Used by for_each_txn_row() to track tables and rows that have been
  96  * processed.  */
  97 static unsigned int serial;
  98
  99 struct ovsdb_txn *
 100 ovsdb_txn_create(struct ovsdb *db)
 101 {
 102     struct ovsdb_txn *txn = xmalloc(sizeof *txn);
 103     txn->db = db;
 104     list_init(&txn->txn_tables);
 105     ds_init(&txn->comment);
 106     return txn;
 107 }
 108
 109 static void
 110 ovsdb_txn_free(struct ovsdb_txn *txn)
 111 {
 112     ovs_assert(list_is_empty(&txn->txn_tables));
 113     ds_destroy(&txn->comment);
 114     free(txn);
 115 }
 116
 117 static struct ovsdb_error *
 118 ovsdb_txn_row_abort(struct ovsdb_txn *txn OVS_UNUSED,
 119                     struct ovsdb_txn_row *txn_row)
 120 {
 121     struct ovsdb_row *old = txn_row->old;
 122     struct ovsdb_row *new = txn_row->new;
 123
 124     ovsdb_txn_row_prefree(txn_row);
 125     if (!old) {
 126         if (new) {
 127             hmap_remove(&new->table->rows, &new->hmap_node);
 128         }
 129     } else if (!new) {
 130         hmap_insert(&old->table->rows, &old->hmap_node, ovsdb_row_hash(old));
 131     } else {
 132         hmap_replace(&new->table->rows, &new->hmap_node, &old->hmap_node);
 133     }
 134     ovsdb_row_destroy(new);
 135     free(txn_row);
 136
 137     return NULL;
 138 }
 139
 140 /* Returns the offset in bytes from the start of an ovsdb_row for 'table' to
 141  * the hmap_node for the index numbered 'i'. */
 142 static size_t
 143 ovsdb_row_index_offset__(const struct ovsdb_table *table, size_t i)
 144 {
 145     size_t n_fields = shash_count(&table->schema->columns);
 146     return (offsetof(struct ovsdb_row, fields)
 147             + n_fields * sizeof(struct ovsdb_datum)
 148             + i * sizeof(struct hmap_node));
 149 }
 150
 151 /* Returns the hmap_node in 'row' for the index numbered 'i'. */
 152 static struct hmap_node *
 153 ovsdb_row_get_index_node(struct ovsdb_row *row, size_t i)
 154 {
 155     return (void *) ((char *) row + ovsdb_row_index_offset__(row->table, i));
 156 }
 157
 158 /* Returns the ovsdb_row given 'index_node', which is a pointer to that row's
 159  * hmap_node for the index numbered 'i' within 'table'. */
 160 static struct ovsdb_row *
 161 ovsdb_row_from_index_node(struct hmap_node *index_node,
 162                           const struct ovsdb_table *table, size_t i)
 163 {
 164     return (void *) ((char *) index_node - ovsdb_row_index_offset__(table, i));
 165 }
 166
 167 void
 168 ovsdb_txn_abort(struct ovsdb_txn *txn)
 169 {
 170     ovsdb_error_assert(for_each_txn_row(txn, ovsdb_txn_row_abort));
 171     ovsdb_txn_free(txn);
 172 }
 173
 174 static struct ovsdb_txn_row *
 175 find_txn_row(const struct ovsdb_table *table, const struct uuid *uuid)
 176 {
 177     struct ovsdb_txn_row *txn_row;
 178
 179     if (!table->txn_table) {
 180         return NULL;
 181     }
 182
 183     HMAP_FOR_EACH_WITH_HASH (txn_row, hmap_node,
 184                              uuid_hash(uuid), &table->txn_table->txn_rows) {
 185         if (uuid_equals(uuid, &txn_row->uuid)) {
 186             return txn_row;
 187         }
 188     }
 189
 190     return NULL;
 191 }
 192
 193 static struct ovsdb_txn_row *
 194 find_or_make_txn_row(struct ovsdb_txn *txn, const struct ovsdb_table *table,
 195                      const struct uuid *uuid)
 196 {
 197     struct ovsdb_txn_row *txn_row = find_txn_row(table, uuid);
 198     if (!txn_row) {
 199         const struct ovsdb_row *row = ovsdb_table_get_row(table, uuid);
 200         if (row) {
 201             txn_row = ovsdb_txn_row_modify(txn, row)->txn_row;
 202         }
 203     }
 204     return txn_row;
 205 }
 206
 207 static struct ovsdb_error * WARN_UNUSED_RESULT
 208 ovsdb_txn_adjust_atom_refs(struct ovsdb_txn *txn, const struct ovsdb_row *r,
 209                            const struct ovsdb_column *c,
 210                            const struct ovsdb_base_type *base,
 211                            const union ovsdb_atom *atoms, unsigned int n,
 212                            int delta)
 213 {
 214     const struct ovsdb_table *table;
 215     unsigned int i;
 216
 217     if (!ovsdb_base_type_is_strong_ref(base)) {
 218         return NULL;
 219     }
 220
 221     table = base->u.uuid.refTable;
 222     for (i = 0; i < n; i++) {
 223         const struct uuid *uuid = &atoms[i].uuid;
 224         struct ovsdb_txn_row *txn_row;
 225
 226         if (uuid_equals(uuid, ovsdb_row_get_uuid(r))) {
 227             /* Self-references don't count. */
 228             continue;
 229         }
 230
 231         txn_row = find_or_make_txn_row(txn, table, uuid);
 232         if (!txn_row) {
 233             return ovsdb_error("referential integrity violation",
 234                                "Table %s column %s row "UUID_FMT" "
 235                                "references nonexistent row "UUID_FMT" in "
 236                                "table %s.",
 237                                r->table->schema->name, c->name,
 238                                UUID_ARGS(ovsdb_row_get_uuid(r)),
 239                                UUID_ARGS(uuid), table->schema->name);
 240         }
 241         txn_row->n_refs += delta;
 242     }
 243
 244     return NULL;
 245 }
 246
 247 static struct ovsdb_error * WARN_UNUSED_RESULT
 248 ovsdb_txn_adjust_row_refs(struct ovsdb_txn *txn, const struct ovsdb_row *r,
 249                           const struct ovsdb_column *column, int delta)
 250 {
 251     const struct ovsdb_datum *field = &r->fields[column->index];
 252     struct ovsdb_error *error;
 253
 254     error = ovsdb_txn_adjust_atom_refs(txn, r, column, &column->type.key,
 255                                        field->keys, field->n, delta);
 256     if (!error) {
 257         error = ovsdb_txn_adjust_atom_refs(txn, r, column, &column->type.value,
 258                                            field->values, field->n, delta);
 259     }
 260     return error;
 261 }
 262
 263 static struct ovsdb_error * WARN_UNUSED_RESULT
 264 update_row_ref_count(struct ovsdb_txn *txn, struct ovsdb_txn_row *r)
 265 {
 266     struct ovsdb_table *table = r->table;
 267     struct shash_node *node;
 268
 269     SHASH_FOR_EACH (node, &table->schema->columns) {
 270         const struct ovsdb_column *column = node->data;
 271         struct ovsdb_error *error;
 272
 273         if (r->old) {
 274             error = ovsdb_txn_adjust_row_refs(txn, r->old, column, -1);
 275             if (error) {
 276                 return OVSDB_WRAP_BUG("error decreasing refcount", error);
 277             }
 278         }
 279         if (r->new) {
 280             error = ovsdb_txn_adjust_row_refs(txn, r->new, column, 1);
 281             if (error) {
 282                 return error;
 283             }
 284         }
 285     }
 286
 287     return NULL;
 288 }
 289
 290 static struct ovsdb_error * WARN_UNUSED_RESULT
 291 check_ref_count(struct ovsdb_txn *txn OVS_UNUSED, struct ovsdb_txn_row *r)
 292 {
 293     if (r->new || !r->n_refs) {
 294         return NULL;
 295     } else {
 296         return ovsdb_error("referential integrity violation",
 297                            "cannot delete %s row "UUID_FMT" because "
 298                            "of %"PRIuSIZE" remaining reference(s)",
 299                            r->table->schema->name, UUID_ARGS(&r->uuid),
 300                            r->n_refs);
 301     }
 302 }
 303
 304 static struct ovsdb_error * WARN_UNUSED_RESULT
 305 delete_row_refs(struct ovsdb_txn *txn, const struct ovsdb_row *row,
 306                 const struct ovsdb_base_type *base,
 307                 const union ovsdb_atom *atoms, unsigned int n)
 308 {
 309     const struct ovsdb_table *table;
 310     unsigned int i;
 311
 312     if (!ovsdb_base_type_is_strong_ref(base)) {
 313         return NULL;
 314     }
 315
 316     table = base->u.uuid.refTable;
 317     for (i = 0; i < n; i++) {
 318         const struct uuid *uuid = &atoms[i].uuid;
 319         struct ovsdb_txn_row *txn_row;
 320
 321         if (uuid_equals(uuid, ovsdb_row_get_uuid(row))) {
 322             /* Self-references don't count. */
 323             continue;
 324         }
 325
 326         txn_row = find_or_make_txn_row(txn, table, uuid);
 327         if (!txn_row) {
 328             return OVSDB_BUG("strong ref target missing");
 329         } else if (!txn_row->n_refs) {
 330             return OVSDB_BUG("strong ref target has zero n_refs");
 331         } else if (!txn_row->new) {
 332             return OVSDB_BUG("deleted strong ref target");
 333         }
 334
 335         if (--txn_row->n_refs == 0) {
 336             struct ovsdb_error *error = delete_garbage_row(txn, txn_row);
 337             if (error) {
 338                 return error;
 339             }
 340         }
 341     }
 342
 343     return NULL;
 344 }
 345
 346 static struct ovsdb_error * WARN_UNUSED_RESULT
 347 delete_garbage_row(struct ovsdb_txn *txn, struct ovsdb_txn_row *txn_row)
 348 {
 349     struct shash_node *node;
 350     struct ovsdb_row *row;
 351
 352     if (txn_row->table->schema->is_root) {
 353         return NULL;
 354     }
 355
 356     row = txn_row->new;
 357     txn_row->new = NULL;
 358     hmap_remove(&txn_row->table->rows, &row->hmap_node);
 359     SHASH_FOR_EACH (node, &txn_row->table->schema->columns) {
 360         const struct ovsdb_column *column = node->data;
 361         const struct ovsdb_datum *field = &row->fields[column->index];
 362         struct ovsdb_error *error;
 363
 364         error = delete_row_refs(txn, row,
 365                                 &column->type.key, field->keys, field->n);
 366         if (error) {
 367             return error;
 368         }
 369
 370         error = delete_row_refs(txn, row,
 371                                 &column->type.value, field->values, field->n);
 372         if (error) {
 373             return error;
 374         }
 375     }
 376     ovsdb_row_destroy(row);
 377
 378     return NULL;
 379 }
 380
 381 static struct ovsdb_error * WARN_UNUSED_RESULT
 382 collect_garbage(struct ovsdb_txn *txn, struct ovsdb_txn_row *txn_row)
 383 {
 384     if (txn_row->new && !txn_row->n_refs) {
 385         return delete_garbage_row(txn, txn_row);
 386     }
 387     return NULL;
 388 }
 389
 390 static struct ovsdb_error * WARN_UNUSED_RESULT
 391 update_ref_counts(struct ovsdb_txn *txn)
 392 {
 393     struct ovsdb_error *error;
 394
 395     error = for_each_txn_row(txn, update_row_ref_count);
 396     if (error) {
 397         return error;
 398     }
 399
 400     return for_each_txn_row(txn, check_ref_count);
 401 }
 402
 403 static struct ovsdb_error *
 404 ovsdb_txn_row_commit(struct ovsdb_txn *txn OVS_UNUSED,
 405                      struct ovsdb_txn_row *txn_row)
 406 {
 407     size_t n_indexes = txn_row->table->schema->n_indexes;
 408
 409     if (txn_row->old) {
 410         size_t i;
 411
 412         for (i = 0; i < n_indexes; i++) {
 413             struct hmap_node *node = ovsdb_row_get_index_node(txn_row->old, i);
 414             hmap_remove(&txn_row->table->indexes[i], node);
 415         }
 416     }
 417     if (txn_row->new) {
 418         size_t i;
 419
 420         for (i = 0; i < n_indexes; i++) {
 421             struct hmap_node *node = ovsdb_row_get_index_node(txn_row->new, i);
 422             hmap_insert(&txn_row->table->indexes[i], node, node->hash);
 423         }
 424     }
 425
 426     ovsdb_txn_row_prefree(txn_row);
 427     if (txn_row->new) {
 428         txn_row->new->n_refs = txn_row->n_refs;
 429     }
 430     ovsdb_row_destroy(txn_row->old);
 431     free(txn_row);
 432
 433     return NULL;
 434 }
 435
 436 static void
 437 add_weak_ref(struct ovsdb_txn *txn,
 438              const struct ovsdb_row *src_, const struct ovsdb_row *dst_)
 439 {
 440     struct ovsdb_row *src = CONST_CAST(struct ovsdb_row *, src_);
 441     struct ovsdb_row *dst = CONST_CAST(struct ovsdb_row *, dst_);
 442     struct ovsdb_weak_ref *weak;
 443
 444     if (src == dst) {
 445         return;
 446     }
 447
 448     dst = ovsdb_txn_row_modify(txn, dst);
 449
 450     if (!list_is_empty(&dst->dst_refs)) {
 451         /* Omit duplicates. */
 452         weak = CONTAINER_OF(list_back(&dst->dst_refs),
 453                             struct ovsdb_weak_ref, dst_node);
 454         if (weak->src == src) {
 455             return;
 456         }
 457     }
 458
 459     weak = xmalloc(sizeof *weak);
 460     weak->src = src;
 461     list_push_back(&dst->dst_refs, &weak->dst_node);
 462     list_push_back(&src->src_refs, &weak->src_node);
 463 }
 464
 465 static struct ovsdb_error * WARN_UNUSED_RESULT
 466 assess_weak_refs(struct ovsdb_txn *txn, struct ovsdb_txn_row *txn_row)
 467 {
 468     struct ovsdb_table *table;
 469     struct shash_node *node;
 470
 471     if (txn_row->old) {
 472         /* Mark rows that have weak references to 'txn_row' as modified, so
 473          * that their weak references will get reassessed. */
 474         struct ovsdb_weak_ref *weak, *next;
 475
 476         LIST_FOR_EACH_SAFE (weak, next, dst_node, &txn_row->old->dst_refs) {
 477             if (!weak->src->txn_row) {
 478                 ovsdb_txn_row_modify(txn, weak->src);
 479             }
 480         }
 481     }
 482
 483     if (!txn_row->new) {
 484         /* We don't have to do anything about references that originate at
 485          * 'txn_row', because ovsdb_row_destroy() will remove those weak
 486          * references. */
 487         return NULL;
 488     }
 489
 490     table = txn_row->table;
 491     SHASH_FOR_EACH (node, &table->schema->columns) {
 492         const struct ovsdb_column *column = node->data;
 493         struct ovsdb_datum *datum = &txn_row->new->fields[column->index];
 494         unsigned int orig_n, i;
 495         bool zero = false;
 496
 497         orig_n = datum->n;
 498
 499         if (ovsdb_base_type_is_weak_ref(&column->type.key)) {
 500             for (i = 0; i < datum->n; ) {
 501                 const struct ovsdb_row *row;
 502
 503                 row = ovsdb_table_get_row(column->type.key.u.uuid.refTable,
 504                                           &datum->keys[i].uuid);
 505                 if (row) {
 506                     add_weak_ref(txn, txn_row->new, row);
 507                     i++;
 508                 } else {
 509                     if (uuid_is_zero(&datum->keys[i].uuid)) {
 510                         zero = true;
 511                     }
 512                     ovsdb_datum_remove_unsafe(datum, i, &column->type);
 513                 }
 514             }
 515         }
 516
 517         if (ovsdb_base_type_is_weak_ref(&column->type.value)) {
 518             for (i = 0; i < datum->n; ) {
 519                 const struct ovsdb_row *row;
 520
 521                 row = ovsdb_table_get_row(column->type.value.u.uuid.refTable,
 522                                           &datum->values[i].uuid);
 523                 if (row) {
 524                     add_weak_ref(txn, txn_row->new, row);
 525                     i++;
 526                 } else {
 527                     if (uuid_is_zero(&datum->values[i].uuid)) {
 528                         zero = true;
 529                     }
 530                     ovsdb_datum_remove_unsafe(datum, i, &column->type);
 531                 }
 532             }
 533         }
 534
 535         if (datum->n != orig_n) {
 536             bitmap_set1(txn_row->changed, column->index);
 537             ovsdb_datum_sort_assert(datum, column->type.key.type);
 538             if (datum->n < column->type.n_min) {
 539                 const struct uuid *row_uuid = ovsdb_row_get_uuid(txn_row->new);
 540                 if (zero && !txn_row->old) {
 541                     return ovsdb_error(
 542                         "constraint violation",
 543                         "Weak reference column \"%s\" in \"%s\" row "UUID_FMT
 544                         " (inserted within this transaction) contained "
 545                         "all-zeros UUID (probably as the default value for "
 546                         "this column) but deleting this value caused a "
 547                         "constraint volation because this column is not "
 548                         "allowed to be empty.", column->name,
 549                         table->schema->name, UUID_ARGS(row_uuid));
 550                 } else {
 551                     return ovsdb_error(
 552                         "constraint violation",
 553                         "Deletion of %u weak reference(s) to deleted (or "
 554                         "never-existing) rows from column \"%s\" in \"%s\" "
 555                         "row "UUID_FMT" %scaused this column to become empty, "
 556                         "but constraints on this column disallow an "
 557                         "empty column.",
 558                         orig_n - datum->n, column->name, table->schema->name,
 559                         UUID_ARGS(row_uuid),
 560                         (txn_row->old
 561                          ? ""
 562                          : "(inserted within this transaction) "));
 563                 }
 564             }
 565         }
 566     }
 567
 568     return NULL;
 569 }
 570
 571 static struct ovsdb_error * WARN_UNUSED_RESULT
 572 determine_changes(struct ovsdb_txn *txn, struct ovsdb_txn_row *txn_row)
 573 {
 574     struct ovsdb_table *table = txn_row->table;
 575
 576     if (txn_row->old && txn_row->new) {
 577         struct shash_node *node;
 578         bool changed = false;
 579
 580         SHASH_FOR_EACH (node, &table->schema->columns) {
 581             const struct ovsdb_column *column = node->data;
 582             const struct ovsdb_type *type = &column->type;
 583             unsigned int idx = column->index;
 584
 585             if (!ovsdb_datum_equals(&txn_row->old->fields[idx],
 586                                     &txn_row->new->fields[idx],
 587                                     type)) {
 588                 bitmap_set1(txn_row->changed, idx);
 589                 changed = true;
 590             }
 591         }
 592
 593         if (!changed) {
 594             /* Nothing actually changed in this row, so drop it. */
 595             ovsdb_txn_row_abort(txn, txn_row);
 596         }
 597     } else {
 598         bitmap_set_multiple(txn_row->changed, 0,
 599                             shash_count(&table->schema->columns), 1);
 600     }
 601
 602     return NULL;
 603 }
 604
 605 static struct ovsdb_error * WARN_UNUSED_RESULT
 606 check_max_rows(struct ovsdb_txn *txn)
 607 {
 608     struct ovsdb_txn_table *t;
 609
 610     LIST_FOR_EACH (t, node, &txn->txn_tables) {
 611         size_t n_rows = hmap_count(&t->table->rows);
 612         unsigned int max_rows = t->table->schema->max_rows;
 613
 614         if (n_rows > max_rows) {
 615             return ovsdb_error("constraint violation",
 616                                "transaction causes \"%s\" table to contain "
 617                                "%"PRIuSIZE" rows, greater than the schema-defined "
 618                                "limit of %u row(s)",
 619                                t->table->schema->name, n_rows, max_rows);
 620         }
 621     }
 622
 623     return NULL;
 624 }
 625
 626 static struct ovsdb_row *
 627 ovsdb_index_search(struct hmap *index, struct ovsdb_row *row, size_t i,
 628                    uint32_t hash)
 629 {
 630     const struct ovsdb_table *table = row->table;
 631     const struct ovsdb_column_set *columns = &table->schema->indexes[i];
 632     struct hmap_node *node;
 633
 634     for (node = hmap_first_with_hash(index, hash); node;
 635          node = hmap_next_with_hash(node)) {
 636         struct ovsdb_row *irow = ovsdb_row_from_index_node(node, table, i);
 637         if (ovsdb_row_equal_columns(row, irow, columns)) {
 638             return irow;
 639         }
 640     }
 641
 642     return NULL;
 643 }
 644
 645 static void
 646 duplicate_index_row__(const struct ovsdb_column_set *index,
 647                       const struct ovsdb_row *row,
 648                       const char *title,
 649                       struct ds *out)
 650 {
 651     size_t n_columns = shash_count(&row->table->schema->columns);
 652
 653     ds_put_format(out, "%s row, with UUID "UUID_FMT", ",
 654                   title, UUID_ARGS(ovsdb_row_get_uuid(row)));
 655     if (!row->txn_row
 656         || bitmap_scan(row->txn_row->changed, 0, n_columns) == n_columns) {
 657         ds_put_cstr(out, "existed in the database before this "
 658                     "transaction and was not modified by the transaction.");
 659     } else if (!row->txn_row->old) {
 660         ds_put_cstr(out, "was inserted by this transaction.");
 661     } else if (ovsdb_row_equal_columns(row->txn_row->old,
 662                                        row->txn_row->new, index)) {
 663         ds_put_cstr(out, "existed in the database before this "
 664                     "transaction, which modified some of the row's columns "
 665                     "but not any columns in this index.");
 666     } else {
 667         ds_put_cstr(out, "had the following index values before the "
 668                     "transaction: ");
 669         ovsdb_row_columns_to_string(row->txn_row->old, index, out);
 670         ds_put_char(out, '.');
 671     }
 672 }
 673
 674 static struct ovsdb_error * WARN_UNUSED_RESULT
 675 duplicate_index_row(const struct ovsdb_column_set *index,
 676                     const struct ovsdb_row *a,
 677                     const struct ovsdb_row *b)
 678 {
 679     struct ovsdb_column_set all_columns;
 680     struct ovsdb_error *error;
 681     char *index_s;
 682     struct ds s;
 683
 684     /* Put 'a' and 'b' in a predictable order to make error messages
 685      * reproducible for testing. */
 686     ovsdb_column_set_init(&all_columns);
 687     ovsdb_column_set_add_all(&all_columns, a->table);
 688     if (ovsdb_row_compare_columns_3way(a, b, &all_columns) < 0) {
 689         const struct ovsdb_row *tmp = a;
 690         a = b;
 691         b = tmp;
 692     }
 693     ovsdb_column_set_destroy(&all_columns);
 694
 695     index_s = ovsdb_column_set_to_string(index);
 696
 697     ds_init(&s);
 698     ds_put_format(&s, "Transaction causes multiple rows in \"%s\" table to "
 699                   "have identical values (", a->table->schema->name);
 700     ovsdb_row_columns_to_string(a, index, &s);
 701     ds_put_format(&s, ") for index on %s.  ", index_s);
 702     duplicate_index_row__(index, a, "First", &s);
 703     ds_put_cstr(&s, "  ");
 704     duplicate_index_row__(index, b, "Second", &s);
 705
 706     free(index_s);
 707
 708     error = ovsdb_error("constraint violation", "%s", ds_cstr(&s));
 709     ds_destroy(&s);
 710     return error;
 711 }
 712
 713 static struct ovsdb_error * WARN_UNUSED_RESULT
 714 check_index_uniqueness(struct ovsdb_txn *txn OVS_UNUSED,
 715                        struct ovsdb_txn_row *txn_row)
 716 {
 717     struct ovsdb_txn_table *txn_table = txn_row->table->txn_table;
 718     struct ovsdb_table *table = txn_row->table;
 719     struct ovsdb_row *row = txn_row->new;
 720     size_t i;
 721
 722     if (!row) {
 723         return NULL;
 724     }
 725
 726     for (i = 0; i < table->schema->n_indexes; i++) {
 727         const struct ovsdb_column_set *index = &table->schema->indexes[i];
 728         struct ovsdb_row *irow;
 729         uint32_t hash;
 730
 731         hash = ovsdb_row_hash_columns(row, index, 0);
 732         irow = ovsdb_index_search(&txn_table->txn_indexes[i], row, i, hash);
 733         if (irow) {
 734             return duplicate_index_row(index, irow, row);
 735         }
 736
 737         irow = ovsdb_index_search(&table->indexes[i], row, i, hash);
 738         if (irow && !irow->txn_row) {
 739             return duplicate_index_row(index, irow, row);
 740         }
 741
 742         hmap_insert(&txn_table->txn_indexes[i],
 743                     ovsdb_row_get_index_node(row, i), hash);
 744     }
 745
 746     return NULL;
 747 }
 748
 749 struct ovsdb_error *
 750 ovsdb_txn_commit(struct ovsdb_txn *txn, bool durable)
 751 {
 752     struct ovsdb_replica *replica;
 753     struct ovsdb_error *error;
 754
 755     /* Figure out what actually changed, and abort early if the transaction
 756      * was really a no-op. */
 757     error = for_each_txn_row(txn, determine_changes);
 758     if (error) {
 759         return OVSDB_WRAP_BUG("can't happen", error);
 760     }
 761     if (list_is_empty(&txn->txn_tables)) {
 762         ovsdb_txn_abort(txn);
 763         return NULL;
 764     }
 765
 766     /* Update reference counts and check referential integrity. */
 767     error = update_ref_counts(txn);
 768     if (error) {
 769         ovsdb_txn_abort(txn);
 770         return error;
 771     }
 772
 773     /* Delete unreferenced, non-root rows. */
 774     error = for_each_txn_row(txn, collect_garbage);
 775     if (error) {
 776         ovsdb_txn_abort(txn);
 777         return OVSDB_WRAP_BUG("can't happen", error);
 778     }
 779
 780     /* Check maximum rows table constraints. */
 781     error = check_max_rows(txn);
 782     if (error) {
 783         ovsdb_txn_abort(txn);
 784         return error;
 785     }
 786
 787     /* Check reference counts and remove bad references for "weak" referential
 788      * integrity. */
 789     error = for_each_txn_row(txn, assess_weak_refs);
 790     if (error) {
 791         ovsdb_txn_abort(txn);
 792         return error;
 793     }
 794
 795     /* Verify that the indexes will still be unique post-transaction. */
 796     error = for_each_txn_row(txn, check_index_uniqueness);
 797     if (error) {
 798         ovsdb_txn_abort(txn);
 799         return error;
 800     }
 801
 802     /* Send the commit to each replica. */
 803     LIST_FOR_EACH (replica, node, &txn->db->replicas) {
 804         error = (replica->class->commit)(replica, txn, durable);
 805         if (error) {
 806             /* We don't support two-phase commit so only the first replica is
 807              * allowed to report an error. */
 808             ovs_assert(&replica->node == txn->db->replicas.next);
 809
 810             ovsdb_txn_abort(txn);
 811             return error;
 812         }
 813     }
 814
 815     /* Finalize commit. */
 816     txn->db->run_triggers = true;
 817     ovsdb_error_assert(for_each_txn_row(txn, ovsdb_txn_row_commit));
 818     ovsdb_txn_free(txn);
 819
 820     return NULL;
 821 }
 822
 823 void
 824 ovsdb_txn_for_each_change(const struct ovsdb_txn *txn,
 825                           ovsdb_txn_row_cb_func *cb, void *aux)
 826 {
 827     struct ovsdb_txn_table *t;
 828     struct ovsdb_txn_row *r;
 829
 830     LIST_FOR_EACH (t, node, &txn->txn_tables) {
 831         HMAP_FOR_EACH (r, hmap_node, &t->txn_rows) {
 832             if ((r->old || r->new) && !cb(r->old, r->new, r->changed, aux)) {
 833                 break;
 834             }
 835         }
 836    }
 837 }
 838
 839 static struct ovsdb_txn_table *
 840 ovsdb_txn_create_txn_table(struct ovsdb_txn *txn, struct ovsdb_table *table)
 841 {
 842     if (!table->txn_table) {
 843         struct ovsdb_txn_table *txn_table;
 844         size_t i;
 845
 846         table->txn_table = txn_table = xmalloc(sizeof *table->txn_table);
 847         txn_table->table = table;
 848         hmap_init(&txn_table->txn_rows);
 849         txn_table->serial = serial - 1;
 850         txn_table->txn_indexes = xmalloc(table->schema->n_indexes
 851                                          * sizeof *txn_table->txn_indexes);
 852         for (i = 0; i < table->schema->n_indexes; i++) {
 853             hmap_init(&txn_table->txn_indexes[i]);
 854         }
 855         list_push_back(&txn->txn_tables, &txn_table->node);
 856     }
 857     return table->txn_table;
 858 }
 859
 860 static struct ovsdb_txn_row *
 861 ovsdb_txn_row_create(struct ovsdb_txn *txn, struct ovsdb_table *table,
 862                      const struct ovsdb_row *old_, struct ovsdb_row *new)
 863 {
 864     const struct ovsdb_row *row = old_ ? old_ : new;
 865     struct ovsdb_row *old = CONST_CAST(struct ovsdb_row *, old_);
 866     size_t n_columns = shash_count(&table->schema->columns);
 867     struct ovsdb_txn_table *txn_table;
 868     struct ovsdb_txn_row *txn_row;
 869
 870     txn_row = xzalloc(offsetof(struct ovsdb_txn_row, changed)
 871                       + bitmap_n_bytes(n_columns));
 872     txn_row->uuid = *ovsdb_row_get_uuid(row);
 873     txn_row->table = row->table;
 874     txn_row->old = old;
 875     txn_row->new = new;
 876     txn_row->n_refs = old ? old->n_refs : 0;
 877     txn_row->serial = serial - 1;
 878
 879     if (old) {
 880         old->txn_row = txn_row;
 881     }
 882     if (new) {
 883         new->txn_row = txn_row;
 884     }
 885
 886     txn_table = ovsdb_txn_create_txn_table(txn, table);
 887     hmap_insert(&txn_table->txn_rows, &txn_row->hmap_node,
 888                 ovsdb_row_hash(old ? old : new));
 889
 890     return txn_row;
 891 }
 892
 893 struct ovsdb_row *
 894 ovsdb_txn_row_modify(struct ovsdb_txn *txn, const struct ovsdb_row *ro_row_)
 895 {
 896     struct ovsdb_row *ro_row = CONST_CAST(struct ovsdb_row *, ro_row_);
 897
 898     if (ro_row->txn_row) {
 899         ovs_assert(ro_row == ro_row->txn_row->new);
 900         return ro_row;
 901     } else {
 902         struct ovsdb_table *table = ro_row->table;
 903         struct ovsdb_row *rw_row;
 904
 905         rw_row = ovsdb_row_clone(ro_row);
 906         rw_row->n_refs = ro_row->n_refs;
 907         uuid_generate(ovsdb_row_get_version_rw(rw_row));
 908         ovsdb_txn_row_create(txn, table, ro_row, rw_row);
 909         hmap_replace(&table->rows, &ro_row->hmap_node, &rw_row->hmap_node);
 910
 911         return rw_row;
 912     }
 913 }
 914
 915 void
 916 ovsdb_txn_row_insert(struct ovsdb_txn *txn, struct ovsdb_row *row)
 917 {
 918     uint32_t hash = ovsdb_row_hash(row);
 919     struct ovsdb_table *table = row->table;
 920
 921     uuid_generate(ovsdb_row_get_version_rw(row));
 922
 923     ovsdb_txn_row_create(txn, table, NULL, row);
 924     hmap_insert(&table->rows, &row->hmap_node, hash);
 925 }
 926
 927 /* 'row' must be assumed destroyed upon return; the caller must not reference
 928  * it again. */
 929 void
 930 ovsdb_txn_row_delete(struct ovsdb_txn *txn, const struct ovsdb_row *row_)
 931 {
 932     struct ovsdb_row *row = CONST_CAST(struct ovsdb_row *, row_);
 933     struct ovsdb_table *table = row->table;
 934     struct ovsdb_txn_row *txn_row = row->txn_row;
 935
 936     hmap_remove(&table->rows, &row->hmap_node);
 937
 938     if (!txn_row) {
 939         ovsdb_txn_row_create(txn, table, row, NULL);
 940     } else {
 941         ovs_assert(txn_row->new == row);
 942         if (txn_row->old) {
 943             txn_row->new = NULL;
 944         } else {
 945             hmap_remove(&table->txn_table->txn_rows, &txn_row->hmap_node);
 946             free(txn_row);
 947         }
 948         ovsdb_row_destroy(row);
 949     }
 950 }
 951
 952 void
 953 ovsdb_txn_add_comment(struct ovsdb_txn *txn, const char *s)
 954 {
 955     if (txn->comment.length) {
 956         ds_put_char(&txn->comment, '\n');
 957     }
 958     ds_put_cstr(&txn->comment, s);
 959 }
 960
 961 const char *
 962 ovsdb_txn_get_comment(const struct ovsdb_txn *txn)
 963 {
 964     return txn->comment.length ? ds_cstr_ro(&txn->comment) : NULL;
 965 }
 966 \f
 967 static void
 968 ovsdb_txn_row_prefree(struct ovsdb_txn_row *txn_row)
 969 {
 970     struct ovsdb_txn_table *txn_table = txn_row->table->txn_table;
 971
 972     txn_table->n_processed--;
 973     hmap_remove(&txn_table->txn_rows, &txn_row->hmap_node);
 974
 975     if (txn_row->old) {
 976         txn_row->old->txn_row = NULL;
 977     }
 978     if (txn_row->new) {
 979         txn_row->new->txn_row = NULL;
 980     }
 981 }
 982
 983 static void
 984 ovsdb_txn_table_destroy(struct ovsdb_txn_table *txn_table)
 985 {
 986     size_t i;
 987
 988     ovs_assert(hmap_is_empty(&txn_table->txn_rows));
 989
 990     for (i = 0; i < txn_table->table->schema->n_indexes; i++) {
 991         hmap_destroy(&txn_table->txn_indexes[i]);
 992     }
 993     free(txn_table->txn_indexes);
 994
 995     txn_table->table->txn_table = NULL;
 996     hmap_destroy(&txn_table->txn_rows);
 997     list_remove(&txn_table->node);
 998     free(txn_table);
 999 }
1000
1001 /* Calls 'cb' for every txn_row within 'txn'.  If 'cb' returns nonnull, this
1002  * aborts the iteration and for_each_txn_row() passes the error up.  Otherwise,
1003  * returns a null pointer after iteration is complete.
1004  *
1005  * 'cb' may insert new txn_rows and new txn_tables into 'txn'.  It may delete
1006  * the txn_row that it is passed in, or txn_rows in txn_tables other than the
1007  * one passed to 'cb'.  It may *not* delete txn_rows other than the one passed
1008  * in within the same txn_table.  It may *not* delete any txn_tables.  As long
1009  * as these rules are followed, 'cb' will be called exactly once for each
1010  * txn_row in 'txn', even those added by 'cb'.
1011  *
1012  * (Even though 'cb' is not allowed to delete some txn_rows, it can still
1013  * delete any actual row by clearing a txn_row's 'new' member.)
1014  */
1015 static struct ovsdb_error * WARN_UNUSED_RESULT
1016 for_each_txn_row(struct ovsdb_txn *txn,
1017                  struct ovsdb_error *(*cb)(struct ovsdb_txn *,
1018                                            struct ovsdb_txn_row *))
1019 {
1020     bool any_work;
1021
1022     serial++;
1023
1024     do {
1025         struct ovsdb_txn_table *t, *next_txn_table;
1026
1027         any_work = false;
1028         LIST_FOR_EACH_SAFE (t, next_txn_table, node, &txn->txn_tables) {
1029             if (t->serial != serial) {
1030                 t->serial = serial;
1031                 t->n_processed = 0;
1032             }
1033
1034             while (t->n_processed < hmap_count(&t->txn_rows)) {
1035                 struct ovsdb_txn_row *r, *next_txn_row;
1036
1037                 HMAP_FOR_EACH_SAFE (r, next_txn_row, hmap_node, &t->txn_rows) {
1038                     if (r->serial != serial) {
1039                         struct ovsdb_error *error;
1040
1041                         r->serial = serial;
1042                         t->n_processed++;
1043                         any_work = true;
1044
1045                         error = cb(txn, r);
1046                         if (error) {
1047                             return error;
1048                         }
1049                     }
1050                 }
1051             }
1052             if (hmap_is_empty(&t->txn_rows)) {
1053                 /* Table is empty.  Drop it. */
1054                 ovsdb_txn_table_destroy(t);
1055             }
1056         }
1057     } while (any_work);
1058
1059     return NULL;
1060 }