test again

Signed-off-by: Jaylin <jaylin@emqx.io>

Author: JaylinYu

src/core/pipe.c

@@ -33,7 +33,7 @@ static void
 pipe_destroy(void *arg)
 {
 	nni_pipe *p = arg;
-	if (p == NULL || p->cache) {
+	if (p == NULL || nni_atomic_get_bool(&p->cache)) {
 		return;
 	}
 
@@ -274,7 +274,7 @@ pipe_create(nni_pipe **pp, nni_sock *sock, nni_sp_tran *tran, void *tran_data)
 	p->p_ref        = 1;
 	// NanoMQ
 	p->packet_id = 0;
-	p->cache     = false;
+	nni_atomic_set_bool(&p->cache, false);
 	p->subinfol = nni_zalloc(sizeof(nni_list));
 	NNI_LIST_INIT(p->subinfol, struct subinfo, node);
 
@@ -456,7 +456,7 @@ nni_pipe_get_conn_param(nni_pipe *p)
 bool
 nni_pipe_get_status(nni_pipe *p)
 {
-	return p->cache;
+	return nni_atomic_get_bool(&p->cache);
 }
 
 uint16_t
@@ -492,7 +492,7 @@ nni_pipe_set_pid(nni_pipe *new_pipe, uint32_t id)
 		rv = nni_id_set(&pipes, id, new_pipe);
 		// Kick out duplicated Client ID
 		nni_mtx_unlock(&pipes_lk);
-		if (!p->cache || rv != 0) {
+		if (!nni_atomic_get_bool(&p->cache) || rv != 0) {
 			log_error("Client ID collision or set ID failed!");
 			// Must close old pipe first to make it like a normal disconnect
 			// so that new pipe can inherit.

src/core/sockimpl.h

@@ -135,13 +135,13 @@ struct nni_pipe {
 #endif
 
 	// NanoMQ
-	void    *conn_param;
-	bool     cache;
-	uint16_t packet_id;
-	nni_list *subinfol;    // additional info for sub
-	//	nano_qos_db stores qos msgs in 'sqlite' or 'nni_id_hash_map'
-	void    *nano_qos_db; // protected by pipe lock.
-	void    *old;
+	void     *conn_param;
+	void     *nano_qos_db; // protected by pipe lock of transport.
+						   //	nano_qos_db stores qos msgs in 'sqlite' or 'nni_id_hash_map'
+	void     *old;
+	uint16_t  packet_id;
+	nni_list *subinfol; // additional info for sub
+	nni_atomic_bool cache;
 };
 
 extern int  nni_sock_add_dialer(nni_sock *, nni_dialer *);

src/sp/protocol/mqtt/nmq_mqtt.c

@@ -191,7 +191,7 @@ nano_pipe_timer_cb(void *arg)
 	}
 	nni_mtx_lock(&p->lk);
 	// TODO pipe lock or sock lock?
-	if (npipe->cache) {
+	if (nni_atomic_get_bool(&npipe->cache)) {
 		nng_time will_intval = p->conn_param->will_delay_interval;
 		nng_time session_int = p->conn_param->session_expiry_interval;
 		p->ka_refresh++;
@@ -211,7 +211,7 @@ nano_pipe_timer_cb(void *arg)
 			old = nni_id_get(&s->cached_sessions, p->pipe->p_id);
 			if (old != NULL) {
 				old->event       = true;
-				old->pipe->cache = false;
+				nni_atomic_set_bool(&old->pipe->cache, false);
 #ifdef NNG_SUPP_SQLITE
 				nni_qos_db_remove_by_pipe(is_sqlite,
 				    old->nano_qos_db, old->pipe->p_id);
@@ -404,8 +404,9 @@ nano_ctx_send(void *arg, nni_aio *aio)
 	if ((p = nni_id_get(&s->pipes, pipe)) == NULL) {
 		// pre-configured session
 		void *qos_db = NULL;
-		if (s->conf->ext_qos_db)
-			 qos_db = nng_id_get(s->conf->ext_qos_db, pipe);
+		if (s->conf->ext_qos_db) {
+			qos_db = nng_id_get(s->conf->ext_qos_db, pipe);
+		}
 		if (qos_db != NULL) {
 			if (nni_msg_get_type(msg) == CMD_PUBLISH &&
 			    nni_msg_get_pub_qos(msg) > 0) {
@@ -442,6 +443,7 @@ nano_ctx_send(void *arg, nni_aio *aio)
 	if ((rv = nni_aio_schedule(aio, nano_ctx_cancel_send, ctx)) != 0) {
 		nni_msg_free(msg);
 		nni_mtx_unlock(&p->lk);
+		nni_aio_set_msg(aio, NULL);
 		return;
 	}
 	log_debug("pipe %d occupied! resending in cb!", pipe);
@@ -552,7 +554,7 @@ static void
 nano_pipe_stop(void *arg)
 {
 	nano_pipe *p = arg;
-	if (p->pipe->cache)
+	if (nni_atomic_get_bool(&p->pipe->cache))
 		return; // your time is yet to come
 
 	log_trace(" ########## nano_pipe_stop ########## ");
@@ -570,7 +572,7 @@ nano_pipe_fini(void *arg)
 	nng_msg   *msg;
 
 	log_trace(" ########## nano_pipe_fini ########## ");
-	if (p->pipe->cache) {
+	if (nni_atomic_get_bool(&p->pipe->cache)) {
 		return; // your time is yet to come
 	}
 	nni_mtx_lock(&s->lk);
@@ -583,16 +585,7 @@ nano_pipe_fini(void *arg)
 		nni_msg_free(msg);
 	}
 
-	//Safely free the msgs in qos_db, only when nano_qos_db is not taken by new pipe
-	void *nano_qos_db = p->pipe->nano_qos_db;
-	if (p->event == true) {
-		if (!p->broker->conf->sqlite.enable && nano_qos_db != NULL) {
-			nni_qos_db_remove_all_msg(false,
-				nano_qos_db, nmq_close_unack_msg_cb);
-			nni_qos_db_fini_id_hash(nano_qos_db);
-			p->pipe->nano_qos_db = NULL;
-		}
-	} else {
+	if (p->event != true) {
 		// we keep all structs in broker layer, except this conn_param
 		conn_param_free(p->conn_param);
 	}
@@ -707,26 +700,30 @@ nano_pipe_start(void *arg)
 		if (old != NULL) {
 			// there should be no msg in this map
 			if (!is_sqlite && p->pipe->nano_qos_db!= NULL) {
+				nni_qos_db_remove_all_msg(false,
+				p->pipe->nano_qos_db, nmq_close_unack_msg_cb);
 				nni_qos_db_fini_id_hash(p->pipe->nano_qos_db);
 				p->pipe->nano_qos_db = NULL;
 			}
 			log_info("resuming session %d with %d", npipe->p_id, old->pipe->p_id);
-			npipe->old = old->pipe;
-			nni_pipe_peer(npipe);
+			// npipe->old = old->pipe;
+			old->pipe->old = npipe;
+			// nni_pipe_peer(npipe);
+			nni_pipe_peer(old->pipe);
 			p->id = nni_pipe_id(npipe);
 			// set event to false so that no notification will be sent
 			p->event = false;
 			// set event of old pipe to false and discard it.
 			old->event       = false;
-			old->pipe->cache = false;
+			// old->pipe->cache = false;
 			nni_id_remove(&s->cached_sessions, p->pipe->p_id);
 		}
 	} else {
 		// clean previous session
 		old = nni_id_get(&s->cached_sessions, p->pipe->p_id);
 		if (old != NULL) {
-			old->event       = true;
-			old->pipe->cache = false;
+			old->event = true;
+			nni_atomic_swap_bool(&old->pipe->cache, false);
 #ifdef NNG_SUPP_SQLITE
 			nni_qos_db_remove_by_pipe(
 			    is_sqlite, old->nano_qos_db, old->pipe->p_id);
@@ -735,8 +732,6 @@ nano_pipe_start(void *arg)
 			nni_qos_db_remove_unused_msg(
 			    is_sqlite, old->nano_qos_db);
 #endif
-			// nni_qos_db_remove_all_msg(is_sqlite, old->nano_qos_db,
-			//     nmq_close_unack_msg_cb);
 			nni_id_remove(&s->cached_sessions, p->pipe->p_id);
 			log_info("cleaning session %d from cache", p->pipe->p_id);
 		}
@@ -761,17 +756,16 @@ nano_pipe_start(void *arg)
 	}
 	nmq_connack_encode(msg, p->conn_param, rv);
 	conn_param_free(p->conn_param);
-	if (old)
-		nni_msg_set_proto_data(msg, NULL, old->pipe);
 	if (rv != 0) {
 		// send connack with reason code 0x05
 		log_warn("Invalid auth info.");
 	}
 
-	// Dont need to manage id_map while enable SQLite.
+	// Recover preset sessions
 	void *qos_db = NULL;
 	if (s->conf->ext_qos_db)
 		qos_db = nng_id_get(s->conf->ext_qos_db, p->pipe->p_id);
+	// Dont need to manage id_map while enable SQLite.
 	if (qos_db != NULL && !s->conf->sqlite.enable) {
 		// check sqlite compatibility
 		if (p->nano_qos_db != NULL)
@@ -851,7 +845,7 @@ nano_pipe_close(void *arg)
 	char      *clientid     = NULL;
 
 	log_trace(" ############## nano_pipe_close [%p] ############## ", p);
-	if (npipe->cache == true) {
+	if (nni_atomic_get_bool(&npipe->cache)) {
 		// not first time we trying to close stored session pipe
 		nni_atomic_swap_bool(&npipe->p_closed, false);
 		return -1;
@@ -866,7 +860,7 @@ nano_pipe_close(void *arg)
 		clientid = (char *) conn_param_get_clientid(p->conn_param);
 	}
 	if (clientid) {
-		nni_pipe *new_pipe;
+		nni_pipe *new_pipe = NULL;
 		if (nni_pipe_find(&new_pipe, npipe->p_id) == 0) {
 			log_debug("keep session id [%s] ", p->conn_param->clientid.body);
 			nni_pipe_rele(new_pipe);
@@ -875,15 +869,15 @@ nano_pipe_close(void *arg)
 			} else {
 				// client with session stored is kicking itself
 				log_info("A keeping Session is kicked out");
+				// also cache kicked session
+				// merging 2 pipes together in pipe start
 			}
 			log_info("session stored %d", npipe->p_id);
 			nni_id_set(&s->cached_sessions, npipe->p_id, p);
-			// set event to false avoid of sending the
-			// disconnecting msg
+			// set event to false avoid of sending the disconnecting msg
 			p->event     = false;
-			npipe->cache = true;
-			// set clean start to 1, prevent caching
-			// session twice
+			nni_atomic_set_bool(&npipe->cache, true);
+			// set clean start to 1, prevent caching session twice
 			p->conn_param->clean_start = 1;
 			nni_atomic_swap_bool(&npipe->p_closed, false);
 			if (nni_list_active(&s->recvpipes, p)) {
@@ -900,24 +894,24 @@ nano_pipe_close(void *arg)
 		}
 
 		// have to close & stop aio timer first, otherwise we hit null qos_db
-		nni_aio_finish_error(&p->aio_timer, NNG_ECANCELED);
-		nni_aio_close(&p->aio_timer);
-		nni_aio_close(&p->aio_send);
-		nni_aio_close(&p->aio_recv);
-		// take params from npipe to new pipe
-		// new_pipe->packet_id = npipe->packet_id;
-		// there should be no msg in this map
-		if (!s->conf->sqlite.enable && new_pipe->nano_qos_db != NULL)
-			nni_qos_db_fini_id_hash(new_pipe->nano_qos_db);
-		// new_pipe->nano_qos_db = npipe->nano_qos_db;
-		// npipe->nano_qos_db = NULL;
-
-		// nni_list *l        = new_pipe->subinfol;
-		// new_pipe->subinfol = npipe->subinfol;
-		// npipe->subinfol    = l;
-		new_pipe->old = npipe;
-		nni_pipe_peer(new_pipe);
-		log_info("client kick itself while keeping session!");
+		// nni_aio_finish_error(&p->aio_timer, NNG_ECANCELED);
+		// nni_aio_close(&p->aio_timer);
+		// nni_aio_close(&p->aio_send);
+		// nni_aio_close(&p->aio_recv);
+		// // take params from npipe to new pipe
+		// // new_pipe->packet_id = npipe->packet_id;
+		// // there should be no msg in this map
+		// if (!s->conf->sqlite.enable && new_pipe->nano_qos_db != NULL)
+		// 	nni_qos_db_fini_id_hash(new_pipe->nano_qos_db);
+		// // new_pipe->nano_qos_db = npipe->nano_qos_db;
+		// // npipe->nano_qos_db = NULL;
+
+		// // nni_list *l        = new_pipe->subinfol;
+		// // new_pipe->subinfol = npipe->subinfol;
+		// // npipe->subinfol    = l;
+		// new_pipe->old = npipe;
+		// nni_pipe_peer(new_pipe);
+		// log_info("client kick itself while keeping session!");
 	} else {
 		nni_aio_close(&p->aio_send);
 		nni_aio_close(&p->aio_recv);
@@ -1178,15 +1172,17 @@ nano_pipe_recv_cb(void *arg)
 		nni_mtx_lock(&p->lk);
 		NNI_GET16(ptr, ackid);
 		p->rid = ackid + 1;
-		if ((qos_msg = nni_qos_db_get(is_sqlite, npipe->nano_qos_db,
-		         npipe->p_id, ackid)) != NULL) {
-			nni_qos_db_remove_msg(
-			    is_sqlite, npipe->nano_qos_db, qos_msg);
-			nni_qos_db_remove(
-			    is_sqlite, npipe->nano_qos_db, npipe->p_id, ackid);
-		} else {
-			log_warn("ACK failed! qos msg %ld not found!", ackid);
-		}
+		// if (npipe->nano_qos_db == NULL)
+		// 	log_error("mem leak!!!!!!!!!!!");
+		// if ((qos_msg = nni_qos_db_get(is_sqlite, npipe->nano_qos_db,
+		//          npipe->p_id, ackid)) != NULL) {
+		// 	nni_qos_db_remove_msg(
+		// 	    is_sqlite, npipe->nano_qos_db, qos_msg);
+		// 	nni_qos_db_remove(
+		// 	    is_sqlite, npipe->nano_qos_db, npipe->p_id, ackid);
+		// } else {
+		// 	log_warn("ACK failed! qos msg %ld not found!", ackid);
+		// }
 		nni_mtx_unlock(&p->lk);
 	case CMD_CONNECT:
 	case CMD_PUBREC: