connmgr: Formalize 'ofproto_mutex' as protecting ofconn monitor data.

'ofproto_mutex' has effectively protected the monitor-related members of
struct ofconn since its introduction, but this was not written down or
systematically annotated.  This commit makes it more systematic and fixes
a few issues found using the annotations.

Signed-off-by: Ben Pfaff <blp@nicira.com>
Acked-by: Ethan Jackson <ethan@nicira.com>

diff --git a/ofproto/connmgr.c b/ofproto/connmgr.c
index 19d109e..84c3a7b 100644 (file)
--- a/ofproto/connmgr.c
+++ b/ofproto/connmgr.c
@@ -52,7 +52,8 @@ static struct vlog_rate_limit rl = VLOG_RATE_LIMIT_INIT(1, 5);
  *
  * 'ofproto_mutex' must be held whenever an ofconn is created or destroyed or,
  * more or less equivalently, whenever an ofconn is added to or removed from a
- * connmgr.  'ofproto_mutex' doesn't protect the data inside the ofconn. */
+ * connmgr.  'ofproto_mutex' doesn't protect the data inside the ofconn, except
+ * as specifically noted below. */
 struct ofconn {
 /* Configuration that persists from one connection to the next. */
 
@@ -98,12 +99,36 @@ struct ofconn {
     uint32_t master_async_config[OAM_N_TYPES]; /* master, other */
     uint32_t slave_async_config[OAM_N_TYPES];  /* slave */
 
-    /* Flow monitors. */
-    struct hmap monitors;       /* Contains "struct ofmonitor"s. */
-    struct list updates;        /* List of "struct ofpbuf"s. */
-    bool sent_abbrev_update;    /* Does 'updates' contain NXFME_ABBREV? */
-    struct rconn_packet_counter *monitor_counter;
-    uint64_t monitor_paused;
+/* Flow monitors (e.g. NXST_FLOW_MONITOR). */
+
+    /* Configuration.  Contains "struct ofmonitor"s. */
+    struct hmap monitors OVS_GUARDED_BY(ofproto_mutex);
+
+    /* Flow control.
+     *
+     * When too many flow monitor notifications back up in the transmit buffer,
+     * we pause the transmission of further notifications.  These members track
+     * the flow control state.
+     *
+     * When notifications are flowing, 'monitor_paused' is 0.  When
+     * notifications are paused, 'monitor_paused' is the value of
+     * 'monitor_seqno' at the point we paused.
+     *
+     * 'monitor_counter' counts the OpenFlow messages and bytes currently in
+     * flight.  This value growing too large triggers pausing. */
+    uint64_t monitor_paused OVS_GUARDED_BY(ofproto_mutex);
+    struct rconn_packet_counter *monitor_counter OVS_GUARDED_BY(ofproto_mutex);
+
+    /* State of monitors for a single ongoing flow_mod.
+     *
+     * 'updates' is a list of "struct ofpbuf"s that contain
+     * NXST_FLOW_MONITOR_REPLY messages representing the changes made by the
+     * current flow_mod.
+     *
+     * When 'updates' is nonempty, 'sent_abbrev_update' is true if 'updates'
+     * contains an update event of type NXFME_ABBREV and false otherwise.. */
+    struct list updates OVS_GUARDED_BY(ofproto_mutex);
+    bool sent_abbrev_update OVS_GUARDED_BY(ofproto_mutex);
 };
 
 static struct ofconn *ofconn_create(struct connmgr *, struct rconn *,
@@ -1801,6 +1826,7 @@ COVERAGE_DEFINE(ofmonitor_resume);
 enum ofperr
 ofmonitor_create(const struct ofputil_flow_monitor_request *request,
                  struct ofconn *ofconn, struct ofmonitor **monitorp)
+    OVS_REQUIRES(ofproto_mutex)
 {
     struct ofmonitor *m;
 
@@ -1826,6 +1852,7 @@ ofmonitor_create(const struct ofputil_flow_monitor_request *request,
 
 struct ofmonitor *
 ofmonitor_lookup(struct ofconn *ofconn, uint32_t id)
+    OVS_REQUIRES(ofproto_mutex)
 {
     struct ofmonitor *m;
 
@@ -1840,6 +1867,7 @@ ofmonitor_lookup(struct ofconn *ofconn, uint32_t id)
 
 void
 ofmonitor_destroy(struct ofmonitor *m)
+    OVS_REQUIRES(ofproto_mutex)
 {
     if (m) {
         minimatch_destroy(&m->match);
@@ -1946,6 +1974,7 @@ ofmonitor_report(struct connmgr *mgr, struct rule *rule,
 
 void
 ofmonitor_flush(struct connmgr *mgr)
+    OVS_REQUIRES(ofproto_mutex)
 {
     struct ofconn *ofconn;
 
@@ -1973,6 +2002,7 @@ ofmonitor_flush(struct connmgr *mgr)
 
 static void
 ofmonitor_resume(struct ofconn *ofconn)
+    OVS_REQUIRES(ofproto_mutex)
 {
     struct rule_collection rules;
     struct ofpbuf *resumed;
@@ -1995,18 +2025,27 @@ ofmonitor_resume(struct ofconn *ofconn)
     ofconn->monitor_paused = 0;
 }
 
+static bool
+ofmonitor_may_resume(const struct ofconn *ofconn)
+    OVS_REQUIRES(ofproto_mutex)
+{
+    return (ofconn->monitor_paused != 0
+            && !rconn_packet_counter_n_packets(ofconn->monitor_counter));
+}
+
 static void
 ofmonitor_run(struct connmgr *mgr)
 {
     struct ofconn *ofconn;
 
+    ovs_mutex_lock(&ofproto_mutex);
     LIST_FOR_EACH (ofconn, node, &mgr->all_conns) {
-        if (ofconn->monitor_paused
-            && !rconn_packet_counter_n_packets(ofconn->monitor_counter)) {
+        if (ofmonitor_may_resume(ofconn)) {
             COVERAGE_INC(ofmonitor_resume);
             ofmonitor_resume(ofconn);
         }
     }
+    ovs_mutex_unlock(&ofproto_mutex);
 }
 
 static void
@@ -2014,10 +2053,11 @@ ofmonitor_wait(struct connmgr *mgr)
 {
     struct ofconn *ofconn;
 
+    ovs_mutex_lock(&ofproto_mutex);
     LIST_FOR_EACH (ofconn, node, &mgr->all_conns) {
-        if (ofconn->monitor_paused
-            && !rconn_packet_counter_n_packets(ofconn->monitor_counter)) {
+        if (ofmonitor_may_resume(ofconn)) {
             poll_immediate_wake();
         }
     }
+    ovs_mutex_unlock(&ofproto_mutex);
 }

diff --git a/ofproto/connmgr.h b/ofproto/connmgr.h
index b31e855..99e8ff5 100644 (file)
--- a/ofproto/connmgr.h
+++ b/ofproto/connmgr.h
@@ -178,21 +178,26 @@ struct ofmonitor {
 struct ofputil_flow_monitor_request;
 
 enum ofperr ofmonitor_create(const struct ofputil_flow_monitor_request *,
-                             struct ofconn *, struct ofmonitor **);
-struct ofmonitor *ofmonitor_lookup(struct ofconn *, uint32_t id);
-void ofmonitor_destroy(struct ofmonitor *);
+                             struct ofconn *, struct ofmonitor **)
+    OVS_REQUIRES(ofproto_mutex);
+struct ofmonitor *ofmonitor_lookup(struct ofconn *, uint32_t id)
+    OVS_REQUIRES(ofproto_mutex);
+void ofmonitor_destroy(struct ofmonitor *)
+    OVS_REQUIRES(ofproto_mutex);
 
 void ofmonitor_report(struct connmgr *, struct rule *,
                       enum nx_flow_update_event, enum ofp_flow_removed_reason,
                       const struct ofconn *abbrev_ofconn, ovs_be32 abbrev_xid)
     OVS_REQUIRES(ofproto_mutex);
-void ofmonitor_flush(struct connmgr *);
+void ofmonitor_flush(struct connmgr *) OVS_REQUIRES(ofproto_mutex);
 
 
 struct rule_collection;
 void ofmonitor_collect_resume_rules(struct ofmonitor *, uint64_t seqno,
-                                    struct rule_collection *);
+                                    struct rule_collection *)
+    OVS_REQUIRES(ofproto_mutex);
 void ofmonitor_compose_refresh_updates(struct rule_collection *rules,
-                                       struct list *msgs);
+                                       struct list *msgs)
+    OVS_REQUIRES(ofproto_mutex);
 
 #endif /* connmgr.h */

diff --git a/ofproto/ofproto.c b/ofproto/ofproto.c
index 282ed0b..ea812ad 100644 (file)
--- a/ofproto/ofproto.c
+++ b/ofproto/ofproto.c
@@ -4751,12 +4751,12 @@ handle_flow_monitor_request(struct ofconn *ofconn, const struct ofp_header *oh)
     return 0;
 
 error:
-    ovs_mutex_unlock(&ofproto_mutex);
-
     for (i = 0; i < n_monitors; i++) {
         ofmonitor_destroy(monitors[i]);
     }
     free(monitors);
+    ovs_mutex_unlock(&ofproto_mutex);
+
     return error;
 }