[ovs-dev] [PATCH v2 8/8] ovsdb-idl: Break into two layers.
Ben Pfaff
blp at ovn.org
Wed Dec 2 06:23:41 UTC 2020
This change breaks the IDL into two layers: the IDL proper, whose
interface to its client is unchanged, and a low-level library called
the OVSDB "client synchronization" (CS) library. There are two
reasons for this change. First, the IDL is big and complicated and
I think that this change factors out some of that complication into
a simpler lower layer. Second, the OVN northd implementation based
on DDlog can benefit from the client synchronization library even
though it would actually be made increasingly complicated by the IDL.
Signed-off-by: Ben Pfaff <blp at ovn.org>
---
lib/ovsdb-cs.c | 1953 ++++++++++++++++++++++++++++++++++
lib/ovsdb-cs.h | 135 ++-
lib/ovsdb-idl-provider.h | 8 +-
lib/ovsdb-idl.c | 2167 +++++++++-----------------------------
4 files changed, 2556 insertions(+), 1707 deletions(-)
diff --git a/lib/ovsdb-cs.c b/lib/ovsdb-cs.c
index f37aa5b04414..70fe10bb33f6 100644
--- a/lib/ovsdb-cs.c
+++ b/lib/ovsdb-cs.c
@@ -39,6 +39,1945 @@
#include "uuid.h"
VLOG_DEFINE_THIS_MODULE(ovsdb_cs);
+
+/* Connection state machine.
+ *
+ * When a JSON-RPC session connects, the CS layer sends a "monitor_cond"
+ * request for the Database table in the _Server database and transitions to
+ * the CS_S_SERVER_MONITOR_REQUESTED state. If the session drops and
+ * reconnects, or if the CS receives a "monitor_canceled" notification for a
+ * table it is monitoring, the CS starts over again in the same way. */
+#define OVSDB_CS_STATES \
+ /* Waits for "get_schema" reply, then sends "monitor_cond" \
+ * request for the Database table in the _Server database, whose \
+ * details are informed by the schema, and transitions to \
+ * CS_S_SERVER_MONITOR_REQUESTED. */ \
+ OVSDB_CS_STATE(SERVER_SCHEMA_REQUESTED) \
+ \
+ /* Waits for "monitor_cond" reply for the Database table: \
+ * \
+ * - If the reply indicates success, and the Database table has a \
+ * row for the CS database: \
+ * \
+ * * If the row indicates that this is a clustered database \
+ * that is not connected to the cluster, closes the \
+ * connection. The next connection attempt has a chance at \
+ * picking a connected server. \
+ * \
+ * * Otherwise, sends a monitoring request for the CS \
+ * database whose details are informed by the schema \
+ * (obtained from the row), and transitions to \
+ * CS_S_DATA_MONITOR_(COND_(SINCE_))REQUESTED. \
+ * \
+ * - If the reply indicates success, but the Database table does \
+ * not have a row for the CS database, transitions to \
+ * CS_S_ERROR. \
+ * \
+ * - If the reply indicates failure, sends a "get_schema" request \
+ * for the CS database and transitions to \
+ * CS_S_DATA_SCHEMA_REQUESTED. */ \
+ OVSDB_CS_STATE(SERVER_MONITOR_REQUESTED) \
+ \
+ /* Waits for "get_schema" reply, then sends "monitor_cond" \
+ * request whose details are informed by the schema, and \
+ * transitions to CS_S_DATA_MONITOR_COND_REQUESTED. */ \
+ OVSDB_CS_STATE(DATA_SCHEMA_REQUESTED) \
+ \
+ /* Waits for "monitor_cond_since" reply. If successful, replaces \
+ * the CS contents by the data carried in the reply and \
+ * transitions to CS_S_MONITORING. On failure, sends a \
+ * "monitor_cond" request and transitions to \
+ * CS_S_DATA_MONITOR_COND_REQUESTED. */ \
+ OVSDB_CS_STATE(DATA_MONITOR_COND_SINCE_REQUESTED) \
+ \
+ /* Waits for "monitor_cond" reply. If successful, replaces the \
+ * CS contents by the data carried in the reply and transitions \
+ * to CS_S_MONITORING. On failure, sends a "monitor" request \
+ * and transitions to CS_S_DATA_MONITOR_REQUESTED. */ \
+ OVSDB_CS_STATE(DATA_MONITOR_COND_REQUESTED) \
+ \
+ /* Waits for "monitor" reply. If successful, replaces the CS \
+ * contents by the data carried in the reply and transitions to \
+ * CS_S_MONITORING. On failure, transitions to CS_S_ERROR. */ \
+ OVSDB_CS_STATE(DATA_MONITOR_REQUESTED) \
+ \
+ /* State that processes "update", "update2" or "update3" \
+ * notifications for the main database (and the Database table \
+ * in _Server if available). \
+ * \
+ * If we're monitoring the Database table and we get notified \
+ * that the CS database has been deleted, we close the \
+ * connection (which will restart the state machine). */ \
+ OVSDB_CS_STATE(MONITORING) \
+ \
+ /* Terminal error state that indicates that nothing useful can be \
+ * done, for example because the database server doesn't actually \
+ * have the desired database. We maintain the session with the \
+ * database server anyway. If it starts serving the database \
+ * that we want, or if someone fixes and restarts the database, \
+ * then it will kill the session and we will automatically \
+ * reconnect and try again. */ \
+ OVSDB_CS_STATE(ERROR) \
+ \
+ /* Terminal state that indicates we connected to a useless server \
+ * in a cluster, e.g. one that is partitioned from the rest of \
+ * the cluster. We're waiting to retry. */ \
+ OVSDB_CS_STATE(RETRY)
+
+enum ovsdb_cs_state {
+#define OVSDB_CS_STATE(NAME) CS_S_##NAME,
+ OVSDB_CS_STATES
+#undef OVSDB_CS_STATE
+};
+
+static const char *
+ovsdb_cs_state_to_string(enum ovsdb_cs_state state)
+{
+ switch (state) {
+#define OVSDB_CS_STATE(NAME) case CS_S_##NAME: return #NAME;
+ OVSDB_CS_STATES
+#undef OVSDB_CS_STATE
+ default: return "<unknown>";
+ }
+}
+
+/* A database being monitored.
+ *
+ * There are two instances of this data structure for each CS instance, one for
+ * the _Server database used for working with clusters, and the other one for
+ * the actual database that the client is interested in. */
+struct ovsdb_cs_db {
+ struct ovsdb_cs *cs;
+
+ /* Data. */
+ const char *db_name; /* Database's name. */
+ struct hmap tables; /* Contains "struct ovsdb_cs_db_table *"s.*/
+ struct json *monitor_id;
+ struct json *schema;
+
+ /* Monitor version. */
+ int max_version; /* Maximum version of monitor request to use. */
+ int monitor_version; /* 0 if not monitoring, 1=monitor,
+ * 2=monitor_cond, 3=monitor_cond_since. */
+
+ /* Condition changes. */
+ bool cond_changed; /* Change not yet sent to server? */
+ unsigned int cond_seqno; /* Increments when condition changes. */
+
+ /* Database locking. */
+ char *lock_name; /* Name of lock we need, NULL if none. */
+ bool has_lock; /* Has db server told us we have the lock? */
+ bool is_lock_contended; /* Has db server told us we can't get lock? */
+ struct json *lock_request_id; /* JSON-RPC ID of in-flight lock request. */
+
+ /* Last db txn id, used for fast resync through monitor_cond_since */
+ struct uuid last_id;
+
+ /* Client interface. */
+ struct ovs_list events;
+ const struct ovsdb_cs_ops *ops;
+ void *ops_aux;
+};
+
+static const struct ovsdb_cs_ops ovsdb_cs_server_ops;
+
+static void ovsdb_cs_db_destroy_tables(struct ovsdb_cs_db *);
+static unsigned int ovsdb_cs_db_set_condition(
+ struct ovsdb_cs_db *, const char *db_name, const struct json *condition);
+
+static void ovsdb_cs_send_schema_request(struct ovsdb_cs *,
+ struct ovsdb_cs_db *);
+static void ovsdb_cs_send_db_change_aware(struct ovsdb_cs *);
+static bool ovsdb_cs_check_server_db(struct ovsdb_cs *);
+static void ovsdb_cs_clear_server_rows(struct ovsdb_cs *);
+static void ovsdb_cs_send_monitor_request(struct ovsdb_cs *,
+ struct ovsdb_cs_db *, int version);
+static void ovsdb_cs_db_ack_condition(struct ovsdb_cs_db *db);
+static void ovsdb_cs_db_sync_condition(struct ovsdb_cs_db *db);
+
+struct ovsdb_cs {
+ struct ovsdb_cs_db server;
+ struct ovsdb_cs_db data;
+
+ /* Session state.
+ *
+ * 'state_seqno' is a snapshot of the session's sequence number as returned
+ * jsonrpc_session_get_seqno(session), so if it differs from the value that
+ * function currently returns then the session has reconnected and the
+ * state machine must restart. */
+ struct jsonrpc_session *session; /* Connection to the server. */
+ char *remote; /* 'session' remote name. */
+ enum ovsdb_cs_state state; /* Current session state. */
+ unsigned int state_seqno; /* See above. */
+ struct json *request_id; /* JSON ID for request awaiting reply. */
+
+ /* IDs of outstanding transactions. */
+ struct json **txns;
+ size_t n_txns, allocated_txns;
+
+ /* Info for the _Server database. */
+ struct uuid cid;
+ struct hmap server_rows;
+
+ /* Clustered servers. */
+ uint64_t min_index; /* Minimum allowed index, to avoid regression. */
+ bool leader_only; /* If true, do not connect to Raft followers. */
+ bool shuffle_remotes; /* If true, connect to servers in random order. */
+};
+
+static void ovsdb_cs_transition_at(struct ovsdb_cs *, enum ovsdb_cs_state,
+ const char *where);
+#define ovsdb_cs_transition(CS, STATE) \
+ ovsdb_cs_transition_at(CS, STATE, OVS_SOURCE_LOCATOR)
+
+static void ovsdb_cs_retry_at(struct ovsdb_cs *, const char *where);
+#define ovsdb_cs_retry(CS) ovsdb_cs_retry_at(CS, OVS_SOURCE_LOCATOR)
+
+static struct vlog_rate_limit syntax_rl = VLOG_RATE_LIMIT_INIT(1, 5);
+
+static void ovsdb_cs_db_parse_monitor_reply(struct ovsdb_cs_db *,
+ const struct json *result,
+ int version);
+static bool ovsdb_cs_db_parse_update_rpc(struct ovsdb_cs_db *,
+ const struct jsonrpc_msg *);
+static bool ovsdb_cs_handle_monitor_canceled(struct ovsdb_cs *,
+ struct ovsdb_cs_db *,
+ const struct jsonrpc_msg *);
+
+static bool ovsdb_cs_db_process_lock_replies(struct ovsdb_cs_db *,
+ const struct jsonrpc_msg *);
+static struct jsonrpc_msg *ovsdb_cs_db_compose_lock_request(
+ struct ovsdb_cs_db *);
+static struct jsonrpc_msg *ovsdb_cs_db_compose_unlock_request(
+ struct ovsdb_cs_db *);
+static void ovsdb_cs_db_parse_lock_reply(struct ovsdb_cs_db *,
+ const struct json *);
+static bool ovsdb_cs_db_parse_lock_notify(struct ovsdb_cs_db *,
+ const struct json *params,
+ bool new_has_lock);
+static void ovsdb_cs_send_cond_change(struct ovsdb_cs *);
+
+static bool ovsdb_cs_db_txn_process_reply(struct ovsdb_cs *,
+ const struct jsonrpc_msg *reply);
+
+/* Events. */
+
+void
+ovsdb_cs_event_destroy(struct ovsdb_cs_event *event)
+{
+ if (event) {
+ switch (event->type) {
+ case OVSDB_CS_EVENT_TYPE_RECONNECT:
+ case OVSDB_CS_EVENT_TYPE_LOCKED:
+ break;
+
+ case OVSDB_CS_EVENT_TYPE_UPDATE:
+ json_destroy(event->update.table_updates);
+ break;
+
+ case OVSDB_CS_EVENT_TYPE_TXN_REPLY:
+ jsonrpc_msg_destroy(event->txn_reply);
+ break;
+ }
+ free(event);
+ }
+}
+
+/* Lifecycle. */
+
+static void
+ovsdb_cs_db_init(struct ovsdb_cs_db *db, const char *db_name,
+ struct ovsdb_cs *parent, int max_version,
+ const struct ovsdb_cs_ops *ops, void *ops_aux)
+{
+ *db = (struct ovsdb_cs_db) {
+ .cs = parent,
+ .db_name = xstrdup(db_name),
+ .tables = HMAP_INITIALIZER(&db->tables),
+ .max_version = max_version,
+ .monitor_id = json_array_create_2(json_string_create("monid"),
+ json_string_create(db_name)),
+ .events = OVS_LIST_INITIALIZER(&db->events),
+ .ops = ops,
+ .ops_aux = ops_aux,
+ };
+}
+
+/* Creates and returns a new client synchronization object. The connection
+ * will monitor remote database 'db_name'. If 'retry' is true, then also
+ * reconnect if the connection fails.
+ *
+ * XXX 'max_version' should ordinarily be 3, to allow use of the most efficient
+ * "monitor_cond_since" method with the database. Currently there's some kind
+ * of bug in the DDlog Rust code that interfaces to that, so instead
+ * ovn-northd-ddlog passes 1 to use plain 'monitor' instead. Once the DDlog
+ * Rust code gets fixed, we might as well just delete 'max_version'
+ * entirely.
+ *
+ * 'ops' is a struct for northd_cs_run() to use, and 'ops_aux' is a pointer
+ * that gets passed into each call.
+ *
+ * Use ovsdb_cs_set_remote() to configure the database to which to connect.
+ * Until a remote is configured, no data can be retrieved.
+ */
+struct ovsdb_cs *
+ovsdb_cs_create(const char *db_name, int max_version,
+ const struct ovsdb_cs_ops *ops, void *ops_aux)
+{
+ struct ovsdb_cs *cs = xzalloc(sizeof *cs);
+ ovsdb_cs_db_init(&cs->server, "_Server", cs, 2, &ovsdb_cs_server_ops, cs);
+ ovsdb_cs_db_init(&cs->data, db_name, cs, max_version, ops, ops_aux);
+ cs->state_seqno = UINT_MAX;
+ cs->request_id = NULL;
+ cs->leader_only = true;
+ cs->shuffle_remotes = true;
+ hmap_init(&cs->server_rows);
+
+ return cs;
+}
+
+static void
+ovsdb_cs_db_destroy(struct ovsdb_cs_db *db)
+{
+ ovsdb_cs_db_destroy_tables(db);
+
+ json_destroy(db->monitor_id);
+ json_destroy(db->schema);
+
+ free(db->lock_name);
+
+ json_destroy(db->lock_request_id);
+
+ /* This list always gets flushed out at the end of ovsdb_cs_run(). */
+ ovs_assert(ovs_list_is_empty(&db->events));
+}
+
+/* Destroys 'cs' and all of the data structures that it manages. */
+void
+ovsdb_cs_destroy(struct ovsdb_cs *cs)
+{
+ if (cs) {
+ ovsdb_cs_db_destroy(&cs->server);
+ ovsdb_cs_db_destroy(&cs->data);
+ jsonrpc_session_close(cs->session);
+ free(cs->remote);
+ json_destroy(cs->request_id);
+
+ for (size_t i = 0; i < cs->n_txns; i++) {
+ json_destroy(cs->txns[i]);
+ }
+ free(cs->txns);
+
+ ovsdb_cs_clear_server_rows(cs);
+ hmap_destroy(&cs->server_rows);
+
+ free(cs);
+ }
+}
+
+static void
+ovsdb_cs_transition_at(struct ovsdb_cs *cs, enum ovsdb_cs_state new_state,
+ const char *where)
+{
+ VLOG_DBG("%s: %s -> %s at %s",
+ cs->session ? jsonrpc_session_get_name(cs->session) : "void",
+ ovsdb_cs_state_to_string(cs->state),
+ ovsdb_cs_state_to_string(new_state),
+ where);
+ cs->state = new_state;
+}
+
+static void
+ovsdb_cs_send_request(struct ovsdb_cs *cs, struct jsonrpc_msg *request)
+{
+ json_destroy(cs->request_id);
+ cs->request_id = json_clone(request->id);
+ if (cs->session) {
+ jsonrpc_session_send(cs->session, request);
+ } else {
+ jsonrpc_msg_destroy(request);
+ }
+}
+
+static void
+ovsdb_cs_retry_at(struct ovsdb_cs *cs, const char *where)
+{
+ ovsdb_cs_force_reconnect(cs);
+ ovsdb_cs_transition_at(cs, CS_S_RETRY, where);
+}
+
+static void
+ovsdb_cs_restart_fsm(struct ovsdb_cs *cs)
+{
+ /* Resync data DB table conditions to avoid missing updates due to
+ * conditions that were in flight or changed locally while the connection
+ * was down.
+ */
+ ovsdb_cs_db_sync_condition(&cs->data);
+
+ ovsdb_cs_send_schema_request(cs, &cs->server);
+ ovsdb_cs_transition(cs, CS_S_SERVER_SCHEMA_REQUESTED);
+ cs->data.monitor_version = 0;
+ cs->server.monitor_version = 0;
+}
+
+static void
+ovsdb_cs_process_response(struct ovsdb_cs *cs, struct jsonrpc_msg *msg)
+{
+ bool ok = msg->type == JSONRPC_REPLY;
+ if (!ok
+ && cs->state != CS_S_SERVER_SCHEMA_REQUESTED
+ && cs->state != CS_S_SERVER_MONITOR_REQUESTED
+ && cs->state != CS_S_DATA_MONITOR_COND_REQUESTED
+ && cs->state != CS_S_DATA_MONITOR_COND_SINCE_REQUESTED) {
+ static struct vlog_rate_limit rl = VLOG_RATE_LIMIT_INIT(5, 5);
+ char *s = jsonrpc_msg_to_string(msg);
+ VLOG_INFO_RL(&rl, "%s: received unexpected %s response in "
+ "%s state: %s", jsonrpc_session_get_name(cs->session),
+ jsonrpc_msg_type_to_string(msg->type),
+ ovsdb_cs_state_to_string(cs->state),
+ s);
+ free(s);
+ ovsdb_cs_retry(cs);
+ return;
+ }
+
+ switch (cs->state) {
+ case CS_S_SERVER_SCHEMA_REQUESTED:
+ if (ok) {
+ json_destroy(cs->server.schema);
+ cs->server.schema = json_clone(msg->result);
+ ovsdb_cs_send_monitor_request(cs, &cs->server,
+ cs->server.max_version);
+ ovsdb_cs_transition(cs, CS_S_SERVER_MONITOR_REQUESTED);
+ } else {
+ ovsdb_cs_send_schema_request(cs, &cs->data);
+ ovsdb_cs_transition(cs, CS_S_DATA_SCHEMA_REQUESTED);
+ }
+ break;
+
+ case CS_S_SERVER_MONITOR_REQUESTED:
+ if (ok) {
+ cs->server.monitor_version = cs->server.max_version;
+ ovsdb_cs_db_parse_monitor_reply(&cs->server, msg->result,
+ cs->server.monitor_version);
+ if (ovsdb_cs_check_server_db(cs)) {
+ ovsdb_cs_send_db_change_aware(cs);
+ }
+ } else {
+ ovsdb_cs_send_schema_request(cs, &cs->data);
+ ovsdb_cs_transition(cs, CS_S_DATA_SCHEMA_REQUESTED);
+ }
+ break;
+
+ case CS_S_DATA_SCHEMA_REQUESTED:
+ json_destroy(cs->data.schema);
+ cs->data.schema = json_clone(msg->result);
+ if (cs->data.max_version >= 2) {
+ ovsdb_cs_send_monitor_request(cs, &cs->data, 2);
+ ovsdb_cs_transition(cs, CS_S_DATA_MONITOR_COND_REQUESTED);
+ } else {
+ ovsdb_cs_send_monitor_request(cs, &cs->data, 1);
+ ovsdb_cs_transition(cs, CS_S_DATA_MONITOR_REQUESTED);
+ }
+ break;
+
+ case CS_S_DATA_MONITOR_COND_SINCE_REQUESTED:
+ if (!ok) {
+ /* "monitor_cond_since" not supported. Try "monitor_cond". */
+ ovsdb_cs_send_monitor_request(cs, &cs->data, 2);
+ ovsdb_cs_transition(cs, CS_S_DATA_MONITOR_COND_REQUESTED);
+ } else {
+ cs->data.monitor_version = 3;
+ ovsdb_cs_transition(cs, CS_S_MONITORING);
+ ovsdb_cs_db_parse_monitor_reply(&cs->data, msg->result, 3);
+ }
+ break;
+
+ case CS_S_DATA_MONITOR_COND_REQUESTED:
+ if (!ok) {
+ /* "monitor_cond" not supported. Try "monitor". */
+ ovsdb_cs_send_monitor_request(cs, &cs->data, 1);
+ ovsdb_cs_transition(cs, CS_S_DATA_MONITOR_REQUESTED);
+ } else {
+ cs->data.monitor_version = 2;
+ ovsdb_cs_transition(cs, CS_S_MONITORING);
+ ovsdb_cs_db_parse_monitor_reply(&cs->data, msg->result, 2);
+ }
+ break;
+
+ case CS_S_DATA_MONITOR_REQUESTED:
+ cs->data.monitor_version = 1;
+ ovsdb_cs_transition(cs, CS_S_MONITORING);
+ ovsdb_cs_db_parse_monitor_reply(&cs->data, msg->result, 1);
+ break;
+
+ case CS_S_MONITORING:
+ /* We don't normally have a request outstanding in this state. If we
+ * do, it's a "monitor_cond_change", which means that the conditional
+ * monitor clauses were updated.
+ *
+ * Mark the last requested conditions as acked and if further
+ * condition changes were pending, send them now. */
+ ovsdb_cs_db_ack_condition(&cs->data);
+ ovsdb_cs_send_cond_change(cs);
+ cs->data.cond_seqno++;
+ break;
+
+ case CS_S_ERROR:
+ case CS_S_RETRY:
+ /* Nothing to do in this state. */
+ break;
+
+ default:
+ OVS_NOT_REACHED();
+ }
+}
+
+static void
+ovsdb_cs_process_msg(struct ovsdb_cs *cs, struct jsonrpc_msg *msg)
+{
+ bool is_response = (msg->type == JSONRPC_REPLY ||
+ msg->type == JSONRPC_ERROR);
+
+ /* Process a reply to an outstanding request. */
+ if (is_response
+ && cs->request_id && json_equal(cs->request_id, msg->id)) {
+ json_destroy(cs->request_id);
+ cs->request_id = NULL;
+ ovsdb_cs_process_response(cs, msg);
+ return;
+ }
+
+ /* Process database contents updates. */
+ if (ovsdb_cs_db_parse_update_rpc(&cs->data, msg)) {
+ return;
+ }
+ if (cs->server.monitor_version
+ && ovsdb_cs_db_parse_update_rpc(&cs->server, msg)) {
+ ovsdb_cs_check_server_db(cs);
+ return;
+ }
+
+ if (ovsdb_cs_handle_monitor_canceled(cs, &cs->data, msg)
+ || (cs->server.monitor_version
+ && ovsdb_cs_handle_monitor_canceled(cs, &cs->server, msg))) {
+ return;
+ }
+
+ /* Process "lock" replies and related notifications. */
+ if (ovsdb_cs_db_process_lock_replies(&cs->data, msg)) {
+ return;
+ }
+
+ /* Process response to a database transaction we submitted. */
+ if (is_response && ovsdb_cs_db_txn_process_reply(cs, msg)) {
+ return;
+ }
+
+ /* Unknown message. Log at a low level because this can happen if
+ * ovsdb_cs_txn_destroy() is called to destroy a transaction
+ * before we receive the reply.
+ *
+ * (We could sort those out from other kinds of unknown messages by
+ * using distinctive IDs for transactions, if it seems valuable to
+ * do so, and then it would be possible to use different log
+ * levels. XXX?) */
+ char *s = jsonrpc_msg_to_string(msg);
+ VLOG_DBG("%s: received unexpected %s message: %s",
+ jsonrpc_session_get_name(cs->session),
+ jsonrpc_msg_type_to_string(msg->type), s);
+ free(s);
+}
+
+static struct ovsdb_cs_event *
+ovsdb_cs_db_add_event(struct ovsdb_cs_db *db, enum ovsdb_cs_event_type type)
+{
+ struct ovsdb_cs_event *event = xmalloc(sizeof *event);
+ event->type = type;
+ ovs_list_push_back(&db->events, &event->list_node);
+ return event;
+}
+
+/* Processes a batch of messages from the database server on 'cs'. This may
+ * cause the CS's contents to change.
+ *
+ * Initializes 'events' with a list of events that occurred on 'cs'. The
+ * caller must process and destroy all of the events. */
+void
+ovsdb_cs_run(struct ovsdb_cs *cs, struct ovs_list *events)
+{
+ ovs_list_init(events);
+ if (!cs->session) {
+ return;
+ }
+
+ ovsdb_cs_send_cond_change(cs);
+
+ jsonrpc_session_run(cs->session);
+
+ unsigned int seqno = jsonrpc_session_get_seqno(cs->session);
+ if (cs->state_seqno != seqno) {
+ cs->state_seqno = seqno;
+ ovsdb_cs_restart_fsm(cs);
+
+ for (size_t i = 0; i < cs->n_txns; i++) {
+ json_destroy(cs->txns[i]);
+ }
+ cs->n_txns = 0;
+
+ ovsdb_cs_db_add_event(&cs->data, OVSDB_CS_EVENT_TYPE_RECONNECT);
+
+ if (cs->data.lock_name) {
+ jsonrpc_session_send(
+ cs->session,
+ ovsdb_cs_db_compose_lock_request(&cs->data));
+ }
+ }
+
+ for (int i = 0; i < 50; i++) {
+ struct jsonrpc_msg *msg = jsonrpc_session_recv(cs->session);
+ if (!msg) {
+ break;
+ }
+ ovsdb_cs_process_msg(cs, msg);
+ jsonrpc_msg_destroy(msg);
+ }
+ ovs_list_push_back_all(events, &cs->data.events);
+}
+
+/* Arranges for poll_block() to wake up when ovsdb_cs_run() has something to
+ * do or when activity occurs on a transaction on 'cs'. */
+void
+ovsdb_cs_wait(struct ovsdb_cs *cs)
+{
+ if (!cs->session) {
+ return;
+ }
+ jsonrpc_session_wait(cs->session);
+ jsonrpc_session_recv_wait(cs->session);
+}
+
+/* Network connection. */
+
+/* Changes the remote and creates a new session.
+ *
+ * If 'retry' is true, the connection to the remote will automatically retry
+ * when it fails. If 'retry' is false, the connection is one-time. */
+void
+ovsdb_cs_set_remote(struct ovsdb_cs *cs, const char *remote, bool retry)
+{
+ if (cs
+ && ((remote != NULL) != (cs->remote != NULL)
+ || (remote && cs->remote && strcmp(remote, cs->remote)))) {
+ /* Close the old session, if any. */
+ if (cs->session) {
+ jsonrpc_session_close(cs->session);
+ cs->session = NULL;
+
+ free(cs->remote);
+ cs->remote = NULL;
+ }
+
+ /* Open new session, if any. */
+ if (remote) {
+ struct svec remotes = SVEC_EMPTY_INITIALIZER;
+ ovsdb_session_parse_remote(remote, &remotes, &cs->cid);
+ if (cs->shuffle_remotes) {
+ svec_shuffle(&remotes);
+ }
+ cs->session = jsonrpc_session_open_multiple(&remotes, retry);
+ svec_destroy(&remotes);
+
+ cs->state_seqno = UINT_MAX;
+
+ cs->remote = xstrdup(remote);
+ }
+ }
+}
+
+/* Reconfigures 'cs' so that it would reconnect to the database, if
+ * connection was dropped. */
+void
+ovsdb_cs_enable_reconnect(struct ovsdb_cs *cs)
+{
+ if (cs->session) {
+ jsonrpc_session_enable_reconnect(cs->session);
+ }
+}
+
+/* Forces 'cs' to drop its connection to the database and reconnect. In the
+ * meantime, the contents of 'cs' will not change. */
+void
+ovsdb_cs_force_reconnect(struct ovsdb_cs *cs)
+{
+ if (cs->session) {
+ jsonrpc_session_force_reconnect(cs->session);
+ }
+}
+
+/* Drops 'cs''s current connection and the cached session. This is useful if
+ * the client notices some kind of inconsistency. */
+void
+ovsdb_cs_flag_inconsistency(struct ovsdb_cs *cs)
+{
+ cs->data.last_id = UUID_ZERO;
+ ovsdb_cs_retry(cs);
+}
+
+/* Returns true if 'cs' is currently connected or will eventually try to
+ * reconnect. */
+bool
+ovsdb_cs_is_alive(const struct ovsdb_cs *cs)
+{
+ return (cs->session
+ && jsonrpc_session_is_alive(cs->session)
+ && cs->state != CS_S_ERROR);
+}
+
+/* Returns true if 'cs' is currently connected to a server. */
+bool
+ovsdb_cs_is_connected(const struct ovsdb_cs *cs)
+{
+ return cs->session && jsonrpc_session_is_connected(cs->session);
+}
+
+/* Returns the last error reported on a connection by 'cs'. The return value
+ * is 0 only if no connection made by 'cs' has ever encountered an error and
+ * a negative response to a schema request has never been received. See
+ * jsonrpc_get_status() for jsonrpc_session_get_last_error() return value
+ * interpretation. */
+int
+ovsdb_cs_get_last_error(const struct ovsdb_cs *cs)
+{
+ int err = cs->session ? jsonrpc_session_get_last_error(cs->session) : 0;
+ if (err) {
+ return err;
+ } else if (cs->state == CS_S_ERROR) {
+ return ENOENT;
+ } else {
+ return 0;
+ }
+}
+
+/* Sets the "probe interval" for 'cs''s current session to 'probe_interval', in
+ * milliseconds. */
+void
+ovsdb_cs_set_probe_interval(const struct ovsdb_cs *cs, int probe_interval)
+{
+ if (cs->session) {
+ jsonrpc_session_set_probe_interval(cs->session, probe_interval);
+ }
+}
+
+/* Conditional monitoring. */
+
+/* A table being monitored.
+ *
+ * At the CS layer, the only thing we care about, table-wise, is the conditions
+ * we're using for monitoring them, so there's little here. We only create
+ * these table structures at all for tables that have conditions. */
+struct ovsdb_cs_db_table {
+ struct hmap_node hmap_node; /* Indexed by 'name'. */
+ const char *name; /* Table name. */
+
+ /* Each of these is a null pointer if it is empty, or JSON [<condition>*]
+ * or [true] or [false] otherwise. [true] could be represented as a null
+ * pointer, but we want to distinguish "empty slot" from "a condition that
+ * is always true" in a slot. */
+ struct json *ack_cond; /* Last condition acked by the server. */
+ struct json *req_cond; /* Last condition requested to the server. */
+ struct json *new_cond; /* Latest condition set by the IDL client. */
+};
+
+/* A kind of condition, so that we can treat equivalent JSON as equivalent. */
+enum condition_type {
+ COND_FALSE, /* [] or [false] */
+ COND_TRUE, /* Null pointer or [true] */
+ COND_OTHER /* Anything else. */
+};
+
+/* Returns the condition_type for 'condition'. */
+static enum condition_type
+condition_classify(const struct json *condition)
+{
+ if (condition) {
+ const struct json_array *a = json_array(condition);
+ switch (a->n) {
+ case 0:
+ return COND_FALSE;
+
+ case 1:
+ return (a->elems[0]->type == JSON_FALSE ? COND_FALSE
+ : a->elems[0]->type == JSON_TRUE ? COND_TRUE
+ : COND_OTHER);
+
+ default:
+ return COND_OTHER;
+ }
+ } else {
+ return COND_TRUE;
+ }
+}
+
+/* Returns true if 'a' and 'b' are the same condition (in an obvious way; we're
+ * not going to compare for boolean equivalence or anything). */
+static bool
+condition_equal(const struct json *a, const struct json *b)
+{
+ enum condition_type type = condition_classify(a);
+ return (type == condition_classify(b)
+ && (type != COND_OTHER || json_equal(a, b)));
+}
+
+/* Returns a clone of 'condition', translating always-true and always-false to
+ * [true] and [false], respectively. */
+static struct json *
+condition_clone(const struct json *condition)
+{
+ switch (condition_classify(condition)) {
+ case COND_TRUE:
+ return json_array_create_1(json_boolean_create(true));
+
+ case COND_FALSE:
+ return json_array_create_1(json_boolean_create(false));
+
+ case COND_OTHER:
+ return json_clone(condition);
+ }
+
+ OVS_NOT_REACHED();
+}
+
+/* Returns the ovsdb_cs_db_table associated with 'table' in 'db', creating an
+ * empty one if necessary. */
+static struct ovsdb_cs_db_table *
+ovsdb_cs_db_get_table(struct ovsdb_cs_db *db, const char *table)
+{
+ uint32_t hash = hash_string(table, 0);
+ struct ovsdb_cs_db_table *t;
+
+ HMAP_FOR_EACH_WITH_HASH (t, hmap_node, hash, &db->tables) {
+ if (!strcmp(t->name, table)) {
+ return t;
+ }
+ }
+
+ t = xzalloc(sizeof *t);
+ t->name = xstrdup(table);
+ t->new_cond = json_array_create_1(json_boolean_create(true));
+ hmap_insert(&db->tables, &t->hmap_node, hash);
+ return t;
+}
+
+static void
+ovsdb_cs_db_destroy_tables(struct ovsdb_cs_db *db)
+{
+ struct ovsdb_cs_db_table *table, *next;
+ HMAP_FOR_EACH_SAFE (table, next, hmap_node, &db->tables) {
+ json_destroy(table->ack_cond);
+ json_destroy(table->req_cond);
+ json_destroy(table->new_cond);
+ hmap_remove(&db->tables, &table->hmap_node);
+ }
+ hmap_destroy(&db->tables);
+}
+
+static unsigned int
+ovsdb_cs_db_set_condition(struct ovsdb_cs_db *db, const char *table,
+ const struct json *condition)
+{
+ /* Compare the new condition to the last known condition which can be
+ * either "new" (not sent yet), "requested" or "acked", in this order. */
+ struct ovsdb_cs_db_table *t = ovsdb_cs_db_get_table(db, table);
+ const struct json *table_cond = (t->new_cond ? t->new_cond
+ : t->req_cond ? t->req_cond
+ : t->ack_cond);
+ if (!condition_equal(condition, table_cond)) {
+ json_destroy(t->new_cond);
+ t->new_cond = condition_clone(condition);
+ db->cond_changed = true;
+ poll_immediate_wake();
+ return db->cond_seqno + 1;
+ } else if (table_cond != t->ack_cond) {
+ /* 'condition' was already set but has not been "acked" yet. The IDL
+ * will be up to date when db->cond_seqno gets incremented. */
+ return db->cond_seqno + 1;
+ } else {
+ return db->cond_seqno;
+ }
+}
+
+/* Sets the replication condition for 'tc' in 'cs' to 'condition' and arranges
+ * to send the new condition to the database server.
+ *
+ * Return the next conditional update sequence number. When this value and
+ * ovsdb_cs_get_condition_seqno() matches, 'cs' contains rows that match the
+ * 'condition'. */
+unsigned int
+ovsdb_cs_set_condition(struct ovsdb_cs *cs, const char *table,
+ const struct json *condition)
+{
+ return ovsdb_cs_db_set_condition(&cs->data, table, condition);
+}
+
+/* Returns a "sequence number" that represents the number of conditional
+ * monitoring updates successfully received by the OVSDB server of a CS
+ * connection.
+ *
+ * ovsdb_cs_set_condition() sets a new condition that is different from the
+ * current condtion, the next expected "sequence number" is returned.
+ *
+ * Whenever ovsdb_cs_get_condition_seqno() returns a value that matches the
+ * return value of ovsdb_cs_set_condition(), the client is assured that:
+ *
+ * - The ovsdb_cs_set_condition() changes has been acknowledged by the OVSDB
+ * server.
+ *
+ * - 'cs' now contains the content matches the new conditions. */
+unsigned int
+ovsdb_cs_get_condition_seqno(const struct ovsdb_cs *cs)
+{
+ return cs->data.cond_seqno;
+}
+
+static struct json *
+ovsdb_cs_create_cond_change_req(const struct json *cond)
+{
+ struct json *monitor_cond_change_request = json_object_create();
+ json_object_put(monitor_cond_change_request, "where", json_clone(cond));
+ return monitor_cond_change_request;
+}
+
+static struct jsonrpc_msg *
+ovsdb_cs_db_compose_cond_change(struct ovsdb_cs_db *db)
+{
+ if (!db->cond_changed) {
+ return NULL;
+ }
+
+ struct json *monitor_cond_change_requests = NULL;
+ struct ovsdb_cs_db_table *table;
+ HMAP_FOR_EACH (table, hmap_node, &db->tables) {
+ /* Always use the most recent conditions set by the CS client when
+ * requesting monitor_cond_change, i.e., table->new_cond.
+ */
+ if (table->new_cond) {
+ struct json *req =
+ ovsdb_cs_create_cond_change_req(table->new_cond);
+ if (req) {
+ if (!monitor_cond_change_requests) {
+ monitor_cond_change_requests = json_object_create();
+ }
+ json_object_put(monitor_cond_change_requests,
+ table->name,
+ json_array_create_1(req));
+ }
+ /* Mark the new condition as requested by moving it to req_cond.
+ * If there's already requested condition that's a bug.
+ */
+ ovs_assert(table->req_cond == NULL);
+ table->req_cond = table->new_cond;
+ table->new_cond = NULL;
+ }
+ }
+
+ if (!monitor_cond_change_requests) {
+ return NULL;
+ }
+
+ db->cond_changed = false;
+ struct json *params = json_array_create_3(json_clone(db->monitor_id),
+ json_clone(db->monitor_id),
+ monitor_cond_change_requests);
+ return jsonrpc_create_request("monitor_cond_change", params, NULL);
+}
+
+/* Marks all requested table conditions in 'db' as acked by the server.
+ * It should be called when the server replies to monitor_cond_change
+ * requests.
+ */
+static void
+ovsdb_cs_db_ack_condition(struct ovsdb_cs_db *db)
+{
+ struct ovsdb_cs_db_table *table;
+ HMAP_FOR_EACH (table, hmap_node, &db->tables) {
+ if (table->req_cond) {
+ json_destroy(table->ack_cond);
+ table->ack_cond = table->req_cond;
+ table->req_cond = NULL;
+ }
+ }
+}
+
+/* Should be called when the CS fsm is restarted and resyncs table conditions
+ * based on the state the DB is in:
+ * - if a non-zero last_id is available for the DB then upon reconnect
+ * the CS should first request acked conditions to avoid missing updates
+ * about records that were added before the transaction with
+ * txn-id == last_id. If there were requested condition changes in flight
+ * (i.e., req_cond not NULL) and the CS client didn't set new conditions
+ * (i.e., new_cond is NULL) then move req_cond to new_cond to trigger a
+ * follow up monitor_cond_change request.
+ * - if there's no last_id available for the DB then it's safe to use the
+ * latest conditions set by the CS client even if they weren't acked yet.
+ */
+static void
+ovsdb_cs_db_sync_condition(struct ovsdb_cs_db *db)
+{
+ bool ack_all = uuid_is_zero(&db->last_id);
+ if (ack_all) {
+ db->cond_changed = false;
+ }
+
+ struct ovsdb_cs_db_table *table;
+ HMAP_FOR_EACH (table, hmap_node, &db->tables) {
+ /* When monitor_cond_since requests will be issued, the
+ * table->ack_cond condition will be added to the "where" clause".
+ * Follow up monitor_cond_change requests will use table->new_cond.
+ */
+ if (ack_all) {
+ if (table->new_cond) {
+ json_destroy(table->req_cond);
+ table->req_cond = table->new_cond;
+ table->new_cond = NULL;
+ }
+
+ if (table->req_cond) {
+ json_destroy(table->ack_cond);
+ table->ack_cond = table->req_cond;
+ table->req_cond = NULL;
+ }
+ } else {
+ /* If there was no "unsent" condition but instead a
+ * monitor_cond_change request was in flight, move table->req_cond
+ * to table->new_cond and set db->cond_changed to trigger a new
+ * monitor_cond_change request.
+ *
+ * However, if a new condition has been set by the CS client,
+ * monitor_cond_change will be sent anyway and will use the most
+ * recent table->new_cond so there's no need to update it here.
+ */
+ if (table->req_cond && !table->new_cond) {
+ table->new_cond = table->req_cond;
+ table->req_cond = NULL;
+ db->cond_changed = true;
+ }
+ }
+ }
+}
+
+static void
+ovsdb_cs_send_cond_change(struct ovsdb_cs *cs)
+{
+ /* When 'cs->request_id' is not NULL, there is an outstanding
+ * conditional monitoring update request that we have not heard
+ * from the server yet. Don't generate another request in this case. */
+ if (!jsonrpc_session_is_connected(cs->session)
+ || cs->data.monitor_version == 1
+ || cs->request_id) {
+ return;
+ }
+
+ struct jsonrpc_msg *msg = ovsdb_cs_db_compose_cond_change(&cs->data);
+ if (msg) {
+ cs->request_id = json_clone(msg->id);
+ jsonrpc_session_send(cs->session, msg);
+ }
+}
+
+/* Clustered servers. */
+
+/* By default, or if 'leader_only' is true, when 'cs' connects to a clustered
+ * database, the CS layer will avoid servers other than the cluster
+ * leader. This ensures that any data that it reads and reports is up-to-date.
+ * If 'leader_only' is false, the CS layer will accept any server in the
+ * cluster, which means that for read-only transactions it can report and act
+ * on stale data (transactions that modify the database are always serialized
+ * even with false 'leader_only'). Refer to Understanding Cluster Consistency
+ * in ovsdb(7) for more information. */
+void
+ovsdb_cs_set_leader_only(struct ovsdb_cs *cs, bool leader_only)
+{
+ cs->leader_only = leader_only;
+ if (leader_only && cs->server.monitor_version) {
+ ovsdb_cs_check_server_db(cs);
+ }
+}
+
+/* Set whether the order of remotes should be shuffled, when there is more than
+ * one remote. The setting doesn't take effect until the next time when
+ * ovsdb_cs_set_remote() is called. */
+void
+ovsdb_cs_set_shuffle_remotes(struct ovsdb_cs *cs, bool shuffle)
+{
+ cs->shuffle_remotes = shuffle;
+}
+
+/* Reset min_index to 0. This prevents a situation where the client
+ * thinks all databases have stale data, when they actually have all
+ * been destroyed and rebuilt from scratch.
+ */
+void
+ovsdb_cs_reset_min_index(struct ovsdb_cs *cs)
+{
+ cs->min_index = 0;
+}
+
+/* Database locks. */
+
+static struct jsonrpc_msg *
+ovsdb_cs_db_set_lock(struct ovsdb_cs_db *db, const char *lock_name)
+{
+ if (db->lock_name
+ && (!lock_name || strcmp(lock_name, db->lock_name))) {
+ /* Release previous lock. */
+ struct jsonrpc_msg *msg = ovsdb_cs_db_compose_unlock_request(db);
+ free(db->lock_name);
+ db->lock_name = NULL;
+ db->is_lock_contended = false;
+ return msg;
+ }
+
+ if (lock_name && !db->lock_name) {
+ /* Acquire new lock. */
+ db->lock_name = xstrdup(lock_name);
+ return ovsdb_cs_db_compose_lock_request(db);
+ }
+
+ return NULL;
+}
+
+/* If 'lock_name' is nonnull, configures 'cs' to obtain the named lock from the
+ * database server and to prevent modifying the database when the lock cannot
+ * be acquired (that is, when another client has the same lock).
+ *
+ * If 'lock_name' is NULL, drops the locking requirement and releases the
+ * lock. */
+void
+ovsdb_cs_set_lock(struct ovsdb_cs *cs, const char *lock_name)
+{
+ for (;;) {
+ struct jsonrpc_msg *msg = ovsdb_cs_db_set_lock(&cs->data, lock_name);
+ if (!msg) {
+ break;
+ }
+ if (cs->session) {
+ jsonrpc_session_send(cs->session, msg);
+ } else {
+ jsonrpc_msg_destroy(msg);
+ }
+ }
+}
+
+/* Returns the name of the lock that 'cs' is trying to obtain, or NULL if none
+ * is configured. */
+const char *
+ovsdb_cs_get_lock(const struct ovsdb_cs *cs)
+{
+ return cs->data.lock_name;
+}
+
+/* Returns true if 'cs' is configured to obtain a lock and owns that lock,
+ * false if it doesn't own the lock or isn't configured to obtain one.
+ *
+ * Locking and unlocking happens asynchronously from the database client's
+ * point of view, so the information is only useful for optimization (e.g. if
+ * the client doesn't have the lock then there's no point in trying to write to
+ * the database). */
+bool
+ovsdb_cs_has_lock(const struct ovsdb_cs *cs)
+{
+ return cs->data.has_lock;
+}
+
+/* Returns true if 'cs' is configured to obtain a lock but the database server
+ * has indicated that some other client already owns the requested lock. */
+bool
+ovsdb_cs_is_lock_contended(const struct ovsdb_cs *cs)
+{
+ return cs->data.is_lock_contended;
+}
+
+static void
+ovsdb_cs_db_update_has_lock(struct ovsdb_cs_db *db, bool new_has_lock)
+{
+ if (new_has_lock && !db->has_lock) {
+ ovsdb_cs_db_add_event(db, OVSDB_CS_EVENT_TYPE_LOCKED);
+ db->is_lock_contended = false;
+ }
+ db->has_lock = new_has_lock;
+}
+
+static bool
+ovsdb_cs_db_process_lock_replies(struct ovsdb_cs_db *db,
+ const struct jsonrpc_msg *msg)
+{
+ if (msg->type == JSONRPC_REPLY
+ && db->lock_request_id
+ && json_equal(db->lock_request_id, msg->id)) {
+ /* Reply to our "lock" request. */
+ ovsdb_cs_db_parse_lock_reply(db, msg->result);
+ return true;
+ }
+
+ if (msg->type == JSONRPC_NOTIFY) {
+ if (!strcmp(msg->method, "locked")) {
+ /* We got our lock. */
+ return ovsdb_cs_db_parse_lock_notify(db, msg->params, true);
+ } else if (!strcmp(msg->method, "stolen")) {
+ /* Someone else stole our lock. */
+ return ovsdb_cs_db_parse_lock_notify(db, msg->params, false);
+ }
+ }
+
+ return false;
+}
+
+static struct jsonrpc_msg *
+ovsdb_cs_db_compose_lock_request__(struct ovsdb_cs_db *db,
+ const char *method)
+{
+ ovsdb_cs_db_update_has_lock(db, false);
+
+ json_destroy(db->lock_request_id);
+ db->lock_request_id = NULL;
+
+ struct json *params = json_array_create_1(json_string_create(
+ db->lock_name));
+ return jsonrpc_create_request(method, params, NULL);
+}
+
+static struct jsonrpc_msg *
+ovsdb_cs_db_compose_lock_request(struct ovsdb_cs_db *db)
+{
+ struct jsonrpc_msg *msg = ovsdb_cs_db_compose_lock_request__(db, "lock");
+ db->lock_request_id = json_clone(msg->id);
+ return msg;
+}
+
+static struct jsonrpc_msg *
+ovsdb_cs_db_compose_unlock_request(struct ovsdb_cs_db *db)
+{
+ return ovsdb_cs_db_compose_lock_request__(db, "unlock");
+}
+
+static void
+ovsdb_cs_db_parse_lock_reply(struct ovsdb_cs_db *db,
+ const struct json *result)
+{
+ bool got_lock;
+
+ json_destroy(db->lock_request_id);
+ db->lock_request_id = NULL;
+
+ if (result->type == JSON_OBJECT) {
+ const struct json *locked;
+
+ locked = shash_find_data(json_object(result), "locked");
+ got_lock = locked && locked->type == JSON_TRUE;
+ } else {
+ got_lock = false;
+ }
+
+ ovsdb_cs_db_update_has_lock(db, got_lock);
+ if (!got_lock) {
+ db->is_lock_contended = true;
+ }
+}
+
+static bool
+ovsdb_cs_db_parse_lock_notify(struct ovsdb_cs_db *db,
+ const struct json *params,
+ bool new_has_lock)
+{
+ if (db->lock_name
+ && params->type == JSON_ARRAY
+ && json_array(params)->n > 0
+ && json_array(params)->elems[0]->type == JSON_STRING) {
+ const char *lock_name = json_string(json_array(params)->elems[0]);
+
+ if (!strcmp(db->lock_name, lock_name)) {
+ ovsdb_cs_db_update_has_lock(db, new_has_lock);
+ if (!new_has_lock) {
+ db->is_lock_contended = true;
+ }
+ return true;
+ }
+ }
+ return false;
+}
+
+/* Transactions. */
+
+static bool
+ovsdb_cs_db_txn_process_reply(struct ovsdb_cs *cs,
+ const struct jsonrpc_msg *reply)
+{
+ bool found = ovsdb_cs_forget_transaction(cs, reply->id);
+ if (found) {
+ struct ovsdb_cs_event *event
+ = ovsdb_cs_db_add_event(&cs->data, OVSDB_CS_EVENT_TYPE_TXN_REPLY);
+ event->txn_reply = jsonrpc_msg_clone(reply);
+ }
+ return found;
+}
+
+/* Returns true if 'cs' can be sent a transaction now, false otherwise. This
+ * is useful for optimization: there is no point in composing and sending a
+ * transaction if it returns false. */
+bool
+ovsdb_cs_may_send_transaction(const struct ovsdb_cs *cs)
+{
+ return (cs->session != NULL
+ && cs->state == CS_S_MONITORING
+ && (!cs->data.lock_name || ovsdb_cs_has_lock(cs)));
+}
+
+/* Attempts to send a transaction with the specified 'operations' to 'cs''s
+ * server. On success, returns the request ID; the caller must eventually free
+ * it. On failure, returns NULL. */
+struct json * OVS_WARN_UNUSED_RESULT
+ovsdb_cs_send_transaction(struct ovsdb_cs *cs, struct json *operations)
+{
+ if (!ovsdb_cs_may_send_transaction(cs)) {
+ json_destroy(operations);
+ return NULL;
+ }
+
+ if (cs->data.lock_name) {
+ struct json *assertion = json_object_create();
+ json_object_put_string(assertion, "op", "assert");
+ json_object_put_string(assertion, "lock", cs->data.lock_name);
+ json_array_add(operations, assertion);
+ }
+
+ struct json *request_id;
+ struct jsonrpc_msg *request = jsonrpc_create_request(
+ "transact", operations, &request_id);
+ int error = jsonrpc_session_send(cs->session, request);
+ if (error) {
+ json_destroy(request_id);
+ return NULL;
+ }
+
+ if (cs->n_txns >= cs->allocated_txns) {
+ cs->txns = x2nrealloc(cs->txns, &cs->allocated_txns,
+ sizeof *cs->txns);
+ }
+ cs->txns[cs->n_txns++] = json_clone(request_id);
+ return request_id;
+}
+
+/* Makes 'cs' drop its record of transaction 'request_id'. If a reply arrives
+ * for it later (which it will, unless the connection drops in the meantime),
+ * it won't be reported through an event.
+ *
+ * Returns true if 'request_id' was known, false otherwise. */
+bool
+ovsdb_cs_forget_transaction(struct ovsdb_cs *cs, const struct json *request_id)
+{
+ for (size_t i = 0; i < cs->n_txns; i++) {
+ if (json_equal(request_id, cs->txns[i])) {
+ cs->txns[i] = cs->txns[--cs->n_txns];
+ return true;
+ }
+ }
+ return false;
+}
+
+static void
+ovsdb_cs_send_schema_request(struct ovsdb_cs *cs,
+ struct ovsdb_cs_db *db)
+{
+ ovsdb_cs_send_request(cs, jsonrpc_create_request(
+ "get_schema",
+ json_array_create_1(json_string_create(
+ db->db_name)),
+ NULL));
+}
+
+static void
+ovsdb_cs_send_db_change_aware(struct ovsdb_cs *cs)
+{
+ struct jsonrpc_msg *msg = jsonrpc_create_request(
+ "set_db_change_aware", json_array_create_1(json_boolean_create(true)),
+ NULL);
+ jsonrpc_session_send(cs->session, msg);
+}
+
+static void
+ovsdb_cs_send_monitor_request(struct ovsdb_cs *cs, struct ovsdb_cs_db *db,
+ int version)
+{
+ struct json *mrs = db->ops->compose_monitor_requests(
+ db->schema, db->ops_aux);
+ /* XXX handle failure */
+ ovs_assert(mrs->type == JSON_OBJECT);
+
+ if (version > 1) {
+ struct ovsdb_cs_db_table *table;
+ HMAP_FOR_EACH (table, hmap_node, &db->tables) {
+ if (table->ack_cond) {
+ struct json *mr = shash_find_data(json_object(mrs),
+ table->name);
+ if (!mr) {
+ mr = json_array_create_empty();
+ json_object_put(mrs, table->name, mr);
+ }
+ ovs_assert(mr->type == JSON_ARRAY);
+
+ struct json *mr0;
+ if (json_array(mr)->n == 0) {
+ mr0 = json_object_create();
+ json_object_put(mr0, "columns", json_array_create_empty());
+ json_array_add(mr, mr0);
+ } else {
+ mr0 = json_array(mr)->elems[0];
+ }
+ ovs_assert(mr0->type == JSON_OBJECT);
+
+ json_object_put(mr0, "where",
+ json_clone(table->ack_cond));
+ }
+ }
+ }
+
+ const char *method = (version == 1 ? "monitor"
+ : version == 2 ? "monitor_cond"
+ : "monitor_cond_since");
+ struct json *params = json_array_create_3(
+ json_string_create(db->db_name),
+ json_clone(db->monitor_id),
+ mrs);
+ if (version == 3) {
+ struct json *json_last_id = json_string_create_nocopy(
+ xasprintf(UUID_FMT, UUID_ARGS(&db->last_id)));
+ json_array_add(params, json_last_id);
+ }
+ ovsdb_cs_send_request(cs, jsonrpc_create_request(method, params, NULL));
+}
+
+static void
+log_parse_update_error(struct ovsdb_error *error)
+{
+ if (!VLOG_DROP_WARN(&syntax_rl)) {
+ char *s = ovsdb_error_to_string(error);
+ VLOG_WARN_RL(&syntax_rl, "%s", s);
+ free(s);
+ }
+ ovsdb_error_destroy(error);
+}
+
+static void
+ovsdb_cs_db_add_update(struct ovsdb_cs_db *db,
+ const struct json *table_updates, int version,
+ bool clear)
+{
+ struct ovsdb_cs_event *event = ovsdb_cs_db_add_event(
+ db, OVSDB_CS_EVENT_TYPE_UPDATE);
+ event->update = (struct ovsdb_cs_update_event) {
+ .table_updates = json_clone(table_updates),
+ .clear = clear,
+ .version = version,
+ };
+}
+
+static void
+ovsdb_cs_db_parse_monitor_reply(struct ovsdb_cs_db *db,
+ const struct json *result, int version)
+{
+ const struct json *table_updates;
+ bool clear;
+ if (version == 3) {
+ struct uuid last_id;
+ if (result->type != JSON_ARRAY || result->array.n != 3
+ || (result->array.elems[0]->type != JSON_TRUE &&
+ result->array.elems[0]->type != JSON_FALSE)
+ || result->array.elems[1]->type != JSON_STRING
+ || !uuid_from_string(&last_id,
+ json_string(result->array.elems[1]))) {
+ struct ovsdb_error *error = ovsdb_syntax_error(
+ result, NULL, "bad monitor_cond_since reply format");
+ log_parse_update_error(error);
+ return;
+ }
+
+ bool found = json_boolean(result->array.elems[0]);
+ clear = !found;
+ table_updates = result->array.elems[2];
+ } else {
+ clear = true;
+ table_updates = result;
+ }
+
+ ovsdb_cs_db_add_update(db, table_updates, version, clear);
+}
+
+static bool
+ovsdb_cs_db_parse_update_rpc(struct ovsdb_cs_db *db,
+ const struct jsonrpc_msg *msg)
+{
+ if (msg->type != JSONRPC_NOTIFY) {
+ return false;
+ }
+
+ int version = (!strcmp(msg->method, "update") ? 1
+ : !strcmp(msg->method, "update2") ? 2
+ : !strcmp(msg->method, "update3") ? 3
+ : 0);
+ if (!version) {
+ return false;
+ }
+
+ struct json *params = msg->params;
+ int n = version == 3 ? 3 : 2;
+ if (params->type != JSON_ARRAY || params->array.n != n) {
+ struct ovsdb_error *error = ovsdb_syntax_error(
+ params, NULL, "%s must be an array with %u elements.",
+ msg->method, n);
+ log_parse_update_error(error);
+ return false;
+ }
+
+ if (!json_equal(params->array.elems[0], db->monitor_id)) {
+ return false;
+ }
+
+ if (version == 3) {
+ const char *last_id = json_string(params->array.elems[1]);
+ if (!uuid_from_string(&db->last_id, last_id)) {
+ struct ovsdb_error *error = ovsdb_syntax_error(
+ params, NULL, "Last-id %s is not in UUID format.", last_id);
+ log_parse_update_error(error);
+ return false;
+ }
+ }
+
+ struct json *table_updates = params->array.elems[version == 3 ? 2 : 1];
+ ovsdb_cs_db_add_update(db, table_updates, version, false);
+ return true;
+}
+
+static bool
+ovsdb_cs_handle_monitor_canceled(struct ovsdb_cs *cs,
+ struct ovsdb_cs_db *db,
+ const struct jsonrpc_msg *msg)
+{
+ if (msg->type != JSONRPC_NOTIFY
+ || strcmp(msg->method, "monitor_canceled")
+ || msg->params->type != JSON_ARRAY
+ || msg->params->array.n != 1
+ || !json_equal(msg->params->array.elems[0], db->monitor_id)) {
+ return false;
+ }
+
+ db->monitor_version = 0;
+
+ /* Cancel the other monitor and restart the FSM from the top.
+ *
+ * Maybe a more sophisticated response would be better in some cases, but
+ * it doesn't seem worth optimizing yet. (Although this is already more
+ * sophisticated than just dropping the connection and reconnecting.) */
+ struct ovsdb_cs_db *other_db
+ = db == &cs->data ? &cs->server : &cs->data;
+ if (other_db->monitor_version) {
+ jsonrpc_session_send(
+ cs->session,
+ jsonrpc_create_request(
+ "monitor_cancel",
+ json_array_create_1(json_clone(other_db->monitor_id)), NULL));
+ other_db->monitor_version = 0;
+ }
+ ovsdb_cs_restart_fsm(cs);
+
+ return true;
+}
+
+/* The _Server database.
+ *
+ * We replicate the Database table in the _Server database because this is the
+ * only way to find out properties we need to know for clustering, such as
+ * whether a database is clustered at all and whether this server is the
+ * leader.
+ *
+ * This code implements a kind of simple IDL-like layer. */
+
+struct server_column {
+ const char *name;
+ struct ovsdb_type type;
+};
+enum server_column_index {
+ COL_NAME,
+ COL_MODEL,
+ COL_CONNECTED,
+ COL_LEADER,
+ COL_SCHEMA,
+ COL_CID,
+ COL_INDEX,
+};
+#define OPTIONAL_COLUMN(TYPE) \
+ { \
+ .key = OVSDB_BASE_##TYPE##_INIT, \
+ .value = OVSDB_BASE_VOID_INIT, \
+ .n_min = 0, \
+ .n_max = 1 \
+ }
+static const struct server_column server_columns[] = {
+ [COL_NAME] = {"name", OPTIONAL_COLUMN(STRING) },
+ [COL_MODEL] = {"model", OPTIONAL_COLUMN(STRING) },
+ [COL_CONNECTED] = {"connected", OPTIONAL_COLUMN(BOOLEAN) },
+ [COL_LEADER] = {"leader", OPTIONAL_COLUMN(BOOLEAN) },
+ [COL_SCHEMA] = {"schema", OPTIONAL_COLUMN(STRING) },
+ [COL_CID] = {"cid", OPTIONAL_COLUMN(UUID) },
+ [COL_INDEX] = {"index", OPTIONAL_COLUMN(INTEGER) },
+};
+#define N_SERVER_COLUMNS ARRAY_SIZE(server_columns)
+struct server_row {
+ struct hmap_node hmap_node;
+ struct uuid uuid;
+ struct ovsdb_datum data[N_SERVER_COLUMNS];
+};
+
+static void
+server_row_destroy(struct server_row *row)
+{
+ if (row) {
+ for (size_t i = 0; i < N_SERVER_COLUMNS; i++) {
+ ovsdb_datum_destroy(&row->data[i], &server_columns[i].type);
+ }
+ free(row);
+ }
+}
+
+static struct server_row *
+ovsdb_cs_find_server_row(struct ovsdb_cs *cs, const struct uuid *uuid)
+{
+ struct server_row *row;
+ HMAP_FOR_EACH (row, hmap_node, &cs->server_rows) {
+ if (uuid_equals(uuid, &row->uuid)) {
+ return row;
+ }
+ }
+ return NULL;
+}
+
+static void
+ovsdb_cs_delete_server_row(struct ovsdb_cs *cs, struct server_row *row)
+{
+ hmap_remove(&cs->server_rows, &row->hmap_node);
+ server_row_destroy(row);
+}
+
+static struct server_row *
+ovsdb_cs_insert_server_row(struct ovsdb_cs *cs, const struct uuid *uuid)
+{
+ struct server_row *row = xmalloc(sizeof *row);
+ hmap_insert(&cs->server_rows, &row->hmap_node, uuid_hash(uuid));
+ row->uuid = *uuid;
+ for (size_t i = 0; i < N_SERVER_COLUMNS; i++) {
+ ovsdb_datum_init_default(&row->data[i], &server_columns[i].type);
+ }
+ return row;
+}
+
+static void
+ovsdb_cs_update_server_row(struct server_row *row,
+ const struct shash *update, bool xor)
+{
+ for (size_t i = 0; i < N_SERVER_COLUMNS; i++) {
+ const struct server_column *column = &server_columns[i];
+ struct shash_node *node = shash_find(update, column->name);
+ if (!node) {
+ continue;
+ }
+ const struct json *json = node->data;
+
+ struct ovsdb_datum *old = &row->data[i];
+ struct ovsdb_datum new;
+ if (!xor) {
+ struct ovsdb_error *error = ovsdb_datum_from_json(
+ &new, &column->type, json, NULL);
+ if (error) {
+ ovsdb_error_destroy(error);
+ continue;
+ }
+ } else {
+ struct ovsdb_datum diff;
+ struct ovsdb_error *error = ovsdb_transient_datum_from_json(
+ &diff, &column->type, json);
+ if (error) {
+ ovsdb_error_destroy(error);
+ continue;
+ }
+
+ error = ovsdb_datum_apply_diff(&new, old, &diff, &column->type);
+ if (error) {
+ ovsdb_error_destroy(error);
+ ovsdb_datum_destroy(&new, &column->type);
+ continue;
+ }
+ ovsdb_datum_destroy(&diff, &column->type);
+ }
+
+ ovsdb_datum_destroy(&row->data[i], &column->type);
+ row->data[i] = new;
+ }
+}
+
+static void
+ovsdb_cs_clear_server_rows(struct ovsdb_cs *cs)
+{
+ struct server_row *row, *next;
+ HMAP_FOR_EACH_SAFE (row, next, hmap_node, &cs->server_rows) {
+ ovsdb_cs_delete_server_row(cs, row);
+ }
+}
+
+static void log_parse_update_error(struct ovsdb_error *);
+
+static void
+ovsdb_cs_process_server_event(struct ovsdb_cs *cs,
+ const struct ovsdb_cs_event *event)
+{
+ ovs_assert(event->type == OVSDB_CS_EVENT_TYPE_UPDATE);
+
+ const struct ovsdb_cs_update_event *update = &event->update;
+ struct ovsdb_cs_db_update *du;
+ struct ovsdb_error *error = ovsdb_cs_parse_db_update(
+ update->table_updates, update->version, &du);
+ if (error) {
+ log_parse_update_error(error);
+ return;
+ }
+
+ if (update->clear) {
+ ovsdb_cs_clear_server_rows(cs);
+ }
+
+ const struct ovsdb_cs_table_update *tu = ovsdb_cs_db_update_find_table(
+ du, "Database");
+ if (tu) {
+ for (size_t i = 0; i < tu->n; i++) {
+ const struct ovsdb_cs_row_update *ru = &tu->row_updates[i];
+ struct server_row *row
+ = ovsdb_cs_find_server_row(cs, &ru->row_uuid);
+ if (ru->type == OVSDB_CS_ROW_DELETE) {
+ ovsdb_cs_delete_server_row(cs, row);
+ } else {
+ if (!row) {
+ row = ovsdb_cs_insert_server_row(cs, &ru->row_uuid);
+ }
+ ovsdb_cs_update_server_row(row, ru->columns,
+ ru->type == OVSDB_CS_ROW_XOR);
+ }
+ }
+ }
+
+ ovsdb_cs_db_update_destroy(du);
+}
+
+static const char *
+server_column_get_string(const struct server_row *row,
+ enum server_column_index index,
+ const char *default_value)
+{
+ ovs_assert(server_columns[index].type.key.type == OVSDB_TYPE_STRING);
+ const struct ovsdb_datum *d = &row->data[index];
+ return d->n == 1 ? d->keys[0].string : default_value;
+}
+
+static bool
+server_column_get_bool(const struct server_row *row,
+ enum server_column_index index,
+ bool default_value)
+{
+ ovs_assert(server_columns[index].type.key.type == OVSDB_TYPE_BOOLEAN);
+ const struct ovsdb_datum *d = &row->data[index];
+ return d->n == 1 ? d->keys[0].boolean : default_value;
+}
+
+static uint64_t
+server_column_get_int(const struct server_row *row,
+ enum server_column_index index,
+ uint64_t default_value)
+{
+ ovs_assert(server_columns[index].type.key.type == OVSDB_TYPE_INTEGER);
+ const struct ovsdb_datum *d = &row->data[index];
+ return d->n == 1 ? d->keys[0].integer : default_value;
+}
+
+static const struct uuid *
+server_column_get_uuid(const struct server_row *row,
+ enum server_column_index index,
+ const struct uuid *default_value)
+{
+ ovs_assert(server_columns[index].type.key.type == OVSDB_TYPE_UUID);
+ const struct ovsdb_datum *d = &row->data[index];
+ return d->n == 1 ? &d->keys[0].uuid : default_value;
+}
+
+static const struct server_row *
+ovsdb_find_server_row(struct ovsdb_cs *cs)
+{
+ const struct server_row *row;
+ HMAP_FOR_EACH (row, hmap_node, &cs->server_rows) {
+ const struct uuid *cid = server_column_get_uuid(row, COL_CID, NULL);
+ const char *name = server_column_get_string(row, COL_NAME, NULL);
+ if (uuid_is_zero(&cs->cid)
+ ? (name && !strcmp(cs->data.db_name, name))
+ : (cid && uuid_equals(cid, &cs->cid))) {
+ return row;
+ }
+ }
+ return NULL;
+}
+
+static void OVS_UNUSED
+ovsdb_log_server_rows(const struct ovsdb_cs *cs)
+{
+ int row_num = 0;
+ const struct server_row *row;
+ HMAP_FOR_EACH (row, hmap_node, &cs->server_rows) {
+ struct ds s = DS_EMPTY_INITIALIZER;
+ for (size_t i = 0; i < N_SERVER_COLUMNS; i++) {
+ ds_put_format(&s, " %s=", server_columns[i].name);
+ if (i == COL_SCHEMA) {
+ ds_put_format(&s, "...");
+ } else {
+ ovsdb_datum_to_string(&row->data[i], &server_columns[i].type,
+ &s);
+ }
+ }
+ VLOG_INFO("row %d:%s", row_num++, ds_cstr(&s));
+ ds_destroy(&s);
+ }
+}
+
+static bool
+ovsdb_cs_check_server_db__(struct ovsdb_cs *cs)
+{
+ struct ovsdb_cs_event *event;
+ LIST_FOR_EACH_POP (event, list_node, &cs->server.events) {
+ ovsdb_cs_process_server_event(cs, event);
+ ovsdb_cs_event_destroy(event);
+ }
+
+ const struct server_row *db_row = ovsdb_find_server_row(cs);
+ static struct vlog_rate_limit rl = VLOG_RATE_LIMIT_INIT(5, 5);
+ const char *server_name = jsonrpc_session_get_name(cs->session);
+ if (!db_row) {
+ VLOG_INFO_RL(&rl, "%s: server does not have %s database",
+ server_name, cs->data.db_name);
+ return false;
+ }
+
+ bool ok = false;
+ const char *model = server_column_get_string(db_row, COL_MODEL, "");
+ const char *schema = server_column_get_string(db_row, COL_SCHEMA, NULL);
+ if (!strcmp(model, "clustered")) {
+ bool connected = server_column_get_bool(db_row, COL_CONNECTED, false);
+ bool leader = server_column_get_bool(db_row, COL_LEADER, false);
+ uint64_t index = server_column_get_int(db_row, COL_INDEX, 0);
+
+ if (!schema) {
+ VLOG_INFO("%s: clustered database server has not yet joined "
+ "cluster; trying another server", server_name);
+ } else if (!connected) {
+ VLOG_INFO("%s: clustered database server is disconnected "
+ "from cluster; trying another server", server_name);
+ } else if (cs->leader_only && !leader) {
+ VLOG_INFO("%s: clustered database server is not cluster "
+ "leader; trying another server", server_name);
+ } else if (index < cs->min_index) {
+ VLOG_WARN("%s: clustered database server has stale data; "
+ "trying another server", server_name);
+ } else {
+ cs->min_index = index;
+ ok = true;
+ }
+ } else {
+ if (!schema) {
+ VLOG_INFO("%s: missing database schema", server_name);
+ } else {
+ ok = true;
+ }
+ }
+ if (!ok) {
+ return false;
+ }
+
+ if (cs->state == CS_S_SERVER_MONITOR_REQUESTED) {
+ json_destroy(cs->data.schema);
+ cs->data.schema = json_from_string(schema);
+ if (cs->data.max_version >= 3) {
+ ovsdb_cs_send_monitor_request(cs, &cs->data, 3);
+ ovsdb_cs_transition(cs, CS_S_DATA_MONITOR_COND_SINCE_REQUESTED);
+ } else if (cs->data.max_version >= 2) {
+ ovsdb_cs_send_monitor_request(cs, &cs->data, 2);
+ ovsdb_cs_transition(cs, CS_S_DATA_MONITOR_COND_REQUESTED);
+ } else {
+ ovsdb_cs_send_monitor_request(cs, &cs->data, 1);
+ ovsdb_cs_transition(cs, CS_S_DATA_MONITOR_REQUESTED);
+ }
+ }
+ return true;
+}
+
+static bool
+ovsdb_cs_check_server_db(struct ovsdb_cs *cs)
+{
+ bool ok = ovsdb_cs_check_server_db__(cs);
+ if (!ok) {
+ ovsdb_cs_retry(cs);
+ }
+ return ok;
+}
+
+static struct json *
+ovsdb_cs_compose_server_monitor_request(const struct json *schema_json,
+ void *cs_)
+{
+ struct ovsdb_cs *cs = cs_;
+ struct shash *schema = ovsdb_cs_parse_schema(schema_json);
+ struct json *monitor_requests = json_object_create();
+
+ const char *table_name = "Database";
+ const struct sset *table_schema
+ = schema ? shash_find_data(schema, table_name) : NULL;
+ if (!table_schema) {
+ VLOG_WARN("%s database lacks %s table "
+ "(database needs upgrade?)",
+ cs->server.db_name, table_name);
+ /* XXX return failure? */
+ } else {
+ struct json *columns = json_array_create_empty();
+ for (size_t j = 0; j < N_SERVER_COLUMNS; j++) {
+ const struct server_column *column = &server_columns[j];
+ bool db_has_column = (table_schema &&
+ sset_contains(table_schema, column->name));
+ if (table_schema && !db_has_column) {
+ VLOG_WARN("%s table in %s database lacks %s column "
+ "(database needs upgrade?)",
+ table_name, cs->server.db_name, column->name);
+ continue;
+ }
+ json_array_add(columns, json_string_create(column->name));
+ }
+
+ struct json *monitor_request = json_object_create();
+ json_object_put(monitor_request, "columns", columns);
+ json_object_put(monitor_requests, table_name,
+ json_array_create_1(monitor_request));
+ }
+ ovsdb_cs_free_schema(schema);
+
+ return monitor_requests;
+}
+
+static const struct ovsdb_cs_ops ovsdb_cs_server_ops = {
+ ovsdb_cs_compose_server_monitor_request
+};
static void
log_error(struct ovsdb_error *error)
@@ -324,3 +2263,17 @@ ovsdb_cs_db_update_destroy(struct ovsdb_cs_db_update *du)
free(du->table_updates);
free(du);
}
+
+const struct ovsdb_cs_table_update *
+ovsdb_cs_db_update_find_table(const struct ovsdb_cs_db_update *du,
+ const char *table_name)
+{
+ for (size_t i = 0; i < du->n; i++) {
+ const struct ovsdb_cs_table_update *tu = &du->table_updates[i];
+ if (!strcmp(tu->table_name, table_name)) {
+ return tu;
+ }
+ }
+ return NULL;
+}
+
diff --git a/lib/ovsdb-cs.h b/lib/ovsdb-cs.h
index e4af32fd70c9..bbe9588435d2 100644
--- a/lib/ovsdb-cs.h
+++ b/lib/ovsdb-cs.h
@@ -17,14 +17,141 @@
#ifndef OVSDB_CS_H
#define OVSDB_CS_H 1
-#include "openvswitch/compiler.h"
+/* Open vSwitch Database client synchronization layer.
+ *
+ * This is a base layer for maintaining an in-memory replica of a database. It
+ * issues RPC requests to an OVSDB database server and passes the semantically
+ * meaningful parts of the stream up to a higher layer. The OVSDB IDL uses
+ * this as a base layer, as well as OVN's DDlog-based northd implementation.
+ */
+
+#include <stdbool.h>
#include "openvswitch/hmap.h"
-#include "openvswitch/json.h"
+#include "openvswitch/list.h"
#include "openvswitch/shash.h"
#include "openvswitch/uuid.h"
-/* Open vSwitch Database client synchronization layer. */
+struct json;
+struct ovsdb_cs;
+
+struct ovsdb_cs_ops {
+ /* Returns <monitor-requests> to use for the specified <schema>. The
+ * implementation might find ovsdb_cs_parse_table_updates() to be a useful
+ * helper.
+ *
+ * The caller might actually use "monitor_cond" or "monitor_cond_since",
+ * rather than plain "monitor". If so, this function's implementation
+ * doesn't need to worry about that, because the caller will add the
+ * conditions itself. */
+ struct json *(*compose_monitor_requests)(const struct json *schema,
+ void *aux);
+};
+
+/* An event is a happening that is worth reporting to the CS client.
+ *
+ * Currently there are three kinds of events:
+ *
+ * - "Reconnect": The connection to the database was lost and it is now
+ * being reconnected. This means that any transactions submitted by the
+ * client will never receive a reply (although it's possible that some of
+ * them were actually committed). This event has no associated data.
+ *
+ * - "Locked": The server granted the lock we requested.
+ *
+ * - "Update": The server sent an update to one or more monitored tables.
+ * The client can use the associated data to update its idea of the
+ * snapshot.
+ *
+ * - "Transaction reply": The server sent a reply to a transaction sent by
+ * the client using ovsdb_cs_send_transaction().
+ */
+struct ovsdb_cs_event {
+ struct ovs_list list_node;
+
+ enum ovsdb_cs_event_type {
+ OVSDB_CS_EVENT_TYPE_RECONNECT, /* Connection lost. */
+ OVSDB_CS_EVENT_TYPE_LOCKED, /* Got the lock we wanted. */
+ OVSDB_CS_EVENT_TYPE_UPDATE, /* Received update notification. */
+ OVSDB_CS_EVENT_TYPE_TXN_REPLY, /* Received reply to transaction. */
+ } type;
+
+ union {
+ /* Represents a <table-updates> or <table-updates2> that contains
+ * either the initial data in a monitor reply or a delta received in an
+ * update notification. The client can use this to update its database
+ * replica.
+ *
+ * If 'clear' is true, then the client should first clear its idea of
+ * what's in the replica before applying the update; otherwise, it's an
+ * incremental update.
+ *
+ * 'table-updates' is a <table-updates> if 'version' if 1, otherwise a
+ * <table-updates2>. The client can use ovsdb_cs_parse_table_updates()
+ * to parse the update.
+ */
+ struct ovsdb_cs_update_event {
+ bool clear;
+ struct json *table_updates;
+ int version;
+ } update;
+ /* The "result" member from a transaction reply. The transaction is
+ * one sent by the client using ovsdb_cs_send_transaction(). The
+ * client can match 'txn_reply->id' against the ID in a transaction it
+ * sent. */
+ struct jsonrpc_msg *txn_reply;
+ };
+};
+void ovsdb_cs_event_destroy(struct ovsdb_cs_event *);
+
+/* Lifecycle. */
+struct ovsdb_cs *ovsdb_cs_create(const char *database, int max_version,
+ const struct ovsdb_cs_ops *ops,
+ void *ops_aux);
+void ovsdb_cs_destroy(struct ovsdb_cs *);
+
+void ovsdb_cs_run(struct ovsdb_cs *, struct ovs_list *events);
+void ovsdb_cs_wait(struct ovsdb_cs *);
+
+/* Network connection. */
+void ovsdb_cs_set_remote(struct ovsdb_cs *, const char *remote, bool retry);
+
+void ovsdb_cs_enable_reconnect(struct ovsdb_cs *);
+void ovsdb_cs_force_reconnect(struct ovsdb_cs *);
+void ovsdb_cs_flag_inconsistency(struct ovsdb_cs *);
+
+bool ovsdb_cs_is_alive(const struct ovsdb_cs *);
+bool ovsdb_cs_is_connected(const struct ovsdb_cs *);
+int ovsdb_cs_get_last_error(const struct ovsdb_cs *);
+
+void ovsdb_cs_set_probe_interval(const struct ovsdb_cs *, int probe_interval);
+
+/* Conditional monitoring (specifying that only rows matching particular
+ * criteria should be monitored).
+ *
+ * Some database servers don't support conditional monitoring; in that case,
+ * the client will get all the rows. */
+unsigned int ovsdb_cs_set_condition(struct ovsdb_cs *, const char *table,
+ const struct json *condition);
+unsigned int ovsdb_cs_get_condition_seqno(const struct ovsdb_cs *);
+
+/* Clustered servers. */
+void ovsdb_cs_set_leader_only(struct ovsdb_cs *, bool leader_only);
+void ovsdb_cs_set_shuffle_remotes(struct ovsdb_cs *, bool shuffle);
+void ovsdb_cs_reset_min_index(struct ovsdb_cs *);
+
+/* Database locks. */
+void ovsdb_cs_set_lock(struct ovsdb_cs *, const char *lock_name);
+const char *ovsdb_cs_get_lock(const struct ovsdb_cs *);
+bool ovsdb_cs_has_lock(const struct ovsdb_cs *);
+bool ovsdb_cs_is_lock_contended(const struct ovsdb_cs *);
+
+/* Transactions. */
+bool ovsdb_cs_may_send_transaction(const struct ovsdb_cs *);
+struct json *ovsdb_cs_send_transaction(struct ovsdb_cs *, struct json *ops)
+ OVS_WARN_UNUSED_RESULT;
+bool ovsdb_cs_forget_transaction(struct ovsdb_cs *, const struct json *);
+
/* Helper for partially parsing the <table-updates> or <table-updates2> that
* appear in struct ovsdb_cs_update_event. The helper leaves the data in JSON
* format, so it doesn't need to know column types. */
@@ -61,6 +188,8 @@ struct ovsdb_error *ovsdb_cs_parse_db_update(
struct ovsdb_cs_db_update **db_updatep)
OVS_WARN_UNUSED_RESULT;
void ovsdb_cs_db_update_destroy(struct ovsdb_cs_db_update *);
+const struct ovsdb_cs_table_update *ovsdb_cs_db_update_find_table(
+ const struct ovsdb_cs_db_update *, const char *table_name);
/* Simple parsing of OVSDB schemas for use by ovsdb_cs clients. */
diff --git a/lib/ovsdb-idl-provider.h b/lib/ovsdb-idl-provider.h
index 00497d940c32..0f38f9b34fdc 100644
--- a/lib/ovsdb-idl-provider.h
+++ b/lib/ovsdb-idl-provider.h
@@ -118,16 +118,10 @@ struct ovsdb_idl_table {
* for replication. */
struct shash columns; /* Contains "const struct ovsdb_idl_column *"s. */
struct hmap rows; /* Contains "struct ovsdb_idl_row"s. */
- struct ovsdb_idl_db *db; /* Containing db. */
+ struct ovsdb_idl *idl; /* Containing IDL instance. */
unsigned int change_seqno[OVSDB_IDL_CHANGE_MAX];
struct ovs_list indexes; /* Contains "struct ovsdb_idl_index"s */
struct ovs_list track_list; /* Tracked rows (ovsdb_idl_row.track_node). */
- struct ovsdb_idl_condition *ack_cond; /* Last condition acked by the
- * server. */
- struct ovsdb_idl_condition *req_cond; /* Last condition requested to the
- * server. */
- struct ovsdb_idl_condition *new_cond; /* Latest condition set by the IDL
- * client. */
};
struct ovsdb_idl_class {
diff --git a/lib/ovsdb-idl.c b/lib/ovsdb-idl.c
index fdc59c5e7142..61be78da33f2 100644
--- a/lib/ovsdb-idl.c
+++ b/lib/ovsdb-idl.c
@@ -83,204 +83,23 @@ struct ovsdb_idl_arc {
struct ovsdb_idl_row *dst; /* Destination row. */
};
-/* Connection state machine.
- *
- * When a JSON-RPC session connects, the IDL sends a "monitor_cond" request for
- * the Database table in the _Server database and transitions to the
- * IDL_S_SERVER_MONITOR_COND_REQUESTED state. If the session drops and
- * reconnects, or if the IDL receives a "monitor_canceled" notification for a
- * table it is monitoring, the IDL starts over again in the same way. */
-#define OVSDB_IDL_STATES \
- /* Waits for "get_schema" reply, then sends "monitor_cond" \
- * request for the Database table in the _Server database, whose \
- * details are informed by the schema, and transitions to \
- * IDL_S_SERVER_MONITOR_COND_REQUESTED. */ \
- OVSDB_IDL_STATE(SERVER_SCHEMA_REQUESTED) \
- \
- /* Waits for "monitor_cond" reply for the Database table: \
- * \
- * - If the reply indicates success, and the Database table has a \
- * row for the IDL database: \
- * \
- * * If the row indicates that this is a clustered database \
- * that is not connected to the cluster, closes the \
- * connection. The next connection attempt has a chance at \
- * picking a connected server. \
- * \
- * * Otherwise, sends a "monitor_cond_since" request for the IDL \
- * database whose details are informed by the schema \
- * (obtained from the row), and transitions to \
- * IDL_S_DATA_MONITOR_COND_SINCE_REQUESTED. \
- * \
- * - If the reply indicates success, but the Database table does \
- * not have a row for the IDL database, transitions to \
- * IDL_S_ERROR. \
- * \
- * - If the reply indicates failure, sends a "get_schema" request \
- * for the IDL database and transitions to \
- * IDL_S_DATA_SCHEMA_REQUESTED. */ \
- OVSDB_IDL_STATE(SERVER_MONITOR_COND_REQUESTED) \
- \
- /* Waits for "get_schema" reply, then sends "monitor_cond" \
- * request whose details are informed by the schema, and \
- * transitions to IDL_S_DATA_MONITOR_COND_REQUESTED. */ \
- OVSDB_IDL_STATE(DATA_SCHEMA_REQUESTED) \
- \
- /* Waits for "monitor_cond_since" reply. If successful, replaces \
- * the IDL contents by the data carried in the reply and \
- * transitions to IDL_S_MONITORING. On failure, sends a \
- * "monitor_cond" request and transitions to \
- * IDL_S_DATA_MONITOR_COND_REQUESTED. */ \
- OVSDB_IDL_STATE(DATA_MONITOR_COND_SINCE_REQUESTED) \
- \
- /* Waits for "monitor_cond" reply. If successful, replaces the \
- * IDL contents by the data carried in the reply and transitions \
- * to IDL_S_MONITORING. On failure, sends a "monitor" request \
- * and transitions to IDL_S_DATA_MONITOR_REQUESTED. */ \
- OVSDB_IDL_STATE(DATA_MONITOR_COND_REQUESTED) \
- \
- /* Waits for "monitor" reply. If successful, replaces the IDL \
- * contents by the data carried in the reply and transitions to \
- * IDL_S_MONITORING. On failure, transitions to IDL_S_ERROR. */ \
- OVSDB_IDL_STATE(DATA_MONITOR_REQUESTED) \
- \
- /* State that processes "update", "update2" or "update3" \
- * notifications for the main database (and the Database table \
- * in _Server if available). \
- * \
- * If we're monitoring the Database table and we get notified \
- * that the IDL database has been deleted, we close the \
- * connection (which will restart the state machine). */ \
- OVSDB_IDL_STATE(MONITORING) \
- \
- /* Terminal error state that indicates that nothing useful can be \
- * done, for example because the database server doesn't actually \
- * have the desired database. We maintain the session with the \
- * database server anyway. If it starts serving the database \
- * that we want, or if someone fixes and restarts the database, \
- * then it will kill the session and we will automatically \
- * reconnect and try again. */ \
- OVSDB_IDL_STATE(ERROR) \
- \
- /* Terminal state that indicates we connected to a useless server \
- * in a cluster, e.g. one that is partitioned from the rest of \
- * the cluster. We're waiting to retry. */ \
- OVSDB_IDL_STATE(RETRY)
-
-enum ovsdb_idl_state {
-#define OVSDB_IDL_STATE(NAME) IDL_S_##NAME,
- OVSDB_IDL_STATES
-#undef OVSDB_IDL_STATE
-};
-
-static const char *ovsdb_idl_state_to_string(enum ovsdb_idl_state);
-
-enum ovsdb_idl_monitor_method {
- OVSDB_IDL_MM_MONITOR,
- OVSDB_IDL_MM_MONITOR_COND,
- OVSDB_IDL_MM_MONITOR_COND_SINCE
-};
-
-enum ovsdb_idl_monitoring {
- OVSDB_IDL_NOT_MONITORING, /* Database is not being monitored. */
- OVSDB_IDL_MONITORING, /* Database has "monitor" outstanding. */
- OVSDB_IDL_MONITORING_COND, /* Database has "monitor_cond" outstanding. */
- OVSDB_IDL_MONITORING_COND_SINCE, /* Database has "monitor_cond_since"
- outstanding. */
-};
-
-struct ovsdb_idl_db {
- struct ovsdb_idl *idl;
-
- /* Data. */
+struct ovsdb_idl {
+ struct ovsdb_cs *cs;
const struct ovsdb_idl_class *class_;
struct shash table_by_name; /* Contains "struct ovsdb_idl_table *"s.*/
struct ovsdb_idl_table *tables; /* Array of ->class_->n_tables elements. */
- struct json *monitor_id;
unsigned int change_seqno;
struct ovsdb_idl_txn *txn;
struct hmap outstanding_txns;
bool verify_write_only;
- struct json *schema;
- enum ovsdb_idl_monitoring monitoring;
-
- /* True if any of the tables' monitoring conditions has changed. */
- bool cond_changed;
-
- unsigned int cond_seqno; /* Keep track of condition clauses changes
- over a single conditional monitoring session.
- Reverts to zero when idl session
- reconnects. */
-
- /* Database locking. */
- char *lock_name; /* Name of lock we need, NULL if none. */
- bool has_lock; /* Has db server told us we have the lock? */
- bool is_lock_contended; /* Has db server told us we can't get lock? */
- struct json *lock_request_id; /* JSON-RPC ID of in-flight lock request. */
-
- /* Last db txn id, used for fast resync through monitor_cond_since */
- struct uuid last_id;
-};
-
-static void ovsdb_idl_db_track_clear(struct ovsdb_idl_db *);
-static void ovsdb_idl_db_add_column(struct ovsdb_idl_db *,
- const struct ovsdb_idl_column *);
-static void ovsdb_idl_db_omit(struct ovsdb_idl_db *,
- const struct ovsdb_idl_column *);
-static void ovsdb_idl_db_omit_alert(struct ovsdb_idl_db *,
- const struct ovsdb_idl_column *);
-static unsigned int ovsdb_idl_db_set_condition(
- struct ovsdb_idl_db *, const struct ovsdb_idl_table_class *,
- const struct ovsdb_idl_condition *);
-
-static void ovsdb_idl_send_schema_request(struct ovsdb_idl *,
- struct ovsdb_idl_db *);
-static void ovsdb_idl_send_db_change_aware(struct ovsdb_idl *);
-static bool ovsdb_idl_check_server_db(struct ovsdb_idl *);
-static void ovsdb_idl_send_monitor_request(struct ovsdb_idl *,
- struct ovsdb_idl_db *,
- enum ovsdb_idl_monitor_method);
-static void ovsdb_idl_db_clear(struct ovsdb_idl_db *db);
-static void ovsdb_idl_db_ack_condition(struct ovsdb_idl_db *db);
-static void ovsdb_idl_db_sync_condition(struct ovsdb_idl_db *db);
-static void ovsdb_idl_condition_move(struct ovsdb_idl_condition **dst,
- struct ovsdb_idl_condition **src);
-
-struct ovsdb_idl {
- struct ovsdb_idl_db server;
- struct ovsdb_idl_db data;
-
- /* Session state.
- *
- *'state_seqno' is a snapshot of the session's sequence number as returned
- * jsonrpc_session_get_seqno(session), so if it differs from the value that
- * function currently returns then the session has reconnected and the
- * state machine must restart. */
- struct jsonrpc_session *session; /* Connection to the server. */
- char *remote; /* 'session' remote name. */
- enum ovsdb_idl_state state; /* Current session state. */
- unsigned int state_seqno; /* See above. */
- struct json *request_id; /* JSON ID for request awaiting reply. */
-
- struct uuid cid;
-
- uint64_t min_index;
- bool leader_only;
- bool shuffle_remotes;
};
-static void ovsdb_idl_transition_at(struct ovsdb_idl *, enum ovsdb_idl_state,
- const char *where);
-#define ovsdb_idl_transition(IDL, STATE) \
- ovsdb_idl_transition_at(IDL, STATE, OVS_SOURCE_LOCATOR)
-
-static void ovsdb_idl_retry_at(struct ovsdb_idl *, const char *where);
-#define ovsdb_idl_retry(IDL) ovsdb_idl_retry_at(IDL, OVS_SOURCE_LOCATOR)
+static struct ovsdb_cs_ops ovsdb_idl_cs_ops;
struct ovsdb_idl_txn {
struct hmap_node hmap_node;
struct json *request_id;
- struct ovsdb_idl_db *db;
+ struct ovsdb_idl *idl;
struct hmap txn_rows;
enum ovsdb_idl_txn_status status;
char *error;
@@ -310,29 +129,24 @@ static struct vlog_rate_limit syntax_rl = VLOG_RATE_LIMIT_INIT(1, 5);
static struct vlog_rate_limit semantic_rl = VLOG_RATE_LIMIT_INIT(1, 5);
static struct vlog_rate_limit other_rl = VLOG_RATE_LIMIT_INIT(1, 5);
-static void ovsdb_idl_clear(struct ovsdb_idl *);
-static void ovsdb_idl_db_parse_monitor_reply(struct ovsdb_idl_db *,
- const struct json *result,
- int version);
-static bool ovsdb_idl_db_parse_update_rpc(struct ovsdb_idl_db *,
- const struct jsonrpc_msg *);
-static bool ovsdb_idl_handle_monitor_canceled(struct ovsdb_idl *,
- struct ovsdb_idl_db *,
- const struct jsonrpc_msg *);
-static void ovsdb_idl_db_parse_update(struct ovsdb_idl_db *,
- const struct json *table_updates,
- int version);
enum update_result {
OVSDB_IDL_UPDATE_DB_CHANGED,
OVSDB_IDL_UPDATE_NO_CHANGES,
OVSDB_IDL_UPDATE_INCONSISTENT,
};
+static void ovsdb_idl_clear(struct ovsdb_idl *);
static enum update_result ovsdb_idl_process_update(
struct ovsdb_idl_table *, const struct ovsdb_cs_row_update *);
-static void ovsdb_idl_insert_row(struct ovsdb_idl_row *, const struct shash *);
+static void ovsdb_idl_insert_row(struct ovsdb_idl_row *,
+ const struct shash *values);
static void ovsdb_idl_delete_row(struct ovsdb_idl_row *);
-static bool ovsdb_idl_modify_row(struct ovsdb_idl_row *, const struct shash *,
- bool xor);
+static bool ovsdb_idl_modify_row(struct ovsdb_idl_row *,
+ const struct shash *values, bool xor);
+static void ovsdb_idl_parse_update(struct ovsdb_idl *,
+ const struct ovsdb_cs_update_event *);
+
+static void ovsdb_idl_txn_process_reply(struct ovsdb_idl *,
+ const struct jsonrpc_msg *);
static bool ovsdb_idl_row_is_orphan(const struct ovsdb_idl_row *);
static struct ovsdb_idl_row *ovsdb_idl_row_create__(
@@ -340,7 +154,7 @@ static struct ovsdb_idl_row *ovsdb_idl_row_create__(
static struct ovsdb_idl_row *ovsdb_idl_row_create(struct ovsdb_idl_table *,
const struct uuid *);
static void ovsdb_idl_row_destroy(struct ovsdb_idl_row *);
-static void ovsdb_idl_row_destroy_postprocess(struct ovsdb_idl_db *);
+static void ovsdb_idl_row_destroy_postprocess(struct ovsdb_idl *);
static void ovsdb_idl_destroy_all_map_op_lists(struct ovsdb_idl_row *);
static void ovsdb_idl_destroy_all_set_op_lists(struct ovsdb_idl_row *);
@@ -350,10 +164,7 @@ static void ovsdb_idl_row_clear_old(struct ovsdb_idl_row *);
static void ovsdb_idl_row_clear_new(struct ovsdb_idl_row *);
static void ovsdb_idl_row_clear_arcs(struct ovsdb_idl_row *, bool destroy_dsts);
-static void ovsdb_idl_db_txn_abort_all(struct ovsdb_idl_db *);
static void ovsdb_idl_txn_abort_all(struct ovsdb_idl *);
-static bool ovsdb_idl_db_txn_process_reply(struct ovsdb_idl_db *,
- const struct jsonrpc_msg *msg);
static bool ovsdb_idl_txn_extract_mutations(struct ovsdb_idl_row *,
struct json *);
static void ovsdb_idl_txn_add_map_op(struct ovsdb_idl_row *,
@@ -365,25 +176,13 @@ static void ovsdb_idl_txn_add_set_op(struct ovsdb_idl_row *,
struct ovsdb_datum *,
enum set_op_type);
-static bool ovsdb_idl_db_process_lock_replies(struct ovsdb_idl_db *,
- const struct jsonrpc_msg *);
-static struct jsonrpc_msg *ovsdb_idl_db_compose_lock_request(
- struct ovsdb_idl_db *);
-static struct jsonrpc_msg *ovsdb_idl_db_compose_unlock_request(
- struct ovsdb_idl_db *);
-static void ovsdb_idl_db_parse_lock_reply(struct ovsdb_idl_db *,
- const struct json *);
-static bool ovsdb_idl_db_parse_lock_notify(struct ovsdb_idl_db *,
- const struct json *params,
- bool new_has_lock);
static struct ovsdb_idl_table *
-ovsdb_idl_db_table_from_class(const struct ovsdb_idl_db *,
+ovsdb_idl_table_from_class(const struct ovsdb_idl *,
const struct ovsdb_idl_table_class *);
static struct ovsdb_idl_table *
ovsdb_idl_table_from_class(const struct ovsdb_idl *,
const struct ovsdb_idl_table_class *);
static bool ovsdb_idl_track_is_set(struct ovsdb_idl_table *table);
-static void ovsdb_idl_send_cond_change(struct ovsdb_idl *idl);
static void ovsdb_idl_destroy_indexes(struct ovsdb_idl_table *);
static void ovsdb_idl_add_to_indexes(const struct ovsdb_idl_row *);
@@ -391,53 +190,6 @@ static void ovsdb_idl_remove_from_indexes(const struct ovsdb_idl_row *);
static int ovsdb_idl_try_commit_loop_txn(struct ovsdb_idl_loop *loop,
bool *may_need_wakeup);
-static void
-ovsdb_idl_db_init(struct ovsdb_idl_db *db, const struct ovsdb_idl_class *class,
- struct ovsdb_idl *parent, bool monitor_everything_by_default)
-{
- memset(db, 0, sizeof *db);
-
- uint8_t default_mode = (monitor_everything_by_default
- ? OVSDB_IDL_MONITOR | OVSDB_IDL_ALERT
- : 0);
-
- db->idl = parent;
- db->class_ = class;
- shash_init(&db->table_by_name);
- db->tables = xmalloc(class->n_tables * sizeof *db->tables);
- for (size_t i = 0; i < class->n_tables; i++) {
- const struct ovsdb_idl_table_class *tc = &class->tables[i];
- struct ovsdb_idl_table *table = &db->tables[i];
-
- shash_add_assert(&db->table_by_name, tc->name, table);
- table->class_ = tc;
- table->modes = xmalloc(tc->n_columns);
- memset(table->modes, default_mode, tc->n_columns);
- table->need_table = false;
- shash_init(&table->columns);
- ovs_list_init(&table->indexes);
- for (size_t j = 0; j < tc->n_columns; j++) {
- const struct ovsdb_idl_column *column = &tc->columns[j];
-
- shash_add_assert(&table->columns, column->name, column);
- }
- hmap_init(&table->rows);
- ovs_list_init(&table->track_list);
- table->change_seqno[OVSDB_IDL_CHANGE_INSERT]
- = table->change_seqno[OVSDB_IDL_CHANGE_MODIFY]
- = table->change_seqno[OVSDB_IDL_CHANGE_DELETE] = 0;
- table->db = db;
- table->ack_cond = NULL;
- table->req_cond = NULL;
- table->new_cond = xmalloc(sizeof *table->new_cond);
- ovsdb_idl_condition_init(table->new_cond);
- ovsdb_idl_condition_add_clause_true(table->new_cond);
- }
- db->monitor_id = json_array_create_2(json_string_create("monid"),
- json_string_create(class->database));
- hmap_init(&db->outstanding_txns);
-}
-
/* Creates and returns a connection to database 'remote', which should be in a
* form acceptable to jsonrpc_session_open(). The connection will maintain an
* in-memory replica of the remote database whose schema is described by
@@ -486,29 +238,44 @@ struct ovsdb_idl *
ovsdb_idl_create_unconnected(const struct ovsdb_idl_class *class,
bool monitor_everything_by_default)
{
- struct ovsdb_idl *idl;
+ struct ovsdb_idl *idl = xmalloc(sizeof *idl);
+ *idl = (struct ovsdb_idl) {
+ .cs = ovsdb_cs_create(class->database, 3, &ovsdb_idl_cs_ops, idl),
+ .class_ = class,
+ .table_by_name = SHASH_INITIALIZER(&idl->table_by_name),
+ .tables = xmalloc(class->n_tables * sizeof *idl->tables),
+ .change_seqno = 0,
+ .txn = NULL,
+ .outstanding_txns = HMAP_INITIALIZER(&idl->outstanding_txns),
+ .verify_write_only = false,
+ };
+
+ uint8_t default_mode = (monitor_everything_by_default
+ ? OVSDB_IDL_MONITOR | OVSDB_IDL_ALERT
+ : 0);
+ for (size_t i = 0; i < class->n_tables; i++) {
+ const struct ovsdb_idl_table_class *tc = &class->tables[i];
+ struct ovsdb_idl_table *table = &idl->tables[i];
- idl = xzalloc(sizeof *idl);
- ovsdb_idl_db_init(&idl->server, &serverrec_idl_class, idl, true);
- ovsdb_idl_db_init(&idl->data, class, idl, monitor_everything_by_default);
- idl->state_seqno = UINT_MAX;
- idl->request_id = NULL;
- idl->leader_only = true;
- idl->shuffle_remotes = true;
+ shash_add_assert(&idl->table_by_name, tc->name, table);
+ table->class_ = tc;
+ table->modes = xmalloc(tc->n_columns);
+ memset(table->modes, default_mode, tc->n_columns);
+ table->need_table = false;
+ shash_init(&table->columns);
+ ovs_list_init(&table->indexes);
+ for (size_t j = 0; j < tc->n_columns; j++) {
+ const struct ovsdb_idl_column *column = &tc->columns[j];
- /* Monitor the Database table in the _Server database.
- *
- * We monitor only the row for 'class', or the row that has the
- * desired 'cid'. */
- struct ovsdb_idl_condition cond;
- ovsdb_idl_condition_init(&cond);
- if (!uuid_is_zero(&idl->cid)) {
- serverrec_database_add_clause_cid(&cond, OVSDB_F_EQ, &idl->cid, 1);
- } else {
- serverrec_database_add_clause_name(&cond, OVSDB_F_EQ, class->database);
+ shash_add_assert(&table->columns, column->name, column);
+ }
+ hmap_init(&table->rows);
+ ovs_list_init(&table->track_list);
+ table->change_seqno[OVSDB_IDL_CHANGE_INSERT]
+ = table->change_seqno[OVSDB_IDL_CHANGE_MODIFY]
+ = table->change_seqno[OVSDB_IDL_CHANGE_DELETE] = 0;
+ table->idl = idl;
}
- ovsdb_idl_db_set_condition(&idl->server, &serverrec_table_database, &cond);
- ovsdb_idl_condition_destroy(&cond);
return idl;
}
@@ -520,35 +287,7 @@ ovsdb_idl_create_unconnected(const struct ovsdb_idl_class *class,
void
ovsdb_idl_set_remote(struct ovsdb_idl *idl, const char *remote, bool retry)
{
- if (idl
- && ((remote != NULL) != (idl->remote != NULL)
- || (remote && idl->remote && strcmp(remote, idl->remote)))) {
- ovs_assert(!idl->data.txn);
-
- /* Close the old session, if any. */
- if (idl->session) {
- jsonrpc_session_close(idl->session);
- idl->session = NULL;
-
- free(idl->remote);
- idl->remote = NULL;
- }
-
- /* Open new session, if any. */
- if (remote) {
- struct svec remotes = SVEC_EMPTY_INITIALIZER;
- ovsdb_session_parse_remote(remote, &remotes, &idl->cid);
- if (idl->shuffle_remotes) {
- svec_shuffle(&remotes);
- }
- idl->session = jsonrpc_session_open_multiple(&remotes, retry);
- svec_destroy(&remotes);
-
- idl->state_seqno = UINT_MAX;
-
- idl->remote = xstrdup(remote);
- }
- }
+ ovsdb_cs_set_remote(idl->cs, remote, retry);
}
/* Set whether the order of remotes should be shuffled, when there
@@ -557,7 +296,7 @@ ovsdb_idl_set_remote(struct ovsdb_idl *idl, const char *remote, bool retry)
void
ovsdb_idl_set_shuffle_remotes(struct ovsdb_idl *idl, bool shuffle)
{
- idl->shuffle_remotes = shuffle;
+ ovsdb_cs_set_shuffle_remotes(idl->cs, shuffle);
}
/* Reset min_index to 0. This prevents a situation where the client
@@ -567,33 +306,7 @@ ovsdb_idl_set_shuffle_remotes(struct ovsdb_idl *idl, bool shuffle)
void
ovsdb_idl_reset_min_index(struct ovsdb_idl *idl)
{
- idl->min_index = 0;
-}
-
-static void
-ovsdb_idl_db_destroy(struct ovsdb_idl_db *db)
-{
- struct ovsdb_idl_condition *null_cond = NULL;
- ovs_assert(!db->txn);
- ovsdb_idl_db_txn_abort_all(db);
- ovsdb_idl_db_clear(db);
- for (size_t i = 0; i < db->class_->n_tables; i++) {
- struct ovsdb_idl_table *table = &db->tables[i];
- ovsdb_idl_condition_move(&table->ack_cond, &null_cond);
- ovsdb_idl_condition_move(&table->req_cond, &null_cond);
- ovsdb_idl_condition_move(&table->new_cond, &null_cond);
- ovsdb_idl_destroy_indexes(table);
- shash_destroy(&table->columns);
- hmap_destroy(&table->rows);
- free(table->modes);
- }
- shash_destroy(&db->table_by_name);
- free(db->tables);
- json_destroy(db->schema);
- hmap_destroy(&db->outstanding_txns);
- free(db->lock_name);
- json_destroy(db->lock_request_id);
- json_destroy(db->monitor_id);
+ ovsdb_cs_reset_min_index(idl->cs);
}
/* Destroys 'idl' and all of the data structures that it manages. */
@@ -601,12 +314,22 @@ void
ovsdb_idl_destroy(struct ovsdb_idl *idl)
{
if (idl) {
+ ovs_assert(!idl->txn);
+
+ ovsdb_idl_txn_abort_all(idl);
+ hmap_destroy(&idl->outstanding_txns);
+
ovsdb_idl_clear(idl);
- jsonrpc_session_close(idl->session);
- ovsdb_idl_db_destroy(&idl->server);
- ovsdb_idl_db_destroy(&idl->data);
- json_destroy(idl->request_id);
- free(idl->remote);
+ ovsdb_cs_destroy(idl->cs);
+ for (size_t i = 0; i < idl->class_->n_tables; i++) {
+ struct ovsdb_idl_table *table = &idl->tables[i];
+ ovsdb_idl_destroy_indexes(table);
+ shash_destroy(&table->columns);
+ hmap_destroy(&table->rows);
+ free(table->modes);
+ }
+ shash_destroy(&idl->table_by_name);
+ free(idl->tables);
free(idl);
}
}
@@ -622,19 +345,15 @@ ovsdb_idl_destroy(struct ovsdb_idl *idl)
void
ovsdb_idl_set_leader_only(struct ovsdb_idl *idl, bool leader_only)
{
- idl->leader_only = leader_only;
- if (leader_only && idl->server.monitoring) {
- ovsdb_idl_check_server_db(idl);
- }
+ ovsdb_cs_set_leader_only(idl->cs, leader_only);
}
static void
-ovsdb_idl_db_clear(struct ovsdb_idl_db *db)
+ovsdb_idl_clear(struct ovsdb_idl *db)
{
bool changed = false;
- size_t i;
- for (i = 0; i < db->class_->n_tables; i++) {
+ for (size_t i = 0; i < db->class_->n_tables; i++) {
struct ovsdb_idl_table *table = &db->tables[i];
struct ovsdb_idl_row *row, *next_row;
@@ -661,287 +380,49 @@ ovsdb_idl_db_clear(struct ovsdb_idl_db *db)
}
ovsdb_idl_row_destroy_postprocess(db);
- db->cond_seqno = 0;
- ovsdb_idl_db_track_clear(db);
+ ovsdb_idl_track_clear(db);
if (changed) {
db->change_seqno++;
}
}
-static const char *
-ovsdb_idl_state_to_string(enum ovsdb_idl_state state)
-{
- switch (state) {
-#define OVSDB_IDL_STATE(NAME) case IDL_S_##NAME: return #NAME;
- OVSDB_IDL_STATES
-#undef OVSDB_IDL_STATE
- default: return "<unknown>";
- }
-}
-
-static void
-ovsdb_idl_retry_at(struct ovsdb_idl *idl, const char *where)
-{
- ovsdb_idl_force_reconnect(idl);
- ovsdb_idl_transition_at(idl, IDL_S_RETRY, where);
-}
-
-static void
-ovsdb_idl_transition_at(struct ovsdb_idl *idl, enum ovsdb_idl_state new_state,
- const char *where)
-{
- VLOG_DBG("%s: %s -> %s at %s",
- idl->session ? jsonrpc_session_get_name(idl->session) : "void",
- ovsdb_idl_state_to_string(idl->state),
- ovsdb_idl_state_to_string(new_state),
- where);
- idl->state = new_state;
-}
-
-static void
-ovsdb_idl_clear(struct ovsdb_idl *idl)
-{
- ovsdb_idl_db_clear(&idl->data);
-}
-
-static void
-ovsdb_idl_send_request(struct ovsdb_idl *idl, struct jsonrpc_msg *request)
-{
- json_destroy(idl->request_id);
- idl->request_id = json_clone(request->id);
- if (idl->session) {
- jsonrpc_session_send(idl->session, request);
- } else {
- jsonrpc_msg_destroy(request);
- }
-}
-
-static void
-ovsdb_idl_restart_fsm(struct ovsdb_idl *idl)
-{
- /* Resync data DB table conditions to avoid missing updates due to
- * conditions that were in flight or changed locally while the connection
- * was down.
- */
- ovsdb_idl_db_sync_condition(&idl->data);
-
- ovsdb_idl_send_schema_request(idl, &idl->server);
- ovsdb_idl_transition(idl, IDL_S_SERVER_SCHEMA_REQUESTED);
- idl->data.monitoring = OVSDB_IDL_NOT_MONITORING;
- idl->server.monitoring = OVSDB_IDL_NOT_MONITORING;
-}
-
-static void
-ovsdb_idl_process_response(struct ovsdb_idl *idl, struct jsonrpc_msg *msg)
-{
- bool ok = msg->type == JSONRPC_REPLY;
- if (!ok
- && idl->state != IDL_S_SERVER_SCHEMA_REQUESTED
- && idl->state != IDL_S_SERVER_MONITOR_COND_REQUESTED
- && idl->state != IDL_S_DATA_MONITOR_COND_REQUESTED
- && idl->state != IDL_S_DATA_MONITOR_COND_SINCE_REQUESTED) {
- static struct vlog_rate_limit rl = VLOG_RATE_LIMIT_INIT(5, 5);
- char *s = jsonrpc_msg_to_string(msg);
- VLOG_INFO_RL(&rl, "%s: received unexpected %s response in "
- "%s state: %s", jsonrpc_session_get_name(idl->session),
- jsonrpc_msg_type_to_string(msg->type),
- ovsdb_idl_state_to_string(idl->state),
- s);
- free(s);
- ovsdb_idl_retry(idl);
- return;
- }
-
- switch (idl->state) {
- case IDL_S_SERVER_SCHEMA_REQUESTED:
- if (ok) {
- json_destroy(idl->server.schema);
- idl->server.schema = json_clone(msg->result);
- ovsdb_idl_send_monitor_request(idl, &idl->server,
- OVSDB_IDL_MM_MONITOR_COND);
- ovsdb_idl_transition(idl, IDL_S_SERVER_MONITOR_COND_REQUESTED);
- } else {
- ovsdb_idl_send_schema_request(idl, &idl->data);
- ovsdb_idl_transition(idl, IDL_S_DATA_SCHEMA_REQUESTED);
- }
- break;
-
- case IDL_S_SERVER_MONITOR_COND_REQUESTED:
- if (ok) {
- idl->server.monitoring = OVSDB_IDL_MONITORING_COND;
- ovsdb_idl_db_parse_monitor_reply(&idl->server, msg->result, 2);
- if (ovsdb_idl_check_server_db(idl)) {
- ovsdb_idl_send_db_change_aware(idl);
- }
- } else {
- ovsdb_idl_send_schema_request(idl, &idl->data);
- ovsdb_idl_transition(idl, IDL_S_DATA_SCHEMA_REQUESTED);
- }
- break;
-
- case IDL_S_DATA_SCHEMA_REQUESTED:
- json_destroy(idl->data.schema);
- idl->data.schema = json_clone(msg->result);
- ovsdb_idl_send_monitor_request(idl, &idl->data,
- OVSDB_IDL_MM_MONITOR_COND);
- ovsdb_idl_transition(idl, IDL_S_DATA_MONITOR_COND_REQUESTED);
- break;
-
- case IDL_S_DATA_MONITOR_COND_SINCE_REQUESTED:
- if (!ok) {
- /* "monitor_cond_since" not supported. Try "monitor_cond". */
- ovsdb_idl_send_monitor_request(idl, &idl->data,
- OVSDB_IDL_MM_MONITOR_COND);
- ovsdb_idl_transition(idl, IDL_S_DATA_MONITOR_COND_REQUESTED);
- } else {
- idl->data.monitoring = OVSDB_IDL_MONITORING_COND_SINCE;
- ovsdb_idl_transition(idl, IDL_S_MONITORING);
- ovsdb_idl_db_parse_monitor_reply(&idl->data, msg->result, 3);
- }
- break;
-
- case IDL_S_DATA_MONITOR_COND_REQUESTED:
- if (!ok) {
- /* "monitor_cond" not supported. Try "monitor". */
- ovsdb_idl_send_monitor_request(idl, &idl->data,
- OVSDB_IDL_MM_MONITOR);
- ovsdb_idl_transition(idl, IDL_S_DATA_MONITOR_REQUESTED);
- } else {
- idl->data.monitoring = OVSDB_IDL_MONITORING_COND;
- ovsdb_idl_transition(idl, IDL_S_MONITORING);
- ovsdb_idl_db_parse_monitor_reply(&idl->data, msg->result, 2);
- }
- break;
-
- case IDL_S_DATA_MONITOR_REQUESTED:
- idl->data.monitoring = OVSDB_IDL_MONITORING;
- ovsdb_idl_transition(idl, IDL_S_MONITORING);
- ovsdb_idl_db_parse_monitor_reply(&idl->data, msg->result, 1);
- idl->data.change_seqno++;
- break;
-
- case IDL_S_MONITORING:
- /* We don't normally have a request outstanding in this state. If we
- * do, it's a "monitor_cond_change", which means that the conditional
- * monitor clauses were updated.
- *
- * Mark the last requested conditions as acked and if further
- * condition changes were pending, send them now. */
- ovsdb_idl_db_ack_condition(&idl->data);
- ovsdb_idl_send_cond_change(idl);
- idl->data.cond_seqno++;
- break;
-
- case IDL_S_ERROR:
- case IDL_S_RETRY:
- /* Nothing to do in this state. */
- break;
-
- default:
- OVS_NOT_REACHED();
- }
-}
-
-static void
-ovsdb_idl_process_msg(struct ovsdb_idl *idl, struct jsonrpc_msg *msg)
-{
- bool is_response = (msg->type == JSONRPC_REPLY ||
- msg->type == JSONRPC_ERROR);
-
- /* Process a reply to an outstanding request. */
- if (is_response
- && idl->request_id && json_equal(idl->request_id, msg->id)) {
- json_destroy(idl->request_id);
- idl->request_id = NULL;
- ovsdb_idl_process_response(idl, msg);
- return;
- }
-
- /* Process database contents updates. */
- if (ovsdb_idl_db_parse_update_rpc(&idl->data, msg)) {
- return;
- }
- if (idl->server.monitoring
- && ovsdb_idl_db_parse_update_rpc(&idl->server, msg)) {
- ovsdb_idl_check_server_db(idl);
- return;
- }
-
- if (ovsdb_idl_handle_monitor_canceled(idl, &idl->data, msg)
- || (idl->server.monitoring
- && ovsdb_idl_handle_monitor_canceled(idl, &idl->server, msg))) {
- return;
- }
-
- /* Process "lock" replies and related notifications. */
- if (ovsdb_idl_db_process_lock_replies(&idl->data, msg)) {
- return;
- }
-
- /* Process response to a database transaction we submitted. */
- if (is_response && ovsdb_idl_db_txn_process_reply(&idl->data, msg)) {
- return;
- }
-
- /* Unknown message. Log at a low level because this can happen if
- * ovsdb_idl_txn_destroy() is called to destroy a transaction
- * before we receive the reply.
- *
- * (We could sort those out from other kinds of unknown messages by
- * using distinctive IDs for transactions, if it seems valuable to
- * do so, and then it would be possible to use different log
- * levels. XXX?) */
- char *s = jsonrpc_msg_to_string(msg);
- VLOG_DBG("%s: received unexpected %s message: %s",
- jsonrpc_session_get_name(idl->session),
- jsonrpc_msg_type_to_string(msg->type), s);
- free(s);
-}
-
/* Processes a batch of messages from the database server on 'idl'. This may
* cause the IDL's contents to change. The client may check for that with
* ovsdb_idl_get_seqno(). */
void
ovsdb_idl_run(struct ovsdb_idl *idl)
{
- if (!idl->session) {
- ovsdb_idl_txn_abort_all(idl);
- return;
- }
-
- int i;
-
- ovs_assert(!idl->data.txn);
+ ovs_assert(!idl->txn);
- ovsdb_idl_send_cond_change(idl);
+ struct ovs_list events;
+ ovsdb_cs_run(idl->cs, &events);
- jsonrpc_session_run(idl->session);
- for (i = 0; jsonrpc_session_is_connected(idl->session) && i < 50; i++) {
- struct jsonrpc_msg *msg;
- unsigned int seqno;
-
- seqno = jsonrpc_session_get_seqno(idl->session);
- if (idl->state_seqno != seqno) {
- idl->state_seqno = seqno;
+ struct ovsdb_cs_event *event;
+ LIST_FOR_EACH_POP (event, list_node, &events) {
+ switch (event->type) {
+ case OVSDB_CS_EVENT_TYPE_RECONNECT:
ovsdb_idl_txn_abort_all(idl);
- ovsdb_idl_restart_fsm(idl);
+ break;
- if (idl->data.lock_name) {
- jsonrpc_session_send(
- idl->session,
- ovsdb_idl_db_compose_lock_request(&idl->data));
- }
- }
+ case OVSDB_CS_EVENT_TYPE_LOCKED:
+ /* If the client couldn't run a transaction because it didn't have
+ * the lock, this will encourage it to try again. */
+ idl->change_seqno++;
+ break;
+
+ case OVSDB_CS_EVENT_TYPE_UPDATE:
+ ovsdb_idl_parse_update(idl, &event->update);
+ break;
- msg = jsonrpc_session_recv(idl->session);
- if (!msg) {
+ case OVSDB_CS_EVENT_TYPE_TXN_REPLY:
+ ovsdb_idl_txn_process_reply(idl, event->txn_reply);
break;
}
- ovsdb_idl_process_msg(idl, msg);
- jsonrpc_msg_destroy(msg);
+ ovsdb_cs_event_destroy(event);
}
- ovsdb_idl_row_destroy_postprocess(&idl->data);
+
+ ovsdb_idl_row_destroy_postprocess(idl);
}
/* Arranges for poll_block() to wake up when ovsdb_idl_run() has something to
@@ -949,11 +430,7 @@ ovsdb_idl_run(struct ovsdb_idl *idl)
void
ovsdb_idl_wait(struct ovsdb_idl *idl)
{
- if (!idl->session) {
- return;
- }
- jsonrpc_session_wait(idl->session);
- jsonrpc_session_recv_wait(idl->session);
+ ovsdb_cs_wait(idl->cs);
}
/* Returns a "sequence number" that represents the state of 'idl'. When
@@ -976,7 +453,7 @@ ovsdb_idl_wait(struct ovsdb_idl *idl)
unsigned int
ovsdb_idl_get_seqno(const struct ovsdb_idl *idl)
{
- return idl->data.change_seqno;
+ return idl->change_seqno;
}
/* Returns a "sequence number" that represents the number of conditional
@@ -996,7 +473,7 @@ ovsdb_idl_get_seqno(const struct ovsdb_idl *idl)
unsigned int
ovsdb_idl_get_condition_seqno(const struct ovsdb_idl *idl)
{
- return idl->data.cond_seqno;
+ return ovsdb_cs_get_condition_seqno(idl->cs);
}
/* Returns true if 'idl' successfully connected to the remote database and
@@ -1018,9 +495,7 @@ ovsdb_idl_has_ever_connected(const struct ovsdb_idl *idl)
void
ovsdb_idl_enable_reconnect(struct ovsdb_idl *idl)
{
- if (idl->session) {
- jsonrpc_session_enable_reconnect(idl->session);
- }
+ ovsdb_cs_enable_reconnect(idl->cs);
}
/* Forces 'idl' to drop its connection to the database and reconnect. In the
@@ -1028,9 +503,7 @@ ovsdb_idl_enable_reconnect(struct ovsdb_idl *idl)
void
ovsdb_idl_force_reconnect(struct ovsdb_idl *idl)
{
- if (idl->session) {
- jsonrpc_session_force_reconnect(idl->session);
- }
+ ovsdb_cs_force_reconnect(idl->cs);
}
/* Some IDL users should only write to write-only columns. Furthermore,
@@ -1040,7 +513,7 @@ ovsdb_idl_force_reconnect(struct ovsdb_idl *idl)
void
ovsdb_idl_verify_write_only(struct ovsdb_idl *idl)
{
- idl->data.verify_write_only = true;
+ idl->verify_write_only = true;
}
/* Returns true if 'idl' is currently connected or trying to connect
@@ -1048,14 +521,13 @@ ovsdb_idl_verify_write_only(struct ovsdb_idl *idl)
bool
ovsdb_idl_is_alive(const struct ovsdb_idl *idl)
{
- return idl->session && jsonrpc_session_is_alive(idl->session) &&
- idl->state != IDL_S_ERROR;
+ return ovsdb_cs_is_alive(idl->cs);
}
bool
ovsdb_idl_is_connected(const struct ovsdb_idl *idl)
{
- return idl->session && jsonrpc_session_is_connected(idl->session);
+ return ovsdb_cs_is_connected(idl->cs);
}
/* Returns the last error reported on a connection by 'idl'. The return value
@@ -1066,14 +538,7 @@ ovsdb_idl_is_connected(const struct ovsdb_idl *idl)
int
ovsdb_idl_get_last_error(const struct ovsdb_idl *idl)
{
- int err = idl->session ? jsonrpc_session_get_last_error(idl->session) : 0;
- if (err) {
- return err;
- } else if (idl->state == IDL_S_ERROR) {
- return ENOENT;
- } else {
- return 0;
- }
+ return ovsdb_cs_get_last_error(idl->cs);
}
/* Sets the "probe interval" for 'idl->session' to 'probe_interval', in
@@ -1082,9 +547,7 @@ ovsdb_idl_get_last_error(const struct ovsdb_idl *idl)
void
ovsdb_idl_set_probe_interval(const struct ovsdb_idl *idl, int probe_interval)
{
- if (idl->session) {
- jsonrpc_session_set_probe_interval(idl->session, probe_interval);
- }
+ ovsdb_cs_set_probe_interval(idl->cs, probe_interval);
}
static size_t
@@ -1155,7 +618,7 @@ void
ovsdb_idl_check_consistency(const struct ovsdb_idl *idl)
{
/* Consistency is broken while a transaction is in progress. */
- if (!idl->data.txn) {
+ if (!idl->txn) {
return;
}
@@ -1164,8 +627,8 @@ ovsdb_idl_check_consistency(const struct ovsdb_idl *idl)
struct uuid *dsts = NULL;
size_t allocated_dsts = 0;
- for (size_t i = 0; i < idl->data.class_->n_tables; i++) {
- const struct ovsdb_idl_table *table = &idl->data.tables[i];
+ for (size_t i = 0; i < idl->class_->n_tables; i++) {
+ const struct ovsdb_idl_table *table = &idl->tables[i];
const struct ovsdb_idl_table_class *class = table->class_;
const struct ovsdb_idl_row *row;
@@ -1208,88 +671,145 @@ ovsdb_idl_check_consistency(const struct ovsdb_idl *idl)
free(dsts);
ovs_assert(ok);
}
-
-const struct ovsdb_idl_class *
-ovsdb_idl_get_class(const struct ovsdb_idl *idl)
-{
- return idl->data.class_;
-}
-
-/* Given 'column' in some table in 'class', returns the table's class. */
-const struct ovsdb_idl_table_class *
-ovsdb_idl_table_class_from_column(const struct ovsdb_idl_class *class,
- const struct ovsdb_idl_column *column)
-{
- for (size_t i = 0; i < class->n_tables; i++) {
- const struct ovsdb_idl_table_class *tc = &class->tables[i];
- if (column >= tc->columns && column < &tc->columns[tc->n_columns]) {
- return tc;
- }
- }
-
- OVS_NOT_REACHED();
-}
-/* Given 'column' in some table in 'db', returns the table. */
-static struct ovsdb_idl_table *
-ovsdb_idl_table_from_column(struct ovsdb_idl_db *db,
- const struct ovsdb_idl_column *column)
-{
- const struct ovsdb_idl_table_class *tc =
- ovsdb_idl_table_class_from_column(db->class_, column);
- return &db->tables[tc - db->class_->tables];
-}
-
-static unsigned char *
-ovsdb_idl_db_get_mode(struct ovsdb_idl_db *db,
- const struct ovsdb_idl_column *column)
+static struct json *
+ovsdb_idl_compose_monitor_request(const struct json *schema_json, void *idl_)
{
- ovs_assert(!db->change_seqno);
+ struct ovsdb_idl *idl = idl_;
- const struct ovsdb_idl_table *table = ovsdb_idl_table_from_column(db,
- column);
- return &table->modes[column - table->class_->columns];
-}
+ struct shash *schema = ovsdb_cs_parse_schema(schema_json);
+ struct json *monitor_requests = json_object_create();
-static void
-ovsdb_idl_db_set_mode(struct ovsdb_idl_db *db,
- const struct ovsdb_idl_column *column,
- unsigned char mode)
-{
- const struct ovsdb_idl_table *table = ovsdb_idl_table_from_column(db,
- column);
- size_t column_idx = column - table->class_->columns;
+ for (size_t i = 0; i < idl->class_->n_tables; i++) {
+ struct ovsdb_idl_table *table = &idl->tables[i];
+ const struct ovsdb_idl_table_class *tc = table->class_;
+ struct json *monitor_request;
+ const struct sset *table_schema
+ = schema ? shash_find_data(schema, table->class_->name) : NULL;
- if (table->modes[column_idx] != mode) {
- *ovsdb_idl_db_get_mode(db, column) = mode;
- }
-}
+ struct json *columns
+ = table->need_table ? json_array_create_empty() : NULL;
+ for (size_t j = 0; j < tc->n_columns; j++) {
+ const struct ovsdb_idl_column *column = &tc->columns[j];
+ bool idl_has_column = (table_schema &&
+ sset_contains(table_schema, column->name));
+ if (column->is_synthetic) {
+ if (idl_has_column) {
+ VLOG_WARN("%s table in %s database has synthetic "
+ "column %s", table->class_->name,
+ idl->class_->database, column->name);
+ }
+ } else if (table->modes[j] & OVSDB_IDL_MONITOR) {
+ if (table_schema && !idl_has_column) {
+ VLOG_WARN("%s table in %s database lacks %s column "
+ "(database needs upgrade?)",
+ table->class_->name, idl->class_->database,
+ column->name);
+ continue;
+ }
+ if (!columns) {
+ columns = json_array_create_empty();
+ }
+ json_array_add(columns, json_string_create(column->name));
+ }
+ }
+
+ if (columns) {
+ if (schema && !table_schema) {
+ VLOG_WARN("%s database lacks %s table "
+ "(database needs upgrade?)",
+ idl->class_->database, table->class_->name);
+ json_destroy(columns);
+ continue;
+ }
+
+ monitor_request = json_object_create();
+ json_object_put(monitor_request, "columns", columns);
+ json_object_put(monitor_requests, tc->name,
+ json_array_create_1(monitor_request));
+ }
+ }
+ ovsdb_cs_free_schema(schema);
+
+ return monitor_requests;
+}
+
+static struct ovsdb_cs_ops ovsdb_idl_cs_ops = {
+ ovsdb_idl_compose_monitor_request,
+};
+
+const struct ovsdb_idl_class *
+ovsdb_idl_get_class(const struct ovsdb_idl *idl)
+{
+ return idl->class_;
+}
+
+/* Given 'column' in some table in 'class', returns the table's class. */
+const struct ovsdb_idl_table_class *
+ovsdb_idl_table_class_from_column(const struct ovsdb_idl_class *class,
+ const struct ovsdb_idl_column *column)
+{
+ for (size_t i = 0; i < class->n_tables; i++) {
+ const struct ovsdb_idl_table_class *tc = &class->tables[i];
+ if (column >= tc->columns && column < &tc->columns[tc->n_columns]) {
+ return tc;
+ }
+ }
+
+ OVS_NOT_REACHED();
+}
+
+/* Given 'column' in some table in 'idl', returns the table. */
+static struct ovsdb_idl_table *
+ovsdb_idl_table_from_column(struct ovsdb_idl *idl,
+ const struct ovsdb_idl_column *column)
+{
+ const struct ovsdb_idl_table_class *tc =
+ ovsdb_idl_table_class_from_column(idl->class_, column);
+ return &idl->tables[tc - idl->class_->tables];
+}
+
+static unsigned char *
+ovsdb_idl_get_mode(struct ovsdb_idl *idl,
+ const struct ovsdb_idl_column *column)
+{
+ ovs_assert(!idl->change_seqno);
+
+ const struct ovsdb_idl_table *table = ovsdb_idl_table_from_column(idl,
+ column);
+ return &table->modes[column - table->class_->columns];
+}
+
+static void
+ovsdb_idl_set_mode(struct ovsdb_idl *idl,
+ const struct ovsdb_idl_column *column,
+ unsigned char mode)
+{
+ const struct ovsdb_idl_table *table = ovsdb_idl_table_from_column(idl,
+ column);
+ size_t column_idx = column - table->class_->columns;
+
+ if (table->modes[column_idx] != mode) {
+ *ovsdb_idl_get_mode(idl, column) = mode;
+ }
+}
static void
-add_ref_table(struct ovsdb_idl_db *db, const struct ovsdb_base_type *base)
+add_ref_table(struct ovsdb_idl *idl, const struct ovsdb_base_type *base)
{
if (base->type == OVSDB_TYPE_UUID && base->uuid.refTableName) {
struct ovsdb_idl_table *table;
- table = shash_find_data(&db->table_by_name, base->uuid.refTableName);
+ table = shash_find_data(&idl->table_by_name, base->uuid.refTableName);
if (table) {
table->need_table = true;
} else {
VLOG_WARN("%s IDL class missing referenced table %s",
- db->class_->database, base->uuid.refTableName);
+ idl->class_->database, base->uuid.refTableName);
}
}
}
-static void
-ovsdb_idl_db_add_column(struct ovsdb_idl_db *db,
- const struct ovsdb_idl_column *column)
-{
- ovsdb_idl_db_set_mode(db, column, OVSDB_IDL_MONITOR | OVSDB_IDL_ALERT);
- add_ref_table(db, &column->type.key);
- add_ref_table(db, &column->type.value);
-}
-
/* Turns on OVSDB_IDL_MONITOR and OVSDB_IDL_ALERT for 'column' in 'idl'. Also
* ensures that any tables referenced by 'column' will be replicated, even if
* no columns in that table are selected for replication (see
@@ -1303,25 +823,9 @@ void
ovsdb_idl_add_column(struct ovsdb_idl *idl,
const struct ovsdb_idl_column *column)
{
- ovsdb_idl_db_add_column(&idl->data, column);
-}
-
-static void
-ovsdb_idl_db_add_table(struct ovsdb_idl_db *db,
- const struct ovsdb_idl_table_class *tc)
-{
- size_t i;
-
- for (i = 0; i < db->class_->n_tables; i++) {
- struct ovsdb_idl_table *table = &db->tables[i];
-
- if (table->class_ == tc) {
- table->need_table = true;
- return;
- }
- }
-
- OVS_NOT_REACHED();
+ ovsdb_idl_set_mode(idl, column, OVSDB_IDL_MONITOR | OVSDB_IDL_ALERT);
+ add_ref_table(idl, &column->type.key);
+ add_ref_table(idl, &column->type.value);
}
/* Ensures that the table with class 'tc' will be replicated on 'idl' even if
@@ -1341,7 +845,16 @@ void
ovsdb_idl_add_table(struct ovsdb_idl *idl,
const struct ovsdb_idl_table_class *tc)
{
- ovsdb_idl_db_add_table(&idl->data, tc);
+ for (size_t i = 0; i < idl->class_->n_tables; i++) {
+ struct ovsdb_idl_table *table = &idl->tables[i];
+
+ if (table->class_ == tc) {
+ table->need_table = true;
+ return;
+ }
+ }
+
+ OVS_NOT_REACHED();
}
/* A single clause within an ovsdb_idl_condition. */
@@ -1502,111 +1015,11 @@ ovsdb_idl_condition_add_clause_true(struct ovsdb_idl_condition *condition)
}
}
-static bool
-ovsdb_idl_condition_equals(const struct ovsdb_idl_condition *a,
- const struct ovsdb_idl_condition *b)
-{
- if (hmap_count(&a->clauses) != hmap_count(&b->clauses)) {
- return false;
- }
- if (a->is_true != b->is_true) {
- return false;
- }
-
- const struct ovsdb_idl_clause *clause;
- HMAP_FOR_EACH (clause, hmap_node, &a->clauses) {
- if (!ovsdb_idl_condition_find_clause(b, clause,
- clause->hmap_node.hash)) {
- return false;
- }
- }
- return true;
-}
-
-static void
-ovsdb_idl_condition_clone(struct ovsdb_idl_condition **dst,
- const struct ovsdb_idl_condition *src)
-{
- if (*dst) {
- ovsdb_idl_condition_destroy(*dst);
- } else {
- *dst = xmalloc(sizeof **dst);
- }
- ovsdb_idl_condition_init(*dst);
-
- (*dst)->is_true = src->is_true;
-
- const struct ovsdb_idl_clause *clause;
- HMAP_FOR_EACH (clause, hmap_node, &src->clauses) {
- ovsdb_idl_condition_add_clause__(*dst, clause, clause->hmap_node.hash);
- }
-}
-
-static void
-ovsdb_idl_condition_move(struct ovsdb_idl_condition **dst,
- struct ovsdb_idl_condition **src)
-{
- if (*dst) {
- ovsdb_idl_condition_destroy(*dst);
- free(*dst);
- }
- *dst = *src;
- *src = NULL;
-}
-
-static unsigned int
-ovsdb_idl_db_set_condition(struct ovsdb_idl_db *db,
- const struct ovsdb_idl_table_class *tc,
- const struct ovsdb_idl_condition *condition)
-{
- struct ovsdb_idl_condition *table_cond;
- struct ovsdb_idl_table *table = ovsdb_idl_db_table_from_class(db, tc);
-
- /* Compare the new condition to the last known condition which can be
- * either "new" (not sent yet), "requested" or "acked", in this order.
- */
- if (table->new_cond) {
- table_cond = table->new_cond;
- } else if (table->req_cond) {
- table_cond = table->req_cond;
- } else {
- table_cond = table->ack_cond;
- }
- ovs_assert(table_cond);
-
- if (!ovsdb_idl_condition_equals(condition, table_cond)) {
- ovsdb_idl_condition_clone(&table->new_cond, condition);
- db->cond_changed = true;
- poll_immediate_wake();
- return db->cond_seqno + 1;
- } else if (table_cond != table->ack_cond) {
- /* 'condition' was already set but has not been "acked" yet. The IDL
- * will be up to date when db->cond_seqno gets incremented. */
- return db->cond_seqno + 1;
- }
-
- return db->cond_seqno;
-}
-
-/* Sets the replication condition for 'tc' in 'idl' to 'condition' and
- * arranges to send the new condition to the database server.
- *
- * Return the next conditional update sequence number. When this
- * value and ovsdb_idl_get_condition_seqno() matches, the 'idl'
- * contains rows that match the 'condition'. */
-unsigned int
-ovsdb_idl_set_condition(struct ovsdb_idl *idl,
- const struct ovsdb_idl_table_class *tc,
- const struct ovsdb_idl_condition *condition)
-{
- return ovsdb_idl_db_set_condition(&idl->data, tc, condition);
-}
-
static struct json *
ovsdb_idl_condition_to_json(const struct ovsdb_idl_condition *cnd)
{
if (cnd->is_true) {
- return json_array_create_empty();
+ return NULL;
}
size_t n = hmap_count(&cnd->clauses);
@@ -1623,158 +1036,22 @@ ovsdb_idl_condition_to_json(const struct ovsdb_idl_condition *cnd)
ovs_assert(i == n);
return json_array_create(clauses, n);
}
-
-static struct json *
-ovsdb_idl_create_cond_change_req(const struct ovsdb_idl_condition *cond)
-{
- struct json *monitor_cond_change_request = json_object_create();
- struct json *cond_json = ovsdb_idl_condition_to_json(cond);
-
- json_object_put(monitor_cond_change_request, "where", cond_json);
-
- return monitor_cond_change_request;
-}
-
-static struct jsonrpc_msg *
-ovsdb_idl_db_compose_cond_change(struct ovsdb_idl_db *db)
-{
- if (!db->cond_changed) {
- return NULL;
- }
-
- struct json *monitor_cond_change_requests = NULL;
- for (size_t i = 0; i < db->class_->n_tables; i++) {
- struct ovsdb_idl_table *table = &db->tables[i];
-
- /* Always use the most recent conditions set by the IDL client when
- * requesting monitor_cond_change, i.e., table->new_cond.
- */
- if (table->new_cond) {
- struct json *req =
- ovsdb_idl_create_cond_change_req(table->new_cond);
- if (req) {
- if (!monitor_cond_change_requests) {
- monitor_cond_change_requests = json_object_create();
- }
- json_object_put(monitor_cond_change_requests,
- table->class_->name,
- json_array_create_1(req));
- }
- /* Mark the new condition as requested by moving it to req_cond.
- * If there's already requested condition that's a bug.
- */
- ovs_assert(table->req_cond == NULL);
- ovsdb_idl_condition_move(&table->req_cond, &table->new_cond);
- }
- }
-
- if (!monitor_cond_change_requests) {
- return NULL;
- }
-
- db->cond_changed = false;
- struct json *params = json_array_create_3(json_clone(db->monitor_id),
- json_clone(db->monitor_id),
- monitor_cond_change_requests);
- return jsonrpc_create_request("monitor_cond_change", params, NULL);
-}
-
-/* Marks all requested table conditions in 'db' as acked by the server.
- * It should be called when the server replies to monitor_cond_change
- * requests.
- */
-static void
-ovsdb_idl_db_ack_condition(struct ovsdb_idl_db *db)
-{
- for (size_t i = 0; i < db->class_->n_tables; i++) {
- struct ovsdb_idl_table *table = &db->tables[i];
-
- if (table->req_cond) {
- ovsdb_idl_condition_move(&table->ack_cond, &table->req_cond);
- }
- }
-}
-
-/* Should be called when the IDL fsm is restarted and resyncs table conditions
- * based on the state the DB is in:
- * - if a non-zero last_id is available for the DB then upon reconnect
- * the IDL should first request acked conditions to avoid missing updates
- * about records that were added before the transaction with
- * txn-id == last_id. If there were requested condition changes in flight
- * (i.e., req_cond not NULL) and the IDL client didn't set new conditions
- * (i.e., new_cond is NULL) then move req_cond to new_cond to trigger a
- * follow up monitor_cond_change request.
- * - if there's no last_id available for the DB then it's safe to use the
- * latest conditions set by the IDL client even if they weren't acked yet.
- */
-static void
-ovsdb_idl_db_sync_condition(struct ovsdb_idl_db *db)
-{
- bool ack_all = uuid_is_zero(&db->last_id);
-
- db->cond_changed = false;
- for (size_t i = 0; i < db->class_->n_tables; i++) {
- struct ovsdb_idl_table *table = &db->tables[i];
-
- /* When monitor_cond_since requests will be issued, the
- * table->ack_cond condition will be added to the "where" clause".
- * Follow up monitor_cond_change requests will use table->new_cond.
- */
- if (ack_all) {
- if (table->new_cond) {
- ovsdb_idl_condition_move(&table->req_cond, &table->new_cond);
- }
-
- if (table->req_cond) {
- ovsdb_idl_condition_move(&table->ack_cond, &table->req_cond);
- }
- } else {
- /* If there was no "unsent" condition but instead a
- * monitor_cond_change request was in flight, move table->req_cond
- * to table->new_cond and set db->cond_changed to trigger a new
- * monitor_cond_change request.
- *
- * However, if a new condition has been set by the IDL client,
- * monitor_cond_change will be sent anyway and will use the most
- * recent table->new_cond so there's no need to update it here.
- */
- if (table->req_cond && !table->new_cond) {
- ovsdb_idl_condition_move(&table->new_cond, &table->req_cond);
- db->cond_changed = true;
- }
- }
- }
-}
-
-static void
-ovsdb_idl_send_cond_change(struct ovsdb_idl *idl)
-{
- /* When 'idl->request_id' is not NULL, there is an outstanding
- * conditional monitoring update request that we have not heard
- * from the server yet. Don't generate another request in this case. */
- if (!jsonrpc_session_is_connected(idl->session)
- || idl->data.monitoring == OVSDB_IDL_MONITORING
- || idl->request_id) {
- return;
- }
- struct jsonrpc_msg *msg = ovsdb_idl_db_compose_cond_change(&idl->data);
- if (msg) {
- idl->request_id = json_clone(msg->id);
- jsonrpc_session_send(idl->session, msg);
- }
-}
-
-/* Turns off OVSDB_IDL_ALERT and OVSDB_IDL_TRACK for 'column' in 'db'.
+/* Sets the replication condition for 'tc' in 'idl' to 'condition' and
+ * arranges to send the new condition to the database server.
*
- * This function should be called between ovsdb_idl_create() and the first call
- * to ovsdb_idl_run().
- */
-static void
-ovsdb_idl_db_omit_alert(struct ovsdb_idl_db *db,
- const struct ovsdb_idl_column *column)
+ * Return the next conditional update sequence number. When this
+ * value and ovsdb_idl_get_condition_seqno() matches, the 'idl'
+ * contains rows that match the 'condition'. */
+unsigned int
+ovsdb_idl_set_condition(struct ovsdb_idl *idl,
+ const struct ovsdb_idl_table_class *tc,
+ const struct ovsdb_idl_condition *condition)
{
- *ovsdb_idl_db_get_mode(db, column) &= ~(OVSDB_IDL_ALERT | OVSDB_IDL_TRACK);
+ struct json *cond_json = ovsdb_idl_condition_to_json(condition);
+ unsigned int seqno = ovsdb_cs_set_condition(idl->cs, tc->name, cond_json);
+ json_destroy(cond_json);
+ return seqno;
}
/* Turns off OVSDB_IDL_ALERT and OVSDB_IDL_TRACK for 'column' in 'idl'.
@@ -1786,14 +1063,7 @@ void
ovsdb_idl_omit_alert(struct ovsdb_idl *idl,
const struct ovsdb_idl_column *column)
{
- ovsdb_idl_db_omit_alert(&idl->data, column);
-}
-
-static void
-ovsdb_idl_db_omit(struct ovsdb_idl_db *db,
- const struct ovsdb_idl_column *column)
-{
- *ovsdb_idl_db_get_mode(db, column) = 0;
+ *ovsdb_idl_get_mode(idl, column) &= ~(OVSDB_IDL_ALERT | OVSDB_IDL_TRACK);
}
/* Sets the mode for 'column' in 'idl' to 0. See the big comment above
@@ -1805,7 +1075,7 @@ ovsdb_idl_db_omit(struct ovsdb_idl_db *db,
void
ovsdb_idl_omit(struct ovsdb_idl *idl, const struct ovsdb_idl_column *column)
{
- ovsdb_idl_db_omit(&idl->data, column);
+ *ovsdb_idl_get_mode(idl, column) = 0;
}
/* Returns the most recent IDL change sequence number that caused a
@@ -1816,7 +1086,7 @@ ovsdb_idl_table_get_seqno(const struct ovsdb_idl *idl,
const struct ovsdb_idl_table_class *table_class)
{
struct ovsdb_idl_table *table
- = ovsdb_idl_db_table_from_class(&idl->data, table_class);
+ = ovsdb_idl_table_from_class(idl, table_class);
unsigned int max_seqno = table->change_seqno[OVSDB_IDL_CHANGE_INSERT];
if (max_seqno < table->change_seqno[OVSDB_IDL_CHANGE_MODIFY]) {
@@ -1852,461 +1122,156 @@ ovsdb_idl_row_get_seqno(const struct ovsdb_idl_row *row,
void
ovsdb_idl_track_add_column(struct ovsdb_idl *idl,
const struct ovsdb_idl_column *column)
-{
- if (!(*ovsdb_idl_db_get_mode(&idl->data, column) & OVSDB_IDL_ALERT)) {
- ovsdb_idl_add_column(idl, column);
- }
- *ovsdb_idl_db_get_mode(&idl->data, column) |= OVSDB_IDL_TRACK;
-}
-
-void
-ovsdb_idl_track_add_all(struct ovsdb_idl *idl)
-{
- size_t i, j;
-
- for (i = 0; i < idl->data.class_->n_tables; i++) {
- const struct ovsdb_idl_table_class *tc = &idl->data.class_->tables[i];
-
- for (j = 0; j < tc->n_columns; j++) {
- const struct ovsdb_idl_column *column = &tc->columns[j];
- ovsdb_idl_track_add_column(idl, column);
- }
- }
-}
-
-/* Returns true if 'table' has any tracked column. */
-static bool
-ovsdb_idl_track_is_set(struct ovsdb_idl_table *table)
-{
- size_t i;
-
- for (i = 0; i < table->class_->n_columns; i++) {
- if (table->modes[i] & OVSDB_IDL_TRACK) {
- return true;
- }
- }
- return false;
-}
-
-/* Returns the first tracked row in table with class 'table_class'
- * for the specified 'idl'. Returns NULL if there are no tracked rows */
-const struct ovsdb_idl_row *
-ovsdb_idl_track_get_first(const struct ovsdb_idl *idl,
- const struct ovsdb_idl_table_class *table_class)
-{
- struct ovsdb_idl_table *table
- = ovsdb_idl_db_table_from_class(&idl->data, table_class);
-
- if (!ovs_list_is_empty(&table->track_list)) {
- return CONTAINER_OF(ovs_list_front(&table->track_list), struct ovsdb_idl_row, track_node);
- }
- return NULL;
-}
-
-/* Returns the next tracked row in table after the specified 'row'
- * (in no particular order). Returns NULL if there are no tracked rows */
-const struct ovsdb_idl_row *
-ovsdb_idl_track_get_next(const struct ovsdb_idl_row *row)
-{
- if (row->track_node.next != &row->table->track_list) {
- return CONTAINER_OF(row->track_node.next, struct ovsdb_idl_row, track_node);
- }
-
- return NULL;
-}
-
-/* Returns true if a tracked 'column' in 'row' was updated by IDL, false
- * otherwise. The tracking data is cleared by ovsdb_idl_track_clear()
- *
- * Function returns false if 'column' is not tracked (see
- * ovsdb_idl_track_add_column()).
- */
-bool
-ovsdb_idl_track_is_updated(const struct ovsdb_idl_row *row,
- const struct ovsdb_idl_column *column)
-{
- const struct ovsdb_idl_table_class *class;
- size_t column_idx;
-
- class = row->table->class_;
- column_idx = column - class->columns;
-
- if (row->updated && bitmap_is_set(row->updated, column_idx)) {
- return true;
- } else {
- return false;
- }
-}
-
-/* Flushes the tracked rows. Client calls this function after calling
- * ovsdb_idl_run() and read all tracked rows with the ovsdb_idl_track_get_*()
- * functions. This is usually done at the end of the client's processing
- * loop when it is ready to do ovsdb_idl_run() again.
- */
-static void
-ovsdb_idl_db_track_clear(struct ovsdb_idl_db *db)
-{
- size_t i;
-
- for (i = 0; i < db->class_->n_tables; i++) {
- struct ovsdb_idl_table *table = &db->tables[i];
-
- if (!ovs_list_is_empty(&table->track_list)) {
- struct ovsdb_idl_row *row, *next;
-
- LIST_FOR_EACH_SAFE(row, next, track_node, &table->track_list) {
- if (row->updated) {
- free(row->updated);
- row->updated = NULL;
- }
-
- row->change_seqno[OVSDB_IDL_CHANGE_INSERT] =
- row->change_seqno[OVSDB_IDL_CHANGE_MODIFY] =
- row->change_seqno[OVSDB_IDL_CHANGE_DELETE] = 0;
-
- ovs_list_remove(&row->track_node);
- ovs_list_init(&row->track_node);
- if (ovsdb_idl_row_is_orphan(row) && row->tracked_old_datum) {
- ovsdb_idl_row_unparse(row);
- const struct ovsdb_idl_table_class *class =
- row->table->class_;
- for (size_t c = 0; c < class->n_columns; c++) {
- ovsdb_datum_destroy(&row->tracked_old_datum[c],
- &class->columns[c].type);
- }
- free(row->tracked_old_datum);
- row->tracked_old_datum = NULL;
- free(row);
- }
- }
- }
- }
-}
-
-/* Flushes the tracked rows. Client calls this function after calling
- * ovsdb_idl_run() and read all tracked rows with the ovsdb_idl_track_get_*()
- * functions. This is usually done at the end of the client's processing
- * loop when it is ready to do ovsdb_idl_run() again.
- */
-void
-ovsdb_idl_track_clear(struct ovsdb_idl *idl)
-{
- ovsdb_idl_db_track_clear(&idl->data);
-}
-
-static void
-ovsdb_idl_send_schema_request(struct ovsdb_idl *idl,
- struct ovsdb_idl_db *db)
-{
- ovsdb_idl_send_request(idl, jsonrpc_create_request(
- "get_schema",
- json_array_create_1(json_string_create(
- db->class_->database)),
- NULL));
-}
-
-static void
-ovsdb_idl_send_db_change_aware(struct ovsdb_idl *idl)
-{
- struct jsonrpc_msg *msg = jsonrpc_create_request(
- "set_db_change_aware", json_array_create_1(json_boolean_create(true)),
- NULL);
- jsonrpc_session_send(idl->session, msg);
-}
-
-static bool
-ovsdb_idl_check_server_db(struct ovsdb_idl *idl)
-{
- const struct serverrec_database *database;
- SERVERREC_DATABASE_FOR_EACH (database, idl) {
- if (uuid_is_zero(&idl->cid)
- ? !strcmp(database->name, idl->data.class_->database)
- : database->n_cid && uuid_equals(database->cid, &idl->cid)) {
- break;
- }
- }
-
- static struct vlog_rate_limit rl = VLOG_RATE_LIMIT_INIT(5, 5);
- const char *server_name = jsonrpc_session_get_name(idl->session);
- bool ok = false;
- if (!database) {
- VLOG_INFO_RL(&rl, "%s: server does not have %s database",
- server_name, idl->data.class_->database);
- } else if (!strcmp(database->model, "clustered")) {
- uint64_t index = database->n_index ? *database->index : 0;
-
- if (!database->schema) {
- VLOG_INFO("%s: clustered database server has not yet joined "
- "cluster; trying another server", server_name);
- } else if (!database->connected) {
- VLOG_INFO("%s: clustered database server is disconnected "
- "from cluster; trying another server", server_name);
- } else if (idl->leader_only && !database->leader) {
- VLOG_INFO("%s: clustered database server is not cluster "
- "leader; trying another server", server_name);
- } else if (index < idl->min_index) {
- VLOG_WARN("%s: clustered database server has stale data; "
- "trying another server", server_name);
- } else {
- idl->min_index = index;
- ok = true;
- }
- } else {
- ok = true;
- }
- if (!ok) {
- ovsdb_idl_retry(idl);
- return false;
- }
-
- if (idl->state == IDL_S_SERVER_MONITOR_COND_REQUESTED) {
- json_destroy(idl->data.schema);
- idl->data.schema = json_from_string(database->schema);
- ovsdb_idl_send_monitor_request(idl, &idl->data,
- OVSDB_IDL_MM_MONITOR_COND_SINCE);
- ovsdb_idl_transition(idl, IDL_S_DATA_MONITOR_COND_SINCE_REQUESTED);
- }
- return true;
-}
-
-static void
-ovsdb_idl_send_monitor_request(struct ovsdb_idl *idl, struct ovsdb_idl_db *db,
- enum ovsdb_idl_monitor_method monitor_method)
-{
- struct shash *schema = ovsdb_cs_parse_schema(db->schema);
- struct json *monitor_requests = json_object_create();
-
- for (size_t i = 0; i < db->class_->n_tables; i++) {
- struct ovsdb_idl_table *table = &db->tables[i];
- const struct ovsdb_idl_table_class *tc = table->class_;
- struct json *monitor_request;
- const struct sset *table_schema
- = schema ? shash_find_data(schema, table->class_->name) : NULL;
-
- struct json *columns
- = table->need_table ? json_array_create_empty() : NULL;
- for (size_t j = 0; j < tc->n_columns; j++) {
- const struct ovsdb_idl_column *column = &tc->columns[j];
- bool db_has_column = (table_schema &&
- sset_contains(table_schema, column->name));
- if (column->is_synthetic) {
- if (db_has_column) {
- VLOG_WARN("%s table in %s database has synthetic "
- "column %s", table->class_->name,
- db->class_->database, column->name);
- }
- } else if (table->modes[j] & OVSDB_IDL_MONITOR) {
- if (table_schema && !db_has_column) {
- VLOG_WARN("%s table in %s database lacks %s column "
- "(database needs upgrade?)",
- table->class_->name, db->class_->database,
- column->name);
- continue;
- }
- if (!columns) {
- columns = json_array_create_empty();
- }
- json_array_add(columns, json_string_create(column->name));
- }
- }
-
- if (columns) {
- if (schema && !table_schema) {
- VLOG_WARN("%s database lacks %s table "
- "(database needs upgrade?)",
- db->class_->database, table->class_->name);
- json_destroy(columns);
- continue;
- }
-
- monitor_request = json_object_create();
- json_object_put(monitor_request, "columns", columns);
-
- /* Always use acked conditions when requesting
- * monitor_cond/monitor_cond_since.
- */
- const struct ovsdb_idl_condition *cond = table->ack_cond;
- if ((monitor_method == OVSDB_IDL_MM_MONITOR_COND ||
- monitor_method == OVSDB_IDL_MM_MONITOR_COND_SINCE) &&
- cond && !ovsdb_idl_condition_is_true(cond)) {
- json_object_put(monitor_request, "where",
- ovsdb_idl_condition_to_json(cond));
- }
- json_object_put(monitor_requests, tc->name,
- json_array_create_1(monitor_request));
- }
- }
- ovsdb_cs_free_schema(schema);
-
- struct json *params = json_array_create_3(
- json_string_create(db->class_->database),
- json_clone(db->monitor_id),
- monitor_requests);
- const char *method;
- switch (monitor_method) {
- case OVSDB_IDL_MM_MONITOR:
- method = "monitor";
- break;
- case OVSDB_IDL_MM_MONITOR_COND:
- method = "monitor_cond";
- break;
- case OVSDB_IDL_MM_MONITOR_COND_SINCE:
- method = "monitor_cond_since";
- struct json *json_last_id = json_string_create_nocopy(
- xasprintf(UUID_FMT, UUID_ARGS(&db->last_id)));
- json_array_add(params, json_last_id);
- break;
- default:
- OVS_NOT_REACHED();
- }
-
- ovsdb_idl_send_request(idl, jsonrpc_create_request(method, params, NULL));
-}
-
-static void
-log_parse_update_error(struct ovsdb_error *error)
-{
- if (!VLOG_DROP_WARN(&syntax_rl)) {
- char *s = ovsdb_error_to_string(error);
- VLOG_WARN_RL(&syntax_rl, "%s", s);
- free(s);
+{
+ if (!(*ovsdb_idl_get_mode(idl, column) & OVSDB_IDL_ALERT)) {
+ ovsdb_idl_add_column(idl, column);
}
- ovsdb_error_destroy(error);
+ *ovsdb_idl_get_mode(idl, column) |= OVSDB_IDL_TRACK;
}
-static void
-ovsdb_idl_db_parse_monitor_reply(struct ovsdb_idl_db *db,
- const struct json *result, int version)
-{
- db->change_seqno++;
- const struct json *table_updates = result;
- bool clear_db = true;
- if (version == 3) {
- if (result->type != JSON_ARRAY || result->array.n != 3) {
- struct ovsdb_error *error = ovsdb_syntax_error(result, NULL,
- "Response of monitor_cond_since must "
- "be an array with 3 elements.");
- log_parse_update_error(error);
- return;
- }
+void
+ovsdb_idl_track_add_all(struct ovsdb_idl *idl)
+{
+ size_t i, j;
- bool found = json_boolean(result->array.elems[0]);
- if (found) {
- clear_db = false;
- }
+ for (i = 0; i < idl->class_->n_tables; i++) {
+ const struct ovsdb_idl_table_class *tc = &idl->class_->tables[i];
- const char *last_id = json_string(result->array.elems[1]);
- if (!uuid_from_string(&db->last_id, last_id)) {
- struct ovsdb_error *error = ovsdb_syntax_error(result, NULL,
- "Last-id %s is not in UUID format.",
- last_id);
- log_parse_update_error(error);
- return;
+ for (j = 0; j < tc->n_columns; j++) {
+ const struct ovsdb_idl_column *column = &tc->columns[j];
+ ovsdb_idl_track_add_column(idl, column);
}
-
- table_updates = result->array.elems[2];
}
- if (clear_db) {
- ovsdb_idl_db_clear(db);
- }
- ovsdb_idl_db_parse_update(db, table_updates, version);
}
+/* Returns true if 'table' has any tracked column. */
static bool
-ovsdb_idl_db_parse_update_rpc(struct ovsdb_idl_db *db,
- const struct jsonrpc_msg *msg)
+ovsdb_idl_track_is_set(struct ovsdb_idl_table *table)
{
- if (msg->type != JSONRPC_NOTIFY) {
- return false;
- }
+ size_t i;
- int version;
- uint8_t n;
- if (!strcmp(msg->method, "update")) {
- version = 1;
- n = 2;
- } else if (!strcmp(msg->method, "update2")) {
- version = 2;
- n = 2;
- } else if (!strcmp(msg->method, "update3")) {
- version = 3;
- n = 3;
- } else {
- return false;
+ for (i = 0; i < table->class_->n_columns; i++) {
+ if (table->modes[i] & OVSDB_IDL_TRACK) {
+ return true;
+ }
}
+ return false;
+}
- struct json *params = msg->params;
- if (params->type != JSON_ARRAY || params->array.n != n) {
- struct ovsdb_error *error = ovsdb_syntax_error(params, NULL,
- "%s must be an array with %u elements.",
- msg->method, n);
- log_parse_update_error(error);
- return false;
- }
+/* Returns the first tracked row in table with class 'table_class'
+ * for the specified 'idl'. Returns NULL if there are no tracked rows */
+const struct ovsdb_idl_row *
+ovsdb_idl_track_get_first(const struct ovsdb_idl *idl,
+ const struct ovsdb_idl_table_class *table_class)
+{
+ struct ovsdb_idl_table *table
+ = ovsdb_idl_table_from_class(idl, table_class);
- if (!json_equal(params->array.elems[0], db->monitor_id)) {
- return false;
+ if (!ovs_list_is_empty(&table->track_list)) {
+ return CONTAINER_OF(ovs_list_front(&table->track_list), struct ovsdb_idl_row, track_node);
}
+ return NULL;
+}
- struct json *table_updates = params->array.elems[1];
- if (!strcmp(msg->method, "update3")) {
- table_updates = params->array.elems[2];
- const char *last_id = json_string(params->array.elems[1]);
- if (!uuid_from_string(&db->last_id, last_id)) {
- struct ovsdb_error *error = ovsdb_syntax_error(params, NULL,
- "Last-id %s is not in UUID format.",
- last_id);
- log_parse_update_error(error);
- return false;
- }
+/* Returns the next tracked row in table after the specified 'row'
+ * (in no particular order). Returns NULL if there are no tracked rows */
+const struct ovsdb_idl_row *
+ovsdb_idl_track_get_next(const struct ovsdb_idl_row *row)
+{
+ if (row->track_node.next != &row->table->track_list) {
+ return CONTAINER_OF(row->track_node.next, struct ovsdb_idl_row, track_node);
}
- ovsdb_idl_db_parse_update(db, table_updates, version);
- return true;
+
+ return NULL;
}
-static bool
-ovsdb_idl_handle_monitor_canceled(struct ovsdb_idl *idl,
- struct ovsdb_idl_db *db,
- const struct jsonrpc_msg *msg)
-{
- if (msg->type != JSONRPC_NOTIFY
- || strcmp(msg->method, "monitor_canceled")
- || msg->params->type != JSON_ARRAY
- || msg->params->array.n != 1
- || !json_equal(msg->params->array.elems[0], db->monitor_id)) {
+/* Returns true if a tracked 'column' in 'row' was updated by IDL, false
+ * otherwise. The tracking data is cleared by ovsdb_idl_track_clear()
+ *
+ * Function returns false if 'column' is not tracked (see
+ * ovsdb_idl_track_add_column()).
+ */
+bool
+ovsdb_idl_track_is_updated(const struct ovsdb_idl_row *row,
+ const struct ovsdb_idl_column *column)
+{
+ const struct ovsdb_idl_table_class *class;
+ size_t column_idx;
+
+ class = row->table->class_;
+ column_idx = column - class->columns;
+
+ if (row->updated && bitmap_is_set(row->updated, column_idx)) {
+ return true;
+ } else {
return false;
}
+}
+
+/* Flushes the tracked rows. Client calls this function after calling
+ * ovsdb_idl_run() and read all tracked rows with the ovsdb_idl_track_get_*()
+ * functions. This is usually done at the end of the client's processing
+ * loop when it is ready to do ovsdb_idl_run() again.
+ */
+void
+ovsdb_idl_track_clear(struct ovsdb_idl *idl)
+{
+ size_t i;
- db->monitoring = OVSDB_IDL_NOT_MONITORING;
+ for (i = 0; i < idl->class_->n_tables; i++) {
+ struct ovsdb_idl_table *table = &idl->tables[i];
- /* Cancel the other monitor and restart the FSM from the top.
- *
- * Maybe a more sophisticated response would be better in some cases, but
- * it doesn't seem worth optimizing yet. (Although this is already more
- * sophisticated than just dropping the connection and reconnecting.) */
- struct ovsdb_idl_db *other_db
- = db == &idl->data ? &idl->server : &idl->data;
- if (other_db->monitoring) {
- jsonrpc_session_send(
- idl->session,
- jsonrpc_create_request(
- "monitor_cancel",
- json_array_create_1(json_clone(other_db->monitor_id)), NULL));
- other_db->monitoring = OVSDB_IDL_NOT_MONITORING;
- }
- ovsdb_idl_restart_fsm(idl);
+ if (!ovs_list_is_empty(&table->track_list)) {
+ struct ovsdb_idl_row *row, *next;
- return true;
+ LIST_FOR_EACH_SAFE(row, next, track_node, &table->track_list) {
+ if (row->updated) {
+ free(row->updated);
+ row->updated = NULL;
+ }
+
+ row->change_seqno[OVSDB_IDL_CHANGE_INSERT] =
+ row->change_seqno[OVSDB_IDL_CHANGE_MODIFY] =
+ row->change_seqno[OVSDB_IDL_CHANGE_DELETE] = 0;
+
+ ovs_list_remove(&row->track_node);
+ ovs_list_init(&row->track_node);
+ if (ovsdb_idl_row_is_orphan(row) && row->tracked_old_datum) {
+ ovsdb_idl_row_unparse(row);
+ const struct ovsdb_idl_table_class *class =
+ row->table->class_;
+ for (size_t c = 0; c < class->n_columns; c++) {
+ ovsdb_datum_destroy(&row->tracked_old_datum[c],
+ &class->columns[c].type);
+ }
+ free(row->tracked_old_datum);
+ row->tracked_old_datum = NULL;
+ free(row);
+ }
+ }
+ }
+ }
+}
+
+static void
+log_parse_update_error(struct ovsdb_error *error)
+{
+ if (!VLOG_DROP_WARN(&syntax_rl)) {
+ char *s = ovsdb_error_to_string(error);
+ VLOG_WARN_RL(&syntax_rl, "%s", s);
+ free(s);
+ }
+ ovsdb_error_destroy(error);
}
static struct ovsdb_error *
-ovsdb_idl_db_parse_update__(struct ovsdb_idl_db *db,
- const struct ovsdb_cs_db_update *du)
+ovsdb_idl_parse_update__(struct ovsdb_idl *idl,
+ const struct ovsdb_cs_db_update *du)
{
for (size_t i = 0; i < du->n; i++) {
const struct ovsdb_cs_table_update *tu = &du->table_updates[i];
- struct ovsdb_idl_table *table = shash_find_data(&db->table_by_name,
+ struct ovsdb_idl_table *table = shash_find_data(&idl->table_by_name,
tu->table_name);
if (!table) {
return ovsdb_syntax_error(
@@ -2317,13 +1282,12 @@ ovsdb_idl_db_parse_update__(struct ovsdb_idl_db *db,
const struct ovsdb_cs_row_update *ru = &tu->row_updates[j];
switch (ovsdb_idl_process_update(table, ru)) {
case OVSDB_IDL_UPDATE_DB_CHANGED:
- db->change_seqno++;
+ idl->change_seqno++;
break;
case OVSDB_IDL_UPDATE_NO_CHANGES:
break;
case OVSDB_IDL_UPDATE_INCONSISTENT:
- memset(&db->last_id, 0, sizeof db->last_id);
- ovsdb_idl_retry(db->idl);
+ ovsdb_cs_flag_inconsistency(idl->cs);
return ovsdb_error(NULL,
"row update received for inconsistent "
"IDL: reconnecting IDL and resync all "
@@ -2336,15 +1300,25 @@ ovsdb_idl_db_parse_update__(struct ovsdb_idl_db *db,
}
static void
-ovsdb_idl_db_parse_update(struct ovsdb_idl_db *db,
- const struct json *json_table_updates,
- int version)
+ovsdb_idl_parse_update(struct ovsdb_idl *idl,
+ const struct ovsdb_cs_update_event *update)
{
struct ovsdb_cs_db_update *du;
struct ovsdb_error *error = ovsdb_cs_parse_db_update(
- json_table_updates, version, &du);
+ update->table_updates, update->version, &du);
if (!error) {
- error = ovsdb_idl_db_parse_update__(db, du);
+ /* If this is the first update notification we've received, then always
+ * indicate a change, even if nothing actually changes, because we
+ * should signal a change for an initially empty database. */
+ if (!idl->change_seqno) {
+ idl->change_seqno++;
+ }
+
+ if (update->clear) {
+ ovsdb_idl_clear(idl);
+ }
+
+ error = ovsdb_idl_parse_update__(idl, du);
}
ovsdb_cs_db_update_destroy(du);
if (error) {
@@ -2457,7 +1431,7 @@ add_tracked_change_for_references(struct ovsdb_idl_row *row)
ref->change_seqno[OVSDB_IDL_CHANGE_MODIFY]
= ref->table->change_seqno[OVSDB_IDL_CHANGE_MODIFY]
- = ref->table->db->change_seqno + 1;
+ = ref->table->idl->change_seqno + 1;
add_tracked_change_for_references(ref);
}
@@ -2479,6 +1453,7 @@ ovsdb_idl_row_change(struct ovsdb_idl_row *row, const struct shash *values,
const struct ovsdb_idl_table_class *class = table->class_;
struct shash_node *node;
bool changed = false;
+
SHASH_FOR_EACH (node, values) {
const char *column_name = node->name;
const struct ovsdb_idl_column *column;
@@ -2520,7 +1495,7 @@ ovsdb_idl_row_change(struct ovsdb_idl_row *row, const struct shash *values,
changed = true;
row->change_seqno[change]
= row->table->change_seqno[change]
- = row->table->db->change_seqno + 1;
+ = row->table->idl->change_seqno + 1;
if (table->modes[column_idx] & OVSDB_IDL_TRACK) {
if (ovs_list_is_empty(&row->track_node) &&
@@ -2686,20 +1661,25 @@ ovsdb_idl_index_generic_comparer(const void *a,
}
}
-static struct ovsdb_idl_index *
-ovsdb_idl_db_index_create(struct ovsdb_idl_db *db,
- const struct ovsdb_idl_index_column *columns,
- size_t n)
+/* Creates a new index for the given 'idl' and with the 'n' specified
+ * 'columns'.
+ *
+ * All indexes must be created before the first call to ovsdb_idl_run(). */
+struct ovsdb_idl_index *
+ovsdb_idl_index_create(struct ovsdb_idl *idl,
+ const struct ovsdb_idl_index_column *columns,
+ size_t n)
{
ovs_assert(n > 0);
struct ovsdb_idl_index *index = xzalloc(sizeof *index);
- index->table = ovsdb_idl_table_from_column(db, columns[0].column);
+ index->table = ovsdb_idl_table_from_column(idl, columns[0].column);
for (size_t i = 0; i < n; i++) {
const struct ovsdb_idl_index_column *c = &columns[i];
- ovs_assert(ovsdb_idl_table_from_column(db, c->column) == index->table);
- ovs_assert(*ovsdb_idl_db_get_mode(db, c->column) & OVSDB_IDL_MONITOR);
+ ovs_assert(ovsdb_idl_table_from_column(idl,
+ c->column) == index->table);
+ ovs_assert(*ovsdb_idl_get_mode(idl, c->column) & OVSDB_IDL_MONITOR);
}
index->columns = xmemdup(columns, n * sizeof *columns);
@@ -2711,18 +1691,6 @@ ovsdb_idl_db_index_create(struct ovsdb_idl_db *db,
return index;
}
-/* Creates a new index for the given 'idl' and with the 'n' specified
- * 'columns'.
- *
- * All indexes must be created before the first call to ovsdb_idl_run(). */
-struct ovsdb_idl_index *
-ovsdb_idl_index_create(struct ovsdb_idl *idl,
- const struct ovsdb_idl_index_column *columns,
- size_t n)
-{
- return ovsdb_idl_db_index_create(&idl->data, columns, n);
-}
-
struct ovsdb_idl_index *
ovsdb_idl_index_create1(struct ovsdb_idl *idl,
const struct ovsdb_idl_column *column1)
@@ -3067,7 +2035,7 @@ ovsdb_idl_row_destroy(struct ovsdb_idl_row *row)
if (ovsdb_idl_track_is_set(row->table)) {
row->change_seqno[OVSDB_IDL_CHANGE_DELETE]
= row->table->change_seqno[OVSDB_IDL_CHANGE_DELETE]
- = row->table->db->change_seqno + 1;
+ = row->table->idl->change_seqno + 1;
}
if (ovs_list_is_empty(&row->track_node)) {
ovs_list_push_back(&row->table->track_list, &row->track_node);
@@ -3118,12 +2086,10 @@ ovsdb_idl_destroy_all_set_op_lists(struct ovsdb_idl_row *row)
}
static void
-ovsdb_idl_row_destroy_postprocess(struct ovsdb_idl_db *db)
+ovsdb_idl_row_destroy_postprocess(struct ovsdb_idl *idl)
{
- size_t i;
-
- for (i = 0; i < db->class_->n_tables; i++) {
- struct ovsdb_idl_table *table = &db->tables[i];
+ for (size_t i = 0; i < idl->class_->n_tables; i++) {
+ struct ovsdb_idl_table *table = &idl->tables[i];
if (!ovs_list_is_empty(&table->track_list)) {
struct ovsdb_idl_row *row, *next;
@@ -3211,26 +2177,12 @@ may_add_arc(const struct ovsdb_idl_row *src, const struct ovsdb_idl_row *dst)
return arc->src != src;
}
-static struct ovsdb_idl_table *
-ovsdb_idl_db_table_from_class(const struct ovsdb_idl_db *db,
- const struct ovsdb_idl_table_class *table_class)
-{
- ptrdiff_t idx = table_class - db->class_->tables;
- return idx >= 0 && idx < db->class_->n_tables ? &db->tables[idx] : NULL;
-}
-
static struct ovsdb_idl_table *
ovsdb_idl_table_from_class(const struct ovsdb_idl *idl,
const struct ovsdb_idl_table_class *table_class)
{
- struct ovsdb_idl_table *table;
-
- table = ovsdb_idl_db_table_from_class(&idl->data, table_class);
- if (!table) {
- table = ovsdb_idl_db_table_from_class(&idl->server, table_class);
- }
-
- return table;
+ ptrdiff_t idx = table_class - idl->class_->tables;
+ return idx >= 0 && idx < idl->class_->n_tables ? &idl->tables[idx] : NULL;
}
/* Called by ovsdb-idlc generated code. */
@@ -3239,14 +2191,14 @@ ovsdb_idl_get_row_arc(struct ovsdb_idl_row *src,
const struct ovsdb_idl_table_class *dst_table_class,
const struct uuid *dst_uuid)
{
- struct ovsdb_idl_db *db = src->table->db;
+ struct ovsdb_idl *idl = src->table->idl;
struct ovsdb_idl_table *dst_table;
struct ovsdb_idl_arc *arc;
struct ovsdb_idl_row *dst;
- dst_table = ovsdb_idl_db_table_from_class(db, dst_table_class);
+ dst_table = ovsdb_idl_table_from_class(idl, dst_table_class);
dst = ovsdb_idl_get_row(dst_table, dst_uuid);
- if (db->txn || is_index_row(src)) {
+ if (idl->txn || is_index_row(src)) {
/* There are two cases we should not update any arcs:
*
* 1. We're being called from ovsdb_idl_txn_write(). We must not update
@@ -3451,10 +2403,10 @@ ovsdb_idl_txn_create(struct ovsdb_idl *idl)
{
struct ovsdb_idl_txn *txn;
- ovs_assert(!idl->data.txn);
- idl->data.txn = txn = xmalloc(sizeof *txn);
+ ovs_assert(!idl->txn);
+ idl->txn = txn = xmalloc(sizeof *txn);
txn->request_id = NULL;
- txn->db = &idl->data;
+ txn->idl = idl;
hmap_init(&txn->txn_rows);
txn->status = TXN_UNCOMMITTED;
txn->error = NULL;
@@ -3544,10 +2496,11 @@ ovsdb_idl_txn_destroy(struct ovsdb_idl_txn *txn)
{
struct ovsdb_idl_txn_insert *insert, *next;
- json_destroy(txn->request_id);
if (txn->status == TXN_INCOMPLETE) {
- hmap_remove(&txn->db->outstanding_txns, &txn->hmap_node);
+ ovsdb_cs_forget_transaction(txn->idl->cs, txn->request_id);
+ hmap_remove(&txn->idl->outstanding_txns, &txn->hmap_node);
}
+ json_destroy(txn->request_id);
ovsdb_idl_txn_abort(txn);
ds_destroy(&txn->comment);
free(txn->error);
@@ -3642,7 +2595,7 @@ ovsdb_idl_txn_disassemble(struct ovsdb_idl_txn *txn)
* ovsdb_idl_column's 'parse' function, which will call
* ovsdb_idl_get_row_arc(), which will seen that the IDL is in a
* transaction and fail to update the graph. */
- txn->db->txn = NULL;
+ txn->idl->txn = NULL;
HMAP_FOR_EACH_SAFE (row, next, txn_node, &txn->txn_rows) {
enum { INSERTED, MODIFIED, DELETED } op
@@ -3935,39 +2888,22 @@ ovsdb_idl_txn_extract_mutations(struct ovsdb_idl_row *row,
enum ovsdb_idl_txn_status
ovsdb_idl_txn_commit(struct ovsdb_idl_txn *txn)
{
- struct ovsdb_idl_row *row;
- struct json *operations;
- bool any_updates;
-
- if (txn != txn->db->txn) {
+ struct ovsdb_idl *idl = txn->idl;
+ if (txn != idl->txn) {
goto coverage_out;
- }
-
- /* If we're still connecting or re-connecting, don't bother sending a
- * transaction. */
- if (txn->db->idl->state != IDL_S_MONITORING) {
+ } else if (!ovsdb_cs_may_send_transaction(idl->cs)) {
txn->status = TXN_TRY_AGAIN;
goto disassemble_out;
- }
-
- /* If we need a lock but don't have it, give up quickly. */
- if (txn->db->lock_name && !txn->db->has_lock) {
+ } else if (ovsdb_cs_get_lock(idl->cs) && !ovsdb_cs_has_lock(idl->cs)) {
txn->status = TXN_NOT_LOCKED;
goto disassemble_out;
}
- operations = json_array_create_1(
- json_string_create(txn->db->class_->database));
-
- /* Assert that we have the required lock (avoiding a race). */
- if (txn->db->lock_name) {
- struct json *op = json_object_create();
- json_array_add(operations, op);
- json_object_put_string(op, "op", "assert");
- json_object_put_string(op, "lock", txn->db->lock_name);
- }
+ struct json *operations = json_array_create_1(
+ json_string_create(idl->class_->database));
/* Add prerequisites and declarations of new rows. */
+ struct ovsdb_idl_row *row;
HMAP_FOR_EACH (row, txn_node, &txn->txn_rows) {
/* XXX check that deleted rows exist even if no prereqs? */
if (row->prereqs) {
@@ -3999,15 +2935,15 @@ ovsdb_idl_txn_commit(struct ovsdb_idl_txn *txn)
}
/* Add updates. */
- any_updates = false;
+ bool any_updates = false;
/* For tables constrained to have only a single row (a fairly common OVSDB
* pattern for storing global data), identify whether we're inserting a
* row. If so, then verify that the table is empty before inserting the
* row. This gives us a clear verification-related failure if there was an
* insertion race with another client. */
- for (size_t i = 0; i < txn->db->class_->n_tables; i++) {
- struct ovsdb_idl_table *table = &txn->db->tables[i];
+ for (size_t i = 0; i < idl->class_->n_tables; i++) {
+ struct ovsdb_idl_table *table = &idl->tables[i];
if (table->class_->is_singleton) {
/* Count the number of rows in the table before and after our
* transaction commits. This is O(n) in the number of rows in the
@@ -4182,18 +3118,15 @@ ovsdb_idl_txn_commit(struct ovsdb_idl_txn *txn)
if (!any_updates) {
txn->status = TXN_UNCHANGED;
json_destroy(operations);
- } else if (!txn->db->idl->session) {
- txn->status = TXN_TRY_AGAIN;
- json_destroy(operations);
- } else if (!jsonrpc_session_send(
- txn->db->idl->session,
- jsonrpc_create_request(
- "transact", operations, &txn->request_id))) {
- hmap_insert(&txn->db->outstanding_txns, &txn->hmap_node,
- json_hash(txn->request_id, 0));
- txn->status = TXN_INCOMPLETE;
} else {
- txn->status = TXN_TRY_AGAIN;
+ txn->request_id = ovsdb_cs_send_transaction(idl->cs, operations);
+ if (txn->request_id) {
+ hmap_insert(&idl->outstanding_txns, &txn->hmap_node,
+ json_hash(txn->request_id, 0));
+ txn->status = TXN_INCOMPLETE;
+ } else {
+ txn->status = TXN_TRY_AGAIN;
+ }
}
disassemble_out:
@@ -4226,8 +3159,8 @@ ovsdb_idl_txn_commit_block(struct ovsdb_idl_txn *txn)
fatal_signal_run();
while ((status = ovsdb_idl_txn_commit(txn)) == TXN_INCOMPLETE) {
- ovsdb_idl_run(txn->db->idl);
- ovsdb_idl_wait(txn->db->idl);
+ ovsdb_idl_run(txn->idl);
+ ovsdb_idl_wait(txn->idl);
ovsdb_idl_txn_wait(txn);
poll_block();
}
@@ -4318,7 +3251,7 @@ ovsdb_idl_txn_complete(struct ovsdb_idl_txn *txn,
enum ovsdb_idl_txn_status status)
{
txn->status = status;
- hmap_remove(&txn->db->outstanding_txns, &txn->hmap_node);
+ hmap_remove(&txn->idl->outstanding_txns, &txn->hmap_node);
}
static void
@@ -4345,7 +3278,7 @@ ovsdb_idl_txn_write__(const struct ovsdb_idl_row *row_,
ovs_assert(row->old_datum == NULL ||
row->table->modes[column_idx] & OVSDB_IDL_MONITOR);
- if (row->table->db->verify_write_only && !write_only) {
+ if (row->table->idl->verify_write_only && !write_only) {
VLOG_ERR("Bug: Attempt to write to a read/write column (%s:%s) when"
" explicitly configured not to.", class->name, column->name);
goto discard_datum;
@@ -4372,7 +3305,7 @@ ovsdb_idl_txn_write__(const struct ovsdb_idl_row *row_,
ovsdb_idl_remove_from_indexes(row);
}
if (hmap_node_is_null(&row->txn_node)) {
- hmap_insert(&row->table->db->txn->txn_rows, &row->txn_node,
+ hmap_insert(&row->table->idl->txn->txn_rows, &row->txn_node,
uuid_hash(&row->uuid));
}
if (row->old_datum == row->new_datum) {
@@ -4498,7 +3431,7 @@ ovsdb_idl_txn_verify(const struct ovsdb_idl_row *row_,
}
if (hmap_node_is_null(&row->txn_node)) {
- hmap_insert(&row->table->db->txn->txn_rows, &row->txn_node,
+ hmap_insert(&row->table->idl->txn->txn_rows, &row->txn_node,
uuid_hash(&row->uuid));
}
if (!row->prereqs) {
@@ -4531,12 +3464,12 @@ ovsdb_idl_txn_delete(const struct ovsdb_idl_row *row_)
ovsdb_idl_row_clear_new(row);
ovs_assert(!row->prereqs);
hmap_remove(&row->table->rows, &row->hmap_node);
- hmap_remove(&row->table->db->txn->txn_rows, &row->txn_node);
+ hmap_remove(&row->table->idl->txn->txn_rows, &row->txn_node);
free(row);
return;
}
if (hmap_node_is_null(&row->txn_node)) {
- hmap_insert(&row->table->db->txn->txn_rows, &row->txn_node,
+ hmap_insert(&row->table->idl->txn->txn_rows, &row->txn_node,
uuid_hash(&row->uuid));
}
ovsdb_idl_row_clear_new(row);
@@ -4569,7 +3502,7 @@ ovsdb_idl_txn_insert(struct ovsdb_idl_txn *txn,
uuid_generate(&row->uuid);
}
- row->table = ovsdb_idl_db_table_from_class(txn->db, class);
+ row->table = ovsdb_idl_table_from_class(txn->idl, class);
row->new_datum = xmalloc(class->n_columns * sizeof *row->new_datum);
hmap_insert(&row->table->rows, &row->hmap_node, uuid_hash(&row->uuid));
hmap_insert(&txn->txn_rows, &row->txn_node, uuid_hash(&row->uuid));
@@ -4579,29 +3512,22 @@ ovsdb_idl_txn_insert(struct ovsdb_idl_txn *txn,
}
static void
-ovsdb_idl_db_txn_abort_all(struct ovsdb_idl_db *db)
+ovsdb_idl_txn_abort_all(struct ovsdb_idl *idl)
{
struct ovsdb_idl_txn *txn;
- HMAP_FOR_EACH (txn, hmap_node, &db->outstanding_txns) {
+ HMAP_FOR_EACH (txn, hmap_node, &idl->outstanding_txns) {
ovsdb_idl_txn_complete(txn, TXN_TRY_AGAIN);
}
}
-static void
-ovsdb_idl_txn_abort_all(struct ovsdb_idl *idl)
-{
- ovsdb_idl_db_txn_abort_all(&idl->server);
- ovsdb_idl_db_txn_abort_all(&idl->data);
-}
-
static struct ovsdb_idl_txn *
-ovsdb_idl_db_txn_find(struct ovsdb_idl_db *db, const struct json *id)
+ovsdb_idl_txn_find(struct ovsdb_idl *idl, const struct json *id)
{
struct ovsdb_idl_txn *txn;
HMAP_FOR_EACH_WITH_HASH (txn, hmap_node,
- json_hash(id, 0), &db->outstanding_txns) {
+ json_hash(id, 0), &idl->outstanding_txns) {
if (json_equal(id, txn->request_id)) {
return txn;
}
@@ -4640,7 +3566,7 @@ ovsdb_idl_txn_process_inc_reply(struct ovsdb_idl_txn *txn,
}
/* We know that this is a JSON object because the loop in
- * ovsdb_idl_db_txn_process_reply() checked. */
+ * ovsdb_idl_txn_process_reply() checked. */
mutate = json_object(results->elems[txn->inc_index]);
count = shash_find_data(mutate, "count");
if (!check_json_type(count, JSON_INTEGER, "\"mutate\" reply \"count\"")) {
@@ -4716,18 +3642,16 @@ ovsdb_idl_txn_process_insert_reply(struct ovsdb_idl_txn_insert *insert,
return true;
}
-static bool
-ovsdb_idl_db_txn_process_reply(struct ovsdb_idl_db *db,
- const struct jsonrpc_msg *msg)
+static void
+ovsdb_idl_txn_process_reply(struct ovsdb_idl *idl,
+ const struct jsonrpc_msg *msg)
{
- struct ovsdb_idl_txn *txn;
- enum ovsdb_idl_txn_status status;
-
- txn = ovsdb_idl_db_txn_find(db, msg->id);
+ struct ovsdb_idl_txn *txn = ovsdb_idl_txn_find(idl, msg->id);
if (!txn) {
- return false;
+ return;
}
+ enum ovsdb_idl_txn_status status;
if (msg->type == JSONRPC_ERROR) {
if (msg->error
&& msg->error->type == JSON_STRING
@@ -4768,7 +3692,7 @@ ovsdb_idl_db_txn_process_reply(struct ovsdb_idl_db *db,
soft_errors++;
} else if (!strcmp(error->string,
"unknown database")) {
- ovsdb_idl_retry(db->idl);
+ ovsdb_cs_flag_inconsistency(idl->cs);
soft_errors++;
} else if (!strcmp(error->string, "not owner")) {
lock_errors++;
@@ -4817,7 +3741,6 @@ ovsdb_idl_db_txn_process_reply(struct ovsdb_idl_db *db,
}
ovsdb_idl_txn_complete(txn, status);
- return true;
}
/* Returns the transaction currently active for 'row''s IDL. A transaction
@@ -4825,7 +3748,7 @@ ovsdb_idl_db_txn_process_reply(struct ovsdb_idl_db *db,
struct ovsdb_idl_txn *
ovsdb_idl_txn_get(const struct ovsdb_idl_row *row)
{
- struct ovsdb_idl_txn *txn = row->table->db->txn;
+ struct ovsdb_idl_txn *txn = row->table->idl->txn;
ovs_assert(txn != NULL);
return txn;
}
@@ -4834,7 +3757,7 @@ ovsdb_idl_txn_get(const struct ovsdb_idl_row *row)
struct ovsdb_idl *
ovsdb_idl_txn_get_idl (struct ovsdb_idl_txn *txn)
{
- return txn->db->idl;
+ return txn->idl;
}
/* Blocks until 'idl' successfully connects to the remote database and
@@ -4852,31 +3775,6 @@ ovsdb_idl_get_initial_snapshot(struct ovsdb_idl *idl)
}
}
-static struct jsonrpc_msg *
-ovsdb_idl_db_set_lock(struct ovsdb_idl_db *db, const char *lock_name)
-{
- ovs_assert(!db->txn);
- ovs_assert(hmap_is_empty(&db->outstanding_txns));
-
- if (db->lock_name
- && (!lock_name || strcmp(lock_name, db->lock_name))) {
- /* Release previous lock. */
- struct jsonrpc_msg *msg = ovsdb_idl_db_compose_unlock_request(db);
- free(db->lock_name);
- db->lock_name = NULL;
- db->is_lock_contended = false;
- return msg;
- }
-
- if (lock_name && !db->lock_name) {
- /* Acquire new lock. */
- db->lock_name = xstrdup(lock_name);
- return ovsdb_idl_db_compose_lock_request(db);
- }
-
- return NULL;
-}
-
/* If 'lock_name' is nonnull, configures 'idl' to obtain the named lock from
* the database server and to avoid modifying the database when the lock cannot
* be acquired (that is, when another client has the same lock).
@@ -4886,17 +3784,7 @@ ovsdb_idl_db_set_lock(struct ovsdb_idl_db *db, const char *lock_name)
void
ovsdb_idl_set_lock(struct ovsdb_idl *idl, const char *lock_name)
{
- for (;;) {
- struct jsonrpc_msg *msg = ovsdb_idl_db_set_lock(&idl->data, lock_name);
- if (!msg) {
- break;
- }
- if (idl->session) {
- jsonrpc_session_send(idl->session, msg);
- } else {
- jsonrpc_msg_destroy(msg);
- }
- }
+ ovsdb_cs_set_lock(idl->cs, lock_name);
}
/* Returns true if 'idl' is configured to obtain a lock and owns that lock.
@@ -4908,7 +3796,7 @@ ovsdb_idl_set_lock(struct ovsdb_idl *idl, const char *lock_name)
bool
ovsdb_idl_has_lock(const struct ovsdb_idl *idl)
{
- return idl->data.has_lock;
+ return ovsdb_cs_has_lock(idl->cs);
}
/* Returns true if 'idl' is configured to obtain a lock but the database server
@@ -4916,122 +3804,7 @@ ovsdb_idl_has_lock(const struct ovsdb_idl *idl)
bool
ovsdb_idl_is_lock_contended(const struct ovsdb_idl *idl)
{
- return idl->data.is_lock_contended;
-}
-
-static void
-ovsdb_idl_db_update_has_lock(struct ovsdb_idl_db *db, bool new_has_lock)
-{
- if (new_has_lock && !db->has_lock) {
- if (db->idl->state == IDL_S_MONITORING) {
- db->change_seqno++;
- } else {
- /* We're setting up a session, so don't signal that the database
- * changed. Finalizing the session will increment change_seqno
- * anyhow. */
- }
- db->is_lock_contended = false;
- }
- db->has_lock = new_has_lock;
-}
-
-static bool
-ovsdb_idl_db_process_lock_replies(struct ovsdb_idl_db *db,
- const struct jsonrpc_msg *msg)
-{
- if (msg->type == JSONRPC_REPLY
- && db->lock_request_id
- && json_equal(db->lock_request_id, msg->id)) {
- /* Reply to our "lock" request. */
- ovsdb_idl_db_parse_lock_reply(db, msg->result);
- return true;
- }
-
- if (msg->type == JSONRPC_NOTIFY) {
- if (!strcmp(msg->method, "locked")) {
- /* We got our lock. */
- return ovsdb_idl_db_parse_lock_notify(db, msg->params, true);
- } else if (!strcmp(msg->method, "stolen")) {
- /* Someone else stole our lock. */
- return ovsdb_idl_db_parse_lock_notify(db, msg->params, false);
- }
- }
-
- return false;
-}
-
-static struct jsonrpc_msg *
-ovsdb_idl_db_compose_lock_request__(struct ovsdb_idl_db *db,
- const char *method)
-{
- ovsdb_idl_db_update_has_lock(db, false);
-
- json_destroy(db->lock_request_id);
- db->lock_request_id = NULL;
-
- struct json *params = json_array_create_1(json_string_create(
- db->lock_name));
- return jsonrpc_create_request(method, params, NULL);
-}
-
-static struct jsonrpc_msg *
-ovsdb_idl_db_compose_lock_request(struct ovsdb_idl_db *db)
-{
- struct jsonrpc_msg *msg = ovsdb_idl_db_compose_lock_request__(db, "lock");
- db->lock_request_id = json_clone(msg->id);
- return msg;
-}
-
-static struct jsonrpc_msg *
-ovsdb_idl_db_compose_unlock_request(struct ovsdb_idl_db *db)
-{
- return ovsdb_idl_db_compose_lock_request__(db, "unlock");
-}
-
-static void
-ovsdb_idl_db_parse_lock_reply(struct ovsdb_idl_db *db,
- const struct json *result)
-{
- bool got_lock;
-
- json_destroy(db->lock_request_id);
- db->lock_request_id = NULL;
-
- if (result->type == JSON_OBJECT) {
- const struct json *locked;
-
- locked = shash_find_data(json_object(result), "locked");
- got_lock = locked && locked->type == JSON_TRUE;
- } else {
- got_lock = false;
- }
-
- ovsdb_idl_db_update_has_lock(db, got_lock);
- if (!got_lock) {
- db->is_lock_contended = true;
- }
-}
-
-static bool
-ovsdb_idl_db_parse_lock_notify(struct ovsdb_idl_db *db,
- const struct json *params,
- bool new_has_lock)
-{
- if (db->lock_name
- && params->type == JSON_ARRAY
- && json_array(params)->n > 0
- && json_array(params)->elems[0]->type == JSON_STRING) {
- const char *lock_name = json_string(json_array(params)->elems[0]);
-
- if (!strcmp(db->lock_name, lock_name)) {
- ovsdb_idl_db_update_has_lock(db, new_has_lock);
- if (!new_has_lock) {
- db->is_lock_contended = true;
- }
- return true;
- }
- }
- return false;
+ return ovsdb_cs_is_lock_contended(idl->cs);
}
/* Inserts a new Map Operation into current transaction. */
@@ -5065,7 +3838,7 @@ ovsdb_idl_txn_add_map_op(struct ovsdb_idl_row *row,
/* Add this row to transaction's list of rows. */
if (hmap_node_is_null(&row->txn_node)) {
- hmap_insert(&row->table->db->txn->txn_rows, &row->txn_node,
+ hmap_insert(&row->table->idl->txn->txn_rows, &row->txn_node,
uuid_hash(&row->uuid));
}
}
@@ -5101,7 +3874,7 @@ ovsdb_idl_txn_add_set_op(struct ovsdb_idl_row *row,
/* Add this row to the transactions's list of rows. */
if (hmap_node_is_null(&row->txn_node)) {
- hmap_insert(&row->table->db->txn->txn_rows, &row->txn_node,
+ hmap_insert(&row->table->idl->txn->txn_rows, &row->txn_node,
uuid_hash(&row->uuid));
}
}
--
2.28.0
More information about the dev
mailing list