From 37532c32464613665d7a9f65faf86f410b44eec7 Mon Sep 17 00:00:00 2001 From: Groene AI <270696204+groeneai@users.noreply.github.com> Date: Tue, 18 Aug 2026 08:08:52 +0000 Subject: [PATCH] libuv adapter: do not let an attach overtake a queued poll removal The adapter defers poll mutation to the event loop thread through a FIFO, but natsLibuv_Attach did not follow that discipline in two ways. First, it wrote nle->socket and nle->events directly. On a reconnect that runs on the nats-io reconnect thread (_doReconnect -> _processConnInit -> evCbs.attach), those writes race the event loop thread's write in uvPollUpdate's close branch. ThreadSanitizer reports it as a data race on a 4-byte field of natsLibuvEvents. Second, it scheduled only when it was called off the loop thread, while Read/Write/Detach also schedule when events are already queued. A loop-thread attach could therefore run ahead of removals already in the list. Both let an attach install a new socket before an older, already queued READ removal is drained. _evStopPolling only requests the removals and then _processOpError starts the reconnect thread, so when the loop thread finally drains that removal, nle->events reaches 0 and uvPollUpdate takes the close branch on the socket the attach just installed: natsConnection_ProcessCloseEvent closes a live, freshly connected fd and sets nle->socket to invalid. The reconnect still reports success (nc->el.attached = true) but the connection never polls again. Carry the socket in the ATTACH event and assign nle->socket and nle->events inside uvAsyncAttach, so both fields are written only by the loop thread, and give the attach the same "or events are queued" guard as the other three callbacks. Ordering is then enforced by the queue instead of by timing, and the race disappears because the fields are no longer shared. No new lock is introduced; nle->lock still guards only head/tail. Reproduced with a standalone harness that links this header against libuv under ThreadSanitizer and drives the interleaving deterministically: before, 25/25 runs close the freshly attached fd and 23/25 concurrent runs report the data race in natsLibuv_Attach; after, 0 in both. --- src/adapters/libuv.h | 79 +++++++++++++++++++++++++++++++++----------- 1 file changed, 60 insertions(+), 19 deletions(-) diff --git a/src/adapters/libuv.h b/src/adapters/libuv.h index 5f212e4be..0eead5fef 100644 --- a/src/adapters/libuv.h +++ b/src/adapters/libuv.h @@ -41,6 +41,8 @@ typedef struct __natsLibuvEvent { int type; bool add; + // Only meaningful for NATS_LIBUV_ATTACH. + natsSock socket; struct __natsLibuvEvent *next; } natsLibuvEvent; @@ -107,18 +109,9 @@ natsLibuv_SetThreadLocalLoop(uv_loop_t *loop) } static natsStatus -uvScheduleToEventLoop(natsLibuvEvents *nle, int eventType, bool add) +uvEnqueueEvent(natsLibuvEvents *nle, natsLibuvEvent *newEvent) { - natsLibuvEvent *newEvent = NULL; - int res; - - newEvent = (natsLibuvEvent*) malloc(sizeof(natsLibuvEvent)); - if (newEvent == NULL) - return NATS_NO_MEMORY; - - newEvent->type = eventType; - newEvent->add = add; - newEvent->next = NULL; + int res; uv_mutex_lock(nle->lock); @@ -142,6 +135,50 @@ uvScheduleToEventLoop(natsLibuvEvents *nle, int eventType, bool add) return (res == 0 ? NATS_OK : NATS_ERR); } +static natsLibuvEvent* +uvNewEvent(int eventType, bool add) +{ + // Zero-initialized so that `socket` has a defined value for the event + // types that do not carry one. + natsLibuvEvent *newEvent = (natsLibuvEvent*) calloc(1, sizeof(natsLibuvEvent)); + + if (newEvent == NULL) + return NULL; + + newEvent->type = eventType; + newEvent->add = add; + newEvent->next = NULL; + + return newEvent; +} + +static natsStatus +uvScheduleToEventLoop(natsLibuvEvents *nle, int eventType, bool add) +{ + natsLibuvEvent *newEvent = uvNewEvent(eventType, add); + + if (newEvent == NULL) + return NATS_NO_MEMORY; + + return uvEnqueueEvent(nle, newEvent); +} + +// The socket travels in the event so that `nle` is not written here. Callers +// pass their own socket argument; reading `nle->socket` off the event loop +// thread is not allowed. +static natsStatus +uvScheduleAttachToEventLoop(natsLibuvEvents *nle, natsSock socket) +{ + natsLibuvEvent *newEvent = uvNewEvent(NATS_LIBUV_ATTACH, true); + + if (newEvent == NULL) + return NATS_NO_MEMORY; + + newEvent->socket = socket; + + return uvEnqueueEvent(nle, newEvent); +} + static void natsLibuvPoll(uv_poll_t* handle, int status, int events) { @@ -204,11 +241,16 @@ uvPollUpdate(natsLibuvEvents *nle, int eventType, bool add) return NATS_OK; } +// Runs on the event loop thread only, which is what makes `nle->socket` and +// `nle->events` single-threaded. static natsStatus -uvAsyncAttach(natsLibuvEvents *nle) +uvAsyncAttach(natsLibuvEvents *nle, natsSock socket) { natsStatus s = NATS_OK; + nle->socket = socket; + nle->events = UV_READABLE; + // Even when this is a reconnect, previous nle->handle has already been // set to NULL (and the memory has or will be freed in uvHandleClosedCb), // so recreate now. @@ -292,7 +334,7 @@ uvAsyncCb(uv_async_t *handle) { case NATS_LIBUV_ATTACH: { - s = uvAsyncAttach(nle); + s = uvAsyncAttach(nle, event->socket); break; } case NATS_LIBUV_READ: @@ -389,13 +431,12 @@ natsLibuv_Attach(void **userData, void *loop, natsConnection *nc, natsSock socke if (s == NATS_OK) { - nle->socket = socket; - nle->events = UV_READABLE; - - if (sched) - s = uvScheduleToEventLoop(nle, NATS_LIBUV_ATTACH, true); + // See comment in natsLibuvRead. Ordering matters here too: a pending + // removal must reach uvPollUpdate before this socket is installed. + if (sched || (nle->head != NULL)) + s = uvScheduleAttachToEventLoop(nle, socket); else - s = uvAsyncAttach(nle); + s = uvAsyncAttach(nle, socket); } if (s == NATS_OK)