ovsdb/transaction.c

   1 /* Copyright (c) 2009, 2010, 2011, 2012, 2013, 2014 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 "perf-counter.h"
  31 #include "uuid.h"
  32
  33 struct ovsdb_txn {
  34     struct ovsdb *db;
  35     struct ovs_list txn_tables; /* Contains "struct ovsdb_txn_table"s. */
  36     struct ds comment;
  37 };
  38
  39 /* A table modified by a transaction. */
  40 struct ovsdb_txn_table {
  41     struct ovs_list node;       /* Element in ovsdb_txn's txn_tables list. */
  42     struct ovsdb_table *table;
  43     struct hmap txn_rows;       /* Contains "struct ovsdb_txn_row"s. */
  44
  45     /* This has the same form as the 'indexes' member of struct ovsdb_table,
  46      * but it is only used or updated at transaction commit time, from
  47      * check_index_uniqueness(). */
  48     struct hmap *txn_indexes;
  49
  50     /* Used by for_each_txn_row(). */
  51     unsigned int serial;        /* Serial number of in-progress iteration. */
  52     unsigned int n_processed;   /* Number of rows processed. */
  53 };
  54
  55 /* A row modified by the transaction:
  56  *
  57  *      - A row added by a transaction will have null 'old' and non-null 'new'.
  58  *
  59  *      - A row deleted by a transaction will have non-null 'old' and null
  60  *        'new'.
  61  *
  62  *      - A row modified by a transaction will have non-null 'old' and 'new'.
  63  *
  64  *      - 'old' and 'new' both null indicates that a row was added then deleted
  65  *        within a single transaction.  Most of the time we instead delete the
  66  *        ovsdb_txn_row entirely, but inside a for_each_txn_row() callback
  67  *        there are restrictions that sometimes mean we have to leave the
  68  *        ovsdb_txn_row in place.
  69  */
  70 struct ovsdb_txn_row {
  71     struct hmap_node hmap_node; /* In ovsdb_txn_table's txn_rows hmap. */
  72     struct ovsdb_row *old;      /* The old row. */
  73     struct ovsdb_row *new;      /* The new row. */
  74     size_t n_refs;              /* Number of remaining references. */
  75
  76     /* These members are the same as the corresponding members of 'old' or
  77      * 'new'.  They are present here for convenience and because occasionally
  78      * there can be an ovsdb_txn_row where both 'old' and 'new' are NULL. */
  79     struct uuid uuid;
  80     struct ovsdb_table *table;
  81
  82     /* Used by for_each_txn_row(). */
  83     unsigned int serial;        /* Serial number of in-progress commit. */
  84
  85     unsigned long changed[];    /* Bits set to 1 for columns that changed. */
  86 };
  87
  88 static struct ovsdb_error * OVS_WARN_UNUSED_RESULT
  89 delete_garbage_row(struct ovsdb_txn *txn, struct ovsdb_txn_row *r);
  90 static void ovsdb_txn_row_prefree(struct ovsdb_txn_row *);
  91 static struct ovsdb_error * OVS_WARN_UNUSED_RESULT
  92 for_each_txn_row(struct ovsdb_txn *txn,
  93                       struct ovsdb_error *(*)(struct ovsdb_txn *,
  94                                               struct ovsdb_txn_row *));
  95
  96 /* Used by for_each_txn_row() to track tables and rows that have been
  97  * processed.  */
  98 static unsigned int serial;
  99
 100 struct ovsdb_txn *
 101 ovsdb_txn_create(struct ovsdb *db)
 102 {
 103     struct ovsdb_txn *txn = xmalloc(sizeof *txn);
 104     txn->db = db;
 105     list_init(&txn->txn_tables);
 106     ds_init(&txn->comment);
 107     return txn;
 108 }
 109
 110 static void
 111 ovsdb_txn_free(struct ovsdb_txn *txn)
 112 {
 113     ovs_assert(list_is_empty(&txn->txn_tables));
 114     ds_destroy(&txn->comment);
 115     free(txn);
 116 }
 117
 118 static struct ovsdb_error *
 119 ovsdb_txn_row_abort(struct ovsdb_txn *txn OVS_UNUSED,
 120                     struct ovsdb_txn_row *txn_row)
 121 {
 122     struct ovsdb_row *old = txn_row->old;
 123     struct ovsdb_row *new = txn_row->new;
 124
 125     ovsdb_txn_row_prefree(txn_row);
 126     if (!old) {
 127         if (new) {
 128             hmap_remove(&new->table->rows, &new->hmap_node);
 129         }
 130     } else if (!new) {
 131         hmap_insert(&old->table->rows, &old->hmap_node, ovsdb_row_hash(old));
 132     } else {
 133         hmap_replace(&new->table->rows, &new->hmap_node, &old->hmap_node);
 134     }
 135     ovsdb_row_destroy(new);
 136     free(txn_row);
 137
 138     return NULL;
 139 }
 140
 141 /* Returns the offset in bytes from the start of an ovsdb_row for 'table' to
 142  * the hmap_node for the index numbered 'i'. */
 143 static size_t
 144 ovsdb_row_index_offset__(const struct ovsdb_table *table, size_t i)
 145 {
 146     size_t n_fields = shash_count(&table->schema->columns);
 147     return (offsetof(struct ovsdb_row, fields)
 148             + n_fields * sizeof(struct ovsdb_datum)
 149             + i * sizeof(struct hmap_node));
 150 }
 151
 152 /* Returns the hmap_node in 'row' for the index numbered 'i'. */
 153 static struct hmap_node *
 154 ovsdb_row_get_index_node(struct ovsdb_row *row, size_t i)
 155 {
 156     return (void *) ((char *) row + ovsdb_row_index_offset__(row->table, i));
 157 }
 158
 159 /* Returns the ovsdb_row given 'index_node', which is a pointer to that row's
 160  * hmap_node for the index numbered 'i' within 'table'. */
 161 static struct ovsdb_row *
 162 ovsdb_row_from_index_node(struct hmap_node *index_node,
 163                           const struct ovsdb_table *table, size_t i)
 164 {
 165     return (void *) ((char *) index_node - ovsdb_row_index_offset__(table, i));
 166 }
 167
 168 void
 169 ovsdb_txn_abort(struct ovsdb_txn *txn)
 170 {
 171     ovsdb_error_assert(for_each_txn_row(txn, ovsdb_txn_row_abort));
 172     ovsdb_txn_free(txn);
 173 }
 174
 175 static struct ovsdb_txn_row *
 176 find_txn_row(const struct ovsdb_table *table, const struct uuid *uuid)
 177 {
 178     struct ovsdb_txn_row *txn_row;
 179
 180     if (!table->txn_table) {
 181         return NULL;
 182     }
 183
 184     HMAP_FOR_EACH_WITH_HASH (txn_row, hmap_node,
 185                              uuid_hash(uuid), &table->txn_table->txn_rows) {
 186         if (uuid_equals(uuid, &txn_row->uuid)) {
 187             return txn_row;
 188         }
 189     }
 190
 191     return NULL;
 192 }
 193
 194 static struct ovsdb_txn_row *
 195 find_or_make_txn_row(struct ovsdb_txn *txn, const struct ovsdb_table *table,
 196                      const struct uuid *uuid)
 197 {
 198     struct ovsdb_txn_row *txn_row = find_txn_row(table, uuid);
 199     if (!txn_row) {
 200         const struct ovsdb_row *row = ovsdb_table_get_row(table, uuid);
 201         if (row) {
 202             txn_row = ovsdb_txn_row_modify(txn, row)->txn_row;
 203         }
 204     }
 205     return txn_row;
 206 }
 207
 208 static struct ovsdb_error * OVS_WARN_UNUSED_RESULT
 209 ovsdb_txn_adjust_atom_refs(struct ovsdb_txn *txn, const struct ovsdb_row *r,
 210                            const struct ovsdb_column *c,
 211                            const struct ovsdb_base_type *base,
 212                            const union ovsdb_atom *atoms, unsigned int n,
 213                            int delta)
 214 {
 215     const struct ovsdb_table *table;
 216     unsigned int i;
 217
 218     if (!ovsdb_base_type_is_strong_ref(base)) {
 219         return NULL;
 220     }
 221
 222     table = base->u.uuid.refTable;
 223     for (i = 0; i < n; i++) {
 224         const struct uuid *uuid = &atoms[i].uuid;
 225         struct ovsdb_txn_row *txn_row;
 226
 227         if (uuid_equals(uuid, ovsdb_row_get_uuid(r))) {
 228             /* Self-references don't count. */
 229             continue;
 230         }
 231
 232         txn_row = find_or_make_txn_row(txn, table, uuid);
 233         if (!txn_row) {
 234             return ovsdb_error("referential integrity violation",
 235                                "Table %s column %s row "UUID_FMT" "
 236                                "references nonexistent row "UUID_FMT" in "
 237                                "table %s.",
 238                                r->table->schema->name, c->name,
 239                                UUID_ARGS(ovsdb_row_get_uuid(r)),
 240                                UUID_ARGS(uuid), table->schema->name);
 241         }
 242         txn_row->n_refs += delta;
 243     }
 244
 245     return NULL;
 246 }
 247
 248 static struct ovsdb_error * OVS_WARN_UNUSED_RESULT
 249 ovsdb_txn_adjust_row_refs(struct ovsdb_txn *txn, const struct ovsdb_row *r,
 250                           const struct ovsdb_column *column, int delta)
 251 {
 252     const struct ovsdb_datum *field = &r->fields[column->index];
 253     struct ovsdb_error *error;
 254
 255     error = ovsdb_txn_adjust_atom_refs(txn, r, column, &column->type.key,
 256                                        field->keys, field->n, delta);
 257     if (!error) {
 258         error = ovsdb_txn_adjust_atom_refs(txn, r, column, &column->type.value,
 259                                            field->values, field->n, delta);
 260     }
 261     return error;
 262 }
 263
 264 static struct ovsdb_error * OVS_WARN_UNUSED_RESULT
 265 update_row_ref_count(struct ovsdb_txn *txn, struct ovsdb_txn_row *r)
 266 {
 267     struct ovsdb_table *table = r->table;
 268     struct shash_node *node;
 269
 270     SHASH_FOR_EACH (node, &table->schema->columns) {
 271         const struct ovsdb_column *column = node->data;
 272         struct ovsdb_error *error;
 273
 274         if (r->old) {
 275             error = ovsdb_txn_adjust_row_refs(txn, r->old, column, -1);
 276             if (error) {
 277                 return OVSDB_WRAP_BUG("error decreasing refcount", error);
 278             }
 279         }
 280         if (r->new) {
 281             error = ovsdb_txn_adjust_row_refs(txn, r->new, column, 1);
 282             if (error) {
 283                 return error;
 284             }
 285         }
 286     }
 287
 288     return NULL;
 289 }
 290
 291 static struct ovsdb_error * OVS_WARN_UNUSED_RESULT
 292 check_ref_count(struct ovsdb_txn *txn OVS_UNUSED, struct ovsdb_txn_row *r)
 293 {
 294     if (r->new || !r->n_refs) {
 295         return NULL;
 296     } else {
 297         return ovsdb_error("referential integrity violation",
 298                            "cannot delete %s row "UUID_FMT" because "
 299                            "of %"PRIuSIZE" remaining reference(s)",
 300                            r->table->schema->name, UUID_ARGS(&r->uuid),
 301                            r->n_refs);
 302     }
 303 }
 304
 305 static struct ovsdb_error * OVS_WARN_UNUSED_RESULT
 306 delete_row_refs(struct ovsdb_txn *txn, const struct ovsdb_row *row,
 307                 const struct ovsdb_base_type *base,
 308                 const union ovsdb_atom *atoms, unsigned int n)
 309 {
 310     const struct ovsdb_table *table;
 311     unsigned int i;
 312
 313     if (!ovsdb_base_type_is_strong_ref(base)) {
 314         return NULL;
 315     }
 316
 317     table = base->u.uuid.refTable;
 318     for (i = 0; i < n; i++) {
 319         const struct uuid *uuid = &atoms[i].uuid;
 320         struct ovsdb_txn_row *txn_row;
 321
 322         if (uuid_equals(uuid, ovsdb_row_get_uuid(row))) {
 323             /* Self-references don't count. */
 324             continue;
 325         }
 326
 327         txn_row = find_or_make_txn_row(txn, table, uuid);
 328         if (!txn_row) {
 329             return OVSDB_BUG("strong ref target missing");
 330         } else if (!txn_row->n_refs) {
 331             return OVSDB_BUG("strong ref target has zero n_refs");
 332         } else if (!txn_row->new) {
 333             return OVSDB_BUG("deleted strong ref target");
 334         }
 335
 336         if (--txn_row->n_refs == 0) {
 337             struct ovsdb_error *error = delete_garbage_row(txn, txn_row);
 338             if (error) {
 339                 return error;
 340             }
 341         }
 342     }
 343
 344     return NULL;
 345 }
 346
 347 static struct ovsdb_error * OVS_WARN_UNUSED_RESULT
 348 delete_garbage_row(struct ovsdb_txn *txn, struct ovsdb_txn_row *txn_row)
 349 {
 350     struct shash_node *node;
 351     struct ovsdb_row *row;
 352
 353     if (txn_row->table->schema->is_root) {
 354         return NULL;
 355     }
 356
 357     row = txn_row->new;
 358     txn_row->new = NULL;
 359     hmap_remove(&txn_row->table->rows, &row->hmap_node);
 360     SHASH_FOR_EACH (node, &txn_row->table->schema->columns) {
 361         const struct ovsdb_column *column = node->data;
 362         const struct ovsdb_datum *field = &row->fields[column->index];
 363         struct ovsdb_error *error;
 364
 365         error = delete_row_refs(txn, row,
 366                                 &column->type.key, field->keys, field->n);
 367         if (error) {
 368             return error;
 369         }
 370
 371         error = delete_row_refs(txn, row,
 372                                 &column->type.value, field->values, field->n);
 373         if (error) {
 374             return error;
 375         }
 376     }
 377     ovsdb_row_destroy(row);
 378
 379     return NULL;
 380 }
 381
 382 static struct ovsdb_error * OVS_WARN_UNUSED_RESULT
 383 collect_garbage(struct ovsdb_txn *txn, struct ovsdb_txn_row *txn_row)
 384 {
 385     if (txn_row->new && !txn_row->n_refs) {
 386         return delete_garbage_row(txn, txn_row);
 387     }
 388     return NULL;
 389 }
 390
 391 static struct ovsdb_error * OVS_WARN_UNUSED_RESULT
 392 update_ref_counts(struct ovsdb_txn *txn)
 393 {
 394     struct ovsdb_error *error;
 395
 396     error = for_each_txn_row(txn, update_row_ref_count);
 397     if (error) {
 398         return error;
 399     }
 400
 401     return for_each_txn_row(txn, check_ref_count);
 402 }
 403
 404 static struct ovsdb_error *
 405 ovsdb_txn_row_commit(struct ovsdb_txn *txn OVS_UNUSED,
 406                      struct ovsdb_txn_row *txn_row)
 407 {
 408     size_t n_indexes = txn_row->table->schema->n_indexes;
 409
 410     if (txn_row->old) {
 411         size_t i;
 412
 413         for (i = 0; i < n_indexes; i++) {
 414             struct hmap_node *node = ovsdb_row_get_index_node(txn_row->old, i);
 415             hmap_remove(&txn_row->table->indexes[i], node);
 416         }
 417     }
 418     if (txn_row->new) {
 419         size_t i;
 420
 421         for (i = 0; i < n_indexes; i++) {
 422             struct hmap_node *node = ovsdb_row_get_index_node(txn_row->new, i);
 423             hmap_insert(&txn_row->table->indexes[i], node, node->hash);
 424         }
 425     }
 426
 427     ovsdb_txn_row_prefree(txn_row);
 428     if (txn_row->new) {
 429         txn_row->new->n_refs = txn_row->n_refs;
 430     }
 431     ovsdb_row_destroy(txn_row->old);
 432     free(txn_row);
 433
 434     return NULL;
 435 }
 436
 437 static void
 438 add_weak_ref(struct ovsdb_txn *txn,
 439              const struct ovsdb_row *src_, const struct ovsdb_row *dst_)
 440 {
 441     struct ovsdb_row *src = CONST_CAST(struct ovsdb_row *, src_);
 442     struct ovsdb_row *dst = CONST_CAST(struct ovsdb_row *, dst_);
 443     struct ovsdb_weak_ref *weak;
 444
 445     if (src == dst) {
 446         return;
 447     }
 448
 449     dst = ovsdb_txn_row_modify(txn, dst);
 450
 451     if (!list_is_empty(&dst->dst_refs)) {
 452         /* Omit duplicates. */
 453         weak = CONTAINER_OF(list_back(&dst->dst_refs),
 454                             struct ovsdb_weak_ref, dst_node);
 455         if (weak->src == src) {
 456             return;
 457         }
 458     }
 459
 460     weak = xmalloc(sizeof *weak);
 461     weak->src = src;
 462     list_push_back(&dst->dst_refs, &weak->dst_node);
 463     list_push_back(&src->src_refs, &weak->src_node);
 464 }
 465
 466 static struct ovsdb_error * OVS_WARN_UNUSED_RESULT
 467 assess_weak_refs(struct ovsdb_txn *txn, struct ovsdb_txn_row *txn_row)
 468 {
 469     struct ovsdb_table *table;
 470     struct shash_node *node;
 471
 472     if (txn_row->old) {
 473         /* Mark rows that have weak references to 'txn_row' as modified, so
 474          * that their weak references will get reassessed. */
 475         struct ovsdb_weak_ref *weak, *next;
 476
 477         LIST_FOR_EACH_SAFE (weak, next, dst_node, &txn_row->old->dst_refs) {
 478             if (!weak->src->txn_row) {
 479                 ovsdb_txn_row_modify(txn, weak->src);
 480             }
 481         }
 482     }
 483
 484     if (!txn_row->new) {
 485         /* We don't have to do anything about references that originate at
 486          * 'txn_row', because ovsdb_row_destroy() will remove those weak
 487          * references. */
 488         return NULL;
 489     }
 490
 491     table = txn_row->table;
 492     SHASH_FOR_EACH (node, &table->schema->columns) {
 493         const struct ovsdb_column *column = node->data;
 494         struct ovsdb_datum *datum = &txn_row->new->fields[column->index];
 495         unsigned int orig_n, i;
 496         bool zero = false;
 497
 498         orig_n = datum->n;
 499
 500         if (ovsdb_base_type_is_weak_ref(&column->type.key)) {
 501             for (i = 0; i < datum->n; ) {
 502                 const struct ovsdb_row *row;
 503
 504                 row = ovsdb_table_get_row(column->type.key.u.uuid.refTable,
 505                                           &datum->keys[i].uuid);
 506                 if (row) {
 507                     add_weak_ref(txn, txn_row->new, row);
 508                     i++;
 509                 } else {
 510                     if (uuid_is_zero(&datum->keys[i].uuid)) {
 511                         zero = true;
 512                     }
 513                     ovsdb_datum_remove_unsafe(datum, i, &column->type);
 514                 }
 515             }
 516         }
 517
 518         if (ovsdb_base_type_is_weak_ref(&column->type.value)) {
 519             for (i = 0; i < datum->n; ) {
 520                 const struct ovsdb_row *row;
 521
 522                 row = ovsdb_table_get_row(column->type.value.u.uuid.refTable,
 523                                           &datum->values[i].uuid);
 524                 if (row) {
 525                     add_weak_ref(txn, txn_row->new, row);
 526                     i++;
 527                 } else {
 528                     if (uuid_is_zero(&datum->values[i].uuid)) {
 529                         zero = true;
 530                     }
 531                     ovsdb_datum_remove_unsafe(datum, i, &column->type);
 532                 }
 533             }
 534         }
 535
 536         if (datum->n != orig_n) {
 537             bitmap_set1(txn_row->changed, OVSDB_COL_VERSION);
 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 * OVS_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 * OVS_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                                "%"PRIuSIZE" 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, 1, 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 * OVS_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 * OVS_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 static 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             ovs_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 struct ovsdb_error *
 826 ovsdb_txn_commit(struct ovsdb_txn *txn, bool durable)
 827 {
 828    struct ovsdb_error *err;
 829
 830    PERF(__func__, err = ovsdb_txn_commit_(txn, durable));
 831    return err;
 832 }
 833
 834 void
 835 ovsdb_txn_for_each_change(const struct ovsdb_txn *txn,
 836                           ovsdb_txn_row_cb_func *cb, void *aux)
 837 {
 838     struct ovsdb_txn_table *t;
 839     struct ovsdb_txn_row *r;
 840
 841     LIST_FOR_EACH (t, node, &txn->txn_tables) {
 842         HMAP_FOR_EACH (r, hmap_node, &t->txn_rows) {
 843             if ((r->old || r->new) && !cb(r->old, r->new, r->changed, aux)) {
 844                 break;
 845             }
 846         }
 847    }
 848 }
 849
 850 static struct ovsdb_txn_table *
 851 ovsdb_txn_create_txn_table(struct ovsdb_txn *txn, struct ovsdb_table *table)
 852 {
 853     if (!table->txn_table) {
 854         struct ovsdb_txn_table *txn_table;
 855         size_t i;
 856
 857         table->txn_table = txn_table = xmalloc(sizeof *table->txn_table);
 858         txn_table->table = table;
 859         hmap_init(&txn_table->txn_rows);
 860         txn_table->serial = serial - 1;
 861         txn_table->txn_indexes = xmalloc(table->schema->n_indexes
 862                                          * sizeof *txn_table->txn_indexes);
 863         for (i = 0; i < table->schema->n_indexes; i++) {
 864             hmap_init(&txn_table->txn_indexes[i]);
 865         }
 866         list_push_back(&txn->txn_tables, &txn_table->node);
 867     }
 868     return table->txn_table;
 869 }
 870
 871 static struct ovsdb_txn_row *
 872 ovsdb_txn_row_create(struct ovsdb_txn *txn, struct ovsdb_table *table,
 873                      const struct ovsdb_row *old_, struct ovsdb_row *new)
 874 {
 875     const struct ovsdb_row *row = old_ ? old_ : new;
 876     struct ovsdb_row *old = CONST_CAST(struct ovsdb_row *, old_);
 877     size_t n_columns = shash_count(&table->schema->columns);
 878     struct ovsdb_txn_table *txn_table;
 879     struct ovsdb_txn_row *txn_row;
 880
 881     txn_row = xzalloc(offsetof(struct ovsdb_txn_row, changed)
 882                       + bitmap_n_bytes(n_columns));
 883     txn_row->uuid = *ovsdb_row_get_uuid(row);
 884     txn_row->table = row->table;
 885     txn_row->old = old;
 886     txn_row->new = new;
 887     txn_row->n_refs = old ? old->n_refs : 0;
 888     txn_row->serial = serial - 1;
 889
 890     if (old) {
 891         old->txn_row = txn_row;
 892     }
 893     if (new) {
 894         new->txn_row = txn_row;
 895     }
 896
 897     txn_table = ovsdb_txn_create_txn_table(txn, table);
 898     hmap_insert(&txn_table->txn_rows, &txn_row->hmap_node,
 899                 ovsdb_row_hash(old ? old : new));
 900
 901     return txn_row;
 902 }
 903
 904 struct ovsdb_row *
 905 ovsdb_txn_row_modify(struct ovsdb_txn *txn, const struct ovsdb_row *ro_row_)
 906 {
 907     struct ovsdb_row *ro_row = CONST_CAST(struct ovsdb_row *, ro_row_);
 908
 909     if (ro_row->txn_row) {
 910         ovs_assert(ro_row == ro_row->txn_row->new);
 911         return ro_row;
 912     } else {
 913         struct ovsdb_table *table = ro_row->table;
 914         struct ovsdb_row *rw_row;
 915
 916         rw_row = ovsdb_row_clone(ro_row);
 917         rw_row->n_refs = ro_row->n_refs;
 918         uuid_generate(ovsdb_row_get_version_rw(rw_row));
 919         ovsdb_txn_row_create(txn, table, ro_row, rw_row);
 920         hmap_replace(&table->rows, &ro_row->hmap_node, &rw_row->hmap_node);
 921
 922         return rw_row;
 923     }
 924 }
 925
 926 void
 927 ovsdb_txn_row_insert(struct ovsdb_txn *txn, struct ovsdb_row *row)
 928 {
 929     uint32_t hash = ovsdb_row_hash(row);
 930     struct ovsdb_table *table = row->table;
 931
 932     uuid_generate(ovsdb_row_get_version_rw(row));
 933
 934     ovsdb_txn_row_create(txn, table, NULL, row);
 935     hmap_insert(&table->rows, &row->hmap_node, hash);
 936 }
 937
 938 /* 'row' must be assumed destroyed upon return; the caller must not reference
 939  * it again. */
 940 void
 941 ovsdb_txn_row_delete(struct ovsdb_txn *txn, const struct ovsdb_row *row_)
 942 {
 943     struct ovsdb_row *row = CONST_CAST(struct ovsdb_row *, row_);
 944     struct ovsdb_table *table = row->table;
 945     struct ovsdb_txn_row *txn_row = row->txn_row;
 946
 947     hmap_remove(&table->rows, &row->hmap_node);
 948
 949     if (!txn_row) {
 950         ovsdb_txn_row_create(txn, table, row, NULL);
 951     } else {
 952         ovs_assert(txn_row->new == row);
 953         if (txn_row->old) {
 954             txn_row->new = NULL;
 955         } else {
 956             hmap_remove(&table->txn_table->txn_rows, &txn_row->hmap_node);
 957             free(txn_row);
 958         }
 959         ovsdb_row_destroy(row);
 960     }
 961 }
 962
 963 void
 964 ovsdb_txn_add_comment(struct ovsdb_txn *txn, const char *s)
 965 {
 966     if (txn->comment.length) {
 967         ds_put_char(&txn->comment, '\n');
 968     }
 969     ds_put_cstr(&txn->comment, s);
 970 }
 971
 972 const char *
 973 ovsdb_txn_get_comment(const struct ovsdb_txn *txn)
 974 {
 975     return txn->comment.length ? ds_cstr_ro(&txn->comment) : NULL;
 976 }
 977 \f
 978 static void
 979 ovsdb_txn_row_prefree(struct ovsdb_txn_row *txn_row)
 980 {
 981     struct ovsdb_txn_table *txn_table = txn_row->table->txn_table;
 982
 983     txn_table->n_processed--;
 984     hmap_remove(&txn_table->txn_rows, &txn_row->hmap_node);
 985
 986     if (txn_row->old) {
 987         txn_row->old->txn_row = NULL;
 988     }
 989     if (txn_row->new) {
 990         txn_row->new->txn_row = NULL;
 991     }
 992 }
 993
 994 static void
 995 ovsdb_txn_table_destroy(struct ovsdb_txn_table *txn_table)
 996 {
 997     size_t i;
 998
 999     ovs_assert(hmap_is_empty(&txn_table->txn_rows));
1000
1001     for (i = 0; i < txn_table->table->schema->n_indexes; i++) {
1002         hmap_destroy(&txn_table->txn_indexes[i]);
1003     }
1004     free(txn_table->txn_indexes);
1005
1006     txn_table->table->txn_table = NULL;
1007     hmap_destroy(&txn_table->txn_rows);
1008     list_remove(&txn_table->node);
1009     free(txn_table);
1010 }
1011
1012 /* Calls 'cb' for every txn_row within 'txn'.  If 'cb' returns nonnull, this
1013  * aborts the iteration and for_each_txn_row() passes the error up.  Otherwise,
1014  * returns a null pointer after iteration is complete.
1015  *
1016  * 'cb' may insert new txn_rows and new txn_tables into 'txn'.  It may delete
1017  * the txn_row that it is passed in, or txn_rows in txn_tables other than the
1018  * one passed to 'cb'.  It may *not* delete txn_rows other than the one passed
1019  * in within the same txn_table.  It may *not* delete any txn_tables.  As long
1020  * as these rules are followed, 'cb' will be called exactly once for each
1021  * txn_row in 'txn', even those added by 'cb'.
1022  *
1023  * (Even though 'cb' is not allowed to delete some txn_rows, it can still
1024  * delete any actual row by clearing a txn_row's 'new' member.)
1025  */
1026 static struct ovsdb_error * OVS_WARN_UNUSED_RESULT
1027 for_each_txn_row(struct ovsdb_txn *txn,
1028                  struct ovsdb_error *(*cb)(struct ovsdb_txn *,
1029                                            struct ovsdb_txn_row *))
1030 {
1031     bool any_work;
1032
1033     serial++;
1034
1035     do {
1036         struct ovsdb_txn_table *t, *next_txn_table;
1037
1038         any_work = false;
1039         LIST_FOR_EACH_SAFE (t, next_txn_table, node, &txn->txn_tables) {
1040             if (t->serial != serial) {
1041                 t->serial = serial;
1042                 t->n_processed = 0;
1043             }
1044
1045             while (t->n_processed < hmap_count(&t->txn_rows)) {
1046                 struct ovsdb_txn_row *r, *next_txn_row;
1047
1048                 HMAP_FOR_EACH_SAFE (r, next_txn_row, hmap_node, &t->txn_rows) {
1049                     if (r->serial != serial) {
1050                         struct ovsdb_error *error;
1051
1052                         r->serial = serial;
1053                         t->n_processed++;
1054                         any_work = true;
1055
1056                         error = cb(txn, r);
1057                         if (error) {
1058                             return error;
1059                         }
1060                     }
1061                 }
1062             }
1063             if (hmap_is_empty(&t->txn_rows)) {
1064                 /* Table is empty.  Drop it. */
1065                 ovsdb_txn_table_destroy(t);
1066             }
1067         }
1068     } while (any_work);
1069
1070     return NULL;
1071 }