ovsdb/transaction.c

   1 /* Copyright (c) 2009, 2010, 2011 Nicira Networks
   2  *
   3  * Licensed under the Apache License, Version 2.0 (the "License");
   4  * you may not use this file except in compliance with the License.
   5  * You may obtain a copy of the License at:
   6  *
   7  *     http://www.apache.org/licenses/LICENSE-2.0
   8  *
   9  * Unless required by applicable law or agreed to in writing, software
  10  * distributed under the License is distributed on an "AS IS" BASIS,
  11  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  12  * See the License for the specific language governing permissions and
  13  * limitations under the License.
  14  */
  15
  16 #include <config.h>
  17
  18 #include "transaction.h"
  19
  20 #include <assert.h>
  21
  22 #include "bitmap.h"
  23 #include "dynamic-string.h"
  24 #include "hash.h"
  25 #include "hmap.h"
  26 #include "json.h"
  27 #include "list.h"
  28 #include "ovsdb-error.h"
  29 #include "ovsdb.h"
  30 #include "row.h"
  31 #include "table.h"
  32 #include "uuid.h"
  33
  34 struct ovsdb_txn {
  35     struct ovsdb *db;
  36     struct list txn_tables;     /* Contains "struct ovsdb_txn_table"s. */
  37     struct ds comment;
  38 };
  39
  40 /* A table modified by a transaction. */
  41 struct ovsdb_txn_table {
  42     struct list node;           /* Element in ovsdb_txn's txn_tables list. */
  43     struct ovsdb_table *table;
  44     struct hmap txn_rows;       /* Contains "struct ovsdb_txn_row"s. */
  45
  46     /* Used by for_each_txn_row(). */
  47     unsigned int serial;        /* Serial number of in-progress iteration. */
  48     unsigned int n_processed;   /* Number of rows processed. */
  49 };
  50
  51 /* A row modified by the transaction:
  52  *
  53  *      - A row added by a transaction will have null 'old' and non-null 'new'.
  54  *
  55  *      - A row deleted by a transaction will have non-null 'old' and null
  56  *        'new'.
  57  *
  58  *      - A row modified by a transaction will have non-null 'old' and 'new'.
  59  *
  60  *      - 'old' and 'new' both null indicates that a row was added then deleted
  61  *        within a single transaction.  Most of the time we instead delete the
  62  *        ovsdb_txn_row entirely, but inside a for_each_txn_row() callback
  63  *        there are restrictions that sometimes mean we have to leave the
  64  *        ovsdb_txn_row in place.
  65  */
  66 struct ovsdb_txn_row {
  67     struct hmap_node hmap_node; /* In ovsdb_txn_table's txn_rows hmap. */
  68     struct ovsdb_row *old;      /* The old row. */
  69     struct ovsdb_row *new;      /* The new row. */
  70     size_t n_refs;              /* Number of remaining references. */
  71
  72     /* These members are the same as the corresponding members of 'old' or
  73      * 'new'.  They are present here for convenience and because occasionally
  74      * there can be an ovsdb_txn_row where both 'old' and 'new' are NULL. */
  75     struct uuid uuid;
  76     struct ovsdb_table *table;
  77
  78     /* Used by for_each_txn_row(). */
  79     unsigned int serial;        /* Serial number of in-progress commit. */
  80
  81     unsigned long changed[];    /* Bits set to 1 for columns that changed. */
  82 };
  83
  84 static struct ovsdb_error * WARN_UNUSED_RESULT
  85 delete_garbage_row(struct ovsdb_txn *txn, struct ovsdb_txn_row *r);
  86 static void ovsdb_txn_row_prefree(struct ovsdb_txn_row *);
  87 static struct ovsdb_error * WARN_UNUSED_RESULT
  88 for_each_txn_row(struct ovsdb_txn *txn,
  89                       struct ovsdb_error *(*)(struct ovsdb_txn *,
  90                                               struct ovsdb_txn_row *));
  91
  92 /* Used by for_each_txn_row() to track tables and rows that have been
  93  * processed.  */
  94 static unsigned int serial;
  95
  96 struct ovsdb_txn *
  97 ovsdb_txn_create(struct ovsdb *db)
  98 {
  99     struct ovsdb_txn *txn = xmalloc(sizeof *txn);
 100     txn->db = db;
 101     list_init(&txn->txn_tables);
 102     ds_init(&txn->comment);
 103     return txn;
 104 }
 105
 106 static void
 107 ovsdb_txn_free(struct ovsdb_txn *txn)
 108 {
 109     assert(list_is_empty(&txn->txn_tables));
 110     ds_destroy(&txn->comment);
 111     free(txn);
 112 }
 113
 114 static struct ovsdb_error *
 115 ovsdb_txn_row_abort(struct ovsdb_txn *txn OVS_UNUSED,
 116                     struct ovsdb_txn_row *txn_row)
 117 {
 118     struct ovsdb_row *old = txn_row->old;
 119     struct ovsdb_row *new = txn_row->new;
 120
 121     ovsdb_txn_row_prefree(txn_row);
 122     if (!old) {
 123         if (new) {
 124             hmap_remove(&new->table->rows, &new->hmap_node);
 125         }
 126     } else if (!new) {
 127         hmap_insert(&old->table->rows, &old->hmap_node, ovsdb_row_hash(old));
 128     } else {
 129         hmap_replace(&new->table->rows, &new->hmap_node, &old->hmap_node);
 130     }
 131     ovsdb_row_destroy(new);
 132     free(txn_row);
 133
 134     return NULL;
 135 }
 136
 137 void
 138 ovsdb_txn_abort(struct ovsdb_txn *txn)
 139 {
 140     ovsdb_error_assert(for_each_txn_row(txn, ovsdb_txn_row_abort));
 141     ovsdb_txn_free(txn);
 142 }
 143
 144 static struct ovsdb_txn_row *
 145 find_txn_row(const struct ovsdb_table *table, const struct uuid *uuid)
 146 {
 147     struct ovsdb_txn_row *txn_row;
 148
 149     if (!table->txn_table) {
 150         return NULL;
 151     }
 152
 153     HMAP_FOR_EACH_WITH_HASH (txn_row, hmap_node,
 154                              uuid_hash(uuid), &table->txn_table->txn_rows) {
 155         if (uuid_equals(uuid, &txn_row->uuid)) {
 156             return txn_row;
 157         }
 158     }
 159
 160     return NULL;
 161 }
 162
 163 static struct ovsdb_txn_row *
 164 find_or_make_txn_row(struct ovsdb_txn *txn, const struct ovsdb_table *table,
 165                      const struct uuid *uuid)
 166 {
 167     struct ovsdb_txn_row *txn_row = find_txn_row(table, uuid);
 168     if (!txn_row) {
 169         const struct ovsdb_row *row = ovsdb_table_get_row(table, uuid);
 170         if (row) {
 171             txn_row = ovsdb_txn_row_modify(txn, row)->txn_row;
 172         }
 173     }
 174     return txn_row;
 175 }
 176
 177 static struct ovsdb_error * WARN_UNUSED_RESULT
 178 ovsdb_txn_adjust_atom_refs(struct ovsdb_txn *txn, const struct ovsdb_row *r,
 179                            const struct ovsdb_column *c,
 180                            const struct ovsdb_base_type *base,
 181                            const union ovsdb_atom *atoms, unsigned int n,
 182                            int delta)
 183 {
 184     const struct ovsdb_table *table;
 185     unsigned int i;
 186
 187     if (!ovsdb_base_type_is_strong_ref(base)) {
 188         return NULL;
 189     }
 190
 191     table = base->u.uuid.refTable;
 192     for (i = 0; i < n; i++) {
 193         const struct uuid *uuid = &atoms[i].uuid;
 194         struct ovsdb_txn_row *txn_row;
 195
 196         if (uuid_equals(uuid, ovsdb_row_get_uuid(r))) {
 197             /* Self-references don't count. */
 198             continue;
 199         }
 200
 201         txn_row = find_or_make_txn_row(txn, table, uuid);
 202         if (!txn_row) {
 203             return ovsdb_error("referential integrity violation",
 204                                "Table %s column %s row "UUID_FMT" "
 205                                "references nonexistent row "UUID_FMT" in "
 206                                "table %s.",
 207                                r->table->schema->name, c->name,
 208                                UUID_ARGS(ovsdb_row_get_uuid(r)),
 209                                UUID_ARGS(uuid), table->schema->name);
 210         }
 211         txn_row->n_refs += delta;
 212     }
 213
 214     return NULL;
 215 }
 216
 217 static struct ovsdb_error * WARN_UNUSED_RESULT
 218 ovsdb_txn_adjust_row_refs(struct ovsdb_txn *txn, const struct ovsdb_row *r,
 219                           const struct ovsdb_column *column, int delta)
 220 {
 221     const struct ovsdb_datum *field = &r->fields[column->index];
 222     struct ovsdb_error *error;
 223
 224     error = ovsdb_txn_adjust_atom_refs(txn, r, column, &column->type.key,
 225                                        field->keys, field->n, delta);
 226     if (!error) {
 227         error = ovsdb_txn_adjust_atom_refs(txn, r, column, &column->type.value,
 228                                            field->values, field->n, delta);
 229     }
 230     return error;
 231 }
 232
 233 static struct ovsdb_error * WARN_UNUSED_RESULT
 234 update_row_ref_count(struct ovsdb_txn *txn, struct ovsdb_txn_row *r)
 235 {
 236     struct ovsdb_table *table = r->table;
 237     struct shash_node *node;
 238
 239     SHASH_FOR_EACH (node, &table->schema->columns) {
 240         const struct ovsdb_column *column = node->data;
 241         struct ovsdb_error *error;
 242
 243         if (r->old) {
 244             error = ovsdb_txn_adjust_row_refs(txn, r->old, column, -1);
 245             if (error) {
 246                 return OVSDB_WRAP_BUG("error decreasing refcount", error);
 247             }
 248         }
 249         if (r->new) {
 250             error = ovsdb_txn_adjust_row_refs(txn, r->new, column, 1);
 251             if (error) {
 252                 return error;
 253             }
 254         }
 255     }
 256
 257     return NULL;
 258 }
 259
 260 static struct ovsdb_error * WARN_UNUSED_RESULT
 261 check_ref_count(struct ovsdb_txn *txn OVS_UNUSED, struct ovsdb_txn_row *r)
 262 {
 263     if (r->new || !r->n_refs) {
 264         return NULL;
 265     } else {
 266         return ovsdb_error("referential integrity violation",
 267                            "cannot delete %s row "UUID_FMT" because "
 268                            "of %zu remaining reference(s)",
 269                            r->table->schema->name, UUID_ARGS(&r->uuid),
 270                            r->n_refs);
 271     }
 272 }
 273
 274 static struct ovsdb_error * WARN_UNUSED_RESULT
 275 delete_row_refs(struct ovsdb_txn *txn, const struct ovsdb_row *row,
 276                 const struct ovsdb_base_type *base,
 277                 const union ovsdb_atom *atoms, unsigned int n)
 278 {
 279     const struct ovsdb_table *table;
 280     unsigned int i;
 281
 282     if (!ovsdb_base_type_is_strong_ref(base)) {
 283         return NULL;
 284     }
 285
 286     table = base->u.uuid.refTable;
 287     for (i = 0; i < n; i++) {
 288         const struct uuid *uuid = &atoms[i].uuid;
 289         struct ovsdb_txn_row *txn_row;
 290
 291         if (uuid_equals(uuid, ovsdb_row_get_uuid(row))) {
 292             /* Self-references don't count. */
 293             continue;
 294         }
 295
 296         txn_row = find_or_make_txn_row(txn, table, uuid);
 297         if (!txn_row) {
 298             return OVSDB_BUG("strong ref target missing");
 299         } else if (!txn_row->n_refs) {
 300             return OVSDB_BUG("strong ref target has zero n_refs");
 301         } else if (!txn_row->new) {
 302             return OVSDB_BUG("deleted strong ref target");
 303         }
 304
 305         if (--txn_row->n_refs == 0) {
 306             struct ovsdb_error *error = delete_garbage_row(txn, txn_row);
 307             if (error) {
 308                 return error;
 309             }
 310         }
 311     }
 312
 313     return NULL;
 314 }
 315
 316 static struct ovsdb_error * WARN_UNUSED_RESULT
 317 delete_garbage_row(struct ovsdb_txn *txn, struct ovsdb_txn_row *txn_row)
 318 {
 319     struct shash_node *node;
 320     struct ovsdb_row *row;
 321
 322     if (txn_row->table->schema->is_root) {
 323         return NULL;
 324     }
 325
 326     row = txn_row->new;
 327     txn_row->new = NULL;
 328     hmap_remove(&txn_row->table->rows, &row->hmap_node);
 329     SHASH_FOR_EACH (node, &txn_row->table->schema->columns) {
 330         const struct ovsdb_column *column = node->data;
 331         const struct ovsdb_datum *field = &row->fields[column->index];
 332         struct ovsdb_error *error;
 333
 334         error = delete_row_refs(txn, row,
 335                                 &column->type.key, field->keys, field->n);
 336         if (error) {
 337             return error;
 338         }
 339
 340         error = delete_row_refs(txn, row,
 341                                 &column->type.value, field->values, field->n);
 342         if (error) {
 343             return error;
 344         }
 345     }
 346     ovsdb_row_destroy(row);
 347
 348     return NULL;
 349 }
 350
 351 static struct ovsdb_error * WARN_UNUSED_RESULT
 352 collect_garbage(struct ovsdb_txn *txn, struct ovsdb_txn_row *txn_row)
 353 {
 354     if (txn_row->new && !txn_row->n_refs) {
 355         return delete_garbage_row(txn, txn_row);
 356     }
 357     return NULL;
 358 }
 359
 360 static struct ovsdb_error * WARN_UNUSED_RESULT
 361 update_ref_counts(struct ovsdb_txn *txn)
 362 {
 363     struct ovsdb_error *error;
 364
 365     error = for_each_txn_row(txn, update_row_ref_count);
 366     if (error) {
 367         return error;
 368     }
 369
 370     return for_each_txn_row(txn, check_ref_count);
 371 }
 372
 373 static struct ovsdb_error *
 374 ovsdb_txn_row_commit(struct ovsdb_txn *txn OVS_UNUSED,
 375                      struct ovsdb_txn_row *txn_row)
 376 {
 377     ovsdb_txn_row_prefree(txn_row);
 378     if (txn_row->new) {
 379         txn_row->new->n_refs = txn_row->n_refs;
 380     }
 381     ovsdb_row_destroy(txn_row->old);
 382     free(txn_row);
 383
 384     return NULL;
 385 }
 386
 387 static void
 388 add_weak_ref(struct ovsdb_txn *txn,
 389              const struct ovsdb_row *src_, const struct ovsdb_row *dst_)
 390 {
 391     struct ovsdb_row *src = (struct ovsdb_row *) src_;
 392     struct ovsdb_row *dst = (struct ovsdb_row *) dst_;
 393     struct ovsdb_weak_ref *weak;
 394
 395     if (src == dst) {
 396         return;
 397     }
 398
 399     dst = ovsdb_txn_row_modify(txn, dst);
 400
 401     if (!list_is_empty(&dst->dst_refs)) {
 402         /* Omit duplicates. */
 403         weak = CONTAINER_OF(list_back(&dst->dst_refs),
 404                             struct ovsdb_weak_ref, dst_node);
 405         if (weak->src == src) {
 406             return;
 407         }
 408     }
 409
 410     weak = xmalloc(sizeof *weak);
 411     weak->src = src;
 412     list_push_back(&dst->dst_refs, &weak->dst_node);
 413     list_push_back(&src->src_refs, &weak->src_node);
 414 }
 415
 416 static struct ovsdb_error * WARN_UNUSED_RESULT
 417 assess_weak_refs(struct ovsdb_txn *txn, struct ovsdb_txn_row *txn_row)
 418 {
 419     struct ovsdb_table *table;
 420     struct shash_node *node;
 421
 422     if (txn_row->old) {
 423         /* Mark rows that have weak references to 'txn_row' as modified, so
 424          * that their weak references will get reassessed. */
 425         struct ovsdb_weak_ref *weak, *next;
 426
 427         LIST_FOR_EACH_SAFE (weak, next, dst_node, &txn_row->old->dst_refs) {
 428             if (!weak->src->txn_row) {
 429                 ovsdb_txn_row_modify(txn, weak->src);
 430             }
 431         }
 432     }
 433
 434     if (!txn_row->new) {
 435         /* We don't have to do anything about references that originate at
 436          * 'txn_row', because ovsdb_row_destroy() will remove those weak
 437          * references. */
 438         return NULL;
 439     }
 440
 441     table = txn_row->table;
 442     SHASH_FOR_EACH (node, &table->schema->columns) {
 443         const struct ovsdb_column *column = node->data;
 444         struct ovsdb_datum *datum = &txn_row->new->fields[column->index];
 445         unsigned int orig_n, i;
 446         bool zero = false;
 447
 448         orig_n = datum->n;
 449
 450         if (ovsdb_base_type_is_weak_ref(&column->type.key)) {
 451             for (i = 0; i < datum->n; ) {
 452                 const struct ovsdb_row *row;
 453
 454                 row = ovsdb_table_get_row(column->type.key.u.uuid.refTable,
 455                                           &datum->keys[i].uuid);
 456                 if (row) {
 457                     add_weak_ref(txn, txn_row->new, row);
 458                     i++;
 459                 } else {
 460                     if (uuid_is_zero(&datum->keys[i].uuid)) {
 461                         zero = true;
 462                     }
 463                     ovsdb_datum_remove_unsafe(datum, i, &column->type);
 464                 }
 465             }
 466         }
 467
 468         if (ovsdb_base_type_is_weak_ref(&column->type.value)) {
 469             for (i = 0; i < datum->n; ) {
 470                 const struct ovsdb_row *row;
 471
 472                 row = ovsdb_table_get_row(column->type.value.u.uuid.refTable,
 473                                           &datum->values[i].uuid);
 474                 if (row) {
 475                     add_weak_ref(txn, txn_row->new, row);
 476                     i++;
 477                 } else {
 478                     if (uuid_is_zero(&datum->values[i].uuid)) {
 479                         zero = true;
 480                     }
 481                     ovsdb_datum_remove_unsafe(datum, i, &column->type);
 482                 }
 483             }
 484         }
 485
 486         if (datum->n != orig_n) {
 487             bitmap_set1(txn_row->changed, column->index);
 488             ovsdb_datum_sort_assert(datum, column->type.key.type);
 489             if (datum->n < column->type.n_min) {
 490                 const struct uuid *row_uuid = ovsdb_row_get_uuid(txn_row->new);
 491                 if (zero && !txn_row->old) {
 492                     return ovsdb_error(
 493                         "constraint violation",
 494                         "Weak reference column \"%s\" in \"%s\" row "UUID_FMT
 495                         " (inserted within this transaction) contained "
 496                         "all-zeros UUID (probably as the default value for "
 497                         "this column) but deleting this value caused a "
 498                         "constraint volation because this column is not "
 499                         "allowed to be empty.", column->name,
 500                         table->schema->name, UUID_ARGS(row_uuid));
 501                 } else {
 502                     return ovsdb_error(
 503                         "constraint violation",
 504                         "Deletion of %u weak reference(s) to deleted (or "
 505                         "never-existing) rows from column \"%s\" in \"%s\" "
 506                         "row "UUID_FMT" %scaused this column to become empty, "
 507                         "but constraints on this column disallow an "
 508                         "empty column.",
 509                         orig_n - datum->n, column->name, table->schema->name,
 510                         UUID_ARGS(row_uuid),
 511                         (txn_row->old
 512                          ? ""
 513                          : "(inserted within this transaction) "));
 514                 }
 515             }
 516         }
 517     }
 518
 519     return NULL;
 520 }
 521
 522 static struct ovsdb_error * WARN_UNUSED_RESULT
 523 determine_changes(struct ovsdb_txn *txn, struct ovsdb_txn_row *txn_row)
 524 {
 525     struct ovsdb_table *table = txn_row->table;
 526
 527     if (txn_row->old && txn_row->new) {
 528         struct shash_node *node;
 529         bool changed = false;
 530
 531         SHASH_FOR_EACH (node, &table->schema->columns) {
 532             const struct ovsdb_column *column = node->data;
 533             const struct ovsdb_type *type = &column->type;
 534             unsigned int idx = column->index;
 535
 536             if (!ovsdb_datum_equals(&txn_row->old->fields[idx],
 537                                     &txn_row->new->fields[idx],
 538                                     type)) {
 539                 bitmap_set1(txn_row->changed, idx);
 540                 changed = true;
 541             }
 542         }
 543
 544         if (!changed) {
 545             /* Nothing actually changed in this row, so drop it. */
 546             ovsdb_txn_row_abort(txn, txn_row);
 547         }
 548     } else {
 549         bitmap_set_multiple(txn_row->changed, 0,
 550                             shash_count(&table->schema->columns), 1);
 551     }
 552
 553     return NULL;
 554 }
 555
 556 static struct ovsdb_error * WARN_UNUSED_RESULT
 557 check_max_rows(struct ovsdb_txn *txn)
 558 {
 559     struct ovsdb_txn_table *t;
 560
 561     LIST_FOR_EACH (t, node, &txn->txn_tables) {
 562         size_t n_rows = hmap_count(&t->table->rows);
 563         unsigned int max_rows = t->table->schema->max_rows;
 564
 565         if (n_rows > max_rows) {
 566             return ovsdb_error("constraint violation",
 567                                "transaction causes \"%s\" table to contain "
 568                                "%zu rows, greater than the schema-defined "
 569                                "limit of %u row(s)",
 570                                t->table->schema->name, n_rows, max_rows);
 571         }
 572     }
 573
 574     return NULL;
 575 }
 576
 577 struct ovsdb_error *
 578 ovsdb_txn_commit(struct ovsdb_txn *txn, bool durable)
 579 {
 580     struct ovsdb_replica *replica;
 581     struct ovsdb_error *error;
 582
 583     /* Figure out what actually changed, and abort early if the transaction
 584      * was really a no-op. */
 585     error = for_each_txn_row(txn, determine_changes);
 586     if (error) {
 587         return OVSDB_WRAP_BUG("can't happen", error);
 588     }
 589     if (list_is_empty(&txn->txn_tables)) {
 590         ovsdb_txn_abort(txn);
 591         return NULL;
 592     }
 593
 594     /* Update reference counts and check referential integrity. */
 595     error = update_ref_counts(txn);
 596     if (error) {
 597         ovsdb_txn_abort(txn);
 598         return error;
 599     }
 600
 601     /* Delete unreferenced, non-root rows. */
 602     error = for_each_txn_row(txn, collect_garbage);
 603     if (error) {
 604         ovsdb_txn_abort(txn);
 605         return OVSDB_WRAP_BUG("can't happen", error);
 606     }
 607
 608     /* Check maximum rows table constraints. */
 609     error = check_max_rows(txn);
 610     if (error) {
 611         ovsdb_txn_abort(txn);
 612         return error;
 613     }
 614
 615     /* Check reference counts and remove bad reference for "weak" referential
 616      * integrity. */
 617     error = for_each_txn_row(txn, assess_weak_refs);
 618     if (error) {
 619         ovsdb_txn_abort(txn);
 620         return error;
 621     }
 622
 623     /* Send the commit to each replica. */
 624     LIST_FOR_EACH (replica, node, &txn->db->replicas) {
 625         error = (replica->class->commit)(replica, txn, durable);
 626         if (error) {
 627             /* We don't support two-phase commit so only the first replica is
 628              * allowed to report an error. */
 629             assert(&replica->node == txn->db->replicas.next);
 630
 631             ovsdb_txn_abort(txn);
 632             return error;
 633         }
 634     }
 635
 636     /* Finalize commit. */
 637     txn->db->run_triggers = true;
 638     ovsdb_error_assert(for_each_txn_row(txn, ovsdb_txn_row_commit));
 639     ovsdb_txn_free(txn);
 640
 641     return NULL;
 642 }
 643
 644 void
 645 ovsdb_txn_for_each_change(const struct ovsdb_txn *txn,
 646                           ovsdb_txn_row_cb_func *cb, void *aux)
 647 {
 648     struct ovsdb_txn_table *t;
 649     struct ovsdb_txn_row *r;
 650
 651     LIST_FOR_EACH (t, node, &txn->txn_tables) {
 652         HMAP_FOR_EACH (r, hmap_node, &t->txn_rows) {
 653             if ((r->old || r->new) && !cb(r->old, r->new, r->changed, aux)) {
 654                 break;
 655             }
 656         }
 657    }
 658 }
 659
 660 static struct ovsdb_txn_table *
 661 ovsdb_txn_create_txn_table(struct ovsdb_txn *txn, struct ovsdb_table *table)
 662 {
 663     if (!table->txn_table) {
 664         struct ovsdb_txn_table *txn_table;
 665
 666         table->txn_table = txn_table = xmalloc(sizeof *table->txn_table);
 667         txn_table->table = table;
 668         hmap_init(&txn_table->txn_rows);
 669         txn_table->serial = serial - 1;
 670         list_push_back(&txn->txn_tables, &txn_table->node);
 671     }
 672     return table->txn_table;
 673 }
 674
 675 static struct ovsdb_txn_row *
 676 ovsdb_txn_row_create(struct ovsdb_txn *txn, struct ovsdb_table *table,
 677                      const struct ovsdb_row *old_, struct ovsdb_row *new)
 678 {
 679     const struct ovsdb_row *row = old_ ? old_ : new;
 680     struct ovsdb_row *old = (struct ovsdb_row *) old_;
 681     size_t n_columns = shash_count(&table->schema->columns);
 682     struct ovsdb_txn_table *txn_table;
 683     struct ovsdb_txn_row *txn_row;
 684
 685     txn_row = xzalloc(offsetof(struct ovsdb_txn_row, changed)
 686                       + bitmap_n_bytes(n_columns));
 687     txn_row->uuid = *ovsdb_row_get_uuid(row);
 688     txn_row->table = row->table;
 689     txn_row->old = old;
 690     txn_row->new = new;
 691     txn_row->n_refs = old ? old->n_refs : 0;
 692     txn_row->serial = serial - 1;
 693
 694     if (old) {
 695         old->txn_row = txn_row;
 696     }
 697     if (new) {
 698         new->txn_row = txn_row;
 699     }
 700
 701     txn_table = ovsdb_txn_create_txn_table(txn, table);
 702     hmap_insert(&txn_table->txn_rows, &txn_row->hmap_node,
 703                 ovsdb_row_hash(old ? old : new));
 704
 705     return txn_row;
 706 }
 707
 708 struct ovsdb_row *
 709 ovsdb_txn_row_modify(struct ovsdb_txn *txn, const struct ovsdb_row *ro_row_)
 710 {
 711     struct ovsdb_row *ro_row = (struct ovsdb_row *) ro_row_;
 712
 713     if (ro_row->txn_row) {
 714         assert(ro_row == ro_row->txn_row->new);
 715         return ro_row;
 716     } else {
 717         struct ovsdb_table *table = ro_row->table;
 718         struct ovsdb_row *rw_row;
 719
 720         rw_row = ovsdb_row_clone(ro_row);
 721         rw_row->n_refs = ro_row->n_refs;
 722         uuid_generate(ovsdb_row_get_version_rw(rw_row));
 723         ovsdb_txn_row_create(txn, table, ro_row, rw_row);
 724         hmap_replace(&table->rows, &ro_row->hmap_node, &rw_row->hmap_node);
 725
 726         return rw_row;
 727     }
 728 }
 729
 730 void
 731 ovsdb_txn_row_insert(struct ovsdb_txn *txn, struct ovsdb_row *row)
 732 {
 733     uint32_t hash = ovsdb_row_hash(row);
 734     struct ovsdb_table *table = row->table;
 735
 736     uuid_generate(ovsdb_row_get_version_rw(row));
 737
 738     ovsdb_txn_row_create(txn, table, NULL, row);
 739     hmap_insert(&table->rows, &row->hmap_node, hash);
 740 }
 741
 742 /* 'row' must be assumed destroyed upon return; the caller must not reference
 743  * it again. */
 744 void
 745 ovsdb_txn_row_delete(struct ovsdb_txn *txn, const struct ovsdb_row *row_)
 746 {
 747     struct ovsdb_row *row = (struct ovsdb_row *) row_;
 748     struct ovsdb_table *table = row->table;
 749     struct ovsdb_txn_row *txn_row = row->txn_row;
 750
 751     hmap_remove(&table->rows, &row->hmap_node);
 752
 753     if (!txn_row) {
 754         ovsdb_txn_row_create(txn, table, row, NULL);
 755     } else {
 756         assert(txn_row->new == row);
 757         if (txn_row->old) {
 758             txn_row->new = NULL;
 759         } else {
 760             hmap_remove(&table->txn_table->txn_rows, &txn_row->hmap_node);
 761             free(txn_row);
 762         }
 763         ovsdb_row_destroy(row);
 764     }
 765 }
 766
 767 void
 768 ovsdb_txn_add_comment(struct ovsdb_txn *txn, const char *s)
 769 {
 770     if (txn->comment.length) {
 771         ds_put_char(&txn->comment, '\n');
 772     }
 773     ds_put_cstr(&txn->comment, s);
 774 }
 775
 776 const char *
 777 ovsdb_txn_get_comment(const struct ovsdb_txn *txn)
 778 {
 779     return txn->comment.length ? ds_cstr_ro(&txn->comment) : NULL;
 780 }
 781 \f
 782 static void
 783 ovsdb_txn_row_prefree(struct ovsdb_txn_row *txn_row)
 784 {
 785     struct ovsdb_txn_table *txn_table = txn_row->table->txn_table;
 786
 787     txn_table->n_processed--;
 788     hmap_remove(&txn_table->txn_rows, &txn_row->hmap_node);
 789
 790     if (txn_row->old) {
 791         txn_row->old->txn_row = NULL;
 792     }
 793     if (txn_row->new) {
 794         txn_row->new->txn_row = NULL;
 795     }
 796 }
 797
 798 static void
 799 ovsdb_txn_table_destroy(struct ovsdb_txn_table *txn_table)
 800 {
 801     assert(hmap_is_empty(&txn_table->txn_rows));
 802     txn_table->table->txn_table = NULL;
 803     hmap_destroy(&txn_table->txn_rows);
 804     list_remove(&txn_table->node);
 805     free(txn_table);
 806 }
 807
 808 /* Calls 'cb' for every txn_row within 'txn'.  If 'cb' returns nonnull, this
 809  * aborts the iteration and for_each_txn_row() passes the error up.  Otherwise,
 810  * returns a null pointer after iteration is complete.
 811  *
 812  * 'cb' may insert new txn_rows and new txn_tables into 'txn'.  It may delete
 813  * the txn_row that it is passed in, or txn_rows in txn_tables other than the
 814  * one passed to 'cb'.  It may *not* delete txn_rows other than the one passed
 815  * in within the same txn_table.  It may *not* delete any txn_tables.  As long
 816  * as these rules are followed, 'cb' will be called exactly once for each
 817  * txn_row in 'txn', even those added by 'cb'.
 818  *
 819  * (Even though 'cb' is not allowed to delete some txn_rows, it can still
 820  * delete any actual row by clearing a txn_row's 'new' member.)
 821  */
 822 static struct ovsdb_error * WARN_UNUSED_RESULT
 823 for_each_txn_row(struct ovsdb_txn *txn,
 824                  struct ovsdb_error *(*cb)(struct ovsdb_txn *,
 825                                            struct ovsdb_txn_row *))
 826 {
 827     bool any_work;
 828
 829     serial++;
 830
 831     do {
 832         struct ovsdb_txn_table *t, *next_txn_table;
 833
 834         any_work = false;
 835         LIST_FOR_EACH_SAFE (t, next_txn_table, node, &txn->txn_tables) {
 836             if (t->serial != serial) {
 837                 t->serial = serial;
 838                 t->n_processed = 0;
 839             }
 840
 841             while (t->n_processed < hmap_count(&t->txn_rows)) {
 842                 struct ovsdb_txn_row *r, *next_txn_row;
 843
 844                 HMAP_FOR_EACH_SAFE (r, next_txn_row, hmap_node, &t->txn_rows) {
 845                     if (r->serial != serial) {
 846                         struct ovsdb_error *error;
 847
 848                         r->serial = serial;
 849                         t->n_processed++;
 850                         any_work = true;
 851
 852                         error = cb(txn, r);
 853                         if (error) {
 854                             return error;
 855                         }
 856                     }
 857                 }
 858             }
 859             if (hmap_is_empty(&t->txn_rows)) {
 860                 /* Table is empty.  Drop it. */
 861                 ovsdb_txn_table_destroy(t);
 862             }
 863         }
 864     } while (any_work);
 865
 866     return NULL;
 867 }