Skip to content

libuv adapter: do not let an attach overtake a queued poll removal #5

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
alexey-milovidov merged 1 commit into from
Sep 13, 2026
Merged

libuv adapter: do not let an attach overtake a queued poll removal #5

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
79 changes: 60 additions & 19 deletions src/adapters/libuv.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -41,6 +41,8 @@ typedef struct __natsLibuvEvent
{
int type;
bool add;
// Only meaningful for NATS_LIBUV_ATTACH.
natsSock socket;
struct __natsLibuvEvent *next;

} natsLibuvEvent;
Expand Down Expand Up @@ -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);

Expand All @@ -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)
{
Expand Down Expand Up @@ -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.
Expand Down Expand Up @@ -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:
Expand Down Expand Up @@ -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)
Expand Down