ofproto: Handle multipart requests with multiple parts.

OpenFlow has a concept of multipart messages, that is, messages that can be
broken into multiple pieces that are sent separately.  Before OpenFlow 1.3,
only replies could actually have multiple pieces.  OpenFlow 1.3 introduced
the idea that requests could have multiple pieces.  This is only useful for
multipart requests that take an array as part of the request, which amounts
to only flow monitoring requests and table features requests.  So far, OVS
hasn't implemented the multipart versions of these (it just reports an
error).  This commit introduces the necessary infastructure to implement
them properly.

Acked-by: Justin Pettit <jpettit@ovn.org>
Signed-off-by: Ben Pfaff <blp@ovn.org>

3 files changed, 50 insertions, 28 deletions

diff --git a/ofproto/connmgr.c b/ofproto/connmgr.c
index 7a7790c0b..a3f49a8ef 100644
--- a/ofproto/connmgr.c
+++ b/ofproto/connmgr.c
@@ -101,6 +101,9 @@ struct ofconn {
     long long int next_op_report;    /* Time to report ops, or LLONG_MAX. */
     long long int op_backoff;        /* Earliest time to report ops again. */
 
+    /* Reassembly of multipart requests. */
+    struct hmap assembler;
+
 /* Flow monitors (e.g. NXST_FLOW_MONITOR). */
 
     /* Configuration.  Contains "struct ofmonitor"s. */
@@ -156,7 +159,7 @@ static void ofconn_reconfigure(struct ofconn *,
 
 static void ofconn_run(struct ofconn *,
                        void (*handle_openflow)(struct ofconn *,
-                                               const struct ofpbuf *ofp_msg));
+                                               const struct ovs_list *msgs));
 static void ofconn_wait(struct ofconn *);
 
 static void ofconn_log_flow_mods(struct ofconn *);
@@ -343,7 +346,7 @@ connmgr_destroy(struct connmgr *mgr)
 void
 connmgr_run(struct connmgr *mgr,
             void (*handle_openflow)(struct ofconn *,
-                                    const struct ofpbuf *ofp_msg))
+                                    const struct ovs_list *msgs))
     OVS_EXCLUDED(ofproto_mutex)
 {
     if (mgr->in_band) {
@@ -1260,6 +1263,8 @@ ofconn_create(struct connmgr *mgr, struct rconn *rconn, enum ofconn_type type,
     hmap_init(&ofconn->bundles);
     ofconn->next_bundle_expiry_check = time_msec() + BUNDLE_EXPIRY_INTERVAL;
 
+    hmap_init(&ofconn->assembler);
+
     ofconn_flush(ofconn);
 
     return ofconn;
@@ -1312,6 +1317,8 @@ ofconn_flush(struct ofconn *ofconn)
     rconn_packet_counter_destroy(ofconn->monitor_counter);
     ofconn->monitor_counter = rconn_packet_counter_create();
     ofpbuf_list_delete(&ofconn->updates); /* ...but it should be empty. */
+
+    ofpmp_assembler_clear(&ofconn->assembler);
 }
 
 static void
@@ -1338,6 +1345,9 @@ ofconn_destroy(struct ofconn *ofconn)
     rconn_packet_counter_destroy(ofconn->packet_in_counter);
     rconn_packet_counter_destroy(ofconn->reply_counter);
     rconn_packet_counter_destroy(ofconn->monitor_counter);
+
+    hmap_destroy(&ofconn->assembler);
+
     free(ofconn);
 }
 
@@ -1375,7 +1385,7 @@ ofconn_may_recv(const struct ofconn *ofconn)
 static void
 ofconn_run(struct ofconn *ofconn,
            void (*handle_openflow)(struct ofconn *,
-                                   const struct ofpbuf *ofp_msg))
+                                   const struct ovs_list *msgs))
 {
     struct connmgr *mgr = ofconn->connmgr;
 
@@ -1399,8 +1409,16 @@ ofconn_run(struct ofconn *ofconn,
             fail_open_maybe_recover(mgr->fail_open);
         }
 
-        handle_openflow(ofconn, of_msg);
-        ofpbuf_delete(of_msg);
+        struct ovs_list msgs;
+        enum ofperr error = ofpmp_assembler_execute(&ofconn->assembler, of_msg,
+                                                    &msgs, time_msec());
+        if (error) {
+            ofconn_send_error(ofconn, of_msg->data, error);
+            ofpbuf_delete(of_msg);
+        } else if (!ovs_list_is_empty(&msgs)) {
+            handle_openflow(ofconn, &msgs);
+            ofpbuf_list_delete(&msgs);
+        }
     }
 
     long long int now = time_msec();
@@ -1414,6 +1432,12 @@ ofconn_run(struct ofconn *ofconn,
         ofconn_log_flow_mods(ofconn);
     }
 
+    struct ofpbuf *error = ofpmp_assembler_run(&ofconn->assembler,
+                                               time_msec());
+    if (error) {
+        ofconn_send(ofconn, error, NULL);
+    }
+
     ovs_mutex_lock(&ofproto_mutex);
     if (!rconn_is_alive(ofconn->rconn)) {
         ofconn_destroy(ofconn);
@@ -1436,6 +1460,7 @@ ofconn_wait(struct ofconn *ofconn)
     if (ofconn->next_op_report != LLONG_MAX) {
         poll_timer_wait_until(ofconn->next_op_report);
     }
+    poll_timer_wait_until(ofpmp_assembler_wait(&ofconn->assembler));
 }
 
 static void
diff --git a/ofproto/connmgr.h b/ofproto/connmgr.h
index 11c8f9a85..7ca7fe67b 100644
--- a/ofproto/connmgr.h
+++ b/ofproto/connmgr.h
@@ -80,7 +80,7 @@ void connmgr_destroy(struct connmgr *)
 
 void connmgr_run(struct connmgr *,
                  void (*handle_openflow)(struct ofconn *,
-                                         const struct ofpbuf *ofp_msg));
+                                         const struct ovs_list *msgs));
 void connmgr_wait(struct connmgr *);
 
 void connmgr_get_memory_usage(const struct connmgr *, struct simap *usage);
diff --git a/ofproto/ofproto.c b/ofproto/ofproto.c
index dfcbc5499..bdc3ac684 100644
--- a/ofproto/ofproto.c
+++ b/ofproto/ofproto.c
@@ -258,7 +258,7 @@ static void delete_flows__(struct rule_collection *,
 static void ofproto_group_delete_all__(struct ofproto *)
     OVS_REQUIRES(ofproto_mutex);
 static bool ofproto_group_exists(const struct ofproto *, uint32_t group_id);
-static void handle_openflow(struct ofconn *, const struct ofpbuf *);
+static void handle_openflow(struct ofconn *, const struct ovs_list *msgs);
 static enum ofperr ofproto_flow_mod_init(struct ofproto *,
                                          struct ofproto_flow_mod *,
                                          const struct ofputil_flow_mod *fm,
@@ -8102,26 +8102,14 @@ handle_tlv_table_request(struct ofconn *ofconn, const struct ofp_header *oh)
     return 0;
 }
 
+/* Processes the single-part OpenFlow message 'oh' that was received on
+ * 'ofconn'.  Returns an ofperr that, if nonzero, the caller should send back
+ * to the controller. */
 static enum ofperr
-handle_openflow__(struct ofconn *ofconn, const struct ofpbuf *msg)
+handle_single_part_openflow(struct ofconn *ofconn, const struct ofp_header *oh,
+                            enum ofptype type)
     OVS_EXCLUDED(ofproto_mutex)
 {
-    const struct ofp_header *oh = msg->data;
-    enum ofptype type;
-    enum ofperr error;
-
-    error = ofptype_decode(&type, oh);
-    if (error) {
-        return error;
-    }
-    if (oh->version >= OFP13_VERSION && ofpmsg_is_stat_request(oh)
-        && ofpmp_more(oh)) {
-        /* We have no buffer implementation for multipart requests.
-         * Report overflow for requests which consists of multiple
-         * messages. */
-        return OFPERR_OFPBRC_MULTIPART_BUFFER_OVERFLOW;
-    }
-
     switch (type) {
         /* OpenFlow requests. */
     case OFPTYPE_ECHO_REQUEST:
@@ -8309,15 +8297,24 @@ handle_openflow__(struct ofconn *ofconn, const struct ofpbuf *msg)
 }
 
 static void
-handle_openflow(struct ofconn *ofconn, const struct ofpbuf *ofp_msg)
+handle_openflow(struct ofconn *ofconn, const struct ovs_list *msgs)
     OVS_EXCLUDED(ofproto_mutex)
 {
-    enum ofperr error = handle_openflow__(ofconn, ofp_msg);
+    COVERAGE_INC(ofproto_recv_openflow);
 
+    struct ofpbuf *msg = ofpbuf_from_list(ovs_list_front(msgs));
+    enum ofptype type;
+    enum ofperr error = ofptype_decode(&type, msg->data);
+    if (!error) {
+        if (!ovs_list_is_short(msgs)) {
+            error = OFPERR_OFPBRC_BAD_STAT;
+        } else {
+            error = handle_single_part_openflow(ofconn, msg->data, type);
+        }
+    }
     if (error) {
-        ofconn_send_error(ofconn, ofp_msg->data, error);
+        ofconn_send_error(ofconn, msg->data, error);
     }
-    COVERAGE_INC(ofproto_recv_openflow);
 }
 
 static uint64_t