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