[ovs-dev,6/8] ofproto: Handle multipart requests with multiple parts.

Message ID 20180830200056.15484-6-blp@ovn.org
State New
Headers show
Series
  • [ovs-dev,1/8] vconn: Avoid null dereference on error path.
Related show

Commit Message

Ben Pfaff Aug. 30, 2018, 8 p.m.
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.

Signed-off-by: Ben Pfaff <blp@ovn.org>
---
 include/openvswitch/ofp-msgs.h |  20 ++-
 lib/ofp-msgs.c                 | 340 +++++++++++++++++++++++++++++++++++------
 ofproto/connmgr.c              |  35 ++++-
 ofproto/connmgr.h              |   2 +-
 ofproto/ofproto.c              |  41 +++--
 5 files changed, 362 insertions(+), 76 deletions(-)

Comments

0-day Robot Aug. 30, 2018, 9:08 p.m. | #1
Bleep bloop.  Greetings Ben Pfaff, I am a robot and I have tried out your patch.
Thanks for your contribution.

I encountered some error that I wasn't expecting.  See the details below.


build:
         enum vlog_level level__ = LEVEL;                                \
              ^
./include/openvswitch/vlog.h:218:31: note: in expansion of macro 'VLOG_RL'
 #define VLOG_WARN_RL(RL, ...) VLOG_RL(RL, VLL_WARN, __VA_ARGS__)
                               ^
lib/ofp-msgs.c:461:13: note: in expansion of macro 'VLOG_WARN_RL'
             VLOG_WARN_RL(&rl, "received %s with incorrect length %u (must be "
             ^
./include/openvswitch/vlog.h:282:14: error: format '%u' expects argument of type 'unsigned int', but argument 7 has type 'size_t' [-Werror=format=]
         enum vlog_level level__ = LEVEL;                                \
              ^
./include/openvswitch/vlog.h:218:31: note: in expansion of macro 'VLOG_RL'
 #define VLOG_WARN_RL(RL, ...) VLOG_RL(RL, VLL_WARN, __VA_ARGS__)
                               ^
lib/ofp-msgs.c:461:13: note: in expansion of macro 'VLOG_WARN_RL'
             VLOG_WARN_RL(&rl, "received %s with incorrect length %u (must be "
             ^
lib/ofp-msgs.c: At top level:
cc1: error: unrecognized command line option "-Wno-null-pointer-arithmetic" [-Werror]
cc1: all warnings being treated as errors
make[2]: *** [lib/ofp-msgs.lo] Error 1
make[2]: Leaving directory `/var/lib/jenkins/jobs/upstream_build_from_pw/workspace'
make[1]: *** [all-recursive] Error 1
make[1]: Leaving directory `/var/lib/jenkins/jobs/upstream_build_from_pw/workspace'
make: *** [all] Error 2


Please check this out.  If you feel there has been an error, please email aconole@bytheb.org

Thanks,
0-day Robot
Ben Pfaff Aug. 30, 2018, 9:15 p.m. | #2
On Thu, Aug 30, 2018 at 05:08:15PM -0400, 0-day Robot wrote:
> Bleep bloop.  Greetings Ben Pfaff, I am a robot and I have tried out your patch.
> Thanks for your contribution.
> 
> I encountered some error that I wasn't expecting.  See the details below.
> 
> 
> build:
>          enum vlog_level level__ = LEVEL;                                \
>               ^
> ./include/openvswitch/vlog.h:218:31: note: in expansion of macro 'VLOG_RL'
>  #define VLOG_WARN_RL(RL, ...) VLOG_RL(RL, VLL_WARN, __VA_ARGS__)
>                                ^
> lib/ofp-msgs.c:461:13: note: in expansion of macro 'VLOG_WARN_RL'
>              VLOG_WARN_RL(&rl, "received %s with incorrect length %u (must be "
>              ^
> ./include/openvswitch/vlog.h:282:14: error: format '%u' expects argument of type 'unsigned int', but argument 7 has type 'size_t' [-Werror=format=]
>          enum vlog_level level__ = LEVEL;                                \
>               ^
> ./include/openvswitch/vlog.h:218:31: note: in expansion of macro 'VLOG_RL'
>  #define VLOG_WARN_RL(RL, ...) VLOG_RL(RL, VLL_WARN, __VA_ARGS__)
>                                ^
> lib/ofp-msgs.c:461:13: note: in expansion of macro 'VLOG_WARN_RL'
>              VLOG_WARN_RL(&rl, "received %s with incorrect length %u (must be "
>              ^

Good point, I'll fold this in:

diff --git a/lib/ofp-msgs.c b/lib/ofp-msgs.c
index 6b5dfee722a5..ddea4f25ef42 100644
--- a/lib/ofp-msgs.c
+++ b/lib/ofp-msgs.c
@@ -433,11 +433,11 @@ ofpraw_decode_assert(const struct ofp_header *oh)
 static enum ofperr
 ofpraw_check_length(const struct raw_info *info,
                     const struct raw_instance *instance,
-                    size_t len)
+                    unsigned int len)
 {
     static struct vlog_rate_limit rl = VLOG_RATE_LIMIT_INIT(1, 5);
 
-    size_t min_len = instance->hdrs_len + info->min_body;
+    unsigned int min_len = instance->hdrs_len + info->min_body;
     switch (info->extra_multiple) {
     case 0:
         if (len != min_len) {

Patch

diff --git a/include/openvswitch/ofp-msgs.h b/include/openvswitch/ofp-msgs.h
index 8a32a3dc69fa..66897b1cdeab 100644
--- a/include/openvswitch/ofp-msgs.h
+++ b/include/openvswitch/ofp-msgs.h
@@ -45,6 +45,7 @@ 
 extern "C" {
 #endif
 
+struct hmap;
 struct ovs_list;
 
 /* Raw identifiers for OpenFlow messages.
@@ -815,10 +816,27 @@  void ofpmp_postappend(struct ovs_list *, size_t start_ofs);
 enum ofp_version ofpmp_version(struct ovs_list *);
 enum ofpraw ofpmp_decode_raw(struct ovs_list *);
 
-/* Decoding multipart replies. */
+/* Decoding multipart messages. */
 uint16_t ofpmp_flags(const struct ofp_header *);
 bool ofpmp_more(const struct ofp_header *);
 
+/* Multipart request assembler.
+ *
+ * OpenFlow 1.3 and later support making multipart requests that span more than
+ * one OpenFlow message.  These functions reassemble such requests.
+ *
+ * A reassembler is simply an hmap.  The following functions manipulate an hmap
+ * used for this purpose. */
+
+void ofpmp_assembler_clear(struct hmap *assembler);
+
+struct ofpbuf *ofpmp_assembler_run(struct hmap *assembler, long long int now)
+    OVS_WARN_UNUSED_RESULT;
+long long int ofpmp_assembler_wait(struct hmap *assembler);
+
+enum ofperr ofpmp_assembler_execute(struct hmap *assembler, struct ofpbuf *msg,
+                                    struct ovs_list *out, long long int now);
+
 #ifdef __cplusplus
 }
 #endif
diff --git a/lib/ofp-msgs.c b/lib/ofp-msgs.c
index 6517210c2cdf..6b5dfee722a5 100644
--- a/lib/ofp-msgs.c
+++ b/lib/ofp-msgs.c
@@ -428,56 +428,16 @@  ofpraw_decode_assert(const struct ofp_header *oh)
     return raw;
 }
 
-/* Determines the OFPRAW_* type of the OpenFlow message in 'msg', which starts
- * at 'msg->data' and has length 'msg->size' bytes.  On success,
- * returns 0 and stores the type into '*rawp'.  On failure, returns an OFPERR_*
- * error code and zeros '*rawp'.
- *
- * This function checks that the message has a valid length for its particular
- * type of message, and returns an error if not.
- *
- * In addition to setting '*rawp', this function pulls off the OpenFlow header
- * (including the stats headers, vendor header, and any subtype header) with
- * ofpbuf_pull().  It also sets 'msg->header' to the start of the OpenFlow
- * header and 'msg->msg' just beyond the headers (that is, to the final value
- * of msg->data). */
-enum ofperr
-ofpraw_pull(enum ofpraw *rawp, struct ofpbuf *msg)
+/* Checks that 'len' is a valid length for an OpenFlow message that corresponds
+ * to 'info' and 'instance'.  Returns 0 if so, otherwise an OpenFlow error. */
+static enum ofperr
+ofpraw_check_length(const struct raw_info *info,
+                    const struct raw_instance *instance,
+                    size_t len)
 {
     static struct vlog_rate_limit rl = VLOG_RATE_LIMIT_INIT(1, 5);
 
-    const struct raw_instance *instance;
-    const struct raw_info *info;
-    struct ofphdrs hdrs;
-
-    unsigned int min_len;
-    unsigned int len;
-
-    enum ofperr error;
-    enum ofpraw raw;
-
-    /* Set default outputs. */
-    msg->header = msg->data;
-    msg->msg = msg->header;
-    *rawp = 0;
-
-    len = msg->size;
-    error = ofphdrs_decode(&hdrs, msg->data, len);
-    if (error) {
-        return error;
-    }
-
-    error = ofpraw_from_ofphdrs(&raw, &hdrs);
-    if (error) {
-        return error;
-    }
-
-    info = raw_info_get(raw);
-    instance = raw_instance_get(info, hdrs.version);
-    msg->header = ofpbuf_pull(msg, instance->hdrs_len);
-    msg->msg = msg->data;
-
-    min_len = instance->hdrs_len + info->min_body;
+    size_t min_len = instance->hdrs_len + info->min_body;
     switch (info->extra_multiple) {
     case 0:
         if (len != min_len) {
@@ -507,6 +467,51 @@  ofpraw_pull(enum ofpraw *rawp, struct ofpbuf *msg)
         break;
     }
 
+    return 0;
+}
+
+/* Determines the OFPRAW_* type of the OpenFlow message in 'msg', which starts
+ * at 'msg->data' and has length 'msg->size' bytes.  On success,
+ * returns 0 and stores the type into '*rawp'.  On failure, returns an OFPERR_*
+ * error code and zeros '*rawp'.
+ *
+ * This function checks that the message has a valid length for its particular
+ * type of message, and returns an error if not.
+ *
+ * In addition to setting '*rawp', this function pulls off the OpenFlow header
+ * (including the stats headers, vendor header, and any subtype header) with
+ * ofpbuf_pull().  It also sets 'msg->header' to the start of the OpenFlow
+ * header and 'msg->msg' just beyond the headers (that is, to the final value
+ * of msg->data). */
+enum ofperr
+ofpraw_pull(enum ofpraw *rawp, struct ofpbuf *msg)
+{
+    /* Set default outputs. */
+    msg->header = msg->data;
+    msg->msg = msg->header;
+    *rawp = 0;
+
+    struct ofphdrs hdrs;
+    enum ofperr error = ofphdrs_decode(&hdrs, msg->data, msg->size);
+    if (error) {
+        return error;
+    }
+
+    enum ofpraw raw;
+    error = ofpraw_from_ofphdrs(&raw, &hdrs);
+    if (error) {
+        return error;
+    }
+
+    const struct raw_info *info = raw_info_get(raw);
+    const struct raw_instance *instance = raw_instance_get(info, hdrs.version);
+    error = ofpraw_check_length(info, instance, msg->size);
+    if (error) {
+        return error;
+    }
+
+    msg->header = ofpbuf_pull(msg, instance->hdrs_len);
+    msg->msg = msg->data;
     *rawp = raw;
     return 0;
 }
@@ -1059,6 +1064,247 @@  ofpmp_more(const struct ofp_header *oh)
     return (ofpmp_flags(oh) & OFPSF_REPLY_MORE) != 0;
 }
 
+/* Multipart request assembler. */
+
+struct ofpmp_partial {
+    struct hmap_node hmap_node; /* In struct ofpmp_assembler's 'msgs'. */
+    ovs_be32 xid;
+    enum ofpraw raw;
+    long long int timeout;
+    struct ovs_list msgs;
+    size_t size;
+    bool has_body;
+};
+
+static uint32_t
+hash_xid(ovs_be32 xid)
+{
+    return hash_int((OVS_FORCE uint32_t) xid, 0);
+}
+
+static struct ofpmp_partial *
+ofpmp_assembler_find(struct hmap *assembler, ovs_be32 xid)
+{
+    if (hmap_is_empty(assembler)) {
+        /* Common case. */
+        return NULL;
+    }
+
+    struct ofpmp_partial *p;
+    HMAP_FOR_EACH_IN_BUCKET (p, hmap_node, hash_xid(xid), assembler) {
+        if (p->xid == xid) {
+            return p;
+        }
+    }
+    return NULL;
+}
+
+static void
+ofpmp_partial_destroy(struct hmap *assembler, struct ofpmp_partial *p)
+{
+    if (p) {
+        hmap_remove(assembler, &p->hmap_node);
+        ofpbuf_list_delete(&p->msgs);
+        free(p);
+    }
+}
+
+static struct ofpbuf *
+ofpmp_partial_error(struct hmap *assembler, struct ofpmp_partial *p,
+                    enum ofperr error)
+{
+    const struct ofpbuf *head = ofpbuf_from_list(ovs_list_back(&p->msgs));
+    const struct ofp_header *oh = head->data;
+    struct ofpbuf *reply = ofperr_encode_reply(error, oh);
+
+    ofpmp_partial_destroy(assembler, p);
+
+    return reply;
+}
+
+/* Clears out and frees any messages currently being reassembled.  Afterward,
+ * the caller may destroy the hmap, with hmap_destroy(), without risk of
+ * leaks. */
+void
+ofpmp_assembler_clear(struct hmap *assembler)
+{
+    struct ofpmp_partial *p, *next;
+    HMAP_FOR_EACH_SAFE (p, next, hmap_node, assembler) {
+        ofpmp_partial_destroy(assembler, p);
+    }
+}
+
+/* Does periodic maintenance on 'assembler'.  If any partially assembled
+ * requests have timed out, returns an appropriate error message for the caller
+ * to send to the controller.
+ *
+ * 'now' should be the current time as returned by time_msec(). */
+struct ofpbuf * OVS_WARN_UNUSED_RESULT
+ofpmp_assembler_run(struct hmap *assembler, long long int now)
+{
+    struct ofpmp_partial *p;
+    HMAP_FOR_EACH (p, hmap_node, assembler) {
+        if (now >= p->timeout) {
+            return ofpmp_partial_error(
+                assembler, p, OFPERR_OFPBRC_MULTIPART_REQUEST_TIMEOUT);
+        }
+    }
+    return NULL;
+}
+
+/* Returns the time at which the next partially assembled request times out.
+ * The caller should pass this time to poll_timer_wait_until(). */
+long long int
+ofpmp_assembler_wait(struct hmap *assembler)
+{
+    long long int timeout = LLONG_MAX;
+
+    struct ofpmp_partial *p;
+    HMAP_FOR_EACH (p, hmap_node, assembler) {
+        timeout = MIN(timeout, p->timeout);
+    }
+
+    return timeout;
+}
+
+/* Submits 'msg' to 'assembler' for reassembly.
+ *
+ * If 'msg' was accepted, returns 0 and initializes 'out' either to an empty
+ * list (if 'msg' is being held for reassembly) or to a list of one or more
+ * reassembled messages.  The reassembler takes ownership of 'msg'; the caller
+ * takes ownership of the messages in 'out'.
+ *
+ * If 'msg' was rejected, returns an OpenFlow error that the caller should
+ * reply to the caller and initializes 'out' as empty.  The caller retains
+ * ownership of 'msg'.
+ *
+ * 'now' should be the current time as returned by time_msec(). */
+enum ofperr
+ofpmp_assembler_execute(struct hmap *assembler, struct ofpbuf *msg,
+                        struct ovs_list *out, long long int now)
+{
+    ovs_list_init(out);
+
+    /* If the message is not a multipart request, pass it along without further
+     * inspection.
+     *
+     * We could also do this kind of early-out for multipart requests that have
+     * only a single piece, or for pre-OF1.3 multipart requests (since only
+     * OF1.3 introduced multipart requests with more than one piece), but we
+     * don't because this allows us to assure code that runs after us that
+     * invariants checked below on correct message lengths are always
+     * satisfied, even if there's only a single piece. */
+    struct ofp_header *oh = msg->data;
+    if (!ofpmsg_is_stat_request(oh)) {
+        ovs_list_push_back(out, &msg->list_node);
+        return 0;
+    }
+
+    /* Decode the multipart request. */
+    struct ofphdrs hdrs;
+    enum ofperr error = ofphdrs_decode(&hdrs, msg->data, msg->size);
+    if (error) {
+        return error;
+    }
+
+    enum ofpraw raw;
+    error = ofpraw_from_ofphdrs(&raw, &hdrs);
+    if (error) {
+        return error;
+    }
+
+    /* If the message has a nonempty body, check that it is a valid length.
+     *
+     * The OpenFlow spec says that pieces with empty bodies are allowed
+     * anywhere in a multipart sequence, so for now we allow such messages even
+     * if the overall multipart request requires a body. */
+    const struct raw_info *info = raw_info_get(raw);
+    const struct raw_instance *instance = raw_instance_get(info, hdrs.version);
+    unsigned int min_len = ofphdrs_len(&hdrs);
+    bool has_body = msg->size > min_len;
+    if (has_body) {
+        error = ofpraw_check_length(info, instance, msg->size);
+        if (error) {
+            return error;
+        }
+    }
+
+    /* Find or create an ofpmp_partial record. */
+    struct ofpmp_partial *p = ofpmp_assembler_find(assembler, oh->xid);
+    if (!p) {
+        p = xzalloc(sizeof *p);
+        hmap_insert(assembler, &p->hmap_node, hash_xid(oh->xid));
+        p->xid = oh->xid;
+        ovs_list_init(&p->msgs);
+        p->raw = raw;
+    }
+    p->timeout = now + 1000;
+
+    /* Check that the type is the same as any previous messages in this
+     * sequence. */
+    if (p->raw != raw) {
+        ofpmp_partial_destroy(assembler, p);
+        return OFPERR_OFPBRC_BAD_STAT;
+    }
+
+    /* Limit the size of a multipart sequence.
+     *
+     * (Table features requests can actually be over 1 MB.) */
+    p->size += msg->size;
+    if (p->size > 4 * 1024 * 1024) {
+        ofpmp_partial_destroy(assembler, p);
+        return OFPERR_OFPBRC_MULTIPART_BUFFER_OVERFLOW;
+    }
+
+    /* If a multipart request type requires a body, ensure that at least one of
+     * the pieces in a multipart request has one. */
+    bool more = oh->version >= OFP13_VERSION && ofpmp_more(oh);
+    if (has_body) {
+        p->has_body = true;
+    }
+    if (!more && !p->has_body && info->min_body) {
+        ofpmp_partial_destroy(assembler, p);
+        return OFPERR_OFPBRC_BAD_LEN;
+    }
+
+    /* Append the part to the list.
+     *
+     * If there are more pieces to come, we're done for now. */
+    ovs_list_push_back(&p->msgs, &msg->list_node);
+    if (more) {
+        return 0;
+    }
+
+    /* This multipart request is complete.  Move the messages from 'p' to 'out'
+     * and discard 'p'. */
+    ovs_list_move(out, &p->msgs);
+    ovs_list_init(&p->msgs);
+    ofpmp_partial_destroy(assembler, p);
+
+    /* Delete pieces with empty bodies from 'out' (but leave at least one
+     * piece).
+     *
+     * Most types of multipart requests have fixed-size bodies.  For example,
+     * OFPMP_PORT_DESCRIPTION has an 8-byte body.  Thus, it doesn't really make
+     * sense for a controller to use multiple pieces for these messages, and
+     * it's simpler to implement OVS as if they weren't really multipart.
+     *
+     * However, the OpenFlow spec says that messages with empty bodies are
+     * allowed anywhere in a multipart sequence, so in theory a controller
+     * could send an OFPMP_PORT_DESCRIPTION with an 8-byte body bracketed
+     * on either side by parts with 0-byte bodies.  We remove the 0-byte
+     * ones here to simplify processing later.
+     */
+    struct ofpbuf *b, *next;
+    LIST_FOR_EACH_SAFE (b, next, list_node, out) {
+        if (b->size <= min_len && !ovs_list_is_short(out)) {
+            ovs_list_remove(&b->list_node);
+            ofpbuf_delete(b);
+        }
+    }
+    return 0;
+}
+
 static void ofpmsgs_init(void);
 
 static const struct raw_info *
diff --git a/ofproto/connmgr.c b/ofproto/connmgr.c
index f78b4c5ff411..1f5964d57cce 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 *);
@@ -347,7 +350,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)
 {
     struct ofconn *ofconn, *next_ofconn;
@@ -1299,6 +1302,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;
@@ -1353,6 +1358,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
@@ -1379,6 +1386,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);
 }
 
@@ -1418,7 +1428,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;
     size_t i;
@@ -1443,8 +1453,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();
@@ -1458,6 +1476,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);
@@ -1482,6 +1506,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 eb3be16686c5..46c1c84d15fd 100644
--- a/ofproto/connmgr.h
+++ b/ofproto/connmgr.h
@@ -79,7 +79,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 9552a585d096..c9d73c10c0ae 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,
@@ -8081,26 +8081,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:
@@ -8288,15 +8276,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