[firefly] [PATCH RFC 6/7] ofproto/connmgr: Queue messages while bundles are active

[Alexandru Copot]

When at least one OF1.4 bundle is open, only bundle related
messages are processed by the switch. The rest are queued and
processed after the bundles are done.

Signed-off-by: Alexandru Copot <alex.mihai.c at gmail.com>
Cc: Daniel Baluta <dbaluta at ixiacom.com>
---
 ofproto/connmgr.c | 126 ++++++++++++++++++++++++++++++++++++++++++++++++++++++
 1 file changed, 126 insertions(+)

diff --git a/ofproto/connmgr.c b/ofproto/connmgr.c
index 369150d..69343f9 100644
--- a/ofproto/connmgr.c
+++ b/ofproto/connmgr.c
@@ -133,6 +133,9 @@ struct ofconn {
     /* Active bundles. Contains "struct ofp_bundle"s. */
     struct hmap bundles;
     struct ovs_mutex bundles_mutex;
+
+    /* Queued OpenFlow messages during the creation of bundles */
+    struct list queued;
 };
 
 static struct ofconn *ofconn_create(struct connmgr *, struct rconn *,
@@ -1185,6 +1188,7 @@ ofconn_create(struct connmgr *mgr, struct rconn *rconn, enum ofconn_type type,
 
     hmap_init(&ofconn->bundles);
     ovs_mutex_init(&ofconn->bundles_mutex);
+    list_init(&ofconn->queued);
 
     ofconn_flush(ofconn);
 
@@ -1329,6 +1333,105 @@ ofconn_may_recv(const struct ofconn *ofconn)
     return (!ofconn->blocked || ofconn->retry) && count < OFCONN_REPLY_MAX;
 }
 
+/* Returns true for the messages that a connection can process while it
+ * has open bundles. */
+static bool
+may_pass_with_bundles(const struct ofpbuf *ofp_msg)
+{
+    struct ofpbuf *clone;
+    const struct ofp_header *oh;
+    enum ofptype type;
+    enum ofperr error;
+
+    clone = ofpbuf_clone(ofp_msg);
+    oh = clone->data;
+
+    error = ofptype_decode(&type, oh);
+    if (error) {
+        return true;
+    }
+
+    switch (type) {
+    case OFPTYPE_HELLO:
+    case OFPTYPE_ECHO_REQUEST:
+    case OFPTYPE_BARRIER_REQUEST:
+    case OFPTYPE_BUNDLE_CONTROL:
+    case OFPTYPE_BUNDLE_ADD_MESSAGE:
+        return true;
+
+    case OFPTYPE_FEATURES_REQUEST:
+    case OFPTYPE_GET_CONFIG_REQUEST:
+    case OFPTYPE_GET_ASYNC_REQUEST:
+    case OFPTYPE_DESC_STATS_REQUEST:
+    case OFPTYPE_FLOW_STATS_REQUEST:
+    case OFPTYPE_AGGREGATE_STATS_REQUEST:
+    case OFPTYPE_TABLE_STATS_REQUEST:
+    case OFPTYPE_PORT_STATS_REQUEST:
+    case OFPTYPE_QUEUE_STATS_REQUEST:
+    case OFPTYPE_PORT_DESC_STATS_REQUEST:
+    case OFPTYPE_FLOW_MONITOR_STATS_REQUEST:
+    case OFPTYPE_METER_STATS_REQUEST:
+    case OFPTYPE_METER_CONFIG_STATS_REQUEST:
+    case OFPTYPE_METER_FEATURES_STATS_REQUEST:
+    case OFPTYPE_GROUP_STATS_REQUEST:
+    case OFPTYPE_GROUP_DESC_STATS_REQUEST:
+    case OFPTYPE_GROUP_FEATURES_STATS_REQUEST:
+    case OFPTYPE_QUEUE_GET_CONFIG_REQUEST:
+
+    case OFPTYPE_SET_CONFIG:
+    case OFPTYPE_PACKET_OUT:
+    case OFPTYPE_PORT_MOD:
+    case OFPTYPE_FLOW_MOD:
+    case OFPTYPE_GROUP_MOD:
+    case OFPTYPE_TABLE_MOD:
+    case OFPTYPE_METER_MOD:
+    case OFPTYPE_ROLE_REQUEST:
+    case OFPTYPE_ECHO_REPLY:
+    case OFPTYPE_FLOW_MOD_TABLE_ID:
+    case OFPTYPE_SET_FLOW_FORMAT:
+    case OFPTYPE_SET_PACKET_IN_FORMAT:
+    case OFPTYPE_SET_CONTROLLER_ID:
+    case OFPTYPE_FLOW_AGE:
+    case OFPTYPE_FLOW_MONITOR_CANCEL:
+    case OFPTYPE_SET_ASYNC_CONFIG:
+
+    case OFPTYPE_ERROR:
+    case OFPTYPE_FEATURES_REPLY:
+    case OFPTYPE_GET_CONFIG_REPLY:
+    case OFPTYPE_PACKET_IN:
+    case OFPTYPE_FLOW_REMOVED:
+    case OFPTYPE_PORT_STATUS:
+    case OFPTYPE_BARRIER_REPLY:
+    case OFPTYPE_QUEUE_GET_CONFIG_REPLY:
+    case OFPTYPE_DESC_STATS_REPLY:
+    case OFPTYPE_FLOW_STATS_REPLY:
+    case OFPTYPE_QUEUE_STATS_REPLY:
+    case OFPTYPE_PORT_STATS_REPLY:
+    case OFPTYPE_TABLE_STATS_REPLY:
+    case OFPTYPE_AGGREGATE_STATS_REPLY:
+    case OFPTYPE_PORT_DESC_STATS_REPLY:
+    case OFPTYPE_ROLE_REPLY:
+    case OFPTYPE_FLOW_MONITOR_PAUSED:
+    case OFPTYPE_FLOW_MONITOR_RESUMED:
+    case OFPTYPE_FLOW_MONITOR_STATS_REPLY:
+    case OFPTYPE_GET_ASYNC_REPLY:
+    case OFPTYPE_GROUP_STATS_REPLY:
+    case OFPTYPE_GROUP_DESC_STATS_REPLY:
+    case OFPTYPE_GROUP_FEATURES_STATS_REPLY:
+    case OFPTYPE_METER_STATS_REPLY:
+    case OFPTYPE_METER_CONFIG_STATS_REPLY:
+    case OFPTYPE_METER_FEATURES_STATS_REPLY:
+    case OFPTYPE_TABLE_FEATURES_STATS_REQUEST:
+    case OFPTYPE_TABLE_FEATURES_STATS_REPLY:
+    case OFPTYPE_ROLE_STATUS:
+        return false;
+    default:
+        return false;
+    }
+
+    return false;
+}
+
 static void
 ofconn_run(struct ofconn *ofconn,
            bool (*handle_openflow)(struct ofconn *,
@@ -1347,6 +1450,24 @@ ofconn_run(struct ofconn *ofconn,
     rconn_run(ofconn->rconn);
 
     if (handle_openflow) {
+        if (!hmap_count(&ofconn->bundles) && !list_is_empty(&ofconn->queued)) {
+            for (i = 0; i < 50 && !list_is_empty(&ofconn->queued); i++) {
+                struct ofpbuf *of_msg;
+
+                of_msg = ofpbuf_from_list(list_pop_front(&ofconn->queued));
+                if (!of_msg) {
+                    break;
+                }
+                if (handle_openflow(ofconn, of_msg)) {
+                    ofpbuf_delete(of_msg);
+                    ofconn->blocked = NULL;
+                } else {
+                    ofconn->blocked = of_msg;
+                    ofconn->retry = false;
+                }
+            }
+        }
+
         /* Limit the number of iterations to avoid starving other tasks. */
         for (i = 0; i < 50 && ofconn_may_recv(ofconn); i++) {
             struct ofpbuf *of_msg;
@@ -1361,6 +1482,11 @@ ofconn_run(struct ofconn *ofconn,
                 fail_open_maybe_recover(mgr->fail_open);
             }
 
+            if (!may_pass_with_bundles(of_msg) && hmap_count(&ofconn->bundles) > 0) {
+                list_push_back(&ofconn->queued, &of_msg->list_node);
+                continue;
+            }
+
             if (handle_openflow(ofconn, of_msg)) {
                 ofpbuf_delete(of_msg);
                 ofconn->blocked = NULL;
-- 
1.9.1