ovsdb/transaction.c

   1 /* Copyright (c) 2009, 2010 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 "dynamic-string.h"
  23 #include "hash.h"
  24 #include "hmap.h"
  25 #include "json.h"
  26 #include "list.h"
  27 #include "ovsdb-error.h"
  28 #include "ovsdb.h"
  29 #include "row.h"
  30 #include "table.h"
  31 #include "uuid.h"
  32
  33 struct ovsdb_txn {
  34     struct ovsdb *db;
  35     struct 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 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     /* Used by for_each_txn_row(). */
  46     unsigned int serial;        /* Serial number of in-progress iteration. */
  47     unsigned int n_processed;   /* Number of rows processed. */
  48 };
  49
  50 /* A row modified by the transaction:
  51  *
  52  *      - A row added by a transaction will have null 'old' and non-null 'new'.
  53  *
  54  *      - A row deleted by a transaction will have non-null 'old' and null
  55  *        'new'.
  56  *
  57  *      - A row modified by a transaction will have non-null 'old' and 'new'.
  58  *
  59  *      - 'old' and 'new' both null is invalid.  It would indicate that a row
  60  *        was added then deleted within a single transaction, but we instead
  61  *        handle that case by deleting the txn_row entirely.
  62  */
  63 struct ovsdb_txn_row {
  64     struct hmap_node hmap_node; /* In ovsdb_txn_table's txn_rows hmap. */
  65     struct ovsdb_row *old;      /* The old row. */
  66     struct ovsdb_row *new;      /* The new row. */
  67     size_t n_refs;              /* Number of remaining references. */
  68
  69     /* Used by for_each_txn_row(). */
  70     unsigned int serial;        /* Serial number of in-progress commit. */
  71 };
  72
  73 static void ovsdb_txn_row_prefree(struct ovsdb_txn_row *);
  74 static struct ovsdb_error * WARN_UNUSED_RESULT
  75 for_each_txn_row(struct ovsdb_txn *txn,
  76                       struct ovsdb_error *(*)(struct ovsdb_txn *,
  77                                               struct ovsdb_txn_row *));
  78
  79 /* Used by for_each_txn_row() to track tables and rows that have been
  80  * processed.  */
  81 static unsigned int serial;
  82
  83 struct ovsdb_txn *
  84 ovsdb_txn_create(struct ovsdb *db)
  85 {
  86     struct ovsdb_txn *txn = xmalloc(sizeof *txn);
  87     txn->db = db;
  88     list_init(&txn->txn_tables);
  89     ds_init(&txn->comment);
  90     return txn;
  91 }
  92
  93 static void
  94 ovsdb_txn_free(struct ovsdb_txn *txn)
  95 {
  96     assert(list_is_empty(&txn->txn_tables));
  97     ds_destroy(&txn->comment);
  98     free(txn);
  99 }
 100
 101 static struct ovsdb_error * WARN_UNUSED_RESULT
 102 ovsdb_txn_row_abort(struct ovsdb_txn *txn OVS_UNUSED,
 103                     struct ovsdb_txn_row *txn_row)
 104 {
 105     struct ovsdb_row *old = txn_row->old;
 106     struct ovsdb_row *new = txn_row->new;
 107
 108     ovsdb_txn_row_prefree(txn_row);
 109     if (!old) {
 110         hmap_remove(&new->table->rows, &new->hmap_node);
 111     } else if (!new) {
 112         hmap_insert(&old->table->rows, &old->hmap_node, ovsdb_row_hash(old));
 113     } else {
 114         hmap_replace(&new->table->rows, &new->hmap_node, &old->hmap_node);
 115     }
 116     ovsdb_row_destroy(new);
 117     free(txn_row);
 118
 119     return NULL;
 120 }
 121
 122 void
 123 ovsdb_txn_abort(struct ovsdb_txn *txn)
 124 {
 125     ovsdb_error_assert(for_each_txn_row(txn, ovsdb_txn_row_abort));
 126     ovsdb_txn_free(txn);
 127 }
 128
 129 static struct ovsdb_txn_row *
 130 find_txn_row(const struct ovsdb_table *table, const struct uuid *uuid)
 131 {
 132     struct ovsdb_txn_row *txn_row;
 133
 134     if (!table->txn_table) {
 135         return NULL;
 136     }
 137
 138     HMAP_FOR_EACH_WITH_HASH (txn_row, struct ovsdb_txn_row, hmap_node,
 139                              uuid_hash(uuid), &table->txn_table->txn_rows) {
 140         const struct ovsdb_row *row;
 141
 142         row = txn_row->old ? txn_row->old : txn_row->new;
 143         if (uuid_equals(uuid, ovsdb_row_get_uuid(row))) {
 144             return txn_row;
 145         }
 146     }
 147
 148     return NULL;
 149 }
 150
 151 static struct ovsdb_error * WARN_UNUSED_RESULT
 152 ovsdb_txn_adjust_atom_refs(struct ovsdb_txn *txn,
 153                            const struct ovsdb_base_type *base,
 154                            const union ovsdb_atom *atoms, unsigned int n,
 155                            int delta)
 156 {
 157     const struct ovsdb_table *table;
 158     unsigned int i;
 159
 160     if (base->type != OVSDB_TYPE_UUID || !base->u.uuid.refTable) {
 161         return NULL;
 162     }
 163
 164     table = base->u.uuid.refTable;
 165     for (i = 0; i < n; i++) {
 166         const struct uuid *uuid = &atoms[i].uuid;
 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             } else {
 173                 return ovsdb_error("referential integrity violation",
 174                                    "reference to nonexistent row "
 175                                    UUID_FMT, UUID_ARGS(uuid));
 176             }
 177         }
 178         txn_row->n_refs += delta;
 179     }
 180
 181     return NULL;
 182 }
 183
 184 static struct ovsdb_error * WARN_UNUSED_RESULT
 185 ovsdb_txn_adjust_row_refs(struct ovsdb_txn *txn, const struct ovsdb_row *r,
 186                           const struct ovsdb_column *column, int delta)
 187 {
 188     const struct ovsdb_datum *field = &r->fields[column->index];
 189     struct ovsdb_error *error;
 190
 191     error = ovsdb_txn_adjust_atom_refs(txn, &column->type.key,
 192                                        field->keys, field->n, delta);
 193     if (!error) {
 194         error = ovsdb_txn_adjust_atom_refs(txn, &column->type.value,
 195                                            field->values, field->n, delta);
 196     }
 197     return error;
 198 }
 199
 200 static struct ovsdb_error * WARN_UNUSED_RESULT
 201 update_row_ref_count(struct ovsdb_txn *txn, struct ovsdb_txn_row *r)
 202 {
 203     struct ovsdb_table *table = r->old ? r->old->table : r->new->table;
 204     struct shash_node *node;
 205
 206     SHASH_FOR_EACH (node, &table->schema->columns) {
 207         const struct ovsdb_column *column = node->data;
 208         struct ovsdb_error *error;
 209
 210         if (r->old) {
 211             error = ovsdb_txn_adjust_row_refs(txn, r->old, column, -1);
 212             if (error) {
 213                 ovsdb_error_destroy(error);
 214                 return OVSDB_BUG("error decreasing refcount");
 215             }
 216         }
 217         if (r->new) {
 218             error = ovsdb_txn_adjust_row_refs(txn, r->new, column, 1);
 219             if (error) {
 220                 return error;
 221             }
 222         }
 223     }
 224
 225     return NULL;
 226 }
 227
 228 static struct ovsdb_error * WARN_UNUSED_RESULT
 229 check_ref_count(struct ovsdb_txn *txn OVS_UNUSED, struct ovsdb_txn_row *r)
 230 {
 231     if (r->new || !r->n_refs) {
 232         return NULL;
 233     } else {
 234         return ovsdb_error("referential integrity violation",
 235                            "cannot delete %s row "UUID_FMT" because "
 236                            "of %zu remaining reference(s)",
 237                            r->old->table->schema->name,
 238                            UUID_ARGS(ovsdb_row_get_uuid(r->old)),
 239                            r->n_refs);
 240     }
 241 }
 242
 243 static struct ovsdb_error * WARN_UNUSED_RESULT
 244 update_ref_counts(struct ovsdb_txn *txn)
 245 {
 246     struct ovsdb_error *error;
 247
 248     error = for_each_txn_row(txn, update_row_ref_count);
 249     if (error) {
 250         return error;
 251     }
 252
 253     return for_each_txn_row(txn, check_ref_count);
 254 }
 255
 256 static struct ovsdb_error * WARN_UNUSED_RESULT
 257 ovsdb_txn_row_commit(struct ovsdb_txn *txn OVS_UNUSED,
 258                      struct ovsdb_txn_row *txn_row)
 259 {
 260     ovsdb_txn_row_prefree(txn_row);
 261     if (txn_row->new) {
 262         txn_row->new->n_refs = txn_row->n_refs;
 263     }
 264     ovsdb_row_destroy(txn_row->old);
 265     free(txn_row);
 266
 267     return NULL;
 268 }
 269
 270 struct ovsdb_error *
 271 ovsdb_txn_commit(struct ovsdb_txn *txn, bool durable)
 272 {
 273     struct ovsdb_replica *replica;
 274     struct ovsdb_error *error;
 275
 276     error = update_ref_counts(txn);
 277     if (error) {
 278         ovsdb_txn_abort(txn);
 279         return error;
 280     }
 281
 282     LIST_FOR_EACH (replica, struct ovsdb_replica, node, &txn->db->replicas) {
 283         error = (replica->class->commit)(replica, txn, durable);
 284         if (error) {
 285             /* We don't support two-phase commit so only the first replica is
 286              * allowed to report an error. */
 287             assert(&replica->node == txn->db->replicas.next);
 288
 289             ovsdb_txn_abort(txn);
 290             return error;
 291         }
 292     }
 293
 294     /* Finalize commit. */
 295     txn->db->run_triggers = true;
 296     ovsdb_error_assert(for_each_txn_row(txn, ovsdb_txn_row_commit));
 297     ovsdb_txn_free(txn);
 298
 299     return NULL;
 300 }
 301
 302 void
 303 ovsdb_txn_for_each_change(const struct ovsdb_txn *txn,
 304                           ovsdb_txn_row_cb_func *cb, void *aux)
 305 {
 306     struct ovsdb_txn_table *t;
 307     struct ovsdb_txn_row *r;
 308
 309     LIST_FOR_EACH (t, struct ovsdb_txn_table, node, &txn->txn_tables) {
 310         HMAP_FOR_EACH (r, struct ovsdb_txn_row, hmap_node, &t->txn_rows) {
 311             if (!cb(r->old, r->new, aux)) {
 312                 break;
 313             }
 314         }
 315    }
 316 }
 317
 318 static struct ovsdb_txn_table *
 319 ovsdb_txn_create_txn_table(struct ovsdb_txn *txn, struct ovsdb_table *table)
 320 {
 321     if (!table->txn_table) {
 322         struct ovsdb_txn_table *txn_table;
 323
 324         table->txn_table = txn_table = xmalloc(sizeof *table->txn_table);
 325         txn_table->table = table;
 326         hmap_init(&txn_table->txn_rows);
 327         txn_table->serial = serial - 1;
 328         list_push_back(&txn->txn_tables, &txn_table->node);
 329     }
 330     return table->txn_table;
 331 }
 332
 333 static struct ovsdb_txn_row *
 334 ovsdb_txn_row_create(struct ovsdb_txn *txn, struct ovsdb_table *table,
 335                      const struct ovsdb_row *old_, struct ovsdb_row *new)
 336 {
 337     struct ovsdb_row *old = (struct ovsdb_row *) old_;
 338     struct ovsdb_txn_table *txn_table;
 339     struct ovsdb_txn_row *txn_row;
 340
 341     txn_row = xmalloc(sizeof *txn_row);
 342     txn_row->old = old;
 343     txn_row->new = new;
 344     txn_row->n_refs = old ? old->n_refs : 0;
 345     txn_row->serial = serial - 1;
 346
 347     if (old) {
 348         old->txn_row = txn_row;
 349     }
 350     if (new) {
 351         new->txn_row = txn_row;
 352     }
 353
 354     txn_table = ovsdb_txn_create_txn_table(txn, table);
 355     hmap_insert(&txn_table->txn_rows, &txn_row->hmap_node,
 356                 ovsdb_row_hash(old ? old : new));
 357
 358     return txn_row;
 359 }
 360
 361 struct ovsdb_row *
 362 ovsdb_txn_row_modify(struct ovsdb_txn *txn, const struct ovsdb_row *ro_row_)
 363 {
 364     struct ovsdb_row *ro_row = (struct ovsdb_row *) ro_row_;
 365
 366     if (ro_row->txn_row) {
 367         assert(ro_row == ro_row->txn_row->new);
 368         return ro_row;
 369     } else {
 370         struct ovsdb_table *table = ro_row->table;
 371         struct ovsdb_row *rw_row;
 372
 373         rw_row = ovsdb_row_clone(ro_row);
 374         rw_row->n_refs = ro_row->n_refs;
 375         uuid_generate(ovsdb_row_get_version_rw(rw_row));
 376         ovsdb_txn_row_create(txn, table, ro_row, rw_row);
 377         hmap_replace(&table->rows, &ro_row->hmap_node, &rw_row->hmap_node);
 378
 379         return rw_row;
 380     }
 381 }
 382
 383 void
 384 ovsdb_txn_row_insert(struct ovsdb_txn *txn, struct ovsdb_row *row)
 385 {
 386     uint32_t hash = ovsdb_row_hash(row);
 387     struct ovsdb_table *table = row->table;
 388
 389     uuid_generate(ovsdb_row_get_version_rw(row));
 390
 391     ovsdb_txn_row_create(txn, table, NULL, row);
 392     hmap_insert(&table->rows, &row->hmap_node, hash);
 393 }
 394
 395 /* 'row' must be assumed destroyed upon return; the caller must not reference
 396  * it again. */
 397 void
 398 ovsdb_txn_row_delete(struct ovsdb_txn *txn, const struct ovsdb_row *row_)
 399 {
 400     struct ovsdb_row *row = (struct ovsdb_row *) row_;
 401     struct ovsdb_table *table = row->table;
 402     struct ovsdb_txn_row *txn_row = row->txn_row;
 403
 404     hmap_remove(&table->rows, &row->hmap_node);
 405
 406     if (!txn_row) {
 407         ovsdb_txn_row_create(txn, table, row, NULL);
 408     } else {
 409         assert(txn_row->new == row);
 410         if (txn_row->old) {
 411             txn_row->new = NULL;
 412         } else {
 413             hmap_remove(&table->txn_table->txn_rows, &txn_row->hmap_node);
 414             free(txn_row);
 415         }
 416         ovsdb_row_destroy(row);
 417     }
 418 }
 419
 420 void
 421 ovsdb_txn_add_comment(struct ovsdb_txn *txn, const char *s)
 422 {
 423     if (txn->comment.length) {
 424         ds_put_char(&txn->comment, '\n');
 425     }
 426     ds_put_cstr(&txn->comment, s);
 427 }
 428
 429 const char *
 430 ovsdb_txn_get_comment(const struct ovsdb_txn *txn)
 431 {
 432     return txn->comment.length ? ds_cstr_ro(&txn->comment) : NULL;
 433 }
 434 \f
 435 static void
 436 ovsdb_txn_row_prefree(struct ovsdb_txn_row *txn_row)
 437 {
 438     struct ovsdb_row *row = txn_row->old ? txn_row->old : txn_row->new;
 439     struct ovsdb_txn_table *txn_table = row->table->txn_table;
 440
 441     txn_table->n_processed--;
 442     hmap_remove(&txn_table->txn_rows, &txn_row->hmap_node);
 443
 444     if (txn_row->old) {
 445         txn_row->old->txn_row = NULL;
 446     }
 447     if (txn_row->new) {
 448         txn_row->new->txn_row = NULL;
 449     }
 450 }
 451
 452 static void
 453 ovsdb_txn_table_destroy(struct ovsdb_txn_table *txn_table)
 454 {
 455     assert(hmap_is_empty(&txn_table->txn_rows));
 456     txn_table->table->txn_table = NULL;
 457     hmap_destroy(&txn_table->txn_rows);
 458     list_remove(&txn_table->node);
 459     free(txn_table);
 460 }
 461
 462 /* Calls 'cb' for every txn_row within 'txn'.  If 'cb' returns nonnull, this
 463  * aborts the iteration and for_each_txn_row() passes the error up.  Otherwise,
 464  * returns a null pointer after iteration is complete.
 465  *
 466  * 'cb' may insert new txn_rows and new txn_tables into 'txn'.  It may delete
 467  * the txn_row that it is passed in, or txn_rows in txn_tables other than the
 468  * one passed to 'cb'.  It may *not* delete txn_rows other than the one passed
 469  * in within the same txn_table.  It may *not* delete any txn_tables.  As long
 470  * as these rules are followed, 'cb' will be called exactly once for each
 471  * txn_row in 'txn', even those added by 'cb'.
 472  */
 473 static struct ovsdb_error * WARN_UNUSED_RESULT
 474 for_each_txn_row(struct ovsdb_txn *txn,
 475                  struct ovsdb_error *(*cb)(struct ovsdb_txn *,
 476                                            struct ovsdb_txn_row *))
 477 {
 478     bool any_work;
 479
 480     serial++;
 481
 482     do {
 483         struct ovsdb_txn_table *t, *next_txn_table;
 484
 485         any_work = false;
 486         LIST_FOR_EACH_SAFE (t, next_txn_table, struct ovsdb_txn_table, node,
 487                             &txn->txn_tables) {
 488             if (t->serial != serial) {
 489                 t->serial = serial;
 490                 t->n_processed = 0;
 491             }
 492
 493             while (t->n_processed < hmap_count(&t->txn_rows)) {
 494                 struct ovsdb_txn_row *r, *next_txn_row;
 495
 496                 HMAP_FOR_EACH_SAFE (r, next_txn_row,
 497                                     struct ovsdb_txn_row, hmap_node,
 498                                     &t->txn_rows) {
 499                     if (r->serial != serial) {
 500                         struct ovsdb_error *error;
 501
 502                         r->serial = serial;
 503                         t->n_processed++;
 504                         any_work = true;
 505
 506                         error = cb(txn, r);
 507                         if (error) {
 508                             return error;
 509                         }
 510                     }
 511                 }
 512             }
 513             if (hmap_is_empty(&t->txn_rows)) {
 514                 /* Table is empty.  Drop it. */
 515                 ovsdb_txn_table_destroy(t);
 516             }
 517         }
 518     } while (any_work);
 519
 520     return NULL;
 521 }