X-Git-Url: https://pintos-os.org/cgi-bin/gitweb.cgi?a=blobdiff_plain;f=ovsdb%2Ftransaction.c;h=c07541ed3a049579d4db1325a98f48d8a77120d5;hb=e7009c364026d69381cdda23941f99ff040d4948;hp=ac9b7f3196916c596f3e1649379c1aa45e83b0dd;hpb=c7d85e0df048a9fda6e1a111cd74e5a82e6b3b91;p=openvswitch diff --git a/ovsdb/transaction.c b/ovsdb/transaction.c index ac9b7f31..c07541ed 100644 --- a/ovsdb/transaction.c +++ b/ovsdb/transaction.c @@ -1,4 +1,4 @@ -/* Copyright (c) 2009, 2010 Nicira Networks +/* Copyright (c) 2009, 2010, 2011 Nicira Networks * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -19,6 +19,7 @@ #include +#include "bitmap.h" #include "dynamic-string.h" #include "hash.h" #include "hmap.h" @@ -56,9 +57,11 @@ struct ovsdb_txn_table { * * - A row modified by a transaction will have non-null 'old' and 'new'. * - * - 'old' and 'new' both null is invalid. It would indicate that a row - * was added then deleted within a single transaction, but we instead - * handle that case by deleting the txn_row entirely. + * - 'old' and 'new' both null indicates that a row was added then deleted + * within a single transaction. Most of the time we instead delete the + * ovsdb_txn_row entirely, but inside a for_each_txn_row() callback + * there are restrictions that sometimes mean we have to leave the + * ovsdb_txn_row in place. */ struct ovsdb_txn_row { struct hmap_node hmap_node; /* In ovsdb_txn_table's txn_rows hmap. */ @@ -66,10 +69,20 @@ struct ovsdb_txn_row { struct ovsdb_row *new; /* The new row. */ size_t n_refs; /* Number of remaining references. */ + /* These members are the same as the corresponding members of 'old' or + * 'new'. They are present here for convenience and because occasionally + * there can be an ovsdb_txn_row where both 'old' and 'new' are NULL. */ + struct uuid uuid; + struct ovsdb_table *table; + /* Used by for_each_txn_row(). */ unsigned int serial; /* Serial number of in-progress commit. */ + + unsigned long changed[]; /* Bits set to 1 for columns that changed. */ }; +static struct ovsdb_error * WARN_UNUSED_RESULT +delete_garbage_row(struct ovsdb_txn *txn, struct ovsdb_txn_row *r); static void ovsdb_txn_row_prefree(struct ovsdb_txn_row *); static struct ovsdb_error * WARN_UNUSED_RESULT for_each_txn_row(struct ovsdb_txn *txn, @@ -98,7 +111,7 @@ ovsdb_txn_free(struct ovsdb_txn *txn) free(txn); } -static struct ovsdb_error * WARN_UNUSED_RESULT +static struct ovsdb_error * ovsdb_txn_row_abort(struct ovsdb_txn *txn OVS_UNUSED, struct ovsdb_txn_row *txn_row) { @@ -107,7 +120,9 @@ ovsdb_txn_row_abort(struct ovsdb_txn *txn OVS_UNUSED, ovsdb_txn_row_prefree(txn_row); if (!old) { - hmap_remove(&new->table->rows, &new->hmap_node); + if (new) { + hmap_remove(&new->table->rows, &new->hmap_node); + } } else if (!new) { hmap_insert(&old->table->rows, &old->hmap_node, ovsdb_row_hash(old)); } else { @@ -135,12 +150,9 @@ find_txn_row(const struct ovsdb_table *table, const struct uuid *uuid) return NULL; } - HMAP_FOR_EACH_WITH_HASH (txn_row, struct ovsdb_txn_row, hmap_node, + HMAP_FOR_EACH_WITH_HASH (txn_row, hmap_node, uuid_hash(uuid), &table->txn_table->txn_rows) { - const struct ovsdb_row *row; - - row = txn_row->old ? txn_row->old : txn_row->new; - if (uuid_equals(uuid, ovsdb_row_get_uuid(row))) { + if (uuid_equals(uuid, &txn_row->uuid)) { return txn_row; } } @@ -148,8 +160,23 @@ find_txn_row(const struct ovsdb_table *table, const struct uuid *uuid) return NULL; } +static struct ovsdb_txn_row * +find_or_make_txn_row(struct ovsdb_txn *txn, const struct ovsdb_table *table, + const struct uuid *uuid) +{ + struct ovsdb_txn_row *txn_row = find_txn_row(table, uuid); + if (!txn_row) { + const struct ovsdb_row *row = ovsdb_table_get_row(table, uuid); + if (row) { + txn_row = ovsdb_txn_row_modify(txn, row)->txn_row; + } + } + return txn_row; +} + static struct ovsdb_error * WARN_UNUSED_RESULT -ovsdb_txn_adjust_atom_refs(struct ovsdb_txn *txn, +ovsdb_txn_adjust_atom_refs(struct ovsdb_txn *txn, const struct ovsdb_row *r, + const struct ovsdb_column *c, const struct ovsdb_base_type *base, const union ovsdb_atom *atoms, unsigned int n, int delta) @@ -157,23 +184,29 @@ ovsdb_txn_adjust_atom_refs(struct ovsdb_txn *txn, const struct ovsdb_table *table; unsigned int i; - if (base->type != OVSDB_TYPE_UUID || !base->u.uuid.refTable) { + if (!ovsdb_base_type_is_strong_ref(base)) { return NULL; } table = base->u.uuid.refTable; for (i = 0; i < n; i++) { const struct uuid *uuid = &atoms[i].uuid; - struct ovsdb_txn_row *txn_row = find_txn_row(table, uuid); + struct ovsdb_txn_row *txn_row; + + if (uuid_equals(uuid, ovsdb_row_get_uuid(r))) { + /* Self-references don't count. */ + continue; + } + + txn_row = find_or_make_txn_row(txn, table, uuid); if (!txn_row) { - const struct ovsdb_row *row = ovsdb_table_get_row(table, uuid); - if (row) { - txn_row = ovsdb_txn_row_modify(txn, row)->txn_row; - } else { - return ovsdb_error("referential integrity violation", - "reference to nonexistent row " - UUID_FMT, UUID_ARGS(uuid)); - } + return ovsdb_error("referential integrity violation", + "Table %s column %s row "UUID_FMT" " + "references nonexistent row "UUID_FMT" in " + "table %s.", + r->table->schema->name, c->name, + UUID_ARGS(ovsdb_row_get_uuid(r)), + UUID_ARGS(uuid), table->schema->name); } txn_row->n_refs += delta; } @@ -188,10 +221,10 @@ ovsdb_txn_adjust_row_refs(struct ovsdb_txn *txn, const struct ovsdb_row *r, const struct ovsdb_datum *field = &r->fields[column->index]; struct ovsdb_error *error; - error = ovsdb_txn_adjust_atom_refs(txn, &column->type.key, + error = ovsdb_txn_adjust_atom_refs(txn, r, column, &column->type.key, field->keys, field->n, delta); if (!error) { - error = ovsdb_txn_adjust_atom_refs(txn, &column->type.value, + error = ovsdb_txn_adjust_atom_refs(txn, r, column, &column->type.value, field->values, field->n, delta); } return error; @@ -200,7 +233,7 @@ ovsdb_txn_adjust_row_refs(struct ovsdb_txn *txn, const struct ovsdb_row *r, static struct ovsdb_error * WARN_UNUSED_RESULT update_row_ref_count(struct ovsdb_txn *txn, struct ovsdb_txn_row *r) { - struct ovsdb_table *table = r->old ? r->old->table : r->new->table; + struct ovsdb_table *table = r->table; struct shash_node *node; SHASH_FOR_EACH (node, &table->schema->columns) { @@ -210,8 +243,7 @@ update_row_ref_count(struct ovsdb_txn *txn, struct ovsdb_txn_row *r) if (r->old) { error = ovsdb_txn_adjust_row_refs(txn, r->old, column, -1); if (error) { - ovsdb_error_destroy(error); - return OVSDB_BUG("error decreasing refcount"); + return OVSDB_WRAP_BUG("error decreasing refcount", error); } } if (r->new) { @@ -234,12 +266,97 @@ check_ref_count(struct ovsdb_txn *txn OVS_UNUSED, struct ovsdb_txn_row *r) return ovsdb_error("referential integrity violation", "cannot delete %s row "UUID_FMT" because " "of %zu remaining reference(s)", - r->old->table->schema->name, - UUID_ARGS(ovsdb_row_get_uuid(r->old)), + r->table->schema->name, UUID_ARGS(&r->uuid), r->n_refs); } } +static struct ovsdb_error * WARN_UNUSED_RESULT +delete_row_refs(struct ovsdb_txn *txn, const struct ovsdb_row *row, + const struct ovsdb_base_type *base, + const union ovsdb_atom *atoms, unsigned int n) +{ + const struct ovsdb_table *table; + unsigned int i; + + if (!ovsdb_base_type_is_strong_ref(base)) { + return NULL; + } + + table = base->u.uuid.refTable; + for (i = 0; i < n; i++) { + const struct uuid *uuid = &atoms[i].uuid; + struct ovsdb_txn_row *txn_row; + + if (uuid_equals(uuid, ovsdb_row_get_uuid(row))) { + /* Self-references don't count. */ + continue; + } + + txn_row = find_or_make_txn_row(txn, table, uuid); + if (!txn_row) { + return OVSDB_BUG("strong ref target missing"); + } else if (!txn_row->n_refs) { + return OVSDB_BUG("strong ref target has zero n_refs"); + } else if (!txn_row->new) { + return OVSDB_BUG("deleted strong ref target"); + } + + if (--txn_row->n_refs == 0) { + struct ovsdb_error *error = delete_garbage_row(txn, txn_row); + if (error) { + return error; + } + } + } + + return NULL; +} + +static struct ovsdb_error * WARN_UNUSED_RESULT +delete_garbage_row(struct ovsdb_txn *txn, struct ovsdb_txn_row *txn_row) +{ + struct shash_node *node; + struct ovsdb_row *row; + + if (txn_row->table->schema->is_root) { + return NULL; + } + + row = txn_row->new; + txn_row->new = NULL; + hmap_remove(&txn_row->table->rows, &row->hmap_node); + SHASH_FOR_EACH (node, &txn_row->table->schema->columns) { + const struct ovsdb_column *column = node->data; + const struct ovsdb_datum *field = &row->fields[column->index]; + struct ovsdb_error *error; + + error = delete_row_refs(txn, row, + &column->type.key, field->keys, field->n); + if (error) { + return error; + } + + error = delete_row_refs(txn, row, + &column->type.value, field->values, field->n); + if (error) { + return error; + } + } + ovsdb_row_destroy(row); + + return NULL; +} + +static struct ovsdb_error * WARN_UNUSED_RESULT +collect_garbage(struct ovsdb_txn *txn, struct ovsdb_txn_row *txn_row) +{ + if (txn_row->new && !txn_row->n_refs) { + return delete_garbage_row(txn, txn_row); + } + return NULL; +} + static struct ovsdb_error * WARN_UNUSED_RESULT update_ref_counts(struct ovsdb_txn *txn) { @@ -253,7 +370,7 @@ update_ref_counts(struct ovsdb_txn *txn) return for_each_txn_row(txn, check_ref_count); } -static struct ovsdb_error * WARN_UNUSED_RESULT +static struct ovsdb_error * ovsdb_txn_row_commit(struct ovsdb_txn *txn OVS_UNUSED, struct ovsdb_txn_row *txn_row) { @@ -267,19 +384,244 @@ ovsdb_txn_row_commit(struct ovsdb_txn *txn OVS_UNUSED, return NULL; } +static void +add_weak_ref(struct ovsdb_txn *txn, + const struct ovsdb_row *src_, const struct ovsdb_row *dst_) +{ + struct ovsdb_row *src = (struct ovsdb_row *) src_; + struct ovsdb_row *dst = (struct ovsdb_row *) dst_; + struct ovsdb_weak_ref *weak; + + if (src == dst) { + return; + } + + dst = ovsdb_txn_row_modify(txn, dst); + + if (!list_is_empty(&dst->dst_refs)) { + /* Omit duplicates. */ + weak = CONTAINER_OF(list_back(&dst->dst_refs), + struct ovsdb_weak_ref, dst_node); + if (weak->src == src) { + return; + } + } + + weak = xmalloc(sizeof *weak); + weak->src = src; + list_push_back(&dst->dst_refs, &weak->dst_node); + list_push_back(&src->src_refs, &weak->src_node); +} + +static struct ovsdb_error * WARN_UNUSED_RESULT +assess_weak_refs(struct ovsdb_txn *txn, struct ovsdb_txn_row *txn_row) +{ + struct ovsdb_table *table; + struct shash_node *node; + + if (txn_row->old) { + /* Mark rows that have weak references to 'txn_row' as modified, so + * that their weak references will get reassessed. */ + struct ovsdb_weak_ref *weak, *next; + + LIST_FOR_EACH_SAFE (weak, next, dst_node, &txn_row->old->dst_refs) { + if (!weak->src->txn_row) { + ovsdb_txn_row_modify(txn, weak->src); + } + } + } + + if (!txn_row->new) { + /* We don't have to do anything about references that originate at + * 'txn_row', because ovsdb_row_destroy() will remove those weak + * references. */ + return NULL; + } + + table = txn_row->table; + SHASH_FOR_EACH (node, &table->schema->columns) { + const struct ovsdb_column *column = node->data; + struct ovsdb_datum *datum = &txn_row->new->fields[column->index]; + unsigned int orig_n, i; + bool zero = false; + + orig_n = datum->n; + + if (ovsdb_base_type_is_weak_ref(&column->type.key)) { + for (i = 0; i < datum->n; ) { + const struct ovsdb_row *row; + + row = ovsdb_table_get_row(column->type.key.u.uuid.refTable, + &datum->keys[i].uuid); + if (row) { + add_weak_ref(txn, txn_row->new, row); + i++; + } else { + if (uuid_is_zero(&datum->keys[i].uuid)) { + zero = true; + } + ovsdb_datum_remove_unsafe(datum, i, &column->type); + } + } + } + + if (ovsdb_base_type_is_weak_ref(&column->type.value)) { + for (i = 0; i < datum->n; ) { + const struct ovsdb_row *row; + + row = ovsdb_table_get_row(column->type.value.u.uuid.refTable, + &datum->values[i].uuid); + if (row) { + add_weak_ref(txn, txn_row->new, row); + i++; + } else { + if (uuid_is_zero(&datum->values[i].uuid)) { + zero = true; + } + ovsdb_datum_remove_unsafe(datum, i, &column->type); + } + } + } + + if (datum->n != orig_n) { + bitmap_set1(txn_row->changed, column->index); + ovsdb_datum_sort_assert(datum, column->type.key.type); + if (datum->n < column->type.n_min) { + const struct uuid *row_uuid = ovsdb_row_get_uuid(txn_row->new); + if (zero && !txn_row->old) { + return ovsdb_error( + "constraint violation", + "Weak reference column \"%s\" in \"%s\" row "UUID_FMT + " (inserted within this transaction) contained " + "all-zeros UUID (probably as the default value for " + "this column) but deleting this value caused a " + "constraint volation because this column is not " + "allowed to be empty.", column->name, + table->schema->name, UUID_ARGS(row_uuid)); + } else { + return ovsdb_error( + "constraint violation", + "Deletion of %u weak reference(s) to deleted (or " + "never-existing) rows from column \"%s\" in \"%s\" " + "row "UUID_FMT" %scaused this column to become empty, " + "but constraints on this column disallow an " + "empty column.", + orig_n - datum->n, column->name, table->schema->name, + UUID_ARGS(row_uuid), + (txn_row->old + ? "" + : "(inserted within this transaction) ")); + } + } + } + } + + return NULL; +} + +static struct ovsdb_error * WARN_UNUSED_RESULT +determine_changes(struct ovsdb_txn *txn, struct ovsdb_txn_row *txn_row) +{ + struct ovsdb_table *table = txn_row->table; + + if (txn_row->old && txn_row->new) { + struct shash_node *node; + bool changed = false; + + SHASH_FOR_EACH (node, &table->schema->columns) { + const struct ovsdb_column *column = node->data; + const struct ovsdb_type *type = &column->type; + unsigned int idx = column->index; + + if (!ovsdb_datum_equals(&txn_row->old->fields[idx], + &txn_row->new->fields[idx], + type)) { + bitmap_set1(txn_row->changed, idx); + changed = true; + } + } + + if (!changed) { + /* Nothing actually changed in this row, so drop it. */ + ovsdb_txn_row_abort(txn, txn_row); + } + } else { + bitmap_set_multiple(txn_row->changed, 0, + shash_count(&table->schema->columns), 1); + } + + return NULL; +} + +static struct ovsdb_error * WARN_UNUSED_RESULT +check_max_rows(struct ovsdb_txn *txn) +{ + struct ovsdb_txn_table *t; + + LIST_FOR_EACH (t, node, &txn->txn_tables) { + size_t n_rows = hmap_count(&t->table->rows); + unsigned int max_rows = t->table->schema->max_rows; + + if (n_rows > max_rows) { + return ovsdb_error("constraint violation", + "transaction causes \"%s\" table to contain " + "%zu rows, greater than the schema-defined " + "limit of %u row(s)", + t->table->schema->name, n_rows, max_rows); + } + } + + return NULL; +} + struct ovsdb_error * ovsdb_txn_commit(struct ovsdb_txn *txn, bool durable) { struct ovsdb_replica *replica; struct ovsdb_error *error; + /* Figure out what actually changed, and abort early if the transaction + * was really a no-op. */ + error = for_each_txn_row(txn, determine_changes); + if (error) { + return OVSDB_WRAP_BUG("can't happen", error); + } + if (list_is_empty(&txn->txn_tables)) { + ovsdb_txn_abort(txn); + return NULL; + } + + /* Update reference counts and check referential integrity. */ error = update_ref_counts(txn); if (error) { ovsdb_txn_abort(txn); return error; } - LIST_FOR_EACH (replica, struct ovsdb_replica, node, &txn->db->replicas) { + /* Delete unreferenced, non-root rows. */ + error = for_each_txn_row(txn, collect_garbage); + if (error) { + ovsdb_txn_abort(txn); + return OVSDB_WRAP_BUG("can't happen", error); + } + + /* Check maximum rows table constraints. */ + error = check_max_rows(txn); + if (error) { + ovsdb_txn_abort(txn); + return error; + } + + /* Check reference counts and remove bad reference for "weak" referential + * integrity. */ + error = for_each_txn_row(txn, assess_weak_refs); + if (error) { + ovsdb_txn_abort(txn); + return error; + } + + /* Send the commit to each replica. */ + LIST_FOR_EACH (replica, node, &txn->db->replicas) { error = (replica->class->commit)(replica, txn, durable); if (error) { /* We don't support two-phase commit so only the first replica is @@ -306,9 +648,9 @@ ovsdb_txn_for_each_change(const struct ovsdb_txn *txn, struct ovsdb_txn_table *t; struct ovsdb_txn_row *r; - LIST_FOR_EACH (t, struct ovsdb_txn_table, node, &txn->txn_tables) { - HMAP_FOR_EACH (r, struct ovsdb_txn_row, hmap_node, &t->txn_rows) { - if (!cb(r->old, r->new, aux)) { + LIST_FOR_EACH (t, node, &txn->txn_tables) { + HMAP_FOR_EACH (r, hmap_node, &t->txn_rows) { + if ((r->old || r->new) && !cb(r->old, r->new, r->changed, aux)) { break; } } @@ -334,11 +676,16 @@ static struct ovsdb_txn_row * ovsdb_txn_row_create(struct ovsdb_txn *txn, struct ovsdb_table *table, const struct ovsdb_row *old_, struct ovsdb_row *new) { + const struct ovsdb_row *row = old_ ? old_ : new; struct ovsdb_row *old = (struct ovsdb_row *) old_; + size_t n_columns = shash_count(&table->schema->columns); struct ovsdb_txn_table *txn_table; struct ovsdb_txn_row *txn_row; - txn_row = xmalloc(sizeof *txn_row); + txn_row = xzalloc(offsetof(struct ovsdb_txn_row, changed) + + bitmap_n_bytes(n_columns)); + txn_row->uuid = *ovsdb_row_get_uuid(row); + txn_row->table = row->table; txn_row->old = old; txn_row->new = new; txn_row->n_refs = old ? old->n_refs : 0; @@ -435,8 +782,7 @@ ovsdb_txn_get_comment(const struct ovsdb_txn *txn) static void ovsdb_txn_row_prefree(struct ovsdb_txn_row *txn_row) { - struct ovsdb_row *row = txn_row->old ? txn_row->old : txn_row->new; - struct ovsdb_txn_table *txn_table = row->table->txn_table; + struct ovsdb_txn_table *txn_table = txn_row->table->txn_table; txn_table->n_processed--; hmap_remove(&txn_table->txn_rows, &txn_row->hmap_node); @@ -469,6 +815,9 @@ ovsdb_txn_table_destroy(struct ovsdb_txn_table *txn_table) * in within the same txn_table. It may *not* delete any txn_tables. As long * as these rules are followed, 'cb' will be called exactly once for each * txn_row in 'txn', even those added by 'cb'. + * + * (Even though 'cb' is not allowed to delete some txn_rows, it can still + * delete any actual row by clearing a txn_row's 'new' member.) */ static struct ovsdb_error * WARN_UNUSED_RESULT for_each_txn_row(struct ovsdb_txn *txn, @@ -483,8 +832,7 @@ for_each_txn_row(struct ovsdb_txn *txn, struct ovsdb_txn_table *t, *next_txn_table; any_work = false; - LIST_FOR_EACH_SAFE (t, next_txn_table, struct ovsdb_txn_table, node, - &txn->txn_tables) { + LIST_FOR_EACH_SAFE (t, next_txn_table, node, &txn->txn_tables) { if (t->serial != serial) { t->serial = serial; t->n_processed = 0; @@ -493,9 +841,7 @@ for_each_txn_row(struct ovsdb_txn *txn, while (t->n_processed < hmap_count(&t->txn_rows)) { struct ovsdb_txn_row *r, *next_txn_row; - HMAP_FOR_EACH_SAFE (r, next_txn_row, - struct ovsdb_txn_row, hmap_node, - &t->txn_rows) { + HMAP_FOR_EACH_SAFE (r, next_txn_row, hmap_node, &t->txn_rows) { if (r->serial != serial) { struct ovsdb_error *error;