ovsdb/transaction.c

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