static void ovsdb_jsonrpc_monitor_flush_all(struct ovsdb_jsonrpc_session *);
static bool ovsdb_jsonrpc_monitor_needs_flush(struct ovsdb_jsonrpc_session *);
static struct json *ovsdb_jsonrpc_monitor_compose_table_update(
- const struct ovsdb_jsonrpc_monitor *monitor, bool initial);
+ struct ovsdb_jsonrpc_monitor *monitor, bool initial);
\f
/* JSON-RPC database server. */
struct hmap_node node; /* In ovsdb_jsonrpc_session's "monitors". */
struct json *monitor_id;
struct ovsdb_monitor *dbmon;
+ uint64_t unflushed; /* The first transaction that has not been
+ flushed to the jsonrpc remote client. */
};
static struct ovsdb_jsonrpc_monitor *
m->session = s;
m->db = db;
m->dbmon = ovsdb_monitor_create(db, m);
+ m->unflushed = 0;
hmap_insert(&s->monitors, &m->node, json_hash(monitor_id, 0));
m->monitor_id = json_clone(monitor_id);
static struct json *
ovsdb_jsonrpc_monitor_compose_table_update(
- const struct ovsdb_jsonrpc_monitor *monitor, bool initial)
+ struct ovsdb_jsonrpc_monitor *monitor, bool initial)
{
- return ovsdb_monitor_compose_table_update(monitor->dbmon, initial);
+ return ovsdb_monitor_compose_table_update(monitor->dbmon, initial,
+ &monitor->unflushed);
}
static bool
struct ovsdb_jsonrpc_monitor *m;
HMAP_FOR_EACH (m, node, &s->monitors) {
- if (ovsdb_monitor_needs_flush(m->dbmon)) {
+ if (ovsdb_monitor_needs_flush(m->dbmon, m->unflushed)) {
return true;
}
}
struct shash tables; /* Holds "struct ovsdb_monitor_table"s. */
struct ovs_list jsonrpc_monitors; /* Contains "jsonrpc_monitor_node"s. */
struct ovsdb *db;
+ uint64_t n_transactions; /* Count number of committed transactions. */
};
struct jsonrpc_monitor_node {
ovsdb_add_replica(db, &dbmon->replica);
list_init(&dbmon->jsonrpc_monitors);
dbmon->db = db;
+ dbmon->n_transactions = 0;
shash_init(&dbmon->tables);
jm = xzalloc(sizeof *jm);
* be used as part of the initial reply to a "monitor" request, false if it is
* going to be used as part of an "update" notification. */
struct json *
-ovsdb_monitor_compose_table_update(
- const struct ovsdb_monitor *dbmon, bool initial)
+ovsdb_monitor_compose_table_update(const struct ovsdb_monitor *dbmon,
+ bool initial, uint64_t *unflushed)
{
struct shash_node *node;
unsigned long int *changed;
struct json *json;
size_t max_columns;
+ *unflushed = dbmon->n_transactions + 1;
+
max_columns = 0;
SHASH_FOR_EACH (node, &dbmon->tables) {
struct ovsdb_monitor_table *mt = node->data;
}
bool
-ovsdb_monitor_needs_flush(struct ovsdb_monitor *dbmon)
+ovsdb_monitor_needs_flush(struct ovsdb_monitor *dbmon,
+ uint64_t next_transaction)
{
- struct shash_node *node;
-
- SHASH_FOR_EACH (node, &dbmon->tables) {
- struct ovsdb_monitor_table *mt = node->data;
-
- if (!hmap_is_empty(&mt->changes)) {
- return true;
- }
- }
- return false;
+ ovs_assert(next_transaction <= dbmon->n_transactions + 1);
+ return (next_transaction <= dbmon->n_transactions);
}
void
ovsdb_monitor_init_aux(&aux, m);
ovsdb_txn_for_each_change(txn, ovsdb_monitor_change_cb, &aux);
+ m->n_transactions++;
return NULL;
}
ovsdb_monitor_table_check_duplicates(struct ovsdb_monitor *,
const struct ovsdb_table *);
-struct json *ovsdb_monitor_compose_table_update(
- const struct ovsdb_monitor *dbmon, bool initial);
+struct json *ovsdb_monitor_compose_table_update(const struct ovsdb_monitor *dbmon,
+ bool initial, uint64_t *unflushed_transaction);
void ovsdb_monitor_table_add_select(struct ovsdb_monitor *dbmon,
const struct ovsdb_table *table,
enum ovsdb_monitor_selection select);
-bool ovsdb_monitor_needs_flush(struct ovsdb_monitor *dbmon);
+bool ovsdb_monitor_needs_flush(struct ovsdb_monitor *dbmon,
+ uint64_t next_transaction);
void ovsdb_monitor_get_initial(const struct ovsdb_monitor *dbmon);
#endif