natas_hw
/
paho
mirror of https://github.com/eclipse/paho.mqtt.c
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Fix some potential locking issues #1169

This commit is contained in:
Ian Craggs 2021-11-16 19:40:46 +00:00
parent baa58602c7
commit 0372393c3f
4 changed files with 22 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
src/MQTTAsync.c
View File

@ -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;

2
src/MQTTAsync.h
View File

@ -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.

16
src/MQTTAsyncUtils.c
View File

@ -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)
}

5
src/Socket.c
View File

@ -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)
{