ovsdb: generate update notifications for monitor_cond session

Hold session's conditions in ovsdb_monitor_session_condition. Pass it
to ovsdb_monitor for generating "update2" notifications.
Add functions that can generate "update2" notification for a
"monitor_cond" session.
JSON cache is enabled only for session's with true condition only.
"monitor_cond" and "monitor_cond_change" are RFC 7047 extensions
described by ovsdb-server(1) manpage.

Performance evaluation:
OVN is the main candidate for conditional monitoring usage. It is clear that
conditional monitoring reduces computation on the ovn-controller (client) side
due to the reduced size of flow tables and update messages. Performance
evaluation shows up to 75% computation reduction.
However, performance evaluation shows also a reduction in computation on the SB
ovsdb-server side proportional to the degree that each logical network is
spread over physical hosts in the DC. Evaluation shows that in a realistic
scenarios there is a computation reduction also in the server side.

Evaluation on simulated environment of 50 hosts and 1000 logical ports shows
the following results (cycles #):

LN spread over # hosts|    master    | patch        | change
-------------------------------------------------------------
            1         | 24597200127  | 24339235374  |  1.0%
            6         | 23788521572  | 19145229352  | 19.5%
           12         | 23886405758  | 17913143176  | 25.0%
           18         | 25812686279  | 23675094540  |  8.2%
           24         | 28414671499  | 24770202308  | 12.8%
           30         | 31487218890  | 28397543436  |  9.8%
           36         | 36116993930  | 34105388739  |  5.5%
           42         | 37898342465  | 38647139083  | -1.9%
           48         | 41637996229  | 41846616306  | -0.5%
           50         | 41679995357  | 43455565977  | -4.2%

Signed-off-by: Liran Schour <[email protected]>
Signed-off-by: Ben Pfaff <[email protected]>

ovsdb/jsonrpc-server.c

@@ -28,6 +28,7 @@
 #include "ovsdb-error.h"
 #include "ovsdb-parser.h"
 #include "ovsdb.h"
+#include "condition.h"
 #include "poll-loop.h"
 #include "reconnect.h"
 #include "row.h"
@@ -865,7 +866,8 @@ ovsdb_jsonrpc_session_got_request(struct ovsdb_jsonrpc_session *s,
             reply = execute_transaction(s, db, request);
         }
     } else if (!strcmp(request->method, "monitor") ||
-               (monitor2_enable__ && !strcmp(request->method, "monitor2"))) {
+               (monitor2_enable__ && !strcmp(request->method, "monitor2")) ||
+               !strcmp(request->method, "monitor_cond")) {
         struct ovsdb *db = ovsdb_jsonrpc_lookup_db(s, request, &reply);
         if (!reply) {
             int l = strlen(request->method) - strlen("monitor");
@@ -1064,6 +1066,7 @@ struct ovsdb_jsonrpc_monitor {
     uint64_t unflushed;         /* The first transaction that has not been
                                        flushed to the jsonrpc remote client. */
     enum ovsdb_monitor_version version;
+    struct ovsdb_monitor_session_condition *condition;/* Session's condition */
 };
 
 static struct ovsdb_jsonrpc_monitor *
@@ -1091,20 +1094,27 @@ parse_bool(struct ovsdb_parser *parser, const char *name, bool default_value)
 }
 
 static struct ovsdb_error * OVS_WARN_UNUSED_RESULT
-ovsdb_jsonrpc_parse_monitor_request(struct ovsdb_monitor *dbmon,
-                                    const struct ovsdb_table *table,
-                                    const struct json *monitor_request)
+ovsdb_jsonrpc_parse_monitor_request(
+                               struct ovsdb_monitor *dbmon,
+                               const struct ovsdb_table *table,
+                               struct ovsdb_monitor_session_condition *cond,
+                               const struct json *monitor_request)
 {
     const struct ovsdb_table_schema *ts = table->schema;
     enum ovsdb_monitor_selection select;
-    const struct json *columns, *select_json;
+    const struct json *columns, *select_json, *where = NULL;
     struct ovsdb_parser parser;
     struct ovsdb_error *error;
 
     ovsdb_parser_init(&parser, monitor_request, "table %s", ts->name);
+    if (cond) {
+        where = ovsdb_parser_member(&parser, "where", OP_ARRAY | OP_OPTIONAL);
+    }
     columns = ovsdb_parser_member(&parser, "columns", OP_ARRAY | OP_OPTIONAL);
+
     select_json = ovsdb_parser_member(&parser, "select",
                                       OP_OBJECT | OP_OPTIONAL);
+
     error = ovsdb_parser_finish(&parser);
     if (error) {
         return error;
@@ -1179,6 +1189,12 @@ ovsdb_jsonrpc_parse_monitor_request(struct ovsdb_monitor *dbmon,
             }
         }
     }
+    if (cond) {
+        error = ovsdb_monitor_table_condition_create(cond, table, where);
+        if (error) {
+            return error;
+        }
+    }
 
     return NULL;
 }
@@ -1217,6 +1233,9 @@ ovsdb_jsonrpc_monitor_create(struct ovsdb_jsonrpc_session *s, struct ovsdb *db,
     m->session = s;
     m->db = db;
     m->dbmon = ovsdb_monitor_create(db, m);
+    if (version == OVSDB_MONITOR_V2) {
+        m->condition = ovsdb_monitor_session_condition_create();
+    }
     m->unflushed = 0;
     m->version = version;
     hmap_insert(&s->monitors, &m->node, json_hash(monitor_id, 0));
@@ -1244,6 +1263,7 @@ ovsdb_jsonrpc_monitor_create(struct ovsdb_jsonrpc_session *s, struct ovsdb *db,
             for (i = 0; i < array->n; i++) {
                 error = ovsdb_jsonrpc_parse_monitor_request(m->dbmon,
                                                             table,
+                                                            m->condition,
                                                             array->elems[i]);
                 if (error) {
                     goto error;
@@ -1252,6 +1272,7 @@ ovsdb_jsonrpc_monitor_create(struct ovsdb_jsonrpc_session *s, struct ovsdb *db,
         } else {
             error = ovsdb_jsonrpc_parse_monitor_request(m->dbmon,
                                                         table,
+                                                        m->condition,
                                                         mr_value);
             if (error) {
                 goto error;
@@ -1267,6 +1288,11 @@ ovsdb_jsonrpc_monitor_create(struct ovsdb_jsonrpc_session *s, struct ovsdb *db,
         m->dbmon = dbmon;
     }
 
+    /* Only now we can bind session's condition to ovsdb_monitor */
+    if (m->condition) {
+        ovsdb_monitor_condition_bind(m->dbmon, m->condition);
+    }
+
     ovsdb_monitor_get_initial(m->dbmon);
     json = ovsdb_jsonrpc_monitor_compose_update(m, true);
     json = json ? json : json_object_create();
@@ -1323,7 +1349,7 @@ ovsdb_jsonrpc_monitor_compose_update(struct ovsdb_jsonrpc_monitor *m,
     }
 
     return ovsdb_monitor_get_update(m->dbmon, initial, &m->unflushed,
-                                    m->version);
+                                    m->condition, m->version);
 }
 
 static bool
@@ -1346,6 +1372,7 @@ ovsdb_jsonrpc_monitor_destroy(struct ovsdb_jsonrpc_monitor *m)
     json_destroy(m->monitor_id);
     hmap_remove(&m->session->monitors, &m->node);
     ovsdb_monitor_remove_jsonrpc_monitor(m->dbmon, m, m->unflushed);
+    ovsdb_monitor_session_condition_destroy(m->condition);
     free(m);
 }