ovsdb/transaction.c

   1 /* Copyright (c) 2009, 2010, 2011, 2012, 2013, 2014, 2015 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 "openvswitch/dynamic-string.h"
  22 #include "hash.h"
  23 #include "openvswitch/hmap.h"
  24 #include "openvswitch/json.h"
  25 #include "openvswitch/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     ovs_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(ovs_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 (bitmap_is_set(r->changed, column->index)) {
 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
 290     return NULL;
 291 }
 292
 293 static struct ovsdb_error * OVS_WARN_UNUSED_RESULT
 294 check_ref_count(struct ovsdb_txn *txn OVS_UNUSED, struct ovsdb_txn_row *r)
 295 {
 296     if (r->new || !r->n_refs) {
 297         return NULL;
 298     } else {
 299         return ovsdb_error("referential integrity violation",
 300                            "cannot delete %s row "UUID_FMT" because "
 301                            "of %"PRIuSIZE" remaining reference(s)",
 302                            r->table->schema->name, UUID_ARGS(&r->uuid),
 303                            r->n_refs);
 304     }
 305 }
 306
 307 static struct ovsdb_error * OVS_WARN_UNUSED_RESULT
 308 delete_row_refs(struct ovsdb_txn *txn, const struct ovsdb_row *row,
 309                 const struct ovsdb_base_type *base,
 310                 const union ovsdb_atom *atoms, unsigned int n)
 311 {
 312     const struct ovsdb_table *table;
 313     unsigned int i;
 314
 315     if (!ovsdb_base_type_is_strong_ref(base)) {
 316         return NULL;
 317     }
 318
 319     table = base->u.uuid.refTable;
 320     for (i = 0; i < n; i++) {
 321         const struct uuid *uuid = &atoms[i].uuid;
 322         struct ovsdb_txn_row *txn_row;
 323
 324         if (uuid_equals(uuid, ovsdb_row_get_uuid(row))) {
 325             /* Self-references don't count. */
 326             continue;
 327         }
 328
 329         txn_row = find_or_make_txn_row(txn, table, uuid);
 330         if (!txn_row) {
 331             return OVSDB_BUG("strong ref target missing");
 332         } else if (!txn_row->n_refs) {
 333             return OVSDB_BUG("strong ref target has zero n_refs");
 334         } else if (!txn_row->new) {
 335             return OVSDB_BUG("deleted strong ref target");
 336         }
 337
 338         if (--txn_row->n_refs == 0) {
 339             struct ovsdb_error *error = delete_garbage_row(txn, txn_row);
 340             if (error) {
 341                 return error;
 342             }
 343         }
 344     }
 345
 346     return NULL;
 347 }
 348
 349 static struct ovsdb_error * OVS_WARN_UNUSED_RESULT
 350 delete_garbage_row(struct ovsdb_txn *txn, struct ovsdb_txn_row *txn_row)
 351 {
 352     struct shash_node *node;
 353     struct ovsdb_row *row;
 354
 355     if (txn_row->table->schema->is_root) {
 356         return NULL;
 357     }
 358
 359     row = txn_row->new;
 360     txn_row->new = NULL;
 361     hmap_remove(&txn_row->table->rows, &row->hmap_node);
 362     SHASH_FOR_EACH (node, &txn_row->table->schema->columns) {
 363         const struct ovsdb_column *column = node->data;
 364         const struct ovsdb_datum *field = &row->fields[column->index];
 365         struct ovsdb_error *error;
 366
 367         error = delete_row_refs(txn, row,
 368                                 &column->type.key, field->keys, field->n);
 369         if (error) {
 370             return error;
 371         }
 372
 373         error = delete_row_refs(txn, row,
 374                                 &column->type.value, field->values, field->n);
 375         if (error) {
 376             return error;
 377         }
 378     }
 379     ovsdb_row_destroy(row);
 380
 381     return NULL;
 382 }
 383
 384 static struct ovsdb_error * OVS_WARN_UNUSED_RESULT
 385 collect_garbage(struct ovsdb_txn *txn, struct ovsdb_txn_row *txn_row)
 386 {
 387     if (txn_row->new && !txn_row->n_refs) {
 388         return delete_garbage_row(txn, txn_row);
 389     }
 390     return NULL;
 391 }
 392
 393 static struct ovsdb_error * OVS_WARN_UNUSED_RESULT
 394 update_ref_counts(struct ovsdb_txn *txn)
 395 {
 396     struct ovsdb_error *error;
 397
 398     error = for_each_txn_row(txn, update_row_ref_count);
 399     if (error) {
 400         return error;
 401     }
 402
 403     return for_each_txn_row(txn, check_ref_count);
 404 }
 405
 406 static struct ovsdb_error *
 407 ovsdb_txn_row_commit(struct ovsdb_txn *txn OVS_UNUSED,
 408                      struct ovsdb_txn_row *txn_row)
 409 {
 410     size_t n_indexes = txn_row->table->schema->n_indexes;
 411
 412     if (txn_row->old) {
 413         size_t i;
 414
 415         for (i = 0; i < n_indexes; i++) {
 416             struct hmap_node *node = ovsdb_row_get_index_node(txn_row->old, i);
 417             hmap_remove(&txn_row->table->indexes[i], node);
 418         }
 419     }
 420     if (txn_row->new) {
 421         size_t i;
 422
 423         for (i = 0; i < n_indexes; i++) {
 424             struct hmap_node *node = ovsdb_row_get_index_node(txn_row->new, i);
 425             hmap_insert(&txn_row->table->indexes[i], node, node->hash);
 426         }
 427     }
 428
 429     ovsdb_txn_row_prefree(txn_row);
 430     if (txn_row->new) {
 431         txn_row->new->n_refs = txn_row->n_refs;
 432     }
 433     ovsdb_row_destroy(txn_row->old);
 434     free(txn_row);
 435
 436     return NULL;
 437 }
 438
 439 static void
 440 add_weak_ref(struct ovsdb_txn *txn,
 441              const struct ovsdb_row *src_, const struct ovsdb_row *dst_)
 442 {
 443     struct ovsdb_row *src = CONST_CAST(struct ovsdb_row *, src_);
 444     struct ovsdb_row *dst = CONST_CAST(struct ovsdb_row *, dst_);
 445     struct ovsdb_weak_ref *weak;
 446
 447     if (src == dst) {
 448         return;
 449     }
 450
 451     dst = ovsdb_txn_row_modify(txn, dst);
 452
 453     if (!ovs_list_is_empty(&dst->dst_refs)) {
 454         /* Omit duplicates. */
 455         weak = CONTAINER_OF(ovs_list_back(&dst->dst_refs),
 456                             struct ovsdb_weak_ref, dst_node);
 457         if (weak->src == src) {
 458             return;
 459         }
 460     }
 461
 462     weak = xmalloc(sizeof *weak);
 463     weak->src = src;
 464     ovs_list_push_back(&dst->dst_refs, &weak->dst_node);
 465     ovs_list_push_back(&src->src_refs, &weak->src_node);
 466 }
 467
 468 static struct ovsdb_error * OVS_WARN_UNUSED_RESULT
 469 assess_weak_refs(struct ovsdb_txn *txn, struct ovsdb_txn_row *txn_row)
 470 {
 471     struct ovsdb_table *table;
 472     struct shash_node *node;
 473
 474     if (txn_row->old) {
 475         /* Mark rows that have weak references to 'txn_row' as modified, so
 476          * that their weak references will get reassessed. */
 477         struct ovsdb_weak_ref *weak, *next;
 478
 479         LIST_FOR_EACH_SAFE (weak, next, dst_node, &txn_row->old->dst_refs) {
 480             if (!weak->src->txn_row) {
 481                 ovsdb_txn_row_modify(txn, weak->src);
 482             }
 483         }
 484     }
 485
 486     if (!txn_row->new) {
 487         /* We don't have to do anything about references that originate at
 488          * 'txn_row', because ovsdb_row_destroy() will remove those weak
 489          * references. */
 490         return NULL;
 491     }
 492
 493     table = txn_row->table;
 494     SHASH_FOR_EACH (node, &table->schema->columns) {
 495         const struct ovsdb_column *column = node->data;
 496         struct ovsdb_datum *datum = &txn_row->new->fields[column->index];
 497         unsigned int orig_n, i;
 498         bool zero = false;
 499
 500         orig_n = datum->n;
 501
 502         if (ovsdb_base_type_is_weak_ref(&column->type.key)) {
 503             for (i = 0; i < datum->n; ) {
 504                 const struct ovsdb_row *row;
 505
 506                 row = ovsdb_table_get_row(column->type.key.u.uuid.refTable,
 507                                           &datum->keys[i].uuid);
 508                 if (row) {
 509                     add_weak_ref(txn, txn_row->new, row);
 510                     i++;
 511                 } else {
 512                     if (uuid_is_zero(&datum->keys[i].uuid)) {
 513                         zero = true;
 514                     }
 515                     ovsdb_datum_remove_unsafe(datum, i, &column->type);
 516                 }
 517             }
 518         }
 519
 520         if (ovsdb_base_type_is_weak_ref(&column->type.value)) {
 521             for (i = 0; i < datum->n; ) {
 522                 const struct ovsdb_row *row;
 523
 524                 row = ovsdb_table_get_row(column->type.value.u.uuid.refTable,
 525                                           &datum->values[i].uuid);
 526                 if (row) {
 527                     add_weak_ref(txn, txn_row->new, row);
 528                     i++;
 529                 } else {
 530                     if (uuid_is_zero(&datum->values[i].uuid)) {
 531                         zero = true;
 532                     }
 533                     ovsdb_datum_remove_unsafe(datum, i, &column->type);
 534                 }
 535             }
 536         }
 537
 538         if (datum->n != orig_n) {
 539             bitmap_set1(txn_row->changed, column->index);
 540             ovsdb_datum_sort_assert(datum, column->type.key.type);
 541             if (datum->n < column->type.n_min) {
 542                 const struct uuid *row_uuid = ovsdb_row_get_uuid(txn_row->new);
 543                 if (zero && !txn_row->old) {
 544                     return ovsdb_error(
 545                         "constraint violation",
 546                         "Weak reference column \"%s\" in \"%s\" row "UUID_FMT
 547                         " (inserted within this transaction) contained "
 548                         "all-zeros UUID (probably as the default value for "
 549                         "this column) but deleting this value caused a "
 550                         "constraint volation because this column is not "
 551                         "allowed to be empty.", column->name,
 552                         table->schema->name, UUID_ARGS(row_uuid));
 553                 } else {
 554                     return ovsdb_error(
 555                         "constraint violation",
 556                         "Deletion of %u weak reference(s) to deleted (or "
 557                         "never-existing) rows from column \"%s\" in \"%s\" "
 558                         "row "UUID_FMT" %scaused this column to become empty, "
 559                         "but constraints on this column disallow an "
 560                         "empty column.",
 561                         orig_n - datum->n, column->name, table->schema->name,
 562                         UUID_ARGS(row_uuid),
 563                         (txn_row->old
 564                          ? ""
 565                          : "(inserted within this transaction) "));
 566                 }
 567             }
 568         }
 569     }
 570
 571     return NULL;
 572 }
 573
 574 static struct ovsdb_error * OVS_WARN_UNUSED_RESULT
 575 determine_changes(struct ovsdb_txn *txn, struct ovsdb_txn_row *txn_row)
 576 {
 577     struct ovsdb_table *table = txn_row->table;
 578
 579     if (txn_row->old && txn_row->new) {
 580         struct shash_node *node;
 581         bool changed = false;
 582
 583         SHASH_FOR_EACH (node, &table->schema->columns) {
 584             const struct ovsdb_column *column = node->data;
 585             const struct ovsdb_type *type = &column->type;
 586             unsigned int idx = column->index;
 587
 588             if (!ovsdb_datum_equals(&txn_row->old->fields[idx],
 589                                     &txn_row->new->fields[idx],
 590                                     type)) {
 591                 bitmap_set1(txn_row->changed, idx);
 592                 changed = true;
 593             }
 594         }
 595
 596         if (!changed) {
 597             /* Nothing actually changed in this row, so drop it. */
 598             ovsdb_txn_row_abort(txn, txn_row);
 599         }
 600     } else {
 601         bitmap_set_multiple(txn_row->changed, 0,
 602                             shash_count(&table->schema->columns), 1);
 603     }
 604
 605     return NULL;
 606 }
 607
 608 static struct ovsdb_error * OVS_WARN_UNUSED_RESULT
 609 check_max_rows(struct ovsdb_txn *txn)
 610 {
 611     struct ovsdb_txn_table *t;
 612
 613     LIST_FOR_EACH (t, node, &txn->txn_tables) {
 614         size_t n_rows = hmap_count(&t->table->rows);
 615         unsigned int max_rows = t->table->schema->max_rows;
 616
 617         if (n_rows > max_rows) {
 618             return ovsdb_error("constraint violation",
 619                                "transaction causes \"%s\" table to contain "
 620                                "%"PRIuSIZE" rows, greater than the schema-defined "
 621                                "limit of %u row(s)",
 622                                t->table->schema->name, n_rows, max_rows);
 623         }
 624     }
 625
 626     return NULL;
 627 }
 628
 629 static struct ovsdb_row *
 630 ovsdb_index_search(struct hmap *index, struct ovsdb_row *row, size_t i,
 631                    uint32_t hash)
 632 {
 633     const struct ovsdb_table *table = row->table;
 634     const struct ovsdb_column_set *columns = &table->schema->indexes[i];
 635     struct hmap_node *node;
 636
 637     for (node = hmap_first_with_hash(index, hash); node;
 638          node = hmap_next_with_hash(node)) {
 639         struct ovsdb_row *irow = ovsdb_row_from_index_node(node, table, i);
 640         if (ovsdb_row_equal_columns(row, irow, columns)) {
 641             return irow;
 642         }
 643     }
 644
 645     return NULL;
 646 }
 647
 648 static void
 649 duplicate_index_row__(const struct ovsdb_column_set *index,
 650                       const struct ovsdb_row *row,
 651                       const char *title,
 652                       struct ds *out)
 653 {
 654     size_t n_columns = shash_count(&row->table->schema->columns);
 655
 656     ds_put_format(out, "%s row, with UUID "UUID_FMT", ",
 657                   title, UUID_ARGS(ovsdb_row_get_uuid(row)));
 658     if (!row->txn_row
 659         || bitmap_scan(row->txn_row->changed, 1, 0, n_columns) == n_columns) {
 660         ds_put_cstr(out, "existed in the database before this "
 661                     "transaction and was not modified by the transaction.");
 662     } else if (!row->txn_row->old) {
 663         ds_put_cstr(out, "was inserted by this transaction.");
 664     } else if (ovsdb_row_equal_columns(row->txn_row->old,
 665                                        row->txn_row->new, index)) {
 666         ds_put_cstr(out, "existed in the database before this "
 667                     "transaction, which modified some of the row's columns "
 668                     "but not any columns in this index.");
 669     } else {
 670         ds_put_cstr(out, "had the following index values before the "
 671                     "transaction: ");
 672         ovsdb_row_columns_to_string(row->txn_row->old, index, out);
 673         ds_put_char(out, '.');
 674     }
 675 }
 676
 677 static struct ovsdb_error * OVS_WARN_UNUSED_RESULT
 678 duplicate_index_row(const struct ovsdb_column_set *index,
 679                     const struct ovsdb_row *a,
 680                     const struct ovsdb_row *b)
 681 {
 682     struct ovsdb_column_set all_columns;
 683     struct ovsdb_error *error;
 684     char *index_s;
 685     struct ds s;
 686
 687     /* Put 'a' and 'b' in a predictable order to make error messages
 688      * reproducible for testing. */
 689     ovsdb_column_set_init(&all_columns);
 690     ovsdb_column_set_add_all(&all_columns, a->table);
 691     if (ovsdb_row_compare_columns_3way(a, b, &all_columns) < 0) {
 692         const struct ovsdb_row *tmp = a;
 693         a = b;
 694         b = tmp;
 695     }
 696     ovsdb_column_set_destroy(&all_columns);
 697
 698     index_s = ovsdb_column_set_to_string(index);
 699
 700     ds_init(&s);
 701     ds_put_format(&s, "Transaction causes multiple rows in \"%s\" table to "
 702                   "have identical values (", a->table->schema->name);
 703     ovsdb_row_columns_to_string(a, index, &s);
 704     ds_put_format(&s, ") for index on %s.  ", index_s);
 705     duplicate_index_row__(index, a, "First", &s);
 706     ds_put_cstr(&s, "  ");
 707     duplicate_index_row__(index, b, "Second", &s);
 708
 709     free(index_s);
 710
 711     error = ovsdb_error("constraint violation", "%s", ds_cstr(&s));
 712     ds_destroy(&s);
 713     return error;
 714 }
 715
 716 static struct ovsdb_error * OVS_WARN_UNUSED_RESULT
 717 check_index_uniqueness(struct ovsdb_txn *txn OVS_UNUSED,
 718                        struct ovsdb_txn_row *txn_row)
 719 {
 720     struct ovsdb_txn_table *txn_table = txn_row->table->txn_table;
 721     struct ovsdb_table *table = txn_row->table;
 722     struct ovsdb_row *row = txn_row->new;
 723     size_t i;
 724
 725     if (!row) {
 726         return NULL;
 727     }
 728
 729     for (i = 0; i < table->schema->n_indexes; i++) {
 730         const struct ovsdb_column_set *index = &table->schema->indexes[i];
 731         struct ovsdb_row *irow;
 732         uint32_t hash;
 733
 734         hash = ovsdb_row_hash_columns(row, index, 0);
 735         irow = ovsdb_index_search(&txn_table->txn_indexes[i], row, i, hash);
 736         if (irow) {
 737             return duplicate_index_row(index, irow, row);
 738         }
 739
 740         irow = ovsdb_index_search(&table->indexes[i], row, i, hash);
 741         if (irow && !irow->txn_row) {
 742             return duplicate_index_row(index, irow, row);
 743         }
 744
 745         hmap_insert(&txn_table->txn_indexes[i],
 746                     ovsdb_row_get_index_node(row, i), hash);
 747     }
 748
 749     return NULL;
 750 }
 751
 752 static struct ovsdb_error * OVS_WARN_UNUSED_RESULT
 753 update_version(struct ovsdb_txn *txn OVS_UNUSED, struct ovsdb_txn_row *txn_row)
 754 {
 755     struct ovsdb_table *table = txn_row->table;
 756     size_t n_columns = shash_count(&table->schema->columns);
 757
 758     if (txn_row->old && txn_row->new
 759         && !bitmap_is_all_zeros(txn_row->changed, n_columns)) {
 760         bitmap_set1(txn_row->changed, OVSDB_COL_VERSION);
 761         uuid_generate(ovsdb_row_get_version_rw(txn_row->new));
 762     }
 763
 764     return NULL;
 765 }
 766
 767 static struct ovsdb_error *
 768 ovsdb_txn_commit_(struct ovsdb_txn *txn, bool durable)
 769 {
 770     struct ovsdb_replica *replica;
 771     struct ovsdb_error *error;
 772
 773     /* Figure out what actually changed, and abort early if the transaction
 774      * was really a no-op. */
 775     error = for_each_txn_row(txn, determine_changes);
 776     if (error) {
 777         return OVSDB_WRAP_BUG("can't happen", error);
 778     }
 779     if (ovs_list_is_empty(&txn->txn_tables)) {
 780         ovsdb_txn_abort(txn);
 781         return NULL;
 782     }
 783
 784     /* Update reference counts and check referential integrity. */
 785     error = update_ref_counts(txn);
 786     if (error) {
 787         ovsdb_txn_abort(txn);
 788         return error;
 789     }
 790
 791     /* Delete unreferenced, non-root rows. */
 792     error = for_each_txn_row(txn, collect_garbage);
 793     if (error) {
 794         ovsdb_txn_abort(txn);
 795         return OVSDB_WRAP_BUG("can't happen", error);
 796     }
 797
 798     /* Check maximum rows table constraints. */
 799     error = check_max_rows(txn);
 800     if (error) {
 801         ovsdb_txn_abort(txn);
 802         return error;
 803     }
 804
 805     /* Check reference counts and remove bad references for "weak" referential
 806      * integrity. */
 807     error = for_each_txn_row(txn, assess_weak_refs);
 808     if (error) {
 809         ovsdb_txn_abort(txn);
 810         return error;
 811     }
 812
 813     /* Verify that the indexes will still be unique post-transaction. */
 814     error = for_each_txn_row(txn, check_index_uniqueness);
 815     if (error) {
 816         ovsdb_txn_abort(txn);
 817         return error;
 818     }
 819
 820     /* Update _version for rows that changed.  */
 821     error = for_each_txn_row(txn, update_version);
 822     if (error) {
 823         return OVSDB_WRAP_BUG("can't happen", error);
 824     }
 825
 826     /* Send the commit to each replica. */
 827     LIST_FOR_EACH (replica, node, &txn->db->replicas) {
 828         error = (replica->class->commit)(replica, txn, durable);
 829         if (error) {
 830             /* We don't support two-phase commit so only the first replica is
 831              * allowed to report an error. */
 832             ovs_assert(&replica->node == txn->db->replicas.next);
 833
 834             ovsdb_txn_abort(txn);
 835             return error;
 836         }
 837     }
 838
 839     /* Finalize commit. */
 840     txn->db->run_triggers = true;
 841     ovsdb_error_assert(for_each_txn_row(txn, ovsdb_txn_row_commit));
 842     ovsdb_txn_free(txn);
 843
 844     return NULL;
 845 }
 846
 847 struct ovsdb_error *
 848 ovsdb_txn_commit(struct ovsdb_txn *txn, bool durable)
 849 {
 850    struct ovsdb_error *err;
 851
 852    PERF(__func__, err = ovsdb_txn_commit_(txn, durable));
 853    return err;
 854 }
 855
 856 void
 857 ovsdb_txn_for_each_change(const struct ovsdb_txn *txn,
 858                           ovsdb_txn_row_cb_func *cb, void *aux)
 859 {
 860     struct ovsdb_txn_table *t;
 861     struct ovsdb_txn_row *r;
 862
 863     LIST_FOR_EACH (t, node, &txn->txn_tables) {
 864         HMAP_FOR_EACH (r, hmap_node, &t->txn_rows) {
 865             if ((r->old || r->new) && !cb(r->old, r->new, r->changed, aux)) {
 866                 break;
 867             }
 868         }
 869    }
 870 }
 871
 872 static struct ovsdb_txn_table *
 873 ovsdb_txn_create_txn_table(struct ovsdb_txn *txn, struct ovsdb_table *table)
 874 {
 875     if (!table->txn_table) {
 876         struct ovsdb_txn_table *txn_table;
 877         size_t i;
 878
 879         table->txn_table = txn_table = xmalloc(sizeof *table->txn_table);
 880         txn_table->table = table;
 881         hmap_init(&txn_table->txn_rows);
 882         txn_table->serial = serial - 1;
 883         txn_table->txn_indexes = xmalloc(table->schema->n_indexes
 884                                          * sizeof *txn_table->txn_indexes);
 885         for (i = 0; i < table->schema->n_indexes; i++) {
 886             hmap_init(&txn_table->txn_indexes[i]);
 887         }
 888         ovs_list_push_back(&txn->txn_tables, &txn_table->node);
 889     }
 890     return table->txn_table;
 891 }
 892
 893 static struct ovsdb_txn_row *
 894 ovsdb_txn_row_create(struct ovsdb_txn *txn, struct ovsdb_table *table,
 895                      const struct ovsdb_row *old_, struct ovsdb_row *new)
 896 {
 897     const struct ovsdb_row *row = old_ ? old_ : new;
 898     struct ovsdb_row *old = CONST_CAST(struct ovsdb_row *, old_);
 899     size_t n_columns = shash_count(&table->schema->columns);
 900     struct ovsdb_txn_table *txn_table;
 901     struct ovsdb_txn_row *txn_row;
 902
 903     txn_row = xzalloc(offsetof(struct ovsdb_txn_row, changed)
 904                       + bitmap_n_bytes(n_columns));
 905     txn_row->uuid = *ovsdb_row_get_uuid(row);
 906     txn_row->table = row->table;
 907     txn_row->old = old;
 908     txn_row->new = new;
 909     txn_row->n_refs = old ? old->n_refs : 0;
 910     txn_row->serial = serial - 1;
 911
 912     if (old) {
 913         old->txn_row = txn_row;
 914     }
 915     if (new) {
 916         new->txn_row = txn_row;
 917     }
 918
 919     txn_table = ovsdb_txn_create_txn_table(txn, table);
 920     hmap_insert(&txn_table->txn_rows, &txn_row->hmap_node,
 921                 ovsdb_row_hash(old ? old : new));
 922
 923     return txn_row;
 924 }
 925
 926 struct ovsdb_row *
 927 ovsdb_txn_row_modify(struct ovsdb_txn *txn, const struct ovsdb_row *ro_row_)
 928 {
 929     struct ovsdb_row *ro_row = CONST_CAST(struct ovsdb_row *, ro_row_);
 930
 931     if (ro_row->txn_row) {
 932         ovs_assert(ro_row == ro_row->txn_row->new);
 933         return ro_row;
 934     } else {
 935         struct ovsdb_table *table = ro_row->table;
 936         struct ovsdb_row *rw_row;
 937
 938         rw_row = ovsdb_row_clone(ro_row);
 939         rw_row->n_refs = ro_row->n_refs;
 940         ovsdb_txn_row_create(txn, table, ro_row, rw_row);
 941         hmap_replace(&table->rows, &ro_row->hmap_node, &rw_row->hmap_node);
 942
 943         return rw_row;
 944     }
 945 }
 946
 947 void
 948 ovsdb_txn_row_insert(struct ovsdb_txn *txn, struct ovsdb_row *row)
 949 {
 950     uint32_t hash = ovsdb_row_hash(row);
 951     struct ovsdb_table *table = row->table;
 952
 953     uuid_generate(ovsdb_row_get_version_rw(row));
 954
 955     ovsdb_txn_row_create(txn, table, NULL, row);
 956     hmap_insert(&table->rows, &row->hmap_node, hash);
 957 }
 958
 959 /* 'row' must be assumed destroyed upon return; the caller must not reference
 960  * it again. */
 961 void
 962 ovsdb_txn_row_delete(struct ovsdb_txn *txn, const struct ovsdb_row *row_)
 963 {
 964     struct ovsdb_row *row = CONST_CAST(struct ovsdb_row *, row_);
 965     struct ovsdb_table *table = row->table;
 966     struct ovsdb_txn_row *txn_row = row->txn_row;
 967
 968     hmap_remove(&table->rows, &row->hmap_node);
 969
 970     if (!txn_row) {
 971         ovsdb_txn_row_create(txn, table, row, NULL);
 972     } else {
 973         ovs_assert(txn_row->new == row);
 974         if (txn_row->old) {
 975             txn_row->new = NULL;
 976         } else {
 977             hmap_remove(&table->txn_table->txn_rows, &txn_row->hmap_node);
 978             free(txn_row);
 979         }
 980         ovsdb_row_destroy(row);
 981     }
 982 }
 983
 984 void
 985 ovsdb_txn_add_comment(struct ovsdb_txn *txn, const char *s)
 986 {
 987     if (txn->comment.length) {
 988         ds_put_char(&txn->comment, '\n');
 989     }
 990     ds_put_cstr(&txn->comment, s);
 991 }
 992
 993 const char *
 994 ovsdb_txn_get_comment(const struct ovsdb_txn *txn)
 995 {
 996     return txn->comment.length ? ds_cstr_ro(&txn->comment) : NULL;
 997 }
 998 \f
 999 static void
1000 ovsdb_txn_row_prefree(struct ovsdb_txn_row *txn_row)
1001 {
1002     struct ovsdb_txn_table *txn_table = txn_row->table->txn_table;
1003
1004     txn_table->n_processed--;
1005     hmap_remove(&txn_table->txn_rows, &txn_row->hmap_node);
1006
1007     if (txn_row->old) {
1008         txn_row->old->txn_row = NULL;
1009     }
1010     if (txn_row->new) {
1011         txn_row->new->txn_row = NULL;
1012     }
1013 }
1014
1015 static void
1016 ovsdb_txn_table_destroy(struct ovsdb_txn_table *txn_table)
1017 {
1018     size_t i;
1019
1020     ovs_assert(hmap_is_empty(&txn_table->txn_rows));
1021
1022     for (i = 0; i < txn_table->table->schema->n_indexes; i++) {
1023         hmap_destroy(&txn_table->txn_indexes[i]);
1024     }
1025     free(txn_table->txn_indexes);
1026
1027     txn_table->table->txn_table = NULL;
1028     hmap_destroy(&txn_table->txn_rows);
1029     ovs_list_remove(&txn_table->node);
1030     free(txn_table);
1031 }
1032
1033 /* Calls 'cb' for every txn_row within 'txn'.  If 'cb' returns nonnull, this
1034  * aborts the iteration and for_each_txn_row() passes the error up.  Otherwise,
1035  * returns a null pointer after iteration is complete.
1036  *
1037  * 'cb' may insert new txn_rows and new txn_tables into 'txn'.  It may delete
1038  * the txn_row that it is passed in, or txn_rows in txn_tables other than the
1039  * one passed to 'cb'.  It may *not* delete txn_rows other than the one passed
1040  * in within the same txn_table.  It may *not* delete any txn_tables.  As long
1041  * as these rules are followed, 'cb' will be called exactly once for each
1042  * txn_row in 'txn', even those added by 'cb'.
1043  *
1044  * (Even though 'cb' is not allowed to delete some txn_rows, it can still
1045  * delete any actual row by clearing a txn_row's 'new' member.)
1046  */
1047 static struct ovsdb_error * OVS_WARN_UNUSED_RESULT
1048 for_each_txn_row(struct ovsdb_txn *txn,
1049                  struct ovsdb_error *(*cb)(struct ovsdb_txn *,
1050                                            struct ovsdb_txn_row *))
1051 {
1052     bool any_work;
1053
1054     serial++;
1055
1056     do {
1057         struct ovsdb_txn_table *t, *next_txn_table;
1058
1059         any_work = false;
1060         LIST_FOR_EACH_SAFE (t, next_txn_table, node, &txn->txn_tables) {
1061             if (t->serial != serial) {
1062                 t->serial = serial;
1063                 t->n_processed = 0;
1064             }
1065
1066             while (t->n_processed < hmap_count(&t->txn_rows)) {
1067                 struct ovsdb_txn_row *r, *next_txn_row;
1068
1069                 HMAP_FOR_EACH_SAFE (r, next_txn_row, hmap_node, &t->txn_rows) {
1070                     if (r->serial != serial) {
1071                         struct ovsdb_error *error;
1072
1073                         r->serial = serial;
1074                         t->n_processed++;
1075                         any_work = true;
1076
1077                         error = cb(txn, r);
1078                         if (error) {
1079                             return error;
1080                         }
1081                     }
1082                 }
1083             }
1084             if (hmap_is_empty(&t->txn_rows)) {
1085                 /* Table is empty.  Drop it. */
1086                 ovsdb_txn_table_destroy(t);
1087             }
1088         }
1089     } while (any_work);
1090
1091     return NULL;
1092 }