fixes #1890 stats could support an inline lock - remove most atomics

This starts by updating UDP to use this approach, since we already
have a convenient lock.  We should look at doing the same for other stats.

4 files changed, 117 insertions, 35 deletions

diff --git a/src/core/stats.c b/src/core/stats.c
index 3b9448e6..5fa9ca41 100644
--- a/src/core/stats.c
+++ b/src/core/stats.c
@@ -111,19 +111,28 @@ nni_stat_unregister(nni_stat_item *item)
 }
 
 void
-nni_stat_init(nni_stat_item *item, const nni_stat_info *info)
+nni_stat_init_lock(
+    nni_stat_item *item, const nni_stat_info *info, nni_mtx *mtx)
 {
 #ifdef NNG_ENABLE_STATS
 	memset(item, 0, sizeof(*item));
 	NNI_LIST_INIT(&item->si_children, nni_stat_item, si_node);
 	item->si_info = info;
+	item->si_mtx  = mtx;
 #else
 	NNI_ARG_UNUSED(item);
 	NNI_ARG_UNUSED(info);
+	NNI_ARG_UNUSED(mtx);
 #endif
 }
 
 void
+nni_stat_init(nni_stat_item *item, const nni_stat_info *info)
+{
+	nni_stat_init_lock(item, info, NULL);
+}
+
+void
 nni_stat_inc(nni_stat_item *item, uint64_t inc)
 {
 #ifdef NNG_ENABLE_STATS
@@ -274,13 +283,26 @@ stat_make_tree(nni_stat_item *item, nni_stat **sp)
 }
 
 static void
-stat_update(nni_stat *stat)
+stat_update(nni_stat *stat, nni_mtx **mtxp)
 {
 	const nni_stat_item *item = stat->s_item;
 	const nni_stat_info *info = item->si_info;
 	char                *old;
 	char                *str;
 
+	if (info->si_lock) {
+		NNI_ASSERT(item->si_mtx != NULL);
+		if (*mtxp != item->si_mtx) {
+			if (*mtxp) {
+				nni_mtx_unlock(*mtxp);
+			}
+			nni_mtx_lock(item->si_mtx);
+			*mtxp = item->si_mtx;
+		}
+	} else if (*mtxp) {
+		nni_mtx_unlock(*mtxp);
+		*mtxp = NULL;
+	}
 	switch (info->si_type) {
 	case NNG_STAT_SCOPE:
 	case NNG_STAT_ID:
@@ -325,12 +347,12 @@ stat_update(nni_stat *stat)
 }
 
 static void
-stat_update_tree(nni_stat *stat)
+stat_update_tree(nni_stat *stat, nni_mtx **mtxp)
 {
 	nni_stat *child;
-	stat_update(stat);
+	stat_update(stat, mtxp);
 	NNI_LIST_FOREACH (&stat->s_children, child) {
-		stat_update_tree(child);
+		stat_update_tree(child, mtxp);
 	}
 }
 
@@ -339,6 +361,7 @@ nni_stat_snapshot(nni_stat **statp, nni_stat_item *item)
 {
 	int       rv;
 	nni_stat *stat;
+	nni_mtx  *mtx = NULL;
 
 	if (item == NULL) {
 		item = &stats_root;
@@ -348,7 +371,10 @@ nni_stat_snapshot(nni_stat **statp, nni_stat_item *item)
 		nni_mtx_unlock(&stats_lock);
 		return (rv);
 	}
-	stat_update_tree(stat);
+	stat_update_tree(stat, &mtx);
+	if (mtx != NULL) {
+		nni_mtx_unlock(mtx);
+	}
 	nni_mtx_unlock(&stats_lock);
 	*statp = stat;
 	return (0);
diff --git a/src/core/stats.h b/src/core/stats.h
index 5a2e2831..ddd8a886 100644
--- a/src/core/stats.h
+++ b/src/core/stats.h
@@ -42,6 +42,7 @@ struct nni_stat_item {
 	nni_list_node        si_node;     // list node, framework use only
 	nni_list             si_children; // children, framework use only
 	const nni_stat_info *si_info;     // statistic description
+	nni_mtx             *si_mtx;      // protects, if flag in info
 	union {
 		uint64_t       sv_number;
 		nni_atomic_u64 sv_atomic;
@@ -53,13 +54,13 @@ struct nni_stat_item {
 };
 
 struct nni_stat_info {
-	const char     *si_name;       // name of statistic
-	const char     *si_desc;       // description of statistic (English)
-	nni_stat_type   si_type;       // statistic type, e.g. NNG_STAT_LEVEL
-	nni_stat_unit   si_unit;       // statistic unit, e.g. NNG_UNIT_MILLIS
-	nni_stat_update si_update;     // update function (can be NULL)
-	bool            si_atomic : 1; // stat is atomic
-	bool            si_alloc : 1;  // stat string is allocated
+	const char   *si_name;       // name of statistic
+	const char   *si_desc;       // description of statistic (English)
+	nni_stat_type si_type;       // statistic type, e.g. NNG_STAT_LEVEL
+	nni_stat_unit si_unit;       // statistic unit, e.g. NNG_UNIT_MILLIS
+	bool          si_atomic : 1; // stat is atomic
+	bool          si_alloc : 1;  // stat string is allocated
+	bool          si_lock : 1;   // stat protected by lock (si_mtx)
 };
 
 #ifdef NNG_ENABLE_STATS
@@ -75,6 +76,9 @@ struct nni_stat_info {
 #define NNI_STAT_ATOMIC(var, name, desc, type, unit)           \
 	NNI_STAT_FIELDS(var, .si_name = name, .si_desc = desc, \
 	    .si_type = type, .si_unit = unit, .si_atomic = true)
+#define NNI_STAT_LOCK(var, name, desc, type, unit)             \
+	NNI_STAT_FIELDS(var, .si_name = name, .si_desc = desc, \
+	    .si_type = type, .si_unit = unit, .si_lock = true)
 
 // nni_stat_add adds a statistic, but the operation is unlocked, and the
 // add is to an unregistered stats tree.
@@ -92,6 +96,7 @@ void nni_stat_set_id(nni_stat_item *, int);
 void nni_stat_set_bool(nni_stat_item *, bool);
 void nni_stat_set_string(nni_stat_item *, const char *);
 void nni_stat_init(nni_stat_item *, const nni_stat_info *);
+void nni_stat_init_lock(nni_stat_item *, const nni_stat_info *, nni_mtx *);
 void nni_stat_inc(nni_stat_item *, uint64_t);
 void nni_stat_dec(nni_stat_item *, uint64_t);
 
diff --git a/src/sp/transport/udp/udp.c b/src/sp/transport/udp/udp.c
index fd513386..165b118d 100644
--- a/src/sp/transport/udp/udp.c
+++ b/src/sp/transport/udp/udp.c
@@ -1347,49 +1347,50 @@ udp_ep_init(udp_ep **epp, nng_url *url, nni_sock *sock, nni_dialer *dialer,
 	nni_aio_init(&ep->resaio, udp_resolv_cb, ep);
 	nni_aio_completions_init(&ep->complq);
 
-	NNI_STAT_ATOMIC(rcv_max_info, "rcv_max", "maximum receive size",
+	NNI_STAT_LOCK(rcv_max_info, "rcv_max", "maximum receive size",
 	    NNG_STAT_LEVEL, NNG_UNIT_BYTES);
-	NNI_STAT_ATOMIC(copy_max_info, "copy_max",
+	NNI_STAT_LOCK(copy_max_info, "copy_max",
 	    "threshold to switch to loan-up", NNG_STAT_LEVEL, NNG_UNIT_BYTES);
-	NNI_STAT_ATOMIC(rcv_reorder_info, "rcv_reorder",
+	NNI_STAT_LOCK(rcv_reorder_info, "rcv_reorder",
 	    "messages received out of order", NNG_STAT_COUNTER,
 	    NNG_UNIT_MESSAGES);
-	NNI_STAT_ATOMIC(rcv_nomatch_info, "rcv_nomatch",
+	NNI_STAT_LOCK(rcv_nomatch_info, "rcv_nomatch",
 	    "messages without a matching connection", NNG_STAT_COUNTER,
 	    NNG_UNIT_MESSAGES);
-	NNI_STAT_ATOMIC(rcv_toobig_info, "rcv_toobig",
+	NNI_STAT_LOCK(rcv_toobig_info, "rcv_toobig",
 	    "received messages rejected because too big", NNG_STAT_COUNTER,
 	    NNG_UNIT_MESSAGES);
-	NNI_STAT_ATOMIC(rcv_copy_info, "rcv_copy",
+	NNI_STAT_LOCK(rcv_copy_info, "rcv_copy",
 	    "received messages copied (small)", NNG_STAT_COUNTER,
 	    NNG_UNIT_MESSAGES);
-	NNI_STAT_ATOMIC(rcv_nocopy_info, "rcv_nocopy",
+	NNI_STAT_LOCK(rcv_nocopy_info, "rcv_nocopy",
 	    "received messages zero copy (large)", NNG_STAT_COUNTER,
 	    NNG_UNIT_MESSAGES);
-	NNI_STAT_ATOMIC(rcv_nobuf_info, "rcv_nobuf",
+	NNI_STAT_LOCK(rcv_nobuf_info, "rcv_nobuf",
 	    "received messages dropped no buffer", NNG_STAT_COUNTER,
 	    NNG_UNIT_MESSAGES);
-	NNI_STAT_ATOMIC(snd_toobig_info, "snd_toobig",
+	NNI_STAT_LOCK(snd_toobig_info, "snd_toobig",
 	    "sent messages rejected because too big", NNG_STAT_COUNTER,
 	    NNG_UNIT_MESSAGES);
-	NNI_STAT_ATOMIC(snd_nobuf_info, "snd_nobuf",
+	NNI_STAT_LOCK(snd_nobuf_info, "snd_nobuf",
 	    "sent messages dropped no buffer", NNG_STAT_COUNTER,
 	    NNG_UNIT_MESSAGES);
-	NNI_STAT_ATOMIC(peer_inactive_info, "peer_inactive",
+	NNI_STAT_LOCK(peer_inactive_info, "peer_inactive",
 	    "connections closed due to inactive peer", NNG_STAT_COUNTER,
 	    NNG_UNIT_EVENTS);
 
-	nni_stat_init(&ep->st_rcv_max, &rcv_max_info);
-	nni_stat_init(&ep->st_copy_max, &copy_max_info);
-	nni_stat_init(&ep->st_rcv_copy, &rcv_copy_info);
-	nni_stat_init(&ep->st_rcv_nocopy, &rcv_nocopy_info);
-	nni_stat_init(&ep->st_rcv_reorder, &rcv_reorder_info);
-	nni_stat_init(&ep->st_rcv_toobig, &rcv_toobig_info);
-	nni_stat_init(&ep->st_rcv_nomatch, &rcv_nomatch_info);
-	nni_stat_init(&ep->st_rcv_nobuf, &rcv_nobuf_info);
-	nni_stat_init(&ep->st_snd_toobig, &snd_toobig_info);
-	nni_stat_init(&ep->st_snd_nobuf, &snd_nobuf_info);
-	nni_stat_init(&ep->st_peer_inactive, &peer_inactive_info);
+	nni_stat_init_lock(&ep->st_rcv_max, &rcv_max_info, &ep->mtx);
+	nni_stat_init_lock(&ep->st_copy_max, &copy_max_info, &ep->mtx);
+	nni_stat_init_lock(&ep->st_rcv_copy, &rcv_copy_info, &ep->mtx);
+	nni_stat_init_lock(&ep->st_rcv_nocopy, &rcv_nocopy_info, &ep->mtx);
+	nni_stat_init_lock(&ep->st_rcv_reorder, &rcv_reorder_info, &ep->mtx);
+	nni_stat_init_lock(&ep->st_rcv_toobig, &rcv_toobig_info, &ep->mtx);
+	nni_stat_init_lock(&ep->st_rcv_nomatch, &rcv_nomatch_info, &ep->mtx);
+	nni_stat_init_lock(&ep->st_rcv_nobuf, &rcv_nobuf_info, &ep->mtx);
+	nni_stat_init_lock(&ep->st_snd_toobig, &snd_toobig_info, &ep->mtx);
+	nni_stat_init_lock(&ep->st_snd_nobuf, &snd_nobuf_info, &ep->mtx);
+	nni_stat_init_lock(
+	    &ep->st_peer_inactive, &peer_inactive_info, &ep->mtx);
 
 	if (listener) {
 		nni_listener_add_stat(listener, &ep->st_rcv_max);
diff --git a/src/sp/transport/udp/udp_tran_test.c b/src/sp/transport/udp/udp_tran_test.c
index fe567e7b..beebb2cd 100644
--- a/src/sp/transport/udp/udp_tran_test.c
+++ b/src/sp/transport/udp/udp_tran_test.c
@@ -309,6 +309,55 @@ test_udp_multi_small_burst(void)
 	NUTS_CLOSE(s1);
 }
 
+void
+test_udp_stats(void)
+{
+	char         msg[256];
+	char         buf[256];
+	nng_socket   s0;
+	nng_socket   s1;
+	nng_listener l;
+	nng_dialer   d;
+	size_t       sz;
+	char        *addr;
+	nng_stat    *stat;
+
+	NUTS_ADDR(addr, "udp");
+
+	NUTS_OPEN(s0);
+	NUTS_PASS(nng_socket_set_ms(s0, NNG_OPT_RECVTIMEO, 100));
+	NUTS_PASS(nng_socket_set_ms(s0, NNG_OPT_SENDTIMEO, 100));
+	NUTS_PASS(nng_listener_create(&l, s0, addr));
+	NUTS_PASS(nng_listener_set_size(l, NNG_OPT_UDP_COPY_MAX, 100));
+	NUTS_PASS(nng_listener_get_size(l, NNG_OPT_UDP_COPY_MAX, &sz));
+	NUTS_TRUE(sz == 100);
+	NUTS_PASS(nng_listener_start(l, 0));
+
+	NUTS_OPEN(s1);
+	NUTS_PASS(nng_socket_set_ms(s1, NNG_OPT_RECVTIMEO, 100));
+	NUTS_PASS(nng_socket_set_ms(s1, NNG_OPT_SENDTIMEO, 100));
+	NUTS_PASS(nng_dialer_create(&d, s1, addr));
+	NUTS_PASS(nng_dialer_set_size(d, NNG_OPT_UDP_COPY_MAX, 100));
+	NUTS_PASS(nng_dialer_get_size(d, NNG_OPT_UDP_COPY_MAX, &sz));
+	NUTS_PASS(nng_dialer_start(d, 0));
+	nng_msleep(100);
+
+	for (int i = 0; i < 50; i++) {
+		NUTS_PASS(nng_send(s1, msg, 95, 0));
+		NUTS_PASS(nng_recv(s0, buf, &sz, 0));
+		NUTS_TRUE(sz == 95);
+		NUTS_PASS(nng_send(s0, msg, 95, 0));
+		NUTS_PASS(nng_recv(s1, buf, &sz, 0));
+		NUTS_TRUE(sz == 95);
+	}
+	NUTS_PASS(nng_stats_get(&stat));
+	nng_stats_dump(stat);
+	nng_stats_free(stat);
+
+	NUTS_CLOSE(s0);
+	NUTS_CLOSE(s1);
+}
+
 NUTS_TESTS = {
 
 	{ "udp wild card connect fail", test_udp_wild_card_connect_fail },
@@ -321,5 +370,6 @@ NUTS_TESTS = {
 	{ "udp recv copy", test_udp_recv_copy },
 	{ "udp multi send recv", test_udp_multi_send_recv },
 	{ "udp multi small burst", test_udp_multi_small_burst },
+	{ "udp stats", test_udp_stats },
 	{ NULL, NULL },
 };