From 0372393c3f7af00026b626ab6b5a5a27c69c9bfc Mon Sep 17 00:00:00 2001
From: Ian Craggs <icraggs@gmail.com>
Date: Tue, 16 Nov 2021 19:40:46 +0000
Subject: [PATCH] Fix some potential locking issues #1169

---
 src/MQTTAsync.c      |  6 ++++--
 src/MQTTAsync.h      |  2 +-
 src/MQTTAsyncUtils.c | 16 ++++++++++++++--
 src/Socket.c         |  5 +++--
 4 files changed, 22 insertions(+), 7 deletions(-)

diff --git a/src/MQTTAsync.c b/src/MQTTAsync.c
index 50940e02..d6fb2181 100644
--- a/src/MQTTAsync.c
+++ b/src/MQTTAsync.c
@@ -1426,6 +1426,7 @@ int MQTTAsync_getPendingTokens(MQTTAsync handle, MQTTAsync_token **tokens)
 
 	FUNC_ENTRY;
 	MQTTAsync_lock_mutex(mqttasync_mutex);
+	MQTTAsync_lock_mutex(mqttcommand_mutex);
 	*tokens = NULL;
 
 	if (m == NULL)
@@ -1439,7 +1440,7 @@ int MQTTAsync_getPendingTokens(MQTTAsync handle, MQTTAsync_token **tokens)
 	{
 		MQTTAsync_queuedCommand* cmd = (MQTTAsync_queuedCommand*)(current->content);
 
-		if (cmd->client == m)
+		if (cmd->client == m && cmd->command.type == PUBLISH)
 			count++;
 	}
 	if (m->c)
@@ -1460,7 +1461,7 @@ int MQTTAsync_getPendingTokens(MQTTAsync handle, MQTTAsync_token **tokens)
 	{
 		MQTTAsync_queuedCommand* cmd = (MQTTAsync_queuedCommand*)(current->content);
 
-		if (cmd->client == m)
+		if (cmd->client == m  && cmd->command.type == PUBLISH)
 			(*tokens)[count++] = cmd->command.token;
 	}
 
@@ -1477,6 +1478,7 @@ int MQTTAsync_getPendingTokens(MQTTAsync handle, MQTTAsync_token **tokens)
 	(*tokens)[count] = -1; /* indicate end of list */
 
 exit:
+	MQTTAsync_unlock_mutex(mqttcommand_mutex);
 	MQTTAsync_unlock_mutex(mqttasync_mutex);
 	FUNC_EXIT_RC(rc);
 	return rc;
diff --git a/src/MQTTAsync.h b/src/MQTTAsync.h
index 360798a2..3d7783fa 100644
--- a/src/MQTTAsync.h
+++ b/src/MQTTAsync.h
@@ -421,7 +421,7 @@ typedef void MQTTAsync_connected(void* context, char* cause);
 
 /**
  * This is a callback function, which will be called when the client
- * library receives a disconnect packet.
+ * library receives a disconnect packet from the server. This applies to MQTT V5 and above only.
  *
  * <b>Note:</b> Neither MQTTAsync_create() nor MQTTAsync_destroy() should be
  * called within this callback.
diff --git a/src/MQTTAsyncUtils.c b/src/MQTTAsyncUtils.c
index 56826aee..98507bdf 100644
--- a/src/MQTTAsyncUtils.c
+++ b/src/MQTTAsyncUtils.c
@@ -1761,11 +1761,18 @@ thread_return_type WINAPI MQTTAsync_sendThread(void* n)
 	while (!MQTTAsync_tostop)
 	{
 		int rc;
+		int command_count = 0;
 
-		while (MQTTAsync_commands->count > 0)
+		MQTTAsync_lock_mutex(mqttcommand_mutex);
+		command_count = MQTTAsync_commands->count;
+		MQTTAsync_unlock_mutex(mqttcommand_mutex);
+		while (command_count > 0)
 		{
 			if (MQTTAsync_processCommand() == 0)
 				break;  /* no commands were processed, so go into a wait */
+			MQTTAsync_lock_mutex(mqttcommand_mutex);
+			command_count = MQTTAsync_commands->count;
+			MQTTAsync_unlock_mutex(mqttcommand_mutex);
 		}
 #if !defined(_WIN32) && !defined(_WIN64)
 		if ((rc = Thread_wait_cond(send_cond, 1)) != 0 && rc != ETIMEDOUT)
@@ -2888,10 +2895,15 @@ static MQTTPacket* MQTTAsync_cycle(int* sock, unsigned long timeout, int* rc)
 	if ((*sock = SSLSocket_getPendingRead()) == -1)
 	{
 #endif
+		int should_stop = 0;
+
 		/* 0 from getReadySocket indicates no work to do, rc -1 == error */
 		*sock = Socket_getReadySocket(0, &tp, socket_mutex, &rc1);
 		*rc = rc1;
-		if (!MQTTAsync_tostop && *sock == 0 && (tp.tv_sec > 0L || tp.tv_usec > 0L))
+		MQTTAsync_lock_mutex(mqttasync_mutex);
+		should_stop = MQTTAsync_tostop;
+		MQTTAsync_unlock_mutex(mqttasync_mutex);
+		if (!should_stop && *sock == 0 && (tp.tv_sec > 0L || tp.tv_usec > 0L))
 			MQTTAsync_sleep(100L);
 #if defined(OPENSSL)
 	}
diff --git a/src/Socket.c b/src/Socket.c
index cc7aac87..189a8412 100644
--- a/src/Socket.c
+++ b/src/Socket.c
@@ -268,14 +268,15 @@ int Socket_getReadySocket(int more_work, struct timeval *tp, mutex_type mutex, i
 
 	if (mod_s.cur_clientsds == NULL)
 	{
-		int rc1;
+		int rc1, maxfdp1_saved;
 		fd_set pwset;
 
 		memcpy((void*)&(mod_s.rset), (void*)&(mod_s.rset_saved), sizeof(mod_s.rset));
 		memcpy((void*)&(pwset), (void*)&(mod_s.pending_wset), sizeof(pwset));
 		/* Prevent performance issue by unlocking the socket_mutex while waiting for a ready socket. */
+		maxfdp1_saved = mod_s.maxfdp1;
 		Thread_unlock_mutex(mutex);
-		*rc = select(mod_s.maxfdp1, &(mod_s.rset), &pwset, NULL, &timeout);
+		*rc = select(maxfdp1_saved, &(mod_s.rset), &pwset, NULL, &timeout);
 		Thread_lock_mutex(mutex);
 		if (*rc == SOCKET_ERROR)
 		{