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

Clean up the connect latency reduction fixes #1430

This commit is contained in:
Ian Craggs 2026-02-10 15:20:42 +00:00
parent 48ed9d0b5e
commit 3a03cf9045
No known key found for this signature in database
GPG Key ID: A7AE1A8F2CCAB186
2 changed files with 46 additions and 33 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
src/MQTTAsyncUtils.c
View File

@ -3067,17 +3067,21 @@ static MQTTPacket* MQTTAsync_cycle(SOCKET* sock, unsigned long timeout, int* rc)
if ((*sock = SSLSocket_getPendingRead()) == -1)
{
#endif
int should_stop = 0;
/*int should_stop = 0;*/
int interrupted = 0;
/* 0 from getReadySocket indicates no work to do, rc -1 == error */
*sock = Socket_getReadySocket(0, (int)timeout, socket_mutex, &rc1, &interrupted);
*rc = rc1;
MQTTAsync_lock_mutex(mqttasync_mutex);
/*MQTTAsync_lock_mutex(mqttasync_mutex);
should_stop = MQTTAsync_tostop;
MQTTAsync_unlock_mutex(mqttasync_mutex);
if (!should_stop && *sock == 0 && (timeout > 0L) && (interrupted == 0))
MQTTAsync_sleep(50L);
MQTTAsync_unlock_mutex(mqttasync_mutex);*/
/* Frank deleted this small delay on the suggestion of Claude.
I want to preserve it as a comment in case we need it in the future.
The preceeding 3 lines of code are only needed if the following line
is reinstated - Ian
if (!should_stop && *sock == 0 && (timeout > 0L) && (interrupted == 0))
MQTTAsync_sleep(50L);*/
#if defined(OPENSSL)
}
#endif

65
src/Socket.c
View File

@ -25,6 +25,10 @@
* \brief Socket related functions
*
* Some other related functions are in the SocketBuffer module
*
* Some portions of this file were generated by Claude:
* - the Windows implementation of socketpair()
*
*/
@ -149,8 +153,9 @@ void SIGPIPE_ignore()
/*
* socketpair() emulation for Windows
* Creates a pair of connected sockets using TCP loopback
*
* This function was generated by Claude
*/
#if defined(_WIN32)
#ifndef AF_LOCAL
@ -560,20 +565,23 @@ int isReady(int index)
}
#endif
int isConnectReady(int index)
/**
* Check whether the socket indicated by the index is in the
* TCP connect pending state.
* @param index
* @return 1 if the socket has a TCP connect pending, otherwise 0
*/
int isConnectPending(int index)
{
int rc = 0;
SOCKET* socket = &mod_s.saved.fds_write[index].fd;
FUNC_ENTRY;
if (ListFindItem(mod_s.connect_pending, socket, intcompare))
Log(TRACE_PROTOCOL, -1, "isConnectReady connect_pending for socket %d, POLLOUT %d",
Log(TRACE_PROTOCOL, -1, "TCP connect is pending for socket %d, POLLOUT %d",
*socket, (mod_s.saved.fds_write[index].revents & POLLOUT));
if (ListFindItem(mod_s.connect_pending, socket, intcompare) /* &&
(mod_s.saved.fds_write[index].revents & POLLOUT)*/)
{
if (ListFindItem(mod_s.connect_pending, socket, intcompare))
rc = 1;
}
FUNC_EXIT_RC(rc);
return rc;
}
@ -797,42 +805,32 @@ SOCKET Socket_getReadySocket(int more_work, int timeout, mutex_type mutex, int*
*rc = SOCKET_ERROR;
goto exit;
}
/* TODO: if there is a writeable socket in the connecting state (see isReady)
* then set the timeout for the next poll to 0, so we can handle the connect
* immediately */
/* if there is a writeable socket in the connecting state then set the timeout for the
* next poll to 0 or a low value, so we can handle the connect immediately */
mod_s.saved.cur_fd = 0;
while (mod_s.saved.cur_fd != -1)
{
if (isConnectReady(mod_s.saved.cur_fd))
if (isConnectPending(mod_s.saved.cur_fd))
{
printf("Socket %d is connecting, reducing timeout from %d, sockets %d\n", mod_s.saved.cur_fd, timeout_ms,
mod_s.saved.nfds);
Log(TRACE_MINIMUM, -1, "Socket %d is connecting, reducing timeout from %d, no of sockets %d\n",
mod_s.saved.cur_fd, timeout_ms, mod_s.saved.nfds);
timeout_ms = (mod_s.saved.nfds > 2) ? 10 : 10;
/* Will this result in excessive CPU use if the TCP connect is taking a long time,
* such as on a satellite link? We could use a backoff algorithm if it turns out to be true */
timeout_ms = 0;
break;
}
mod_s.saved.cur_fd = (mod_s.saved.cur_fd == mod_s.saved.nfds - 1) ? -1 : mod_s.saved.cur_fd + 1;
}
/* Prevent performance issue by unlocking the socket_mutex while waiting for a ready socket. */
Paho_thread_unlock_mutex(mutex);
/* clear interrupt socket */
/* clear interrupt socket - make sure it is unset */
{
static char dbuf[1];
/*int irc = */recv(sockfd[0], dbuf, 1, 0);
//printf("rc %d from read for interrupt\n", irc);
}
Log(TRACE_PROTOCOL, -1, "Start poll, sockets %d", mod_s.saved.nfds);
/* Prevent performance issue by unlocking the socket_mutex while waiting for a ready socket. */
Paho_thread_unlock_mutex(mutex);
*rc = poll(mod_s.saved.fds_read, mod_s.saved.nfds, timeout_ms);
Log(TRACE_PROTOCOL, -1, "Return code %d from poll", *rc);
/* clear interrupt socket */
{
static char dbuf[1];
int irc = recv(sockfd[0], dbuf, 1, 0);
//printf("rc %d from read for interrupt\n", irc);
if (irc != -1)
*interrupted = 1;
}
Paho_thread_lock_mutex(mutex);
if (*rc == SOCKET_ERROR)
{
@ -841,6 +839,17 @@ SOCKET Socket_getReadySocket(int more_work, int timeout, mutex_type mutex, int*
}
Log(TRACE_MAX, -1, "Return code %d from poll", *rc);
/* check interrupt socket to see if we were interrupted */
{
static char dbuf[1];
int irc = recv(sockfd[0], dbuf, 1, 0);
if (irc > 0)
{
*interrupted = 1;
Log(TRACE_MINIMUM, -1, "Poll was interrupted, rc %d", irc);
}
}
if (rc1 == 0 && *rc == 0)
{
sock = 0;