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
  46 /* A row modified by the transaction:
  47  *
  48  *      - A row added by a transaction will have null 'old' and non-null 'new'.
  49  *
  50  *      - A row deleted by a transaction will have non-null 'old' and null
  51  *        'new'.
  52  *
  53  *      - A row modified by a transaction will have non-null 'old' and 'new'.
  54  *
  55  *      - 'old' and 'new' both null is invalid.  It would indicate that a row
  56  *        was added then deleted within a single transaction, but we instead
  57  *        handle that case by deleting the txn_row entirely.
  58  */
  59 struct ovsdb_txn_row {
  60     struct hmap_node hmap_node; /* In ovsdb_txn_table's txn_rows hmap. */
  61     struct ovsdb_row *old;      /* The old row. */
  62     struct ovsdb_row *new;      /* The new row. */
  63 };
  64
  65 struct ovsdb_txn *
  66 ovsdb_txn_create(struct ovsdb *db)
  67 {
  68     struct ovsdb_txn *txn = xmalloc(sizeof *txn);
  69     txn->db = db;
  70     list_init(&txn->txn_tables);
  71     ds_init(&txn->comment);
  72     return txn;
  73 }
  74
  75 static void
  76 ovsdb_txn_destroy(struct ovsdb_txn *txn, void (*cb)(struct ovsdb_txn_row *))
  77 {
  78     struct ovsdb_txn_table *txn_table, *next_txn_table;
  79
  80     LIST_FOR_EACH_SAFE (txn_table, next_txn_table,
  81                         struct ovsdb_txn_table, node, &txn->txn_tables) {
  82         struct ovsdb_txn_row *txn_row, *next_txn_row;
  83
  84         HMAP_FOR_EACH_SAFE (txn_row, next_txn_row,
  85                             struct ovsdb_txn_row, hmap_node,
  86                             &txn_table->txn_rows)
  87         {
  88             if (txn_row->old) {
  89                 txn_row->old->txn_row = NULL;
  90             }
  91             if (txn_row->new) {
  92                 txn_row->new->txn_row = NULL;
  93             }
  94             cb(txn_row);
  95             free(txn_row);
  96         }
  97
  98         txn_table->table->txn_table = NULL;
  99         hmap_destroy(&txn_table->txn_rows);
 100         free(txn_table);
 101     }
 102     ds_destroy(&txn->comment);
 103     free(txn);
 104 }
 105
 106 static void
 107 ovsdb_txn_row_abort(struct ovsdb_txn_row *txn_row)
 108 {
 109     struct ovsdb_row *old = txn_row->old;
 110     struct ovsdb_row *new = txn_row->new;
 111
 112     if (!old) {
 113         hmap_remove(&new->table->rows, &new->hmap_node);
 114     } else if (!new) {
 115         hmap_insert(&old->table->rows, &old->hmap_node, ovsdb_row_hash(old));
 116     } else {
 117         hmap_replace(&new->table->rows, &new->hmap_node, &old->hmap_node);
 118     }
 119     ovsdb_row_destroy(new);
 120 }
 121
 122 void
 123 ovsdb_txn_abort(struct ovsdb_txn *txn)
 124 {
 125     ovsdb_txn_destroy(txn, ovsdb_txn_row_abort);
 126 }
 127
 128 static void
 129 ovsdb_txn_row_commit(struct ovsdb_txn_row *txn_row)
 130 {
 131     ovsdb_row_destroy(txn_row->old);
 132 }
 133
 134 static struct ovsdb_txn_row *
 135 find_txn_row(const struct ovsdb_table *table, const struct uuid *uuid)
 136 {
 137     struct ovsdb_txn_row *txn_row;
 138
 139     if (!table->txn_table) {
 140         return NULL;
 141     }
 142
 143     HMAP_FOR_EACH_WITH_HASH (txn_row, struct ovsdb_txn_row, hmap_node,
 144                              uuid_hash(uuid), &table->txn_table->txn_rows) {
 145         const struct ovsdb_row *row;
 146
 147         row = txn_row->old ? txn_row->old : txn_row->new;
 148         if (uuid_equals(uuid, ovsdb_row_get_uuid(row))) {
 149             return txn_row;
 150         }
 151     }
 152
 153     return NULL;
 154 }
 155
 156 static void
 157 ovsdb_txn_adjust_atom_refs(const union ovsdb_atom *atoms, unsigned int n,
 158                            const struct ovsdb_table *table,
 159                            int delta, struct ovsdb_error **errorp)
 160 {
 161     unsigned int i;
 162
 163     for (i = 0; i < n; i++) {
 164         const struct uuid *uuid = &atoms[i].uuid;
 165         struct ovsdb_txn_row *txn_row = find_txn_row(table, uuid);
 166         if (txn_row) {
 167             if (txn_row->old) {
 168                 txn_row->old->n_refs += delta;
 169             }
 170             if (txn_row->new) {
 171                 txn_row->new->n_refs += delta;
 172             }
 173         } else {
 174             const struct ovsdb_row *row_ = ovsdb_table_get_row(table, uuid);
 175             if (row_) {
 176                 struct ovsdb_row *row = (struct ovsdb_row *) row_;
 177                 row->n_refs += delta;
 178             } else if (errorp) {
 179                 if (!*errorp) {
 180                     *errorp = ovsdb_error("referential integrity violation",
 181                                           "reference to nonexistent row "
 182                                           UUID_FMT, UUID_ARGS(uuid));
 183                 }
 184             } else {
 185                 NOT_REACHED();
 186             }
 187         }
 188     }
 189 }
 190
 191 static void
 192 ovsdb_txn_adjust_row_refs(const struct ovsdb_row *r,
 193                           const struct ovsdb_column *column, int delta,
 194                           struct ovsdb_error **errorp)
 195 {
 196     const struct ovsdb_datum *field = &r->fields[column->index];
 197     const struct ovsdb_type *type = &column->type;
 198
 199     if (type->key.type == OVSDB_TYPE_UUID && type->key.u.uuid.refTable) {
 200         ovsdb_txn_adjust_atom_refs(field->keys, field->n,
 201                                    type->key.u.uuid.refTable, delta, errorp);
 202     }
 203     if (type->value.type == OVSDB_TYPE_UUID && type->value.u.uuid.refTable) {
 204         ovsdb_txn_adjust_atom_refs(field->values, field->n,
 205                                    type->value.u.uuid.refTable, delta, errorp);
 206     }
 207 }
 208
 209 static struct ovsdb_error * WARN_UNUSED_RESULT
 210 ovsdb_txn_adjust_ref_counts__(struct ovsdb_txn *txn, int delta)
 211 {
 212     struct ovsdb_txn_table *t;
 213     struct ovsdb_error *error;
 214
 215     error = NULL;
 216     LIST_FOR_EACH (t, struct ovsdb_txn_table, node, &txn->txn_tables) {
 217         struct ovsdb_table *table = t->table;
 218         struct ovsdb_txn_row *r;
 219
 220         HMAP_FOR_EACH (r, struct ovsdb_txn_row, hmap_node, &t->txn_rows) {
 221             struct shash_node *node;
 222
 223             SHASH_FOR_EACH (node, &table->schema->columns) {
 224                 const struct ovsdb_column *column = node->data;
 225
 226                 if (r->old) {
 227                     ovsdb_txn_adjust_row_refs(r->old, column, -delta, NULL);
 228                 }
 229                 if (r->new) {
 230                     ovsdb_txn_adjust_row_refs(r->new, column, delta, &error);
 231                 }
 232             }
 233         }
 234     }
 235     return error;
 236 }
 237
 238 static void
 239 ovsdb_txn_rollback_counts(struct ovsdb_txn *txn)
 240 {
 241     ovsdb_error_destroy(ovsdb_txn_adjust_ref_counts__(txn, -1));
 242 }
 243
 244 static struct ovsdb_error * WARN_UNUSED_RESULT
 245 ovsdb_txn_commit_ref_counts(struct ovsdb_txn *txn)
 246 {
 247     struct ovsdb_error *error = ovsdb_txn_adjust_ref_counts__(txn, 1);
 248     if (error) {
 249         ovsdb_txn_rollback_counts(txn);
 250     }
 251     return error;
 252 }
 253
 254 static struct ovsdb_error * WARN_UNUSED_RESULT
 255 update_ref_counts(struct ovsdb_txn *txn)
 256 {
 257     struct ovsdb_error *error;
 258     struct ovsdb_txn_table *t;
 259
 260     error = ovsdb_txn_commit_ref_counts(txn);
 261     if (error) {
 262         return error;
 263     }
 264
 265     LIST_FOR_EACH (t, struct ovsdb_txn_table, node, &txn->txn_tables) {
 266         struct ovsdb_txn_row *r;
 267
 268         HMAP_FOR_EACH (r, struct ovsdb_txn_row, hmap_node, &t->txn_rows) {
 269             if (!r->new && r->old->n_refs) {
 270                 error = ovsdb_error("referential integrity violation",
 271                                     "cannot delete %s row "UUID_FMT" because "
 272                                     "of %zu remaining reference(s)",
 273                                     t->table->schema->name,
 274                                     UUID_ARGS(ovsdb_row_get_uuid(r->old)),
 275                                     r->old->n_refs);
 276                 ovsdb_txn_rollback_counts(txn);
 277                 return error;
 278             }
 279         }
 280     }
 281
 282     return NULL;
 283 }
 284
 285 struct ovsdb_error *
 286 ovsdb_txn_commit(struct ovsdb_txn *txn, bool durable)
 287 {
 288     struct ovsdb_replica *replica;
 289     struct ovsdb_error *error;
 290
 291     error = update_ref_counts(txn);
 292     if (error) {
 293         ovsdb_txn_abort(txn);
 294         return error;
 295     }
 296
 297     LIST_FOR_EACH (replica, struct ovsdb_replica, node, &txn->db->replicas) {
 298         error = (replica->class->commit)(replica, txn, durable);
 299         if (error) {
 300             /* We don't support two-phase commit so only the first replica is
 301              * allowed to report an error. */
 302             assert(&replica->node == txn->db->replicas.next);
 303
 304             ovsdb_txn_abort(txn);
 305             return error;
 306         }
 307     }
 308
 309     txn->db->run_triggers = true;
 310     ovsdb_txn_destroy(txn, ovsdb_txn_row_commit);
 311     return NULL;
 312 }
 313
 314 void
 315 ovsdb_txn_for_each_change(const struct ovsdb_txn *txn,
 316                           ovsdb_txn_row_cb_func *cb, void *aux)
 317 {
 318     struct ovsdb_txn_table *t;
 319     struct ovsdb_txn_row *r;
 320
 321     LIST_FOR_EACH (t, struct ovsdb_txn_table, node, &txn->txn_tables) {
 322         HMAP_FOR_EACH (r, struct ovsdb_txn_row, hmap_node, &t->txn_rows) {
 323             if (!cb(r->old, r->new, aux)) {
 324                 break;
 325             }
 326         }
 327    }
 328 }
 329
 330 static struct ovsdb_txn_table *
 331 ovsdb_txn_create_txn_table(struct ovsdb_txn *txn, struct ovsdb_table *table)
 332 {
 333     if (!table->txn_table) {
 334         struct ovsdb_txn_table *txn_table;
 335
 336         table->txn_table = txn_table = xmalloc(sizeof *table->txn_table);
 337         txn_table->table = table;
 338         hmap_init(&txn_table->txn_rows);
 339         list_push_back(&txn->txn_tables, &txn_table->node);
 340     }
 341     return table->txn_table;
 342 }
 343
 344 static struct ovsdb_txn_row *
 345 ovsdb_txn_row_create(struct ovsdb_txn *txn, struct ovsdb_table *table,
 346                      const struct ovsdb_row *old_, struct ovsdb_row *new)
 347 {
 348     struct ovsdb_row *old = (struct ovsdb_row *) old_;
 349     struct ovsdb_txn_table *txn_table;
 350     struct ovsdb_txn_row *txn_row;
 351
 352     txn_row = xmalloc(sizeof *txn_row);
 353     txn_row->old = old;
 354     txn_row->new = new;
 355
 356     if (old) {
 357         old->txn_row = txn_row;
 358     }
 359     if (new) {
 360         new->txn_row = txn_row;
 361     }
 362
 363     txn_table = ovsdb_txn_create_txn_table(txn, table);
 364     hmap_insert(&txn_table->txn_rows, &txn_row->hmap_node,
 365                 ovsdb_row_hash(old ? old : new));
 366
 367     return txn_row;
 368 }
 369
 370 struct ovsdb_row *
 371 ovsdb_txn_row_modify(struct ovsdb_txn *txn, const struct ovsdb_row *ro_row_)
 372 {
 373     struct ovsdb_row *ro_row = (struct ovsdb_row *) ro_row_;
 374
 375     if (ro_row->txn_row) {
 376         assert(ro_row == ro_row->txn_row->new);
 377         return ro_row;
 378     } else {
 379         struct ovsdb_table *table = ro_row->table;
 380         struct ovsdb_row *rw_row;
 381
 382         rw_row = ovsdb_row_clone(ro_row);
 383         rw_row->n_refs = ro_row->n_refs;
 384         uuid_generate(ovsdb_row_get_version_rw(rw_row));
 385         ovsdb_txn_row_create(txn, table, ro_row, rw_row);
 386         hmap_replace(&table->rows, &ro_row->hmap_node, &rw_row->hmap_node);
 387
 388         return rw_row;
 389     }
 390 }
 391
 392 void
 393 ovsdb_txn_row_insert(struct ovsdb_txn *txn, struct ovsdb_row *row)
 394 {
 395     uint32_t hash = ovsdb_row_hash(row);
 396     struct ovsdb_table *table = row->table;
 397
 398     uuid_generate(ovsdb_row_get_version_rw(row));
 399
 400     ovsdb_txn_row_create(txn, table, NULL, row);
 401     hmap_insert(&table->rows, &row->hmap_node, hash);
 402 }
 403
 404 /* 'row' must be assumed destroyed upon return; the caller must not reference
 405  * it again. */
 406 void
 407 ovsdb_txn_row_delete(struct ovsdb_txn *txn, const struct ovsdb_row *row_)
 408 {
 409     struct ovsdb_row *row = (struct ovsdb_row *) row_;
 410     struct ovsdb_table *table = row->table;
 411     struct ovsdb_txn_row *txn_row = row->txn_row;
 412
 413     hmap_remove(&table->rows, &row->hmap_node);
 414
 415     if (!txn_row) {
 416         ovsdb_txn_row_create(txn, table, row, NULL);
 417     } else {
 418         assert(txn_row->new == row);
 419         if (txn_row->old) {
 420             txn_row->new = NULL;
 421         } else {
 422             hmap_remove(&table->txn_table->txn_rows, &txn_row->hmap_node);
 423             free(txn_row);
 424         }
 425         ovsdb_row_destroy(row);
 426     }
 427 }
 428
 429 void
 430 ovsdb_txn_add_comment(struct ovsdb_txn *txn, const char *s)
 431 {
 432     if (txn->comment.length) {
 433         ds_put_char(&txn->comment, '\n');
 434     }
 435     ds_put_cstr(&txn->comment, s);
 436 }
 437
 438 const char *
 439 ovsdb_txn_get_comment(const struct ovsdb_txn *txn)
 440 {
 441     return txn->comment.length ? ds_cstr_ro(&txn->comment) : NULL;
 442 }