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 timing issues in some scenarios for the poll implementation

This commit is contained in:
Ian Craggs 2022-05-26 21:26:39 +01:00
parent 22690b8881
commit 56c6151e9b
2 changed files with 102 additions and 43 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

139
src/Socket.c
View File

@ -153,10 +153,12 @@ void Socket_outInitialize(void)
memcpy((void*)&(mod_s.rset_saved), (void*)&(mod_s.rset), sizeof(mod_s.rset_saved));
#else
mod_s.nfds = 0;
mod_s.fds = NULL;
mod_s.fds_read = NULL;
mod_s.fds_write = NULL;
mod_s.saved.cur_fd = -1;
mod_s.saved.fds = NULL;
mod_s.saved.fds_write = NULL;
mod_s.saved.fds_read = NULL;
mod_s.saved.nfds = 0;
#endif
FUNC_EXIT;
@ -174,10 +176,14 @@ void Socket_outTerminate(void)
#if defined(USE_SELECT)
ListFree(mod_s.clientsds);
#else
if (mod_s.fds)
free(mod_s.fds);
if (mod_s.saved.fds)
free(mod_s.saved.fds);
if (mod_s.fds_read)
free(mod_s.fds_read);
if (mod_s.fds_write)
free(mod_s.fds_write);
if (mod_s.saved.fds_write)
free(mod_s.saved.fds_write);
if (mod_s.saved.fds_read)
free(mod_s.saved.fds_read);
#endif
SocketBuffer_terminate();
#if defined(_WIN32) || defined(_WIN64)
@ -263,25 +269,38 @@ int Socket_addSocket(SOCKET newSd)
FUNC_ENTRY;
mod_s.nfds++;
if (mod_s.fds)
mod_s.fds = realloc(mod_s.fds, mod_s.nfds * sizeof(mod_s.fds[0]));
if (mod_s.fds_read)
mod_s.fds_read = realloc(mod_s.fds_read, mod_s.nfds * sizeof(mod_s.fds_read[0]));
else
mod_s.fds = malloc(mod_s.nfds * sizeof(mod_s.fds[0]));
if (!mod_s.fds)
mod_s.fds_read = malloc(mod_s.nfds * sizeof(mod_s.fds_read[0]));
if (!mod_s.fds_read)
{
rc = PAHO_MEMORY_ERROR;
goto exit;
}
if (mod_s.fds_write)
mod_s.fds_write = realloc(mod_s.fds_write, mod_s.nfds * sizeof(mod_s.fds_write[0]));
else
mod_s.fds_write = malloc(mod_s.nfds * sizeof(mod_s.fds_write[0]));
if (!mod_s.fds_read)
{
rc = PAHO_MEMORY_ERROR;
goto exit;
}
mod_s.fds[mod_s.nfds - 1].fd = newSd;
mod_s.fds_read[mod_s.nfds - 1].fd = newSd;
mod_s.fds_write[mod_s.nfds - 1].fd = newSd;
#if defined(_WIN32) || defined(_WIN64)
mod_s.fds[mod_s.nfds - 1].events = POLLIN | POLLOUT;
mod_s.fds_read[mod_s.nfds - 1].events = POLLIN;
mod_s.fds_write[mod_s.nfds - 1].events = POLLOUT;
#else
mod_s.fds[mod_s.nfds - 1].events = POLLIN | POLLOUT | POLLNVAL;
mod_s.fds_read[mod_s.nfds - 1].events = POLLIN | POLLNVAL;
mod_s.fds_write[mod_s.nfds - 1].events = POLLOUT;
#endif
/* sort the poll fds array by socket number */
qsort(mod_s.fds, (size_t)mod_s.nfds, sizeof(mod_s.fds[0]), cmpfds);
qsort(mod_s.fds_read, (size_t)mod_s.nfds, sizeof(mod_s.fds_read[0]), cmpfds);
qsort(mod_s.fds_write, (size_t)mod_s.nfds, sizeof(mod_s.fds_write[0]), cmpfds);
rc = Socket_setnonblocking(newSd);
if (rc == SOCKET_ERROR)
@ -325,19 +344,20 @@ int isReady(int socket, fd_set* read_set, fd_set* write_set)
int isReady(int index)
{
int rc = 1;
SOCKET* socket = &mod_s.saved.fds[index].fd;
SOCKET* socket = &mod_s.saved.fds_write[index].fd;
FUNC_ENTRY;
if ((mod_s.saved.fds[index].revents & POLLHUP) || (mod_s.saved.fds[index].revents & POLLNVAL))
if ((mod_s.saved.fds_read[index].revents & POLLHUP) || (mod_s.saved.fds_read[index].revents & POLLNVAL))
; /* signal work to be done if there is an error on the socket */
else if (ListFindItem(mod_s.connect_pending, socket, intcompare) &&
(mod_s.saved.fds[index].revents & POLLOUT))
(mod_s.saved.fds_write[index].revents & POLLOUT))
ListRemoveItem(mod_s.connect_pending, socket, intcompare);
else
rc = (mod_s.saved.fds[index].revents & POLLIN) &&
(mod_s.saved.fds[index].revents & POLLOUT) &&
rc = (mod_s.saved.fds_read[index].revents & POLLIN) &&
(mod_s.saved.fds_write[index].revents & POLLOUT) &&
Socket_noPendingWrites(*socket);
FUNC_EXIT_RC(rc);
return rc;
}
@ -487,15 +507,22 @@ SOCKET Socket_getReadySocket(int more_work, int timeout, mutex_type mutex, int*
if (mod_s.saved.cur_fd == -1)
{
int rc1 = 0;
if (mod_s.nfds != mod_s.saved.nfds)
{
mod_s.saved.nfds = mod_s.nfds;
if (mod_s.saved.fds)
mod_s.saved.fds = realloc(mod_s.saved.fds, mod_s.nfds * sizeof(struct pollfd));
if (mod_s.saved.fds_read)
mod_s.saved.fds_read = realloc(mod_s.saved.fds_read, mod_s.nfds * sizeof(struct pollfd));
else
mod_s.saved.fds = malloc(mod_s.nfds * sizeof(struct pollfd));
mod_s.saved.fds_read = malloc(mod_s.nfds * sizeof(struct pollfd));
if (mod_s.saved.fds_write)
mod_s.saved.fds_write = realloc(mod_s.saved.fds_write, mod_s.nfds * sizeof(struct pollfd));
else
mod_s.saved.fds_write = malloc(mod_s.nfds * sizeof(struct pollfd));
}
memcpy(mod_s.saved.fds, mod_s.fds, mod_s.nfds * sizeof(struct pollfd));
memcpy(mod_s.saved.fds_read, mod_s.fds_read, mod_s.nfds * sizeof(struct pollfd));
memcpy(mod_s.saved.fds_write, mod_s.fds_write, mod_s.nfds * sizeof(struct pollfd));
if (mod_s.saved.nfds == 0)
{
@ -503,9 +530,17 @@ SOCKET Socket_getReadySocket(int more_work, int timeout, mutex_type mutex, int*
goto exit; /* no work to do */
}
/* Check pending write set for writeable sockets */
rc1 = poll(mod_s.saved.fds_write, mod_s.saved.nfds, 0);
if (rc1 > 0 && Socket_continueWrites(&sock, mutex) == SOCKET_ERROR)
{
*rc = SOCKET_ERROR;
goto exit;
}
/* Prevent performance issue by unlocking the socket_mutex while waiting for a ready socket. */
Thread_unlock_mutex(mutex);
*rc = poll(mod_s.saved.fds, mod_s.saved.nfds, timeout_ms);
*rc = poll(mod_s.saved.fds_read, mod_s.saved.nfds, timeout_ms);
Thread_lock_mutex(mutex);
if (*rc == SOCKET_ERROR)
{
@ -514,13 +549,7 @@ SOCKET Socket_getReadySocket(int more_work, int timeout, mutex_type mutex, int*
}
Log(TRACE_MAX, -1, "Return code %d from poll", *rc);
if (Socket_continueWrites(&sock, mutex) == SOCKET_ERROR)
{
*rc = SOCKET_ERROR;
goto exit;
}
if (*rc == 0)
if (rc1 == 0 && *rc == 0)
{
sock = 0;
goto exit; /* no work to do */
@ -540,7 +569,7 @@ SOCKET Socket_getReadySocket(int more_work, int timeout, mutex_type mutex, int*
sock = 0;
else
{
sock = mod_s.saved.fds[mod_s.saved.cur_fd].fd;
sock = mod_s.saved.fds_read[mod_s.saved.cur_fd].fd;
mod_s.saved.cur_fd = (mod_s.saved.cur_fd == mod_s.saved.nfds - 1) ? -1 : mod_s.saved.cur_fd + 1;
}
exit:
@ -915,25 +944,54 @@ int Socket_close(SOCKET socket)
ListRemoveItem(mod_s.connect_pending, &socket, intcompare);
ListRemoveItem(mod_s.write_pending, &socket, intcompare);
fd = bsearch(&socket, mod_s.fds, (size_t)mod_s.nfds, sizeof(mod_s.fds[0]), cmpsockfds);
fd = bsearch(&socket, mod_s.fds_read, (size_t)mod_s.nfds, sizeof(mod_s.fds_read[0]), cmpsockfds);
if (fd)
{
struct pollfd* last_fd = &mod_s.fds[mod_s.nfds - 1];
struct pollfd* last_fd = &mod_s.fds_read[mod_s.nfds - 1];
if (--mod_s.nfds == 0)
{
free(mod_s.fds);
mod_s.fds = NULL;
free(mod_s.fds_read);
mod_s.fds_read = NULL;
}
else
{
if (fd != last_fd)
{
/* shift array to remove the socket in question */
memmove(fd, fd + 1, (mod_s.nfds - (fd - mod_s.fds)) * sizeof(mod_s.fds[0]));
memmove(fd, fd + 1, (mod_s.nfds - (fd - mod_s.fds_read)) * sizeof(mod_s.fds_read[0]));
}
mod_s.fds = realloc(mod_s.fds, sizeof(mod_s.fds[0]) * mod_s.nfds);
if (mod_s.fds == NULL)
mod_s.fds_read = realloc(mod_s.fds_read, sizeof(mod_s.fds_read[0]) * mod_s.nfds);
if (mod_s.fds_read == NULL)
{
rc = PAHO_MEMORY_ERROR;
goto exit;
}
}
Log(TRACE_MIN, -1, "Removed socket %d", socket);
}
else
Log(LOG_ERROR, -1, "Failed to remove socket %d", socket);
fd = bsearch(&socket, mod_s.fds_write, (size_t)(mod_s.nfds+1), sizeof(mod_s.fds_write[0]), cmpsockfds);
if (fd)
{
struct pollfd* last_fd = &mod_s.fds_write[mod_s.nfds];
if (mod_s.nfds == 0)
{
free(mod_s.fds_write);
mod_s.fds_write = NULL;
}
else
{
if (fd != last_fd)
{
/* shift array to remove the socket in question */
memmove(fd, fd + 1, (mod_s.nfds - (fd - mod_s.fds_write)) * sizeof(mod_s.fds_write[0]));
}
mod_s.fds_write = realloc(mod_s.fds_write, sizeof(mod_s.fds_write[0]) * mod_s.nfds);
if (mod_s.fds_write == NULL)
{
rc = PAHO_MEMORY_ERROR;
goto exit;
@ -1142,7 +1200,6 @@ exit:
return rc;
}
static Socket_writeComplete* writecomplete = NULL;
void Socket_setWriteCompleteCallback(Socket_writeComplete* mywritecomplete)
@ -1311,7 +1368,7 @@ int Socket_continueWrites(SOCKET* sock, mutex_type mutex)
struct pollfd* fd;
/* find the socket in the fds structure */
fd = bsearch(&socket, mod_s.saved.fds, (size_t)mod_s.saved.nfds, sizeof(mod_s.saved.fds[0]), cmpsockfds);
fd = bsearch(&socket, mod_s.saved.fds_write, (size_t)mod_s.saved.nfds, sizeof(mod_s.saved.fds_write[0]), cmpsockfds);
if ((fd->revents & POLLOUT) && ((rc = Socket_continueWrite(socket)) != 0))
#endif

6
src/Socket.h
View File

@ -123,12 +123,14 @@ typedef struct
fd_set pending_wset; /**< socket pending write set for select */
#else
unsigned int nfds; /**< no of file descriptors for poll */
struct pollfd* fds; /**< poll read file descriptors */
struct pollfd* fds_read; /**< poll read file descriptors */
struct pollfd* fds_write;
struct {
int cur_fd; /**< index into the fds_saved array */
unsigned int nfds; /**< number of fds in the fds_saved array */
struct pollfd* fds;
struct pollfd* fds_write;
struct pollfd* fds_read;
} saved;
#endif
} Sockets;