From a2061b3dc7394144419be1231705bcee504e7ecc Mon Sep 17 00:00:00 2001
From: JaylinYu <letrangerjaylin@gmail.com>
Date: Tue, 15 Aug 2023 16:05:33 +0800
Subject: [PATCH] * FIX [nng/core] prevent collateral damage of issue
 https://github.com/emqx/nanomq/issues/1411     This is not final fix.

Signed-off-by: JaylinYu <letrangerjaylin@gmail.com>
---
 src/core/list.c                 |  1 +
 src/core/pipe.c                 | 17 ++++++++++++-----
 src/core/sockimpl.h             |  1 +
 src/sp/protocol/mqtt/nmq_mqtt.c |  6 ++----
 4 files changed, 16 insertions(+), 9 deletions(-)

diff --git a/src/core/list.c b/src/core/list.c
index 65e866704..7f6ec378e 100644
--- a/src/core/list.c
+++ b/src/core/list.c
@@ -153,6 +153,7 @@ int
 nni_list_empty(nni_list *list)
 {
 	// The first check ensures that we treat an uninitialized list
+
 	// as empty.  This use useful for statically initialized lists.
 	return ((list->ll_head.ln_next == NULL) ||
 	    (list->ll_head.ln_next == &list->ll_head));
diff --git a/src/core/pipe.c b/src/core/pipe.c
index 1bd79624e..34e12b867 100644
--- a/src/core/pipe.c
+++ b/src/core/pipe.c
@@ -36,6 +36,9 @@ pipe_destroy(void *arg)
 	if (p == NULL || p->cache) {
 		return;
 	}
+	// for a reaper bug
+	// if(p->guard != 1)
+	// 	return;
 
 	nni_pipe_run_cb(p, NNG_PIPE_EV_REM_POST);
 
@@ -59,14 +62,17 @@ pipe_destroy(void *arg)
 	}
 
 	// Freed here
-	struct subinfo * s = NULL;
+	struct subinfo *s = NULL;
 	while (!nni_list_empty(p->subinfol)) {
 		s = nni_list_last(p->subinfol);
-		nni_list_remove(p->subinfol, s);
-		nng_free(s->topic, strlen(s->topic));
-		nng_free(s, sizeof(*s));
+		if (s && s->topic != NULL) {
+			nni_list_remove(p->subinfol, s);
+			nng_free(s->topic, strlen(s->topic));
+			nng_free(s, sizeof(*s));
+		}
 	}
-	nni_free(p->subinfol, sizeof(nni_list));
+	if (p->subinfol != NULL)
+		nni_free(p->subinfol, sizeof(nni_list));
 
 #ifdef NNG_ENABLE_STATS
 	nni_stat_unregister(&p->st_root);
@@ -255,6 +261,7 @@ pipe_create(nni_pipe **pp, nni_sock *sock, nni_sp_tran *tran, void *tran_data)
 		return (NNG_ENOMEM);
 	}
 
+	// p->guard = 1;
 	p->p_size       = sz;
 	p->p_proto_data = p + 1;
 	p->p_tran_ops   = *tran->tran_pipe;
diff --git a/src/core/sockimpl.h b/src/core/sockimpl.h
index aa8d2802b..36bc34722 100644
--- a/src/core/sockimpl.h
+++ b/src/core/sockimpl.h
@@ -106,6 +106,7 @@ struct subinfo {
 
 struct nni_pipe {
 	uint32_t           p_id;
+	// uint32_t           guard;
 	nni_sp_pipe_ops    p_tran_ops;
 	nni_proto_pipe_ops p_proto_ops;
 	size_t             p_size;
diff --git a/src/sp/protocol/mqtt/nmq_mqtt.c b/src/sp/protocol/mqtt/nmq_mqtt.c
index 177798b73..d7148aa9c 100644
--- a/src/sp/protocol/mqtt/nmq_mqtt.c
+++ b/src/sp/protocol/mqtt/nmq_mqtt.c
@@ -553,6 +553,7 @@ nano_pipe_stop(void *arg)
 		return; // your time is yet to come
 
 	log_trace(" ########## nano_pipe_stop ########## ");
+	nni_aio_abort(&p->aio_send, NNG_ECANCELED);
 	nni_aio_stop(&p->aio_send);
 	nni_aio_stop(&p->aio_timer);
 	nni_aio_stop(&p->aio_recv);
@@ -575,7 +576,6 @@ nano_pipe_fini(void *arg)
 		nni_aio_set_msg(&p->aio_send, NULL);
 		nni_msg_free(msg);
 	}
-
 	void *nano_qos_db = p->pipe->nano_qos_db;
 
 	//Safely free the msgs in qos_db
@@ -1026,10 +1026,8 @@ nano_pipe_recv_cb(void *arg)
 	bool is_sqlite = s->conf->sqlite.enable;
 
 	if ((rv = nni_aio_result(&p->aio_recv)) != 0) {
-		// unexpected disconnect
-		nni_mtx_lock(&p->lk);
+		// unexpected disconnect, dont mind the TSAN here
 		p->reason_code = rv;
-		nni_mtx_unlock(&p->lk);
 		nni_pipe_close(p->pipe);
 		return;
 	}