mir/ovs
2
0
Fork 0
mirror of https://github.com/openvswitch/ovs synced 2025-08-31 06:15:47 +00:00
Code Issues Releases Wiki Activity
Files
9167cb52fa8708ff524be9fc41a8f5ccdfa0a15d
ovs/ovsdb/monitor.c
Han Zhou 9167cb52fa ovsdb-monitor: Support monitor_cond_since. 
Support the new monitor method monitor_cond_since so that a client
can request monitoring start from a specific point instead of always
from beginning. This will reduce the cost at scenarios when server
is restarted/failed-over but client still has all existing data. In
these scenarios only new changes (and in most cases no change) needed
to be transfered to client. When ovsdb-server restarted, history
transactions are read from disk file; when ovsdb-server failed over,
history transactions exists already in the memory of the new server.

There are situations that the requested transaction may not be found.
For example, a transaction that is too old and has been discarded
from the maintained history list in memory, or the transactions on
disk has been compacted during ovsdb compaction. In those situations
the server fall backs to transfer all data start from begining.

For more details of the protocol change, see
Documentation/ref/ovsdb-server.7.rst.

This change includes both server side and ovsdb-client side changes
with the new protocol. IDLs using this capability will be added in
future patches.

Now the feature takes effect only for cluster mode of ovsdb-server,
because cluster mode is the only mode that supports unique transcation
uuid today. For other modes, the monitor_cond_since always fall back
to transfer all data with found = false. Support for those modes can
be added in the future.

Signed-off-by: Han Zhou <hzhou8@ebay.com>
Signed-off-by: Ben Pfaff <blp@ovn.org>
2019-02-28 10:26:18 -08:00

1804 lines
59 KiB
C
Raw Blame History

/*
* Copyright (c) 2015, 2017 Nicira, Inc.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at:
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
#include <config.h>
#include <errno.h>
#include "bitmap.h"
#include "column.h"
#include "openvswitch/dynamic-string.h"
#include "openvswitch/json.h"
#include "jsonrpc.h"
#include "ovsdb-error.h"
#include "ovsdb-parser.h"
#include "ovsdb.h"
#include "row.h"
#include "condition.h"
#include "simap.h"
#include "hash.h"
#include "table.h"
#include "hash.h"
#include "timeval.h"
#include "transaction.h"
#include "jsonrpc-server.h"
#include "monitor.h"
#include "util.h"
#include "openvswitch/vlog.h"
VLOG_DEFINE_THIS_MODULE(ovsdb_monitor);
static struct hmap ovsdb_monitors = HMAP_INITIALIZER(&ovsdb_monitors);
/* Keep state of session's conditions */
struct ovsdb_monitor_session_condition {
bool conditional; /* True iff every table's condition is true. */
struct shash tables; /* Contains
* "struct ovsdb_monitor_table_condition *"s. */
};
/* Monitored table session's conditions */
struct ovsdb_monitor_table_condition {
const struct ovsdb_table *table;
struct ovsdb_monitor_table *mt;
struct ovsdb_condition old_condition;
struct ovsdb_condition new_condition;
};
/* Backend monitor.
*
* ovsdb_monitor keep track of the ovsdb changes.
*/
/* A collection of tables being monitored. */
struct ovsdb_monitor {
struct ovs_list list_node; /* In struct ovsdb's "monitors" list. */
struct shash tables; /* Holds "struct ovsdb_monitor_table"s. */
struct ovs_list jsonrpc_monitors; /* Contains "jsonrpc_monitor_node"s. */
struct ovsdb *db;
/* Contains "ovsdb_monitor_change_set". Each change set contains changes
* from some start point up to the latest committed transaction. There can
* be different change sets for the same struct ovsdb_monitor because there
* are different clients pending on changes starting from different points.
* The different change sets are maintained as a list. */
struct ovs_list change_sets;
/* The new change set that is to be populated for future transactions. */
struct ovsdb_monitor_change_set *new_change_set;
/* The change set that starts from the first transaction of the DB, which
* is used for populating the initial data for new clients. */
struct ovsdb_monitor_change_set *init_change_set;
struct hmap_node hmap_node; /* Elements within ovsdb_monitors. */
struct hmap json_cache; /* Contains "ovsdb_monitor_json_cache_node"s.*/
};
/* A json object of updates for the ovsdb_monitor_change_set and the given
* monitor version. */
struct ovsdb_monitor_json_cache_node {
struct hmap_node hmap_node; /* Elements in json cache. */
enum ovsdb_monitor_version version;
struct uuid change_set_uuid;
struct json *json; /* Null, or a cloned of json */
};
struct jsonrpc_monitor_node {
struct ovs_list node;
struct ovsdb_jsonrpc_monitor *jsonrpc_monitor;
};
/* A particular column being monitored. */
struct ovsdb_monitor_column {
const struct ovsdb_column *column;
enum ovsdb_monitor_selection select;
bool monitored;
};
/* A row that has changed in a monitored table. */
struct ovsdb_monitor_row {
struct hmap_node hmap_node; /* In ovsdb_monitor_change_set_for_table. */
struct uuid uuid; /* UUID of row that changed. */
struct ovsdb_datum *old; /* Old data, NULL for an inserted row. */
struct ovsdb_datum *new; /* New data, NULL for a deleted row. */
};
/* Contains a set of changes that are not yet flushed to all the jsonrpc
* connections.
*
* 'n_refs' represent the number of jsonrpc connections that depend on this
* change set (have not received updates). Generate the update for the last
* jsonprc connection will also destroy the whole "struct
* ovsdb_monitor_change_set" object.
*/
struct ovsdb_monitor_change_set {
/* Element in change_sets of ovsdb_monitor. */
struct ovs_list list_node;
/* Internally generated uuid that identifies this data structure. */
struct uuid uuid;
/* Contains struct ovsdb_monitor_change_set_for_table. */
struct ovs_list change_set_for_tables;
int n_refs;
/* The previous txn id before this change set's start point. */
struct uuid prev_txn;
};
/* Contains 'struct ovsdb_monitor_row's for rows in a specific table
* of struct ovsdb_monitor_change_set. It can also be searched from
* member 'change_sets' of struct ovsdb_monitor_table. */
struct ovsdb_monitor_change_set_for_table {
/* Element in ovsdb_monitor_tables' change_sets list. */
struct ovs_list list_in_mt;
/* Element in ovsdb_monitor_change_sets' change_set_for_tables list. */
struct ovs_list list_in_change_set;
struct ovsdb_monitor_table *mt;
struct ovsdb_monitor_change_set *mcs;
/* Contains struct ovsdb_monitor_row. */
struct hmap rows;
/* Save the mt->n_columns that is used when creating the changes.
* It can be different from the current mt->n_columns because
* mt->n_columns can be increased when there are condition changes
* from any of the clients sharing the dbmon. */
size_t n_columns;
};
/* A particular table being monitored. */
struct ovsdb_monitor_table {
const struct ovsdb_table *table;
/* This is the union (bitwise-OR) of the 'select' values in all of the
* members of 'columns' below. */
enum ovsdb_monitor_selection select;
/* Columns being monitored. */
struct ovsdb_monitor_column *columns;
size_t n_columns;
size_t n_monitored_columns;
size_t allocated_columns;
/* Columns in ovsdb_monitor_row have different indexes then in
* ovsdb_row. This field maps between column->index to the index in the
* ovsdb_monitor_row. It is used for condition evaluation. */
unsigned int *columns_index_map;
/* Contains 'ovsdb_monitor_change_set_for_table'. */
struct ovs_list change_sets;
};
enum ovsdb_monitor_row_type {
OVSDB_ROW,
OVSDB_MONITOR_ROW
};
typedef struct json *
(*compose_row_update_cb_func)
(const struct ovsdb_monitor_table *mt,
const struct ovsdb_monitor_session_condition * condition,
enum ovsdb_monitor_row_type row_type,
const void *,
bool initial, unsigned long int *changed,
size_t n_columns);
static void ovsdb_monitor_destroy(struct ovsdb_monitor *);
static struct ovsdb_monitor_change_set * ovsdb_monitor_add_change_set(
struct ovsdb_monitor *, bool init_only, const struct uuid *prev_txn);
static struct ovsdb_monitor_change_set * ovsdb_monitor_find_change_set(
const struct ovsdb_monitor *, const struct uuid *prev_txn);
static void ovsdb_monitor_change_set_destroy(
struct ovsdb_monitor_change_set *);
static void ovsdb_monitor_track_new_change_set(struct ovsdb_monitor *);
static uint32_t
json_cache_hash(enum ovsdb_monitor_version version,
struct ovsdb_monitor_change_set *change_set)
{
return hash_uint64_basis(version, uuid_hash(&change_set->uuid));
}
static struct ovsdb_monitor_json_cache_node *
ovsdb_monitor_json_cache_search(const struct ovsdb_monitor *dbmon,
enum ovsdb_monitor_version version,
struct ovsdb_monitor_change_set *change_set)
{
struct ovsdb_monitor_json_cache_node *node;
uint32_t hash = json_cache_hash(version, change_set);
HMAP_FOR_EACH_WITH_HASH(node, hmap_node, hash, &dbmon->json_cache) {
if (uuid_equals(&node->change_set_uuid, &change_set->uuid) &&
node->version == version) {
return node;
}
}
return NULL;
}
static void
ovsdb_monitor_json_cache_insert(struct ovsdb_monitor *dbmon,
enum ovsdb_monitor_version version,
struct ovsdb_monitor_change_set *change_set,
struct json *json)
{
struct ovsdb_monitor_json_cache_node *node;
uint32_t hash = json_cache_hash(version, change_set);
node = xmalloc(sizeof *node);
node->version = version;
node->change_set_uuid = change_set->uuid;
node->json = json ? json_clone(json) : NULL;
hmap_insert(&dbmon->json_cache, &node->hmap_node, hash);
}
static void
ovsdb_monitor_json_cache_flush(struct ovsdb_monitor *dbmon)
{
struct ovsdb_monitor_json_cache_node *node;
HMAP_FOR_EACH_POP(node, hmap_node, &dbmon->json_cache) {
json_destroy(node->json);
free(node);
}
}
/* Free all versions of json cache for a given change_set.*/
static void
ovsdb_monitor_json_cache_destroy(struct ovsdb_monitor *dbmon,
struct ovsdb_monitor_change_set *change_set)
{
enum ovsdb_monitor_version v;
for (v = OVSDB_MONITOR_V1; v < OVSDB_MONITOR_VERSION_MAX; v++) {
struct ovsdb_monitor_json_cache_node *node
= ovsdb_monitor_json_cache_search(dbmon, v, change_set);
if (node) {
hmap_remove(&dbmon->json_cache, &node->hmap_node);
json_destroy(node->json);
free(node);
}
}
}
static int
compare_ovsdb_monitor_column(const void *a_, const void *b_)
{
const struct ovsdb_monitor_column *a = a_;
const struct ovsdb_monitor_column *b = b_;
/* put all monitored columns at the begining */
if (a->monitored != b->monitored) {
return a->monitored ? -1 : 1;
}
return a->column < b->column ? -1 : a->column > b->column;
}
/* Finds and returns the ovsdb_monitor_row in 'mt->changes->rows' for the
* given 'uuid', or NULL if there is no such row. */
static struct ovsdb_monitor_row *
ovsdb_monitor_changes_row_find(
const struct ovsdb_monitor_change_set_for_table *changes,
const struct uuid *uuid)
{
struct ovsdb_monitor_row *row;
HMAP_FOR_EACH_WITH_HASH (row, hmap_node, uuid_hash(uuid),
&changes->rows) {
if (uuid_equals(uuid, &row->uuid)) {
return row;
}
}
return NULL;
}
/* Allocates an array of 'n_columns' ovsdb_datums and initializes them as
* copies of the data in 'row' drawn from the columns represented by
* mt->columns[]. Returns the array.
*
* If 'row' is NULL, returns NULL. */
static struct ovsdb_datum *
clone_monitor_row_data(const struct ovsdb_monitor_table *mt,
const struct ovsdb_row *row,
size_t n_columns)
{
struct ovsdb_datum *data;
size_t i;
if (!row) {
return NULL;
}
data = xmalloc(n_columns * sizeof *data);
for (i = 0; i < n_columns; i++) {
const struct ovsdb_column *c = mt->columns[i].column;
const struct ovsdb_datum *src = &row->fields[c->index];
struct ovsdb_datum *dst = &data[i];
const struct ovsdb_type *type = &c->type;
ovsdb_datum_clone(dst, src, type);
}
return data;
}
/* Replaces the n_columns ovsdb_datums in row[] by copies of the data from
* in 'row' drawn from the columns represented by mt->columns[]. */
static void
update_monitor_row_data(const struct ovsdb_monitor_table *mt,
const struct ovsdb_row *row,
struct ovsdb_datum *data,
size_t n_columns)
{
size_t i;
for (i = 0; i < n_columns; i++) {
const struct ovsdb_column *c = mt->columns[i].column;
const struct ovsdb_datum *src = &row->fields[c->index];
struct ovsdb_datum *dst = &data[i];
const struct ovsdb_type *type = &c->type;
if (!ovsdb_datum_equals(src, dst, type)) {
ovsdb_datum_destroy(dst, type);
ovsdb_datum_clone(dst, src, type);
}
}
}
/* Frees all of the n_columns ovsdb_datums in data[], using the types taken
* from mt->columns[], plus 'data' itself. */
static void
free_monitor_row_data(const struct ovsdb_monitor_table *mt,
struct ovsdb_datum *data,
size_t n_columns)
{
if (data) {
size_t i;
for (i = 0; i < n_columns; i++) {
const struct ovsdb_column *c = mt->columns[i].column;
ovsdb_datum_destroy(&data[i], &c->type);
}
free(data);
}
}
/* Frees 'row', which must have been created from 'mt'. */
static void
ovsdb_monitor_row_destroy(const struct ovsdb_monitor_table *mt,
struct ovsdb_monitor_row *row,
size_t n_columns)
{
if (row) {
free_monitor_row_data(mt, row->old, n_columns);
free_monitor_row_data(mt, row->new, n_columns);
free(row);
}
}
static void
ovsdb_monitor_columns_sort(struct ovsdb_monitor *dbmon)
{
int i;
struct shash_node *node;
SHASH_FOR_EACH (node, &dbmon->tables) {
struct ovsdb_monitor_table *mt = node->data;
qsort(mt->columns, mt->n_columns, sizeof *mt->columns,
compare_ovsdb_monitor_column);
for (i = 0; i < mt->n_columns; i++) {
/* re-set index map due to sort */
mt->columns_index_map[mt->columns[i].column->index] = i;
}
}
}
void
ovsdb_monitor_add_jsonrpc_monitor(struct ovsdb_monitor *dbmon,
struct ovsdb_jsonrpc_monitor *jsonrpc_monitor)
{
struct jsonrpc_monitor_node *jm;
jm = xzalloc(sizeof *jm);
jm->jsonrpc_monitor = jsonrpc_monitor;
ovs_list_push_back(&dbmon->jsonrpc_monitors, &jm->node);
}
struct ovsdb_monitor *
ovsdb_monitor_create(struct ovsdb *db,
struct ovsdb_jsonrpc_monitor *jsonrpc_monitor)
{
struct ovsdb_monitor *dbmon;
dbmon = xzalloc(sizeof *dbmon);
ovs_list_push_back(&db->monitors, &dbmon->list_node);
ovs_list_init(&dbmon->jsonrpc_monitors);
dbmon->db = db;
ovs_list_init(&dbmon->change_sets);
shash_init(&dbmon->tables);
hmap_node_nullify(&dbmon->hmap_node);
hmap_init(&dbmon->json_cache);
ovsdb_monitor_add_jsonrpc_monitor(dbmon, jsonrpc_monitor);
return dbmon;
}
void
ovsdb_monitor_add_table(struct ovsdb_monitor *m,
const struct ovsdb_table *table)
{
struct ovsdb_monitor_table *mt;
int i;
size_t n_columns = shash_count(&table->schema->columns);
mt = xzalloc(sizeof *mt);
mt->table = table;
shash_add(&m->tables, table->schema->name, mt);
ovs_list_init(&mt->change_sets);
mt->columns_index_map =
xmalloc(sizeof *mt->columns_index_map * n_columns);
for (i = 0; i < n_columns; i++) {
mt->columns_index_map[i] = -1;
}
}
const char *
ovsdb_monitor_add_column(struct ovsdb_monitor *dbmon,
const struct ovsdb_table *table,
const struct ovsdb_column *column,
enum ovsdb_monitor_selection select,
bool monitored)
{
struct ovsdb_monitor_table *mt;
struct ovsdb_monitor_column *c;
mt = shash_find_data(&dbmon->tables, table->schema->name);
/* Check for column duplication. Return duplicated column name. */
if (mt->columns_index_map[column->index] != -1) {
return column->name;
}
if (mt->n_columns >= mt->allocated_columns) {
mt->columns = x2nrealloc(mt->columns, &mt->allocated_columns,
sizeof *mt->columns);
}
mt->select |= select;
mt->columns_index_map[column->index] = mt->n_columns;
c = &mt->columns[mt->n_columns++];
c->column = column;
c->select = select;
c->monitored = monitored;
if (monitored) {
mt->n_monitored_columns++;
}
return NULL;
}
static void
ovsdb_monitor_condition_add_columns(struct ovsdb_monitor *dbmon,
const struct ovsdb_table *table,
struct ovsdb_condition *condition)
{
size_t n_columns;
int i;
const struct ovsdb_column **columns =
ovsdb_condition_get_columns(condition, &n_columns);
for (i = 0; i < n_columns; i++) {
ovsdb_monitor_add_column(dbmon, table, columns[i],
OJMS_NONE, false);
}
free(columns);
}
/* Bind this session's condition to ovsdb_monitor */
void
ovsdb_monitor_condition_bind(struct ovsdb_monitor *dbmon,
struct ovsdb_monitor_session_condition *cond)
{
struct shash_node *node;
SHASH_FOR_EACH(node, &cond->tables) {
struct ovsdb_monitor_table_condition *mtc = node->data;
struct ovsdb_monitor_table *mt =
shash_find_data(&dbmon->tables, mtc->table->schema->name);
mtc->mt = mt;
ovsdb_monitor_condition_add_columns(dbmon, mtc->table,
&mtc->new_condition);
}
}
bool
ovsdb_monitor_table_exists(struct ovsdb_monitor *m,
const struct ovsdb_table *table)
{
return shash_find_data(&m->tables, table->schema->name);
}
static struct ovsdb_monitor_change_set *
ovsdb_monitor_add_change_set(struct ovsdb_monitor *dbmon,
bool init_only, const struct uuid *prev_txn)
{
struct ovsdb_monitor_change_set *change_set = xzalloc(sizeof *change_set);
change_set->uuid = uuid_random();
ovs_list_push_back(&(dbmon->change_sets), &change_set->list_node);
ovs_list_init(&change_set->change_set_for_tables);
change_set->n_refs = 1;
change_set->prev_txn = prev_txn ? *prev_txn : UUID_ZERO;
struct shash_node *node;
SHASH_FOR_EACH (node, &dbmon->tables) {
struct ovsdb_monitor_table *mt = node->data;
if (!init_only || (mt->select & OJMS_INITIAL)) {
struct ovsdb_monitor_change_set_for_table *mcst =
xzalloc(sizeof *mcst);
mcst->mt = mt;
mcst->n_columns = mt->n_columns;
mcst->mcs = change_set;
hmap_init(&mcst->rows);
ovs_list_push_back(&mt->change_sets, &mcst->list_in_mt);
ovs_list_push_back(&change_set->change_set_for_tables,
&mcst->list_in_change_set);
}
}
return change_set;
};
static struct ovsdb_monitor_change_set *
ovsdb_monitor_find_change_set(const struct ovsdb_monitor *dbmon,
const struct uuid *prev_txn)
{
struct ovsdb_monitor_change_set *cs;
LIST_FOR_EACH (cs, list_node, &dbmon->change_sets) {
if (uuid_equals(&cs->prev_txn, prev_txn)) {
/* Check n_columns for each table in dbmon, in case it is changed
* after the change set is populated. */
bool n_col_is_equal = true;
struct ovsdb_monitor_change_set_for_table *mcst;
LIST_FOR_EACH (mcst, list_in_change_set,
&cs->change_set_for_tables) {
struct ovsdb_monitor_table *mt = mcst->mt;
if (mt->n_columns != mcst->n_columns) {
n_col_is_equal = false;
break;
}
}
if (n_col_is_equal) {
return cs;
}
}
}
return NULL;
}
static void
ovsdb_monitor_untrack_change_set(struct ovsdb_monitor *dbmon,
struct ovsdb_monitor_change_set *mcs)
{
ovs_assert(mcs);
if (--mcs->n_refs == 0) {
if (mcs == dbmon->init_change_set) {
dbmon->init_change_set = NULL;
} else if (mcs == dbmon->new_change_set) {
dbmon->new_change_set = NULL;
}
ovsdb_monitor_json_cache_destroy(dbmon, mcs);
ovsdb_monitor_change_set_destroy(mcs);
}
}
static void
ovsdb_monitor_track_new_change_set(struct ovsdb_monitor *dbmon)
{
struct ovsdb_monitor_change_set *change_set = dbmon->new_change_set;
if (change_set) {
change_set->n_refs++;
} else {
change_set = ovsdb_monitor_add_change_set(dbmon, false,
ovsdb_monitor_get_last_txnid(dbmon));
dbmon->new_change_set = change_set;
}
}
static void
ovsdb_monitor_change_set_destroy(struct ovsdb_monitor_change_set *mcs)
{
ovs_list_remove(&mcs->list_node);
struct ovsdb_monitor_change_set_for_table *mcst, *next_mcst;
LIST_FOR_EACH_SAFE (mcst, next_mcst, list_in_change_set,
&mcs->change_set_for_tables) {
ovs_list_remove(&mcst->list_in_change_set);
ovs_list_remove(&mcst->list_in_mt);
struct ovsdb_monitor_row *row, *next;
HMAP_FOR_EACH_SAFE (row, next, hmap_node, &mcst->rows) {
hmap_remove(&mcst->rows, &row->hmap_node);
ovsdb_monitor_row_destroy(mcst->mt, row, mcst->n_columns);
}
hmap_destroy(&mcst->rows);
free(mcst);
}
free(mcs);
}
static enum ovsdb_monitor_selection
ovsdb_monitor_row_update_type(bool initial, const bool old, const bool new)
{
return initial ? OJMS_INITIAL
: !old ? OJMS_INSERT
: !new ? OJMS_DELETE
: OJMS_MODIFY;
}
/* Set conditional monitoring mode only if we have non-empty condition in one
* of the tables at least */
static inline void
ovsdb_monitor_session_condition_set_mode(
struct ovsdb_monitor_session_condition *cond)
{
struct shash_node *node;
SHASH_FOR_EACH (node, &cond->tables) {
struct ovsdb_monitor_table_condition *mtc = node->data;
if (!ovsdb_condition_is_true(&mtc->new_condition)) {
cond->conditional = true;
return;
}
}
cond->conditional = false;
}
/* Returnes an empty allocated session's condition state holder */
struct ovsdb_monitor_session_condition *
ovsdb_monitor_session_condition_create(void)
{
struct ovsdb_monitor_session_condition *condition =
xzalloc(sizeof *condition);
condition->conditional = false;
shash_init(&condition->tables);
return condition;
}
void
ovsdb_monitor_session_condition_destroy(
struct ovsdb_monitor_session_condition *condition)
{
struct shash_node *node, *next;
if (!condition) {
return;
}
SHASH_FOR_EACH_SAFE (node, next, &condition->tables) {
struct ovsdb_monitor_table_condition *mtc = node->data;
ovsdb_condition_destroy(&mtc->new_condition);
ovsdb_condition_destroy(&mtc->old_condition);
shash_delete(&condition->tables, node);
free(mtc);
}
shash_destroy(&condition->tables);
free(condition);
}
struct ovsdb_error *
ovsdb_monitor_table_condition_create(
struct ovsdb_monitor_session_condition *condition,
const struct ovsdb_table *table,
const struct json *json_cnd)
{
struct ovsdb_monitor_table_condition *mtc;
struct ovsdb_error *error;
mtc = xzalloc(sizeof *mtc);
mtc->table = table;
ovsdb_condition_init(&mtc->old_condition);
ovsdb_condition_init(&mtc->new_condition);
if (json_cnd) {
error = ovsdb_condition_from_json(table->schema,
json_cnd,
NULL,
&mtc->old_condition);
if (error) {
free(mtc);
return error;
}
}
shash_add(&condition->tables, table->schema->name, mtc);
/* On session startup old == new condition */
ovsdb_condition_clone(&mtc->new_condition, &mtc->old_condition);
ovsdb_monitor_session_condition_set_mode(condition);
return NULL;
}
static bool
ovsdb_monitor_get_table_conditions(
const struct ovsdb_monitor_table *mt,
const struct ovsdb_monitor_session_condition *condition,
struct ovsdb_condition **old_condition,
struct ovsdb_condition **new_condition)
{
if (!condition) {
return false;
}
struct ovsdb_monitor_table_condition *mtc =
shash_find_data(&condition->tables, mt->table->schema->name);
if (!mtc) {
return false;
}
*old_condition = &mtc->old_condition;
*new_condition = &mtc->new_condition;
return true;
}
struct ovsdb_error *
ovsdb_monitor_table_condition_update(
struct ovsdb_monitor *dbmon,
struct ovsdb_monitor_session_condition *condition,
const struct ovsdb_table *table,
const struct json *cond_json)
{
if (!condition) {
return NULL;
}
struct ovsdb_monitor_table_condition *mtc =
shash_find_data(&condition->tables, table->schema->name);
struct ovsdb_error *error;
struct ovsdb_condition cond = OVSDB_CONDITION_INITIALIZER(&cond);
error = ovsdb_condition_from_json(table->schema, cond_json,
NULL, &cond);
if (error) {
return error;
}
ovsdb_condition_destroy(&mtc->new_condition);
ovsdb_condition_clone(&mtc->new_condition, &cond);
ovsdb_condition_destroy(&cond);
ovsdb_monitor_condition_add_columns(dbmon,
table,
&mtc->new_condition);
return NULL;
}
static void
ovsdb_monitor_table_condition_updated(struct ovsdb_monitor_table *mt,
struct ovsdb_monitor_session_condition *condition)
{
struct ovsdb_monitor_table_condition *mtc =
shash_find_data(&condition->tables, mt->table->schema->name);
if (mtc) {
/* If conditional monitoring - set old condition to new condition */
if (ovsdb_condition_cmp_3way(&mtc->old_condition,
&mtc->new_condition)) {
ovsdb_condition_destroy(&mtc->old_condition);
ovsdb_condition_clone(&mtc->old_condition, &mtc->new_condition);
ovsdb_monitor_session_condition_set_mode(condition);
}
}
}
static enum ovsdb_monitor_selection
ovsdb_monitor_row_update_type_condition(
const struct ovsdb_monitor_table *mt,
const struct ovsdb_monitor_session_condition *condition,
bool initial,
enum ovsdb_monitor_row_type row_type,
const struct ovsdb_datum *old,
const struct ovsdb_datum *new)
{
struct ovsdb_condition *old_condition, *new_condition;
enum ovsdb_monitor_selection type =
ovsdb_monitor_row_update_type(initial, old, new);
if (ovsdb_monitor_get_table_conditions(mt,
condition,
&old_condition,
&new_condition)) {
bool old_cond = !old ? false
: ovsdb_condition_empty_or_match_any(old,
old_condition,
row_type == OVSDB_MONITOR_ROW ?
mt->columns_index_map :
NULL);
bool new_cond = !new ? false
: ovsdb_condition_empty_or_match_any(new,
new_condition,
row_type == OVSDB_MONITOR_ROW ?
mt->columns_index_map :
NULL);
if (!old_cond && !new_cond) {
type = OJMS_NONE;
}
switch (type) {
case OJMS_INITIAL:
case OJMS_INSERT:
if (!new_cond) {
type = OJMS_NONE;
}
break;
case OJMS_MODIFY:
type = !old_cond ? OJMS_INSERT : !new_cond
? OJMS_DELETE : OJMS_MODIFY;
break;
case OJMS_DELETE:
if (!old_cond) {
type = OJMS_NONE;
}
break;
case OJMS_NONE:
break;
}
}
return type;
}
static bool
ovsdb_monitor_row_skip_update(const struct ovsdb_monitor_table *mt,
enum ovsdb_monitor_row_type row_type,
const struct ovsdb_datum *old,
const struct ovsdb_datum *new,
enum ovsdb_monitor_selection type,
unsigned long int *changed,
size_t n_columns)
{
if (!(mt->select & type)) {
return true;
}
if (type == OJMS_MODIFY) {
size_t i, n_changes;
n_changes = 0;
memset(changed, 0, bitmap_n_bytes(n_columns));
for (i = 0; i < n_columns; i++) {
const struct ovsdb_column *c = mt->columns[i].column;
size_t index = row_type == OVSDB_ROW ? c->index : i;
if (!ovsdb_datum_equals(&old[index], &new[index], &c->type)) {
bitmap_set1(changed, i);
n_changes++;
}
}
if (!n_changes) {
/* No actual changes: presumably a row changed and then
* changed back later. */
return true;
}
}
return false;
}
/* Returns JSON for a <row-update> (as described in RFC 7047) for 'row' within
* 'mt', or NULL if no row update should be sent.
*
* The caller should specify 'initial' as true if the returned JSON is going to
* be used as part of the initial reply to a "monitor" request, false if it is
* going to be used as part of an "update" notification.
*
* 'changed' must be a scratch buffer for internal use that is at least
* bitmap_n_bytes(n_columns) bytes long. */
static struct json *
ovsdb_monitor_compose_row_update(
const struct ovsdb_monitor_table *mt,
const struct ovsdb_monitor_session_condition *condition OVS_UNUSED,
enum ovsdb_monitor_row_type row_type OVS_UNUSED,
const void *_row,
bool initial, unsigned long int *changed,
size_t n_columns OVS_UNUSED)
{
const struct ovsdb_monitor_row *row = _row;
enum ovsdb_monitor_selection type;
struct json *old_json, *new_json;
struct json *row_json;
size_t i;
ovs_assert(row_type == OVSDB_MONITOR_ROW);
type = ovsdb_monitor_row_update_type(initial, row->old, row->new);
if (ovsdb_monitor_row_skip_update(mt, row_type, row->old,
row->new, type, changed,
mt->n_columns)) {
return NULL;
}
row_json = json_object_create();
old_json = new_json = NULL;
if (type & (OJMS_DELETE | OJMS_MODIFY)) {
old_json = json_object_create();
json_object_put(row_json, "old", old_json);
}
if (type & (OJMS_INITIAL | OJMS_INSERT | OJMS_MODIFY)) {
new_json = json_object_create();
json_object_put(row_json, "new", new_json);
}
for (i = 0; i < mt->n_monitored_columns; i++) {
const struct ovsdb_monitor_column *c = &mt->columns[i];
if (!c->monitored || !(type & c->select)) {
/* We don't care about this type of change for this
* particular column (but we will care about it for some
* other column). */
continue;
}
if ((type == OJMS_MODIFY && bitmap_is_set(changed, i))
|| type == OJMS_DELETE) {
json_object_put(old_json, c->column->name,
ovsdb_datum_to_json(&row->old[i],
&c->column->type));
}
if (type & (OJMS_INITIAL | OJMS_INSERT | OJMS_MODIFY)) {
json_object_put(new_json, c->column->name,
ovsdb_datum_to_json(&row->new[i],
&c->column->type));
}
}
return row_json;
}
/* Returns JSON for a <row-update2> (as described in ovsdb-server(1) mapage)
* for 'row' within * 'mt', or NULL if no row update should be sent.
*
* The caller should specify 'initial' as true if the returned JSON is
* going to be used as part of the initial reply to a "monitor_cond" request,
* false if it is going to be used as part of an "update2" notification.
*
* 'changed' must be a scratch buffer for internal use that is at least
* bitmap_n_bytes(n_columns) bytes long. */
static struct json *
ovsdb_monitor_compose_row_update2(
const struct ovsdb_monitor_table *mt,
const struct ovsdb_monitor_session_condition *condition,
enum ovsdb_monitor_row_type row_type,
const void *_row,
bool initial, unsigned long int *changed,
size_t n_columns)
{
enum ovsdb_monitor_selection type;
struct json *row_update2, *diff_json;
const struct ovsdb_datum *old, *new;
size_t i;
if (row_type == OVSDB_MONITOR_ROW) {
old = ((const struct ovsdb_monitor_row *)_row)->old;;
new = ((const struct ovsdb_monitor_row *)_row)->new;
} else {
old = new = ((const struct ovsdb_row *)_row)->fields;
}
type = ovsdb_monitor_row_update_type_condition(mt, condition, initial,
row_type, old, new);
if (ovsdb_monitor_row_skip_update(mt, row_type, old, new, type, changed,
n_columns)) {
return NULL;
}
row_update2 = json_object_create();
if (type == OJMS_DELETE) {
json_object_put(row_update2, "delete", json_null_create());
} else {
diff_json = json_object_create();
const char *op;
for (i = 0; i < mt->n_monitored_columns; i++) {
const struct ovsdb_monitor_column *c = &mt->columns[i];
size_t index = row_type == OVSDB_ROW ? c->column->index : i;
if (!c->monitored || !(type & c->select)) {
/* We don't care about this type of change for this
* particular column (but we will care about it for some
* other column). */
continue;
}
if (type == OJMS_MODIFY) {
struct ovsdb_datum diff;
if (!bitmap_is_set(changed, i)) {
continue;
}
ovsdb_datum_diff(&diff ,&old[index], &new[index],
&c->column->type);
json_object_put(diff_json, c->column->name,
ovsdb_datum_to_json(&diff, &c->column->type));
ovsdb_datum_destroy(&diff, &c->column->type);
} else {
if (!ovsdb_datum_is_default(&new[index], &c->column->type)) {
json_object_put(diff_json, c->column->name,
ovsdb_datum_to_json(&new[index],
&c->column->type));
}
}
}
op = type == OJMS_INITIAL ? "initial"
: type == OJMS_MODIFY ? "modify" : "insert";
json_object_put(row_update2, op, diff_json);
}
return row_update2;
}
static size_t
ovsdb_monitor_max_columns(struct ovsdb_monitor *dbmon)
{
struct shash_node *node;
size_t max_columns = 0;
SHASH_FOR_EACH (node, &dbmon->tables) {
struct ovsdb_monitor_table *mt = node->data;
max_columns = MAX(max_columns, mt->n_columns);
}
return max_columns;
}
static void
ovsdb_monitor_add_json_row(struct json **json, const char *table_name,
struct json **table_json, struct json *row_json,
const struct uuid *row_uuid)
{
char uuid[UUID_LEN + 1];
/* Create JSON object for transaction overall. */
if (!*json) {
*json = json_object_create();
}
/* Create JSON object for transaction on this table. */
if (!*table_json) {
*table_json = json_object_create();
json_object_put(*json, table_name, *table_json);
}
/* Add JSON row to JSON table. */
snprintf(uuid, sizeof uuid, UUID_FMT, UUID_ARGS(row_uuid));
json_object_put(*table_json, uuid, row_json);
}
/* Constructs and returns JSON for a <table-updates> object (as described in
* RFC 7047) for all the outstanding changes within 'monitor', starting from
* 'transaction'. */
static struct json*
ovsdb_monitor_compose_update(
struct ovsdb_monitor *dbmon,
bool initial, struct ovsdb_monitor_change_set *mcs,
const struct ovsdb_monitor_session_condition *condition,
compose_row_update_cb_func row_update)
{
struct json *json;
size_t max_columns = ovsdb_monitor_max_columns(dbmon);
unsigned long int *changed = xmalloc(bitmap_n_bytes(max_columns));
json = NULL;
struct ovsdb_monitor_change_set_for_table *mcst;
LIST_FOR_EACH (mcst, list_in_change_set, &mcs->change_set_for_tables) {
struct ovsdb_monitor_row *row, *next;
struct json *table_json = NULL;
struct ovsdb_monitor_table *mt = mcst->mt;
HMAP_FOR_EACH_SAFE (row, next, hmap_node, &mcst->rows) {
struct json *row_json;
row_json = (*row_update)(mt, condition, OVSDB_MONITOR_ROW, row,
initial, changed, mcst->n_columns);
if (row_json) {
ovsdb_monitor_add_json_row(&json, mt->table->schema->name,
&table_json, row_json,
&row->uuid);
}
}
}
free(changed);
return json;
}
static struct json*
ovsdb_monitor_compose_cond_change_update(
struct ovsdb_monitor *dbmon,
struct ovsdb_monitor_session_condition *condition)
{
struct shash_node *node;
struct json *json = NULL;
size_t max_columns = ovsdb_monitor_max_columns(dbmon);
unsigned long int *changed = xmalloc(bitmap_n_bytes(max_columns));
SHASH_FOR_EACH (node, &dbmon->tables) {
struct ovsdb_monitor_table *mt = node->data;
struct ovsdb_row *row;
struct json *table_json = NULL;
struct ovsdb_condition *old_condition, *new_condition;
if (!ovsdb_monitor_get_table_conditions(mt,
condition,
&old_condition,
&new_condition) ||
!ovsdb_condition_cmp_3way(old_condition, new_condition)) {
/* Nothing to update on this table */
continue;
}
/* Iterate over all rows in table */
HMAP_FOR_EACH (row, hmap_node, &mt->table->rows) {
struct json *row_json;
row_json = ovsdb_monitor_compose_row_update2(mt, condition,
OVSDB_ROW, row,
false, changed,
mt->n_columns);
if (row_json) {
ovsdb_monitor_add_json_row(&json, mt->table->schema->name,
&table_json, row_json,
ovsdb_row_get_uuid(row));
}
}
ovsdb_monitor_table_condition_updated(mt, condition);
}
free(changed);
return json;
}
/* Returns JSON for a <table-updates> object (as described in RFC 7047)
* for all the outstanding changes in dbmon that are tracked by the change set
* *p_mcs.
*
* If cond_updated is true all rows in the db that match conditions will be
* sent.
*
* The caller should specify 'initial' as true if the returned JSON is going to
* be used as part of the initial reply to a "monitor" request, false if it is
* going to be used as part of an "update" notification. */
struct json *
ovsdb_monitor_get_update(
struct ovsdb_monitor *dbmon,
bool initial, bool cond_updated,
struct ovsdb_monitor_session_condition *condition,
enum ovsdb_monitor_version version,
struct ovsdb_monitor_change_set **p_mcs)
{
struct ovsdb_monitor_json_cache_node *cache_node = NULL;
struct json *json;
struct ovsdb_monitor_change_set *mcs = *p_mcs;
ovs_assert(cond_updated ? mcs == dbmon->new_change_set : true);
/* Return a clone of cached json if one exists. Otherwise,
* generate a new one and add it to the cache. */
if (!condition || (!condition->conditional && !cond_updated)) {
cache_node = ovsdb_monitor_json_cache_search(dbmon, version,
mcs);
}
if (cache_node) {
json = cache_node->json ? json_clone(cache_node->json) : NULL;
} else {
if (version == OVSDB_MONITOR_V1) {
json =
ovsdb_monitor_compose_update(dbmon, initial, mcs,
condition,
ovsdb_monitor_compose_row_update);
} else {
ovs_assert(version == OVSDB_MONITOR_V2 ||
version == OVSDB_MONITOR_V3);
if (!cond_updated) {
json = ovsdb_monitor_compose_update(dbmon, initial, mcs,
condition,
ovsdb_monitor_compose_row_update2);
if (!condition || !condition->conditional) {
ovsdb_monitor_json_cache_insert(dbmon, version, mcs,
json);
}
} else {
/* Compose update on whole db due to condition update.
Session must be flushed (change list is empty)*/
json =
ovsdb_monitor_compose_cond_change_update(dbmon, condition);
}
}
}
/* Maintain tracking change set. */
ovsdb_monitor_untrack_change_set(dbmon, mcs);
ovsdb_monitor_track_new_change_set(dbmon);
*p_mcs = dbmon->new_change_set;
return json;
}
bool
ovsdb_monitor_needs_flush(struct ovsdb_monitor *dbmon,
struct ovsdb_monitor_change_set *change_set)
{
ovs_assert(change_set);
return (change_set != dbmon->new_change_set);
}
void
ovsdb_monitor_table_add_select(struct ovsdb_monitor *dbmon,
const struct ovsdb_table *table,
enum ovsdb_monitor_selection select)
{
struct ovsdb_monitor_table * mt;
mt = shash_find_data(&dbmon->tables, table->schema->name);
mt->select |= select;
}
/*
* If a row's change type (insert, delete or modify) matches that of
* the monitor, they should be sent to the monitor's clients as updates.
* Of cause, the monitor should also internally update with this change.
*
* When a change type does not require client side update, the monitor
* may still need to keep track of certain changes in order to generate
* correct future updates. For example, the monitor internal state should
* be updated whenever a new row is inserted, in order to generate the
* correct initial state, regardless if a insert change type is being
* monitored.
*
* On the other hand, if a transaction only contains changes to columns
* that are not monitored, this transaction can be safely ignored by the
* monitor.
*
* Thus, the order of the declaration is important:
* 'OVSDB_CHANGES_REQUIRE_EXTERNAL_UPDATE' always implies
* 'OVSDB_CHANGES_REQUIRE_INTERNAL_UPDATE', but not vice versa. */
enum ovsdb_monitor_changes_efficacy {
OVSDB_CHANGES_NO_EFFECT, /* Monitor does not care about this
change. */
OVSDB_CHANGES_REQUIRE_INTERNAL_UPDATE, /* Monitor internal updates. */
OVSDB_CHANGES_REQUIRE_EXTERNAL_UPDATE, /* Client needs to be updated. */
};
struct ovsdb_monitor_aux {
const struct ovsdb_monitor *monitor;
struct ovsdb_monitor_table *mt;
enum ovsdb_monitor_changes_efficacy efficacy;
};
static void
ovsdb_monitor_init_aux(struct ovsdb_monitor_aux *aux,
const struct ovsdb_monitor *m)
{
aux->monitor = m;
aux->mt = NULL;
aux->efficacy = OVSDB_CHANGES_NO_EFFECT;
}
static void
ovsdb_monitor_changes_update(const struct ovsdb_row *old,
const struct ovsdb_row *new,
const struct ovsdb_monitor_table *mt,
struct ovsdb_monitor_change_set_for_table *mcst)
{
const struct uuid *uuid = ovsdb_row_get_uuid(new ? new : old);
struct ovsdb_monitor_row *change;
change = ovsdb_monitor_changes_row_find(mcst, uuid);
if (!change) {
change = xzalloc(sizeof *change);
hmap_insert(&mcst->rows, &change->hmap_node, uuid_hash(uuid));
change->uuid = *uuid;
change->old = clone_monitor_row_data(mt, old, mcst->n_columns);
change->new = clone_monitor_row_data(mt, new, mcst->n_columns);
} else {
if (new) {
if (!change->new) {
/* Reinsert the row that was just deleted.
*
* This path won't be hit without replication. Whenever OVSDB
* server inserts a new row, It always generates a new UUID
* that is different from the row just deleted.
*
* With replication, this path can be hit in a corner
* case when two OVSDB servers are set up to replicate
* each other. Not that is a useful set up, but can
* happen in practice.
*
* An example of how this path can be hit is documented below.
* The details is not as important to the correctness of the
* logic, but added here to convince ourselves that this path
* can be hit.
*
* Imagine two OVSDB servers that replicates from each
* other. For each replication session, there is a
* corresponding monitor at the other end of the replication
* JSONRPC connection.
*
* The events can lead to a back to back deletion and
* insertion operation of the same row for the monitor of
* the first server are:
*
* 1. A row is inserted in the first OVSDB server.
* 2. The row is then replicated to the remote OVSDB server.
* 3. The row is now deleted by the local OVSDB server. This
* deletion operation is replicated to the local monitor
* of the OVSDB server.
* 4. The monitor now receives the same row, as an insertion,
* from the replication server. Because of
* replication, the row carries the same UUID as the row
* just deleted.
*/
change->new = clone_monitor_row_data(mt, new, mcst->n_columns);
} else {
update_monitor_row_data(mt, new, change->new, mcst->n_columns);
}
} else {
free_monitor_row_data(mt, change->new, mcst->n_columns);
change->new = NULL;
if (!change->old) {
/* This row was added then deleted. Forget about it. */
hmap_remove(&mcst->rows, &change->hmap_node);
free(change);
}
}
}
}
static bool
ovsdb_monitor_columns_changed(const struct ovsdb_monitor_table *mt,
const unsigned long int *changed)
{
size_t i;
for (i = 0; i < mt->n_columns; i++) {
size_t column_index = mt->columns[i].column->index;
if (bitmap_is_set(changed, column_index)) {
return true;
}
}
return false;
}
/* Return the efficacy of a row's change to a monitor table.
*
* Please see the block comment above 'ovsdb_monitor_changes_efficacy'
* definition form more information. */
static enum ovsdb_monitor_changes_efficacy
ovsdb_monitor_changes_classify(enum ovsdb_monitor_selection type,
const struct ovsdb_monitor_table *mt,
const unsigned long int *changed)
{
if (type == OJMS_MODIFY &&
!ovsdb_monitor_columns_changed(mt, changed)) {
return OVSDB_CHANGES_NO_EFFECT;
}
if (type == OJMS_MODIFY) {
/* Condition might turn a modify operation to insert or delete */
type |= OJMS_INSERT | OJMS_DELETE;
}
return (mt->select & type)
? OVSDB_CHANGES_REQUIRE_EXTERNAL_UPDATE
: OVSDB_CHANGES_REQUIRE_INTERNAL_UPDATE;
}
static bool
ovsdb_monitor_change_cb(const struct ovsdb_row *old,
const struct ovsdb_row *new,
const unsigned long int *changed,
void *aux_)
{
struct ovsdb_monitor_aux *aux = aux_;
const struct ovsdb_monitor *m = aux->monitor;
struct ovsdb_table *table = new ? new->table : old->table;
struct ovsdb_monitor_table *mt;
struct ovsdb_monitor_change_set_for_table *mcst;
if (!aux->mt || table != aux->mt->table) {
aux->mt = shash_find_data(&m->tables, table->schema->name);
if (!aux->mt) {
/* We don't care about rows in this table at all. Tell the caller
* to skip it. */
return false;
}
}
mt = aux->mt;
enum ovsdb_monitor_selection type =
ovsdb_monitor_row_update_type(false, old, new);
enum ovsdb_monitor_changes_efficacy efficacy =
ovsdb_monitor_changes_classify(type, mt, changed);
if (efficacy > OVSDB_CHANGES_NO_EFFECT) {
LIST_FOR_EACH (mcst, list_in_mt, &mt->change_sets) {
ovsdb_monitor_changes_update(old, new, mt, mcst);
}
}
if (aux->efficacy < efficacy) {
aux->efficacy = efficacy;
}
return true;
}
void
ovsdb_monitor_get_initial(struct ovsdb_monitor *dbmon,
struct ovsdb_monitor_change_set **p_mcs)
{
if (!dbmon->init_change_set) {
struct ovsdb_monitor_change_set *change_set =
ovsdb_monitor_add_change_set(dbmon, true, NULL);
dbmon->init_change_set = change_set;
struct ovsdb_monitor_change_set_for_table *mcst;
LIST_FOR_EACH (mcst, list_in_change_set,
&change_set->change_set_for_tables) {
if (mcst->mt->select & OJMS_INITIAL) {
struct ovsdb_row *row;
HMAP_FOR_EACH (row, hmap_node, &mcst->mt->table->rows) {
ovsdb_monitor_changes_update(NULL, row, mcst->mt, mcst);
}
}
}
} else {
dbmon->init_change_set->n_refs++;
}
*p_mcs = dbmon->init_change_set;
}
static bool
ovsdb_monitor_history_change_cb(const struct ovsdb_row *old,
const struct ovsdb_row *new,
const unsigned long int *changed,
void *aux)
{
struct ovsdb_monitor_change_set *change_set = aux;
struct ovsdb_table *table = new ? new->table : old->table;
struct ovsdb_monitor_change_set_for_table *mcst;
enum ovsdb_monitor_selection type =
ovsdb_monitor_row_update_type(false, old, new);
LIST_FOR_EACH (mcst, list_in_change_set,
&change_set->change_set_for_tables) {
if (mcst->mt->table == table) {
enum ovsdb_monitor_changes_efficacy efficacy =
ovsdb_monitor_changes_classify(type, mcst->mt, changed);
if (efficacy > OVSDB_CHANGES_NO_EFFECT) {
ovsdb_monitor_changes_update(old, new, mcst->mt, mcst);
}
return true;
}
}
return false;
}
void
ovsdb_monitor_get_changes_after(const struct uuid *txn_uuid,
struct ovsdb_monitor *dbmon,
struct ovsdb_monitor_change_set **p_mcs)
{
ovs_assert(*p_mcs == NULL);
ovs_assert(!uuid_is_zero(txn_uuid));
struct ovsdb_monitor_change_set *change_set =
ovsdb_monitor_find_change_set(dbmon, txn_uuid);
if (change_set) {
change_set->n_refs++;
*p_mcs = change_set;
return;
}
struct ovsdb_txn_history_node *h_node;
bool found = false;
LIST_FOR_EACH (h_node, node, &dbmon->db->txn_history) {
struct ovsdb_txn *txn = h_node->txn;
if (!found) {
/* find the txn with last_id in history */
if (uuid_equals(ovsdb_txn_get_txnid(txn), txn_uuid)) {
found = true;
change_set = ovsdb_monitor_add_change_set(dbmon, false,
txn_uuid);
}
} else {
/* Already found. Add changes in each follow up transaction to
* the new change_set. */
ovsdb_txn_for_each_change(txn, ovsdb_monitor_history_change_cb,
change_set);
}
}
*p_mcs = change_set;
}
void
ovsdb_monitor_remove_jsonrpc_monitor(struct ovsdb_monitor *dbmon,
struct ovsdb_jsonrpc_monitor *jsonrpc_monitor,
struct ovsdb_monitor_change_set *change_set)
{
struct jsonrpc_monitor_node *jm;
if (ovs_list_is_empty(&dbmon->jsonrpc_monitors)) {
ovsdb_monitor_destroy(dbmon);
return;
}
/* Find and remove the jsonrpc monitor from the list. */
LIST_FOR_EACH(jm, node, &dbmon->jsonrpc_monitors) {
if (jm->jsonrpc_monitor == jsonrpc_monitor) {
/* Release the tracked changes. */
if (change_set) {
ovsdb_monitor_untrack_change_set(dbmon, change_set);
}
ovs_list_remove(&jm->node);
free(jm);
/* Destroy ovsdb monitor if this is the last user. */
if (ovs_list_is_empty(&dbmon->jsonrpc_monitors)) {
ovsdb_monitor_destroy(dbmon);
}
return;
};
}
/* Should never reach here. jsonrpc_monitor should be on the list. */
OVS_NOT_REACHED();
}
static bool
ovsdb_monitor_table_equal(const struct ovsdb_monitor_table *a,
const struct ovsdb_monitor_table *b)
{
size_t i;
ovs_assert(b->n_columns == b->n_monitored_columns);
if ((a->table != b->table) ||
(a->select != b->select) ||
(a->n_monitored_columns != b->n_monitored_columns)) {
return false;
}
/* Compare only monitored columns that must be sorted already */
for (i = 0; i < a->n_monitored_columns; i++) {
if ((a->columns[i].column != b->columns[i].column) ||
(a->columns[i].select != b->columns[i].select)) {
return false;
}
}
return true;
}
static bool
ovsdb_monitor_equal(const struct ovsdb_monitor *a,
const struct ovsdb_monitor *b)
{
struct shash_node *node;
if (shash_count(&a->tables) != shash_count(&b->tables)) {
return false;
}
SHASH_FOR_EACH(node, &a->tables) {
const struct ovsdb_monitor_table *mta = node->data;
const struct ovsdb_monitor_table *mtb;
mtb = shash_find_data(&b->tables, node->name);
if (!mtb) {
return false;
}
if (!ovsdb_monitor_table_equal(mta, mtb)) {
return false;
}
}
return true;
}
static size_t
ovsdb_monitor_hash(const struct ovsdb_monitor *dbmon, size_t basis)
{
const struct shash_node **nodes;
size_t i, j, n;
nodes = shash_sort(&dbmon->tables);
n = shash_count(&dbmon->tables);
for (i = 0; i < n; i++) {
struct ovsdb_monitor_table *mt = nodes[i]->data;
basis = hash_pointer(mt->table, basis);
basis = hash_3words(mt->select, mt->n_columns, basis);
for (j = 0; j < mt->n_columns; j++) {
basis = hash_pointer(mt->columns[j].column, basis);
basis = hash_2words(mt->columns[j].select, basis);
}
}
free(nodes);
return basis;
}
struct ovsdb_monitor *
ovsdb_monitor_add(struct ovsdb_monitor *new_dbmon)
{
struct ovsdb_monitor *dbmon;
size_t hash;
/* New_dbmon should be associated with only one jsonrpc
* connections. */
ovs_assert(ovs_list_is_singleton(&new_dbmon->jsonrpc_monitors));
ovsdb_monitor_columns_sort(new_dbmon);
hash = ovsdb_monitor_hash(new_dbmon, 0);
HMAP_FOR_EACH_WITH_HASH(dbmon, hmap_node, hash, &ovsdb_monitors) {
if (ovsdb_monitor_equal(dbmon, new_dbmon)) {
return dbmon;
}
}
hmap_insert(&ovsdb_monitors, &new_dbmon->hmap_node, hash);
return new_dbmon;
}
static void
ovsdb_monitor_destroy(struct ovsdb_monitor *dbmon)
{
struct shash_node *node;
ovs_list_remove(&dbmon->list_node);
if (!hmap_node_is_null(&dbmon->hmap_node)) {
hmap_remove(&ovsdb_monitors, &dbmon->hmap_node);
}
ovsdb_monitor_json_cache_flush(dbmon);
hmap_destroy(&dbmon->json_cache);
struct ovsdb_monitor_change_set *cs, *cs_next;
LIST_FOR_EACH_SAFE (cs, cs_next, list_node, &dbmon->change_sets) {
ovsdb_monitor_change_set_destroy(cs);
}
SHASH_FOR_EACH (node, &dbmon->tables) {
struct ovsdb_monitor_table *mt = node->data;
ovs_assert(ovs_list_is_empty(&mt->change_sets));
free(mt->columns);
free(mt->columns_index_map);
free(mt);
}
shash_destroy(&dbmon->tables);
free(dbmon);
}
static void
ovsdb_monitor_commit(struct ovsdb_monitor *m, const struct ovsdb_txn *txn)
{
struct ovsdb_monitor_aux aux;
ovsdb_monitor_init_aux(&aux, m);
ovsdb_txn_for_each_change(txn, ovsdb_monitor_change_cb, &aux);
if (aux.efficacy > OVSDB_CHANGES_NO_EFFECT) {
/* The transaction is has impact to the monitor.
* Reset new_change_set, so that a new change set will be
* created for future trackings. */
m->new_change_set = NULL;
if (aux.efficacy == OVSDB_CHANGES_REQUIRE_EXTERNAL_UPDATE) {
ovsdb_monitor_json_cache_flush(m);
}
}
}
void
ovsdb_monitors_commit(struct ovsdb *db, const struct ovsdb_txn *txn)
{
struct ovsdb_monitor *m;
LIST_FOR_EACH (m, list_node, &db->monitors) {
ovsdb_monitor_commit(m, txn);
}
}
void
ovsdb_monitors_remove(struct ovsdb *db)
{
struct ovsdb_monitor *m, *next_m;
LIST_FOR_EACH_SAFE (m, next_m, list_node, &db->monitors) {
struct jsonrpc_monitor_node *jm, *next_jm;
/* Delete all front-end monitors. Removing the last front-end monitor
* will also destroy the corresponding ovsdb_monitor. */
LIST_FOR_EACH_SAFE (jm, next_jm, node, &m->jsonrpc_monitors) {
ovsdb_jsonrpc_monitor_destroy(jm->jsonrpc_monitor, false);
}
}
}
/* Add some memory usage statics for monitors into 'usage', for use with
* memory_report(). */
void
ovsdb_monitor_get_memory_usage(struct simap *usage)
{
struct ovsdb_monitor *dbmon;
simap_put(usage, "monitors", hmap_count(&ovsdb_monitors));
HMAP_FOR_EACH(dbmon, hmap_node, &ovsdb_monitors) {
simap_increase(usage, "json-caches", hmap_count(&dbmon->json_cache));
}
}
void
ovsdb_monitor_prereplace_db(struct ovsdb *db)
{
struct ovsdb_monitor *m, *next_m;
LIST_FOR_EACH_SAFE (m, next_m, list_node, &db->monitors) {
struct jsonrpc_monitor_node *jm, *next_jm;
/* Delete all front-end monitors. Removing the last front-end monitor
* will also destroy the corresponding ovsdb_monitor. */
LIST_FOR_EACH_SAFE (jm, next_jm, node, &m->jsonrpc_monitors) {
ovsdb_jsonrpc_monitor_destroy(jm->jsonrpc_monitor, true);
}
}
}
const struct uuid *
ovsdb_monitor_get_last_txnid(struct ovsdb_monitor *dbmon) {
static struct uuid dummy = { .parts = { 0, 0, 0, 0 } };
if (dbmon->db->n_txn_history) {
struct ovsdb_txn_history_node *thn = CONTAINER_OF(
ovs_list_back(&dbmon->db->txn_history),
struct ovsdb_txn_history_node, node);
return ovsdb_txn_get_txnid(thn->txn);
}
return &dummy;
}
Reference in New Issue View Git Blame Copy Permalink