fixes #1523 rare SEGV in sub nni_list_remove

Credit goes to Wu Xuan (@willwu1217) for diagnosing and proposing
a fix as part of #1695.  This approach takes a revised approach
to avoid adding extra memory, and it also is slightly faster as we
do not need to update both pointers in the linked list, by reusing
the reap node.

As part of this a new internal API, nni_aio_completions, is introduced.
In all likelihood we will be able to use this to solve some similar
crashes in other areas of the code.

1 files changed, 32 insertions, 35 deletions

diff --git a/src/sp/protocol/pubsub0/sub.c b/src/sp/protocol/pubsub0/sub.c
index 10f42724..e7540dee 100644
--- a/src/sp/protocol/pubsub0/sub.c
+++ b/src/sp/protocol/pubsub0/sub.c
@@ -1,5 +1,5 @@
 //
-// Copyright 2021 Staysail Systems, Inc. <info@staysail.tech>
+// Copyright 2023 Staysail Systems, Inc. <info@staysail.tech>
 // Copyright 2018 Capitar IT Group BV <info@capitar.com>
 // Copyright 2019 Nathan Kent <nate@nkent.net>
 //
@@ -44,14 +44,14 @@ static void sub0_pipe_fini(void *);
 struct sub0_topic {
 	nni_list_node node;
 	size_t        len;
-	void *        buf;
+	void         *buf;
 };
 
 // sub0_ctx is a context for a SUB socket.  The advantage of contexts is
 // that different contexts can maintain different subscriptions.
 struct sub0_ctx {
 	nni_list_node node;
-	sub0_sock *   sock;
+	sub0_sock    *sock;
 	nni_list      topics;     // TODO: Consider patricia trie
 	nni_list      recv_queue; // can have multiple pending receives
 	nni_lmq       lmq;
@@ -71,7 +71,7 @@ struct sub0_sock {
 
 // sub0_pipe is our per-pipe protocol private structure.
 struct sub0_pipe {
-	nni_pipe * pipe;
+	nni_pipe  *pipe;
 	sub0_sock *sub;
 	nni_aio    aio_recv;
 };
@@ -79,7 +79,7 @@ struct sub0_pipe {
 static void
 sub0_ctx_cancel(nng_aio *aio, void *arg, int rv)
 {
-	sub0_ctx * ctx  = arg;
+	sub0_ctx  *ctx  = arg;
 	sub0_sock *sock = ctx->sock;
 	nni_mtx_lock(&sock->lk);
 	if (nni_list_active(&ctx->recv_queue, aio)) {
@@ -92,9 +92,9 @@ sub0_ctx_cancel(nng_aio *aio, void *arg, int rv)
 static void
 sub0_ctx_recv(void *arg, nni_aio *aio)
 {
-	sub0_ctx * ctx  = arg;
+	sub0_ctx  *ctx  = arg;
 	sub0_sock *sock = ctx->sock;
-	nni_msg *  msg;
+	nni_msg   *msg;
 
 	if (nni_aio_begin(aio) != 0) {
 		return;
@@ -140,9 +140,9 @@ sub0_ctx_send(void *arg, nni_aio *aio)
 static void
 sub0_ctx_close(void *arg)
 {
-	sub0_ctx * ctx  = arg;
+	sub0_ctx  *ctx  = arg;
 	sub0_sock *sock = ctx->sock;
-	nni_aio *  aio;
+	nni_aio   *aio;
 
 	nni_mtx_lock(&sock->lk);
 	while ((aio = nni_list_first(&ctx->recv_queue)) != NULL) {
@@ -155,8 +155,8 @@ sub0_ctx_close(void *arg)
 static void
 sub0_ctx_fini(void *arg)
 {
-	sub0_ctx *  ctx  = arg;
-	sub0_sock * sock = ctx->sock;
+	sub0_ctx   *ctx  = arg;
+	sub0_sock  *sock = ctx->sock;
 	sub0_topic *topic;
 
 	sub0_ctx_close(ctx);
@@ -179,7 +179,7 @@ static void
 sub0_ctx_init(void *ctx_arg, void *sock_arg)
 {
 	sub0_sock *sock = sock_arg;
-	sub0_ctx * ctx  = ctx_arg;
+	sub0_ctx  *ctx  = ctx_arg;
 	size_t     len;
 	bool       prefer_new;
 
@@ -311,22 +311,22 @@ sub0_matches(sub0_ctx *ctx, uint8_t *body, size_t len)
 static void
 sub0_recv_cb(void *arg)
 {
-	sub0_pipe *p    = arg;
-	sub0_sock *sock = p->sub;
-	sub0_ctx * ctx;
-	nni_msg *  msg;
-	size_t     len;
-	uint8_t *  body;
-	nni_list   finish;
-	nng_aio *  aio;
-	nni_msg *  dup_msg;
+	sub0_pipe          *p    = arg;
+	sub0_sock          *sock = p->sub;
+	sub0_ctx           *ctx;
+	nni_msg            *msg;
+	size_t              len;
+	uint8_t            *body;
+	nng_aio            *aio;
+	nni_msg            *dup_msg;
+	nni_aio_completions finish;
 
 	if (nni_aio_result(&p->aio_recv) != 0) {
 		nni_pipe_close(p->pipe);
 		return;
 	}
 
-	nni_aio_list_init(&finish);
+	nni_aio_completions_init(&finish);
 
 	msg = nni_aio_get_msg(&p->aio_recv);
 	nni_aio_set_msg(&p->aio_recv, NULL);
@@ -370,7 +370,7 @@ sub0_recv_cb(void *arg)
 			nni_aio_set_msg(aio, dup_msg);
 
 			// Save for synchronous completion
-			nni_list_append(&finish, aio);
+			nni_aio_completions_add(&finish, aio, 0, len);
 		} else if (nni_lmq_full(&ctx->lmq)) {
 			// Make space for the new message.
 			nni_msg *old;
@@ -401,10 +401,7 @@ sub0_recv_cb(void *arg)
 		nni_msg_free(msg);
 	}
 
-	while ((aio = nni_list_first(&finish)) != NULL) {
-		nni_list_remove(&finish, aio);
-		nni_aio_finish_sync(aio, 0, len);
-	}
+	nni_aio_completions_run(&finish);
 
 	nni_pipe_recv(p->pipe, &p->aio_recv);
 }
@@ -412,7 +409,7 @@ sub0_recv_cb(void *arg)
 static int
 sub0_ctx_get_recv_buf_len(void *arg, void *buf, size_t *szp, nni_type t)
 {
-	sub0_ctx * ctx  = arg;
+	sub0_ctx  *ctx  = arg;
 	sub0_sock *sock = ctx->sock;
 	int        val;
 	nni_mtx_lock(&sock->lk);
@@ -425,7 +422,7 @@ sub0_ctx_get_recv_buf_len(void *arg, void *buf, size_t *szp, nni_type t)
 static int
 sub0_ctx_set_recv_buf_len(void *arg, const void *buf, size_t sz, nni_type t)
 {
-	sub0_ctx * ctx  = arg;
+	sub0_ctx  *ctx  = arg;
 	sub0_sock *sock = ctx->sock;
 	int        val;
 	int        rv;
@@ -456,8 +453,8 @@ sub0_ctx_set_recv_buf_len(void *arg, const void *buf, size_t sz, nni_type t)
 static int
 sub0_ctx_subscribe(void *arg, const void *buf, size_t sz, nni_type t)
 {
-	sub0_ctx *  ctx  = arg;
-	sub0_sock * sock = ctx->sock;
+	sub0_ctx   *ctx  = arg;
+	sub0_sock  *sock = ctx->sock;
 	sub0_topic *topic;
 	sub0_topic *new_topic;
 	NNI_ARG_UNUSED(t);
@@ -494,8 +491,8 @@ sub0_ctx_subscribe(void *arg, const void *buf, size_t sz, nni_type t)
 static int
 sub0_ctx_unsubscribe(void *arg, const void *buf, size_t sz, nni_type t)
 {
-	sub0_ctx *  ctx  = arg;
-	sub0_sock * sock = ctx->sock;
+	sub0_ctx   *ctx  = arg;
+	sub0_sock  *sock = ctx->sock;
 	sub0_topic *topic;
 	size_t      len;
 	NNI_ARG_UNUSED(t);
@@ -540,7 +537,7 @@ sub0_ctx_unsubscribe(void *arg, const void *buf, size_t sz, nni_type t)
 static int
 sub0_ctx_get_prefer_new(void *arg, void *buf, size_t *szp, nni_type t)
 {
-	sub0_ctx * ctx  = arg;
+	sub0_ctx  *ctx  = arg;
 	sub0_sock *sock = ctx->sock;
 	bool       val;
 
@@ -554,7 +551,7 @@ sub0_ctx_get_prefer_new(void *arg, void *buf, size_t *szp, nni_type t)
 static int
 sub0_ctx_set_prefer_new(void *arg, const void *buf, size_t sz, nni_type t)
 {
-	sub0_ctx * ctx  = arg;
+	sub0_ctx  *ctx  = arg;
 	sub0_sock *sock = ctx->sock;
 	bool       val;
 	int        rv;