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