dpif-netdev: Use new "ovsthread_counter" to track dp statistics.

ovsthread_counter is an abstract interface that could be implemented
different ways.  The initial implementation is simple but less than
optimally efficient.

Signed-off-by: Ben Pfaff <blp@nicira.com>

lib/dpif-netdev.c

@@ -102,9 +102,9 @@ struct dp_netdev {
     struct seq *queue_seq;      /* Incremented whenever a packet is queued. */
 
     /* Statistics. */
-    long long int n_hit;        /* Number of flow table matches. */
-    long long int n_missed;     /* Number of flow table misses. */
-    long long int n_lost;       /* Number of misses not passed to client. */
+    struct ovsthread_counter *n_hit;    /* Number of flow table matches. */
+    struct ovsthread_counter *n_missed; /* Number of flow table misses. */
+    struct ovsthread_counter *n_lost;   /* Number of misses not passed up. */
 
     /* Ports. */
     struct dp_netdev_port *ports[MAX_PORTS];
@@ -293,6 +293,11 @@ create_dp_netdev(const char *name, const struct dpif_class *class,
     dp->queue_seq = seq_create();
     classifier_init(&dp->cls, NULL);
     hmap_init(&dp->flow_table);
+
+    dp->n_hit = ovsthread_counter_create();
+    dp->n_missed = ovsthread_counter_create();
+    dp->n_lost = ovsthread_counter_create();
+
     list_init(&dp->port_list);
     dp->port_seq = seq_create();
 
@@ -357,6 +362,9 @@ dp_netdev_free(struct dp_netdev *dp)
     LIST_FOR_EACH_SAFE (port, next, node, &dp->port_list) {
         do_del_port(dp, port->port_no);
     }
+    ovsthread_counter_destroy(dp->n_hit);
+    ovsthread_counter_destroy(dp->n_missed);
+    ovsthread_counter_destroy(dp->n_lost);
     dp_netdev_purge_queues(dp);
     seq_destroy(dp->queue_seq);
     classifier_destroy(&dp->cls);
@@ -402,9 +410,9 @@ dpif_netdev_get_stats(const struct dpif *dpif, struct dpif_dp_stats *stats)
 
     ovs_mutex_lock(&dp_netdev_mutex);
     stats->n_flows = hmap_count(&dp->flow_table);
-    stats->n_hit = dp->n_hit;
-    stats->n_missed = dp->n_missed;
-    stats->n_lost = dp->n_lost;
+    stats->n_hit = ovsthread_counter_read(dp->n_hit);
+    stats->n_missed = ovsthread_counter_read(dp->n_missed);
+    stats->n_lost = ovsthread_counter_read(dp->n_lost);
     stats->n_masks = UINT32_MAX;
     stats->n_mask_hit = UINT64_MAX;
     ovs_mutex_unlock(&dp_netdev_mutex);
@@ -1262,9 +1270,9 @@ dp_netdev_port_input(struct dp_netdev *dp, struct ofpbuf *packet,
         dp_netdev_execute_actions(dp, &key, packet, md,
                                   netdev_flow->actions,
                                   netdev_flow->actions_len);
-        dp->n_hit++;
+        ovsthread_counter_inc(dp->n_hit, 1);
     } else {
-        dp->n_missed++;
+        ovsthread_counter_inc(dp->n_missed, 1);
         dp_netdev_output_userspace(dp, packet, DPIF_UC_MISS, &key, NULL);
     }
 }
@@ -1383,7 +1391,7 @@ dp_netdev_output_userspace(struct dp_netdev *dp, struct ofpbuf *packet,
 
         return 0;
     } else {
-        dp->n_lost++;
+        ovsthread_counter_inc(dp->n_lost, 1);
         return ENOBUFS;
     }
 }

lib/ovs-thread.c

@@ -21,6 +21,7 @@
 #include <stdlib.h>
 #include <unistd.h>
 #include "compiler.h"
+#include "hash.h"
 #include "poll-loop.h"
 #include "socket-util.h"
 #include "util.h"
@@ -310,6 +311,84 @@ may_fork(void)
     return !must_not_fork;
 }
 
+/* ovsthread_counter.
+ *
+ * We implement the counter as an array of N_COUNTERS individual counters, each
+ * with its own lock.  Each thread uses one of the counters chosen based on a
+ * hash of the thread's ID, the idea being that, statistically, different
+ * threads will tend to use different counters and therefore avoid
+ * interfering with each other.
+ *
+ * Undoubtedly, better implementations are possible. */
+
+/* Basic counter structure. */
+struct ovsthread_counter__ {
+    struct ovs_mutex mutex;
+    unsigned long long int value;
+};
+
+/* Pad the basic counter structure to 64 bytes to avoid cache line
+ * interference. */
+struct ovsthread_counter {
+    struct ovsthread_counter__ c;
+    char pad[ROUND_UP(sizeof(struct ovsthread_counter__), 64)
+             - sizeof(struct ovsthread_counter__)];
+};
+
+#define N_COUNTERS 16
+
+struct ovsthread_counter *
+ovsthread_counter_create(void)
+{
+    struct ovsthread_counter *c;
+    int i;
+
+    c = xmalloc(N_COUNTERS * sizeof *c);
+    for (i = 0; i < N_COUNTERS; i++) {
+        ovs_mutex_init(&c[i].c.mutex);
+        c[i].c.value = 0;
+    }
+    return c;
+}
+
+void
+ovsthread_counter_destroy(struct ovsthread_counter *c)
+{
+    if (c) {
+        int i;
+
+        for (i = 0; i < N_COUNTERS; i++) {
+            ovs_mutex_destroy(&c[i].c.mutex);
+        }
+        free(c);
+    }
+}
+
+void
+ovsthread_counter_inc(struct ovsthread_counter *c, unsigned long long int n)
+{
+    c = &c[hash_int(ovsthread_id_self(), 0) % N_COUNTERS];
+
+    ovs_mutex_lock(&c->c.mutex);
+    c->c.value += n;
+    ovs_mutex_unlock(&c->c.mutex);
+}
+
+unsigned long long int
+ovsthread_counter_read(const struct ovsthread_counter *c)
+{
+    unsigned long long int sum;
+    int i;
+
+    sum = 0;
+    for (i = 0; i < N_COUNTERS; i++) {
+        ovs_mutex_lock(&c[i].c.mutex);
+        sum += c[i].c.value;
+        ovs_mutex_unlock(&c[i].c.mutex);
+    }
+    return sum;
+}
+
 /* Parses /proc/cpuinfo for the total number of physical cores on this system
  * across all CPU packages, not counting hyper-threads.
  *

lib/ovs-thread.h

@@ -494,6 +494,25 @@ ovsthread_id_self(void)
     return *ovsthread_id_get();
 }
 
+/* Simulated global counter.
+ *
+ * Incrementing such a counter is meant to be cheaper than incrementing a
+ * global counter protected by a lock.  It is probably more expensive than
+ * incrementing a truly thread-local variable, but such a variable has no
+ * straightforward way to get the sum.
+ *
+ *
+ * Thread-safety
+ * =============
+ *
+ * Fully thread-safe. */
+
+struct ovsthread_counter *ovsthread_counter_create(void);
+void ovsthread_counter_destroy(struct ovsthread_counter *);
+void ovsthread_counter_inc(struct ovsthread_counter *, unsigned long long int);
+unsigned long long int ovsthread_counter_read(
+    const struct ovsthread_counter *);
+
 void assert_single_threaded_at(const char *where);
 #define assert_single_threaded() assert_single_threaded_at(SOURCE_LOCATOR)