ovsdb/transaction.c

   1 /* Copyright (c) 2009 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 "ovsdb-error.h"
  27 #include "ovsdb.h"
  28 #include "row.h"
  29 #include "table.h"
  30 #include "uuid.h"
  31
  32 struct ovsdb_txn {
  33     struct ovsdb *db;
  34     struct hmap txn_tables;     /* Contains "struct ovsdb_txn_table"s. */
  35     struct ds comment;
  36 };
  37
  38 /* A table modified by a transaction. */
  39 struct ovsdb_txn_table {
  40     struct hmap_node hmap_node; /* Element in ovsdb_txn's txn_tables hmap. */
  41     struct ovsdb_table *table;
  42     struct hmap txn_rows;       /* Contains "struct ovsdb_txn_row"s. */
  43 };
  44
  45 /* A row modified by the transaction:
  46  *
  47  *      - A row added by a transaction will have null 'old' and non-null 'new'.
  48  *
  49  *      - A row deleted by a transaction will have non-null 'old' and null
  50  *        'new'.
  51  *
  52  *      - A row modified by a transaction will have non-null 'old' and 'new'.
  53  *
  54  *      - 'old' and 'new' both null is invalid.  It would indicate that a row
  55  *        was added then deleted within a single transaction, but we instead
  56  *        handle that case by deleting the txn_row entirely.
  57  */
  58 struct ovsdb_txn_row {
  59     struct hmap_node hmap_node; /* In ovsdb_txn_table's txn_rows hmap. */
  60     struct ovsdb_row *old;      /* The old row. */
  61     struct ovsdb_row *new;      /* The new row. */
  62 };
  63
  64 struct ovsdb_txn *
  65 ovsdb_txn_create(struct ovsdb *db)
  66 {
  67     struct ovsdb_txn *txn = xmalloc(sizeof *txn);
  68     txn->db = db;
  69     hmap_init(&txn->txn_tables);
  70     ds_init(&txn->comment);
  71     return txn;
  72 }
  73
  74 static void
  75 ovsdb_txn_destroy(struct ovsdb_txn *txn, void (*cb)(struct ovsdb_txn_row *))
  76 {
  77     struct ovsdb_txn_table *txn_table, *next_txn_table;
  78
  79     HMAP_FOR_EACH_SAFE (txn_table, next_txn_table,
  80                         struct ovsdb_txn_table, hmap_node, &txn->txn_tables)
  81     {
  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         hmap_destroy(&txn_table->txn_rows);
  99         free(txn_table);
 100     }
 101     hmap_destroy(&txn->txn_tables);
 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 struct ovsdb_error *
 135 ovsdb_txn_commit(struct ovsdb_txn *txn, bool durable)
 136 {
 137     struct ovsdb_replica *replica;
 138     struct ovsdb_error *error;
 139
 140     LIST_FOR_EACH (replica, struct ovsdb_replica, node, &txn->db->replicas) {
 141         error = (replica->class->commit)(replica, txn, durable);
 142         if (error) {
 143             /* We don't support two-phase commit so only the first replica is
 144              * allowed to report an error. */
 145             assert(&replica->node == txn->db->replicas.next);
 146
 147             ovsdb_txn_abort(txn);
 148             return error;
 149         }
 150     }
 151
 152     txn->db->run_triggers = true;
 153     ovsdb_txn_destroy(txn, ovsdb_txn_row_commit);
 154     return NULL;
 155 }
 156
 157 void
 158 ovsdb_txn_for_each_change(const struct ovsdb_txn *txn,
 159                           ovsdb_txn_row_cb_func *cb, void *aux)
 160 {
 161     struct ovsdb_txn_table *t;
 162     struct ovsdb_txn_row *r;
 163
 164     HMAP_FOR_EACH (t, struct ovsdb_txn_table, hmap_node, &txn->txn_tables) {
 165         HMAP_FOR_EACH (r, struct ovsdb_txn_row, hmap_node, &t->txn_rows) {
 166             if (!cb(r->old, r->new, aux)) {
 167                 break;
 168             }
 169         }
 170    }
 171 }
 172
 173 static struct ovsdb_txn_table *
 174 ovsdb_txn_get_txn_table__(struct ovsdb_txn *txn,
 175                           const struct ovsdb_table *table,
 176                           uint32_t hash)
 177 {
 178     struct ovsdb_txn_table *txn_table;
 179
 180     HMAP_FOR_EACH_IN_BUCKET (txn_table, struct ovsdb_txn_table, hmap_node,
 181                              hash, &txn->txn_tables) {
 182         if (txn_table->table == table) {
 183             return txn_table;
 184         }
 185     }
 186
 187     return NULL;
 188 }
 189
 190 static struct ovsdb_txn_table *
 191 ovsdb_txn_get_txn_table(struct ovsdb_txn *txn, const struct ovsdb_table *table)
 192 {
 193     return ovsdb_txn_get_txn_table__(txn, table, hash_pointer(table, 0));
 194 }
 195
 196 static struct ovsdb_txn_table *
 197 ovsdb_txn_create_txn_table(struct ovsdb_txn *txn,
 198                            struct ovsdb_table *table)
 199 {
 200     uint32_t hash = hash_pointer(table, 0);
 201     struct ovsdb_txn_table *txn_table;
 202
 203     txn_table = ovsdb_txn_get_txn_table__(txn, table, hash);
 204     if (!txn_table) {
 205         txn_table = xmalloc(sizeof *txn_table);
 206         txn_table->table = table;
 207         hmap_init(&txn_table->txn_rows);
 208         hmap_insert(&txn->txn_tables, &txn_table->hmap_node, hash);
 209     }
 210     return txn_table;
 211 }
 212
 213 static struct ovsdb_txn_row *
 214 ovsdb_txn_row_create(struct ovsdb_txn_table *txn_table,
 215                      const struct ovsdb_row *old, struct ovsdb_row *new)
 216 {
 217     uint32_t hash = ovsdb_row_hash(old ? old : new);
 218     struct ovsdb_txn_row *txn_row;
 219
 220     txn_row = xmalloc(sizeof *txn_row);
 221     txn_row->old = (struct ovsdb_row *) old;
 222     txn_row->new = new;
 223     hmap_insert(&txn_table->txn_rows, &txn_row->hmap_node, hash);
 224
 225     return txn_row;
 226 }
 227
 228 struct ovsdb_row *
 229 ovsdb_txn_row_modify(struct ovsdb_txn *txn, const struct ovsdb_row *ro_row_)
 230 {
 231     struct ovsdb_row *ro_row = (struct ovsdb_row *) ro_row_;
 232
 233     if (ro_row->txn_row) {
 234         assert(ro_row == ro_row->txn_row->new);
 235         return ro_row;
 236     } else {
 237         struct ovsdb_table *table = ro_row->table;
 238         struct ovsdb_txn_table *txn_table;
 239         struct ovsdb_row *rw_row;
 240
 241         txn_table = ovsdb_txn_create_txn_table(txn, table);
 242         rw_row = ovsdb_row_clone(ro_row);
 243         uuid_generate(ovsdb_row_get_version_rw(rw_row));
 244         rw_row->txn_row = ovsdb_txn_row_create(txn_table, ro_row, rw_row);
 245         hmap_replace(&table->rows, &ro_row->hmap_node, &rw_row->hmap_node);
 246
 247         return rw_row;
 248     }
 249 }
 250
 251 void
 252 ovsdb_txn_row_insert(struct ovsdb_txn *txn, struct ovsdb_row *row)
 253 {
 254     uint32_t hash = ovsdb_row_hash(row);
 255     struct ovsdb_table *table = row->table;
 256     struct ovsdb_txn_table *txn_table;
 257
 258     uuid_generate(ovsdb_row_get_version_rw(row));
 259
 260     txn_table = ovsdb_txn_create_txn_table(txn, table);
 261     row->txn_row = ovsdb_txn_row_create(txn_table, NULL, row);
 262     hmap_insert(&table->rows, &row->hmap_node, hash);
 263 }
 264
 265 /* 'row' must be assumed destroyed upon return; the caller must not reference
 266  * it again. */
 267 void
 268 ovsdb_txn_row_delete(struct ovsdb_txn *txn, const struct ovsdb_row *row_)
 269 {
 270     struct ovsdb_row *row = (struct ovsdb_row *) row_;
 271     struct ovsdb_table *table = row->table;
 272     struct ovsdb_txn_row *txn_row = row->txn_row;
 273     struct ovsdb_txn_table *txn_table;
 274
 275     hmap_remove(&table->rows, &row->hmap_node);
 276
 277     if (!txn_row) {
 278         txn_table = ovsdb_txn_create_txn_table(txn, table);
 279         row->txn_row = ovsdb_txn_row_create(txn_table, row, NULL);
 280     } else {
 281         assert(txn_row->new == row);
 282         if (txn_row->old) {
 283             txn_row->new = NULL;
 284         } else {
 285             txn_table = ovsdb_txn_get_txn_table(txn, table);
 286             hmap_remove(&txn_table->txn_rows, &txn_row->hmap_node);
 287             free(txn_row);
 288         }
 289         ovsdb_row_destroy(row);
 290     }
 291 }
 292
 293 void
 294 ovsdb_txn_add_comment(struct ovsdb_txn *txn, const char *s)
 295 {
 296     if (txn->comment.length) {
 297         ds_put_char(&txn->comment, '\n');
 298     }
 299     ds_put_cstr(&txn->comment, s);
 300 }
 301
 302 const char *
 303 ovsdb_txn_get_comment(const struct ovsdb_txn *txn)
 304 {
 305     return txn->comment.length ? ds_cstr_ro(&txn->comment) : NULL;
 306 }