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 "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, column->index);
 538             ovsdb_datum_sort_assert(datum, column->type.key.type);
 539             if (datum->n < column->type.n_min) {
 540                 const struct uuid *row_uuid = ovsdb_row_get_uuid(txn_row->new);
 541                 if (zero && !txn_row->old) {
 542                     return ovsdb_error(
 543                         "constraint violation",
 544                         "Weak reference column \"%s\" in \"%s\" row "UUID_FMT
 545                         " (inserted within this transaction) contained "
 546                         "all-zeros UUID (probably as the default value for "
 547                         "this column) but deleting this value caused a "
 548                         "constraint volation because this column is not "
 549                         "allowed to be empty.", column->name,
 550                         table->schema->name, UUID_ARGS(row_uuid));
 551                 } else {
 552                     return ovsdb_error(
 553                         "constraint violation",
 554                         "Deletion of %u weak reference(s) to deleted (or "
 555                         "never-existing) rows from column \"%s\" in \"%s\" "
 556                         "row "UUID_FMT" %scaused this column to become empty, "
 557                         "but constraints on this column disallow an "
 558                         "empty column.",
 559                         orig_n - datum->n, column->name, table->schema->name,
 560                         UUID_ARGS(row_uuid),
 561                         (txn_row->old
 562                          ? ""
 563                          : "(inserted within this transaction) "));
 564                 }
 565             }
 566         }
 567     }
 568
 569     return NULL;
 570 }
 571
 572 static struct ovsdb_error * OVS_WARN_UNUSED_RESULT
 573 determine_changes(struct ovsdb_txn *txn, struct ovsdb_txn_row *txn_row)
 574 {
 575     struct ovsdb_table *table = txn_row->table;
 576
 577     if (txn_row->old && txn_row->new) {
 578         struct shash_node *node;
 579         bool changed = false;
 580
 581         SHASH_FOR_EACH (node, &table->schema->columns) {
 582             const struct ovsdb_column *column = node->data;
 583             const struct ovsdb_type *type = &column->type;
 584             unsigned int idx = column->index;
 585
 586             if (!ovsdb_datum_equals(&txn_row->old->fields[idx],
 587                                     &txn_row->new->fields[idx],
 588                                     type)) {
 589                 bitmap_set1(txn_row->changed, idx);
 590                 changed = true;
 591             }
 592         }
 593
 594         if (!changed) {
 595             /* Nothing actually changed in this row, so drop it. */
 596             ovsdb_txn_row_abort(txn, txn_row);
 597         }
 598     } else {
 599         bitmap_set_multiple(txn_row->changed, 0,
 600                             shash_count(&table->schema->columns), 1);
 601     }
 602
 603     return NULL;
 604 }
 605
 606 static struct ovsdb_error * OVS_WARN_UNUSED_RESULT
 607 check_max_rows(struct ovsdb_txn *txn)
 608 {
 609     struct ovsdb_txn_table *t;
 610
 611     LIST_FOR_EACH (t, node, &txn->txn_tables) {
 612         size_t n_rows = hmap_count(&t->table->rows);
 613         unsigned int max_rows = t->table->schema->max_rows;
 614
 615         if (n_rows > max_rows) {
 616             return ovsdb_error("constraint violation",
 617                                "transaction causes \"%s\" table to contain "
 618                                "%"PRIuSIZE" rows, greater than the schema-defined "
 619                                "limit of %u row(s)",
 620                                t->table->schema->name, n_rows, max_rows);
 621         }
 622     }
 623
 624     return NULL;
 625 }
 626
 627 static struct ovsdb_row *
 628 ovsdb_index_search(struct hmap *index, struct ovsdb_row *row, size_t i,
 629                    uint32_t hash)
 630 {
 631     const struct ovsdb_table *table = row->table;
 632     const struct ovsdb_column_set *columns = &table->schema->indexes[i];
 633     struct hmap_node *node;
 634
 635     for (node = hmap_first_with_hash(index, hash); node;
 636          node = hmap_next_with_hash(node)) {
 637         struct ovsdb_row *irow = ovsdb_row_from_index_node(node, table, i);
 638         if (ovsdb_row_equal_columns(row, irow, columns)) {
 639             return irow;
 640         }
 641     }
 642
 643     return NULL;
 644 }
 645
 646 static void
 647 duplicate_index_row__(const struct ovsdb_column_set *index,
 648                       const struct ovsdb_row *row,
 649                       const char *title,
 650                       struct ds *out)
 651 {
 652     size_t n_columns = shash_count(&row->table->schema->columns);
 653
 654     ds_put_format(out, "%s row, with UUID "UUID_FMT", ",
 655                   title, UUID_ARGS(ovsdb_row_get_uuid(row)));
 656     if (!row->txn_row
 657         || bitmap_scan(row->txn_row->changed, 1, 0, n_columns) == n_columns) {
 658         ds_put_cstr(out, "existed in the database before this "
 659                     "transaction and was not modified by the transaction.");
 660     } else if (!row->txn_row->old) {
 661         ds_put_cstr(out, "was inserted by this transaction.");
 662     } else if (ovsdb_row_equal_columns(row->txn_row->old,
 663                                        row->txn_row->new, index)) {
 664         ds_put_cstr(out, "existed in the database before this "
 665                     "transaction, which modified some of the row's columns "
 666                     "but not any columns in this index.");
 667     } else {
 668         ds_put_cstr(out, "had the following index values before the "
 669                     "transaction: ");
 670         ovsdb_row_columns_to_string(row->txn_row->old, index, out);
 671         ds_put_char(out, '.');
 672     }
 673 }
 674
 675 static struct ovsdb_error * OVS_WARN_UNUSED_RESULT
 676 duplicate_index_row(const struct ovsdb_column_set *index,
 677                     const struct ovsdb_row *a,
 678                     const struct ovsdb_row *b)
 679 {
 680     struct ovsdb_column_set all_columns;
 681     struct ovsdb_error *error;
 682     char *index_s;
 683     struct ds s;
 684
 685     /* Put 'a' and 'b' in a predictable order to make error messages
 686      * reproducible for testing. */
 687     ovsdb_column_set_init(&all_columns);
 688     ovsdb_column_set_add_all(&all_columns, a->table);
 689     if (ovsdb_row_compare_columns_3way(a, b, &all_columns) < 0) {
 690         const struct ovsdb_row *tmp = a;
 691         a = b;
 692         b = tmp;
 693     }
 694     ovsdb_column_set_destroy(&all_columns);
 695
 696     index_s = ovsdb_column_set_to_string(index);
 697
 698     ds_init(&s);
 699     ds_put_format(&s, "Transaction causes multiple rows in \"%s\" table to "
 700                   "have identical values (", a->table->schema->name);
 701     ovsdb_row_columns_to_string(a, index, &s);
 702     ds_put_format(&s, ") for index on %s.  ", index_s);
 703     duplicate_index_row__(index, a, "First", &s);
 704     ds_put_cstr(&s, "  ");
 705     duplicate_index_row__(index, b, "Second", &s);
 706
 707     free(index_s);
 708
 709     error = ovsdb_error("constraint violation", "%s", ds_cstr(&s));
 710     ds_destroy(&s);
 711     return error;
 712 }
 713
 714 static struct ovsdb_error * OVS_WARN_UNUSED_RESULT
 715 check_index_uniqueness(struct ovsdb_txn *txn OVS_UNUSED,
 716                        struct ovsdb_txn_row *txn_row)
 717 {
 718     struct ovsdb_txn_table *txn_table = txn_row->table->txn_table;
 719     struct ovsdb_table *table = txn_row->table;
 720     struct ovsdb_row *row = txn_row->new;
 721     size_t i;
 722
 723     if (!row) {
 724         return NULL;
 725     }
 726
 727     for (i = 0; i < table->schema->n_indexes; i++) {
 728         const struct ovsdb_column_set *index = &table->schema->indexes[i];
 729         struct ovsdb_row *irow;
 730         uint32_t hash;
 731
 732         hash = ovsdb_row_hash_columns(row, index, 0);
 733         irow = ovsdb_index_search(&txn_table->txn_indexes[i], row, i, hash);
 734         if (irow) {
 735             return duplicate_index_row(index, irow, row);
 736         }
 737
 738         irow = ovsdb_index_search(&table->indexes[i], row, i, hash);
 739         if (irow && !irow->txn_row) {
 740             return duplicate_index_row(index, irow, row);
 741         }
 742
 743         hmap_insert(&txn_table->txn_indexes[i],
 744                     ovsdb_row_get_index_node(row, i), hash);
 745     }
 746
 747     return NULL;
 748 }
 749
 750 static struct ovsdb_error * OVS_WARN_UNUSED_RESULT
 751 update_version(struct ovsdb_txn *txn OVS_UNUSED, struct ovsdb_txn_row *txn_row)
 752 {
 753     struct ovsdb_table *table = txn_row->table;
 754     size_t n_columns = shash_count(&table->schema->columns);
 755
 756     if (txn_row->old && txn_row->new
 757         && !bitmap_is_all_zeros(txn_row->changed, n_columns)) {
 758         bitmap_set1(txn_row->changed, OVSDB_COL_VERSION);
 759         uuid_generate(ovsdb_row_get_version_rw(txn_row->new));
 760     }
 761
 762     return NULL;
 763 }
 764
 765 static struct ovsdb_error *
 766 ovsdb_txn_commit_(struct ovsdb_txn *txn, bool durable)
 767 {
 768     struct ovsdb_replica *replica;
 769     struct ovsdb_error *error;
 770
 771     /* Figure out what actually changed, and abort early if the transaction
 772      * was really a no-op. */
 773     error = for_each_txn_row(txn, determine_changes);
 774     if (error) {
 775         return OVSDB_WRAP_BUG("can't happen", error);
 776     }
 777     if (list_is_empty(&txn->txn_tables)) {
 778         ovsdb_txn_abort(txn);
 779         return NULL;
 780     }
 781
 782     /* Update reference counts and check referential integrity. */
 783     error = update_ref_counts(txn);
 784     if (error) {
 785         ovsdb_txn_abort(txn);
 786         return error;
 787     }
 788
 789     /* Delete unreferenced, non-root rows. */
 790     error = for_each_txn_row(txn, collect_garbage);
 791     if (error) {
 792         ovsdb_txn_abort(txn);
 793         return OVSDB_WRAP_BUG("can't happen", error);
 794     }
 795
 796     /* Check maximum rows table constraints. */
 797     error = check_max_rows(txn);
 798     if (error) {
 799         ovsdb_txn_abort(txn);
 800         return error;
 801     }
 802
 803     /* Check reference counts and remove bad references for "weak" referential
 804      * integrity. */
 805     error = for_each_txn_row(txn, assess_weak_refs);
 806     if (error) {
 807         ovsdb_txn_abort(txn);
 808         return error;
 809     }
 810
 811     /* Verify that the indexes will still be unique post-transaction. */
 812     error = for_each_txn_row(txn, check_index_uniqueness);
 813     if (error) {
 814         ovsdb_txn_abort(txn);
 815         return error;
 816     }
 817
 818     /* Update _version for rows that changed.  */
 819     error = for_each_txn_row(txn, update_version);
 820     if (error) {
 821         return OVSDB_WRAP_BUG("can't happen", error);
 822     }
 823
 824     /* Send the commit to each replica. */
 825     LIST_FOR_EACH (replica, node, &txn->db->replicas) {
 826         error = (replica->class->commit)(replica, txn, durable);
 827         if (error) {
 828             /* We don't support two-phase commit so only the first replica is
 829              * allowed to report an error. */
 830             ovs_assert(&replica->node == txn->db->replicas.next);
 831
 832             ovsdb_txn_abort(txn);
 833             return error;
 834         }
 835     }
 836
 837     /* Finalize commit. */
 838     txn->db->run_triggers = true;
 839     ovsdb_error_assert(for_each_txn_row(txn, ovsdb_txn_row_commit));
 840     ovsdb_txn_free(txn);
 841
 842     return NULL;
 843 }
 844
 845 struct ovsdb_error *
 846 ovsdb_txn_commit(struct ovsdb_txn *txn, bool durable)
 847 {
 848    struct ovsdb_error *err;
 849
 850    PERF(__func__, err = ovsdb_txn_commit_(txn, durable));
 851    return err;
 852 }
 853
 854 void
 855 ovsdb_txn_for_each_change(const struct ovsdb_txn *txn,
 856                           ovsdb_txn_row_cb_func *cb, void *aux)
 857 {
 858     struct ovsdb_txn_table *t;
 859     struct ovsdb_txn_row *r;
 860
 861     LIST_FOR_EACH (t, node, &txn->txn_tables) {
 862         HMAP_FOR_EACH (r, hmap_node, &t->txn_rows) {
 863             if ((r->old || r->new) && !cb(r->old, r->new, r->changed, aux)) {
 864                 break;
 865             }
 866         }
 867    }
 868 }
 869
 870 static struct ovsdb_txn_table *
 871 ovsdb_txn_create_txn_table(struct ovsdb_txn *txn, struct ovsdb_table *table)
 872 {
 873     if (!table->txn_table) {
 874         struct ovsdb_txn_table *txn_table;
 875         size_t i;
 876
 877         table->txn_table = txn_table = xmalloc(sizeof *table->txn_table);
 878         txn_table->table = table;
 879         hmap_init(&txn_table->txn_rows);
 880         txn_table->serial = serial - 1;
 881         txn_table->txn_indexes = xmalloc(table->schema->n_indexes
 882                                          * sizeof *txn_table->txn_indexes);
 883         for (i = 0; i < table->schema->n_indexes; i++) {
 884             hmap_init(&txn_table->txn_indexes[i]);
 885         }
 886         list_push_back(&txn->txn_tables, &txn_table->node);
 887     }
 888     return table->txn_table;
 889 }
 890
 891 static struct ovsdb_txn_row *
 892 ovsdb_txn_row_create(struct ovsdb_txn *txn, struct ovsdb_table *table,
 893                      const struct ovsdb_row *old_, struct ovsdb_row *new)
 894 {
 895     const struct ovsdb_row *row = old_ ? old_ : new;
 896     struct ovsdb_row *old = CONST_CAST(struct ovsdb_row *, old_);
 897     size_t n_columns = shash_count(&table->schema->columns);
 898     struct ovsdb_txn_table *txn_table;
 899     struct ovsdb_txn_row *txn_row;
 900
 901     txn_row = xzalloc(offsetof(struct ovsdb_txn_row, changed)
 902                       + bitmap_n_bytes(n_columns));
 903     txn_row->uuid = *ovsdb_row_get_uuid(row);
 904     txn_row->table = row->table;
 905     txn_row->old = old;
 906     txn_row->new = new;
 907     txn_row->n_refs = old ? old->n_refs : 0;
 908     txn_row->serial = serial - 1;
 909
 910     if (old) {
 911         old->txn_row = txn_row;
 912     }
 913     if (new) {
 914         new->txn_row = txn_row;
 915     }
 916
 917     txn_table = ovsdb_txn_create_txn_table(txn, table);
 918     hmap_insert(&txn_table->txn_rows, &txn_row->hmap_node,
 919                 ovsdb_row_hash(old ? old : new));
 920
 921     return txn_row;
 922 }
 923
 924 struct ovsdb_row *
 925 ovsdb_txn_row_modify(struct ovsdb_txn *txn, const struct ovsdb_row *ro_row_)
 926 {
 927     struct ovsdb_row *ro_row = CONST_CAST(struct ovsdb_row *, ro_row_);
 928
 929     if (ro_row->txn_row) {
 930         ovs_assert(ro_row == ro_row->txn_row->new);
 931         return ro_row;
 932     } else {
 933         struct ovsdb_table *table = ro_row->table;
 934         struct ovsdb_row *rw_row;
 935
 936         rw_row = ovsdb_row_clone(ro_row);
 937         rw_row->n_refs = ro_row->n_refs;
 938         ovsdb_txn_row_create(txn, table, ro_row, rw_row);
 939         hmap_replace(&table->rows, &ro_row->hmap_node, &rw_row->hmap_node);
 940
 941         return rw_row;
 942     }
 943 }
 944
 945 void
 946 ovsdb_txn_row_insert(struct ovsdb_txn *txn, struct ovsdb_row *row)
 947 {
 948     uint32_t hash = ovsdb_row_hash(row);
 949     struct ovsdb_table *table = row->table;
 950
 951     uuid_generate(ovsdb_row_get_version_rw(row));
 952
 953     ovsdb_txn_row_create(txn, table, NULL, row);
 954     hmap_insert(&table->rows, &row->hmap_node, hash);
 955 }
 956
 957 /* 'row' must be assumed destroyed upon return; the caller must not reference
 958  * it again. */
 959 void
 960 ovsdb_txn_row_delete(struct ovsdb_txn *txn, const struct ovsdb_row *row_)
 961 {
 962     struct ovsdb_row *row = CONST_CAST(struct ovsdb_row *, row_);
 963     struct ovsdb_table *table = row->table;
 964     struct ovsdb_txn_row *txn_row = row->txn_row;
 965
 966     hmap_remove(&table->rows, &row->hmap_node);
 967
 968     if (!txn_row) {
 969         ovsdb_txn_row_create(txn, table, row, NULL);
 970     } else {
 971         ovs_assert(txn_row->new == row);
 972         if (txn_row->old) {
 973             txn_row->new = NULL;
 974         } else {
 975             hmap_remove(&table->txn_table->txn_rows, &txn_row->hmap_node);
 976             free(txn_row);
 977         }
 978         ovsdb_row_destroy(row);
 979     }
 980 }
 981
 982 void
 983 ovsdb_txn_add_comment(struct ovsdb_txn *txn, const char *s)
 984 {
 985     if (txn->comment.length) {
 986         ds_put_char(&txn->comment, '\n');
 987     }
 988     ds_put_cstr(&txn->comment, s);
 989 }
 990
 991 const char *
 992 ovsdb_txn_get_comment(const struct ovsdb_txn *txn)
 993 {
 994     return txn->comment.length ? ds_cstr_ro(&txn->comment) : NULL;
 995 }
 996 \f
 997 static void
 998 ovsdb_txn_row_prefree(struct ovsdb_txn_row *txn_row)
 999 {
1000     struct ovsdb_txn_table *txn_table = txn_row->table->txn_table;
1001
1002     txn_table->n_processed--;
1003     hmap_remove(&txn_table->txn_rows, &txn_row->hmap_node);
1004
1005     if (txn_row->old) {
1006         txn_row->old->txn_row = NULL;
1007     }
1008     if (txn_row->new) {
1009         txn_row->new->txn_row = NULL;
1010     }
1011 }
1012
1013 static void
1014 ovsdb_txn_table_destroy(struct ovsdb_txn_table *txn_table)
1015 {
1016     size_t i;
1017
1018     ovs_assert(hmap_is_empty(&txn_table->txn_rows));
1019
1020     for (i = 0; i < txn_table->table->schema->n_indexes; i++) {
1021         hmap_destroy(&txn_table->txn_indexes[i]);
1022     }
1023     free(txn_table->txn_indexes);
1024
1025     txn_table->table->txn_table = NULL;
1026     hmap_destroy(&txn_table->txn_rows);
1027     list_remove(&txn_table->node);
1028     free(txn_table);
1029 }
1030
1031 /* Calls 'cb' for every txn_row within 'txn'.  If 'cb' returns nonnull, this
1032  * aborts the iteration and for_each_txn_row() passes the error up.  Otherwise,
1033  * returns a null pointer after iteration is complete.
1034  *
1035  * 'cb' may insert new txn_rows and new txn_tables into 'txn'.  It may delete
1036  * the txn_row that it is passed in, or txn_rows in txn_tables other than the
1037  * one passed to 'cb'.  It may *not* delete txn_rows other than the one passed
1038  * in within the same txn_table.  It may *not* delete any txn_tables.  As long
1039  * as these rules are followed, 'cb' will be called exactly once for each
1040  * txn_row in 'txn', even those added by 'cb'.
1041  *
1042  * (Even though 'cb' is not allowed to delete some txn_rows, it can still
1043  * delete any actual row by clearing a txn_row's 'new' member.)
1044  */
1045 static struct ovsdb_error * OVS_WARN_UNUSED_RESULT
1046 for_each_txn_row(struct ovsdb_txn *txn,
1047                  struct ovsdb_error *(*cb)(struct ovsdb_txn *,
1048                                            struct ovsdb_txn_row *))
1049 {
1050     bool any_work;
1051
1052     serial++;
1053
1054     do {
1055         struct ovsdb_txn_table *t, *next_txn_table;
1056
1057         any_work = false;
1058         LIST_FOR_EACH_SAFE (t, next_txn_table, node, &txn->txn_tables) {
1059             if (t->serial != serial) {
1060                 t->serial = serial;
1061                 t->n_processed = 0;
1062             }
1063
1064             while (t->n_processed < hmap_count(&t->txn_rows)) {
1065                 struct ovsdb_txn_row *r, *next_txn_row;
1066
1067                 HMAP_FOR_EACH_SAFE (r, next_txn_row, hmap_node, &t->txn_rows) {
1068                     if (r->serial != serial) {
1069                         struct ovsdb_error *error;
1070
1071                         r->serial = serial;
1072                         t->n_processed++;
1073                         any_work = true;
1074
1075                         error = cb(txn, r);
1076                         if (error) {
1077                             return error;
1078                         }
1079                     }
1080                 }
1081             }
1082             if (hmap_is_empty(&t->txn_rows)) {
1083                 /* Table is empty.  Drop it. */
1084                 ovsdb_txn_table_destroy(t);
1085             }
1086         }
1087     } while (any_work);
1088
1089     return NULL;
1090 }