rconn: Make queued packet counting harder to screw up.

The semantics of the 'n_queued' parameter to rconn_send() and
rconn_send_with_limit() were too easy to screw up: if the memory area in
which the passed-in data lived was destroyed before the rconn was
destroyed, then rconn_destroy() (or simply flushing out the transmission
queue) would access invalid memory or, worse, decrement a random integer
in reused memory.  It was possible to avoid this by destroying the rconn
before destroying the queue count data area, but this is difficult to
remember and not always possible in the general case.

This commit changes to using a reference-counted structure, which is harder
to get wrong.

diff --git a/lib/learning-switch.c b/lib/learning-switch.c
index 197e33a438755c0eefab8b982ef86913facb64e8..2e91143fd332d68cc3f67256eab65406ec301bee 100644 (file)
--- a/lib/learning-switch.c
+++ b/lib/learning-switch.c
@@ -76,7 +76,7 @@ struct lswitch {
     struct mac_learning *ml;    /* NULL to act as hub instead of switch. */
 
     /* Number of outgoing queued packets on the rconn. */
-    int n_queued;
+    struct rconn_packet_counter *queued;
 
     /* Spanning tree protocol implementation.
      *
@@ -132,6 +132,7 @@ lswitch_create(struct rconn *rconn, bool learn_macs, int max_idle)
     sw->datapath_id = 0;
     sw->last_features_request = time_now() - 1;
     sw->ml = learn_macs ? mac_learning_create() : NULL;
+    sw->queued = rconn_packet_counter_create();
     sw->next_query = LLONG_MIN;
     sw->last_query = LLONG_MIN;
     sw->last_reply = LLONG_MIN;
@@ -148,6 +149,7 @@ lswitch_destroy(struct lswitch *sw)
 {
     if (sw) {
         mac_learning_destroy(sw->ml);
+        rconn_packet_counter_destroy(sw->queued);
         free(sw);
     }
 }
@@ -358,7 +360,7 @@ send_features_request(struct lswitch *sw, struct rconn *rconn)
 static void
 queue_tx(struct lswitch *sw, struct rconn *rconn, struct ofpbuf *b)
 {
-    int retval = rconn_send_with_limit(rconn, b, &sw->n_queued, 10);
+    int retval = rconn_send_with_limit(rconn, b, sw->queued, 10);
     if (retval && retval != ENOTCONN) {
         if (retval == EAGAIN) {
             VLOG_INFO_RL(&rl, "%012llx: %s: tx queue overflow",

diff --git a/lib/rconn.c b/lib/rconn.c
index 0ec847470b11de5fffe1bd0d5a21757a0efd1fe4..78f86567bfe045284ce1fb5a5feb82ee9eab7497 100644 (file)
--- a/lib/rconn.c
+++ b/lib/rconn.c
@@ -498,23 +498,24 @@ rconn_recv_wait(struct rconn *rc)
  * destroyed), or ENOTCONN if 'rc' is not currently connected (in which case
  * the caller retains ownership of 'b').
  *
- * If 'n_queued' is non-null, then '*n_queued' will be incremented while the
+ * If 'counter' is non-null, then 'counter' will be incremented while the
  * packet is in flight, then decremented when it has been sent (or discarded
  * due to disconnection).  Because 'b' may be sent (or discarded) before this
  * function returns, the caller may not be able to observe any change in
- * '*n_queued'.
+ * 'counter'.
  *
  * There is no rconn_send_wait() function: an rconn has a send queue that it
  * takes care of sending if you call rconn_run(), which will have the side
  * effect of waking up poll_block(). */
 int
-rconn_send(struct rconn *rc, struct ofpbuf *b, int *n_queued)
+rconn_send(struct rconn *rc, struct ofpbuf *b,
+           struct rconn_packet_counter *counter)
 {
     if (rconn_is_connected(rc)) {
         copy_to_monitor(rc, b);
-        b->private = n_queued;
-        if (n_queued) {
-            ++*n_queued;
+        b->private = counter;
+        if (counter) {
+            rconn_packet_counter_inc(counter);
         }
         queue_push_tail(&rc->txq, b);
 
@@ -531,24 +532,24 @@ rconn_send(struct rconn *rc, struct ofpbuf *b, int *n_queued)
     }
 }
 
-/* Sends 'b' on 'rc'.  Increments '*n_queued' while the packet is in flight; it
+/* Sends 'b' on 'rc'.  Increments 'counter' while the packet is in flight; it
  * will be decremented when it has been sent (or discarded due to
- * disconnection).  Returns 0 if successful, EAGAIN if '*n_queued' is already
+ * disconnection).  Returns 0 if successful, EAGAIN if 'counter->n' is already
  * at least as large as 'queue_limit', or ENOTCONN if 'rc' is not currently
  * connected.  Regardless of return value, 'b' is destroyed.
  *
  * Because 'b' may be sent (or discarded) before this function returns, the
- * caller may not be able to observe any change in '*n_queued'.
+ * caller may not be able to observe any change in 'counter'.
  *
  * There is no rconn_send_wait() function: an rconn has a send queue that it
  * takes care of sending if you call rconn_run(), which will have the side
  * effect of waking up poll_block(). */
 int
 rconn_send_with_limit(struct rconn *rc, struct ofpbuf *b,
-                      int *n_queued, int queue_limit)
+                      struct rconn_packet_counter *counter, int queue_limit)
 {
     int retval;
-    retval = *n_queued >= queue_limit ? EAGAIN : rconn_send(rc, b, n_queued);
+    retval = counter->n >= queue_limit ? EAGAIN : rconn_send(rc, b, counter);
     if (retval) {
         ofpbuf_delete(b);
     }
@@ -713,6 +714,41 @@ rconn_get_connection_seqno(const struct rconn *rc)
     return rc->seqno;
 }
 \f
+struct rconn_packet_counter *
+rconn_packet_counter_create(void)
+{
+    struct rconn_packet_counter *c = xmalloc(sizeof *c);
+    c->n = 0;
+    c->ref_cnt = 1;
+    return c;
+}
+
+void
+rconn_packet_counter_destroy(struct rconn_packet_counter *c)
+{
+    if (c) {
+        assert(c->ref_cnt > 0);
+        if (!--c->ref_cnt && !c->n) {
+            free(c);
+        }
+    }
+}
+
+void
+rconn_packet_counter_inc(struct rconn_packet_counter *c)
+{
+    c->n++;
+}
+
+void
+rconn_packet_counter_dec(struct rconn_packet_counter *c)
+{
+    assert(c->n > 0);
+    if (!--c->n && !c->ref_cnt) {
+        free(c);
+    }
+}
+\f
 /* Tries to send a packet from 'rc''s send buffer.  Returns 0 if successful,
  * otherwise a positive errno value. */
 static int
@@ -720,7 +756,7 @@ try_send(struct rconn *rc)
 {
     int retval = 0;
     struct ofpbuf *next = rc->txq.head->next;
-    int *n_queued = rc->txq.head->private;
+    struct rconn_packet_counter *counter = rc->txq.head->private;
     retval = vconn_send(rc->vconn, rc->txq.head);
     if (retval) {
         if (retval != EAGAIN) {
@@ -729,8 +765,8 @@ try_send(struct rconn *rc)
         return retval;
     }
     rc->packets_sent++;
-    if (n_queued) {
-        --*n_queued;
+    if (counter) {
+        rconn_packet_counter_dec(counter);
     }
     queue_advance_head(&rc->txq, next);
     return 0;
@@ -788,9 +824,9 @@ flush_queue(struct rconn *rc)
     }
     while (rc->txq.n > 0) {
         struct ofpbuf *b = queue_pop_head(&rc->txq);
-        int *n_queued = b->private;
-        if (n_queued) {
-            --*n_queued;
+        struct rconn_packet_counter *counter = b->private;
+        if (counter) {
+            rconn_packet_counter_dec(counter);
         }
         ofpbuf_delete(b);
     }

diff --git a/lib/rconn.h b/lib/rconn.h
index 694665ef4e166fbbc3cece0ac44f85abf098f848..587eafc1423156142784b05d2b94cbe3346c4ed2 100644 (file)
--- a/lib/rconn.h
+++ b/lib/rconn.h
@@ -1,4 +1,4 @@
-/* Copyright (c) 2008 The Board of Trustees of The Leland Stanford
+/* Copyright (c) 2008, 2009 The Board of Trustees of The Leland Stanford
  * Junior University
  * 
  * We are making the OpenFlow specification and associated documentation
@@ -52,6 +52,7 @@
  */
 
 struct vconn;
+struct rconn_packet_counter;
 
 struct rconn *rconn_new(const char *name, 
                         int inactivity_probe_interval, int max_backoff);
@@ -67,9 +68,9 @@ void rconn_run(struct rconn *);
 void rconn_run_wait(struct rconn *);
 struct ofpbuf *rconn_recv(struct rconn *);
 void rconn_recv_wait(struct rconn *);
-int rconn_send(struct rconn *, struct ofpbuf *, int *n_queued);
+int rconn_send(struct rconn *, struct ofpbuf *, struct rconn_packet_counter *);
 int rconn_send_with_limit(struct rconn *, struct ofpbuf *,
-                          int *n_queued, int queue_limit);
+                          struct rconn_packet_counter *, int queue_limit);
 unsigned int rconn_packets_sent(const struct rconn *);
 unsigned int rconn_packets_received(const struct rconn *);
 
@@ -93,4 +94,15 @@ int rconn_get_backoff(const struct rconn *);
 unsigned int rconn_get_state_elapsed(const struct rconn *);
 unsigned int rconn_get_connection_seqno(const struct rconn *);
 
+/* Counts the number of packets queued into an rconn by a given source. */
+struct rconn_packet_counter {
+    int n;                      /* Number of packets queued. */
+    int ref_cnt;                /* Number of owners. */
+};
+
+struct rconn_packet_counter *rconn_packet_counter_create(void);
+void rconn_packet_counter_destroy(struct rconn_packet_counter *);
+void rconn_packet_counter_inc(struct rconn_packet_counter *);
+void rconn_packet_counter_dec(struct rconn_packet_counter *);
+
 #endif /* rconn.h */

diff --git a/vswitchd/bridge.c b/vswitchd/bridge.c
index 64a45c05ada36432b4bb45bef8bb528bf168b197..cf4dc45a2d4dc7ce998c506f2fdd1ed9c916f2e7 100644 (file)
--- a/vswitchd/bridge.c
+++ b/vswitchd/bridge.c
@@ -145,7 +145,7 @@ struct port {
 struct bridge {
     struct list node;           /* Node in global list of bridges. */
     char *name;                 /* User-specified arbitrary name. */
-    int txqlen;                 /* # of messages queued to send on 'rconn'. */
+    struct rconn_packet_counter *txqlen; /* # queued to send on 'rconn'. */
     struct mac_learning *ml;    /* MAC learning table, or null not to learn. */
     int flow_idle_time;         /* Idle time for flows we set up. */
     bool sent_config_request;   /* Successfully sent config request? */
@@ -552,7 +552,7 @@ bridge_create(const char *name)
     assert(!bridge_lookup(name));
     br = xcalloc(1, sizeof *br);
     br->name = xstrdup(name);
-    br->txqlen = 0;
+    br->txqlen = rconn_packet_counter_create();
     br->ml = mac_learning_create();
     br->flow_idle_time = 5;
     br->sent_config_request = false;
@@ -795,6 +795,7 @@ bridge_destroy(struct bridge *br)
         }
         process_destroy(br->secchan);
         rconn_destroy(br->rconn);
+        rconn_packet_counter_destroy(br->txqlen);
         free(br->controller);
         svec_destroy(&br->secchan_opts);
         ft_destroy(br->ft);
@@ -850,7 +851,7 @@ send_set_config_request(struct bridge *br)
     osc = make_openflow(sizeof *osc, OFPT_SET_CONFIG, &msg);
     osc->flags = htons(OFPC_SEND_FLOW_EXP | OFPC_FRAG_NORMAL);
     osc->miss_send_len = htons(OFP_DEFAULT_MISS_SEND_LEN);
-    if (!rconn_send_with_limit(br->rconn, msg, &br->txqlen, INT_MAX)) {
+    if (!rconn_send_with_limit(br->rconn, msg, br->txqlen, INT_MAX)) {
         br->sent_config_request = true;
     }
 }
@@ -862,7 +863,7 @@ send_features_request(struct bridge *br)
     struct ofpbuf *msg;
 
     oh = make_openflow(sizeof *oh, OFPT_FEATURES_REQUEST, &msg);
-    if (!rconn_send_with_limit(br->rconn, msg, &br->txqlen, INT_MAX)) {
+    if (!rconn_send_with_limit(br->rconn, msg, br->txqlen, INT_MAX)) {
         br->sent_features_request = true;
     }
 }
@@ -1054,7 +1055,7 @@ bridge_get_all_ifaces(const struct bridge *br, struct svec *ifaces)
 static bool
 bridge_is_backlogged(const struct bridge *br)
 {
-    return br->txqlen >= 100;
+    return br->txqlen->n >= 100;
 }
 
 /* For robustness, in case the administrator moves around datapath ports behind
@@ -1172,7 +1173,7 @@ queue_tx(struct bridge *br, struct ofpbuf *msg)
 {
     int retval;
     update_openflow_length(msg);
-    retval = rconn_send(br->rconn, msg, &br->txqlen);
+    retval = rconn_send(br->rconn, msg, br->txqlen);
     if (retval) {
         ofpbuf_delete(msg);
         /* No point in logging: rconn_send() only fails due to disconnection,

diff --git a/vswitchd/stats.c b/vswitchd/stats.c
index 0155c0e835d486d39d11f27f3612efb04f6970af..26eea8ebb1b92a584d69e917153af65fd658e760 100644 (file)
--- a/vswitchd/stats.c
+++ b/vswitchd/stats.c
@@ -51,7 +51,7 @@ struct stats_mgr {
     struct list requests;
     struct rconn *rconn;
     unsigned int rconn_seqno;
-    int txqlen;
+    struct rconn_packet_counter *txqlen;
     time_t timeout;
 };
 
@@ -88,7 +88,7 @@ stats_mgr_create(struct rconn *rconn)
     list_init(&mgr->requests);
     mgr->rconn = rconn;
     mgr->rconn_seqno = rconn_get_connection_seqno(rconn);
-    mgr->txqlen = 0;
+    mgr->txqlen = rconn_packet_counter_create();
     return mgr;
 }
 
@@ -102,6 +102,7 @@ void
 stats_mgr_destroy(struct stats_mgr *mgr)
 {
     if (mgr) {
+        rconn_packet_counter_destroy(mgr->txqlen);
         cancel_all_requests(mgr, ECONNABORTED);
         free(mgr);
     }
@@ -286,13 +287,13 @@ send_stats_request(struct stats_mgr *mgr)
         || !rconn_is_connected(mgr->rconn)) {
         mgr->rconn_seqno = rconn_get_connection_seqno(mgr->rconn);
         cancel_all_requests(mgr, ENOTCONN);
-    } else if (!mgr->txqlen && !mgr->rq && !list_is_empty(&mgr->requests)) {
+    } else if (!mgr->txqlen->n && !mgr->rq && !list_is_empty(&mgr->requests)) {
         struct stats_request *rq;
         int retval;
 
         rq = mgr->rq = CONTAINER_OF(list_pop_front(&mgr->requests),
                                   struct stats_request, node);
-        retval = rconn_send(mgr->rconn, rq->osr, &mgr->txqlen);
+        retval = rconn_send(mgr->rconn, rq->osr, mgr->txqlen);
         if (!retval) {
             /* rconn_send() consumed the message. */
             rq->osr = NULL;