diff --git a/src/adapters/libevent.h b/src/adapters/libevent.h index c9863242f..6952bb44c 100644 --- a/src/adapters/libevent.h +++ b/src/adapters/libevent.h @@ -158,6 +158,17 @@ natsLibevent_Attach(void **userData, void *loop, natsConnection *nc, natsSock so return s; } +static void +_closeCb(evutil_socket_t fd, short event, void *arg) +{ + natsSock socket = (natsSock) fd; + + // We have stopped polling for the "READ" event and are now in the + // event loop thread and invoke this so that the NATS C client + // library can proceed with the close of the socket/connection. + natsConnection_ProcessCloseEvent(&socket); +} + /** \brief Start or stop polling on READ events. * * This callback is invoked to notify that the event library should start @@ -175,7 +186,16 @@ natsLibevent_Read(void *userData, bool add) if (add) res = event_add(nle->read, NULL); else + { + int socket = event_get_fd(nle->read); res = event_del_noblock(nle->read); + if (res == 0) + { + // This will schedule a one-time event that guarantees that the + // callback `_closeCb` will be invoked from the event loop thread. + res = event_base_once(nle->loop, socket, EV_TIMEOUT, _closeCb, (void*) nle, NULL); + } + } return (res == 0 ? NATS_OK : NATS_ERR); } diff --git a/src/adapters/libuv.h b/src/adapters/libuv.h index 50499e98b..5f212e4be 100644 --- a/src/adapters/libuv.h +++ b/src/adapters/libuv.h @@ -130,9 +130,14 @@ uvScheduleToEventLoop(natsLibuvEvents *nle, int eventType, bool add) nle->tail = newEvent; - uv_mutex_unlock(nle->lock); - + // We need to wake up the event loop thread under our lock because + // due to signal coalescing (and the reason we have a list), it is + // possible that the detach that we have just added is processed + // after we release the lock, freeing the `nle` structure. Calling + // `uv_async_send(nle->scheduler)` outside this lock would then + // cause a crash or race. res = uv_async_send(nle->scheduler); + uv_mutex_unlock(nle->lock); return (res == 0 ? NATS_OK : NATS_ERR); } @@ -158,11 +163,15 @@ natsLibuvPoll(uv_poll_t* handle, int status, int events) natsConnection_ProcessWriteEvent(nle->nc); } +static void +uvHandleClosedCb(uv_handle_t *handle) +{ + free(handle); +} + static natsStatus uvPollUpdate(natsLibuvEvents *nle, int eventType, bool add) { - int res; - if (eventType == NATS_LIBUV_READ) { if (add) @@ -179,34 +188,30 @@ uvPollUpdate(natsLibuvEvents *nle, int eventType, bool add) } if (nle->events) - res = uv_poll_start(nle->handle, nle->events, natsLibuvPoll); - else - res = uv_poll_stop(nle->handle); - - if (res != 0) - return NATS_ERR; + { + int res = uv_poll_start(nle->handle, nle->events, natsLibuvPoll); + return (res == 0 ? NATS_OK : NATS_ERR); + } + // Both read and write events have been removed, this signal that the socket + // should be closed prior to a reconnect or during natsConnection_Close(). + uv_close((uv_handle_t*) nle->handle, uvHandleClosedCb); + nle->handle = NULL; + // We have stopped polling for events for this socket and are in the event + // loop thread, so we invoke this so that the NATS C client library can + // proceed with closing the socket. + natsConnection_ProcessCloseEvent(&(nle->socket)); return NATS_OK; } -static void -uvHandleClosedCb(uv_handle_t *handle) -{ - free(handle); -} - static natsStatus uvAsyncAttach(natsLibuvEvents *nle) { natsStatus s = NATS_OK; - // We are reconnecting, destroy the old handle, create a new one - if (nle->handle != NULL) - { - uv_close((uv_handle_t*) nle->handle, uvHandleClosedCb); - nle->handle = NULL; - } - + // 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. nle->handle = (uv_poll_t*) malloc(sizeof(uv_poll_t)); if (nle->handle == NULL) s = NATS_NO_MEMORY; @@ -232,9 +237,9 @@ uvAsyncAttach(natsLibuvEvents *nle) } static void -finalCloseCb(uv_handle_t* handle) +uvFinalCloseCb(uv_handle_t* handle) { - natsLibuvEvents *nle = (natsLibuvEvents*)handle->data; + natsLibuvEvents *nle = (natsLibuvEvents*) handle->data; natsLibuvEvent *event; while ((event = nle->head) != NULL) @@ -242,25 +247,16 @@ finalCloseCb(uv_handle_t* handle) nle->head = event->next; free(event); } - free(nle->handle); free(nle->scheduler); uv_mutex_destroy(nle->lock); free(nle->lock); free(nle); } -static void -closeSchedulerCb(uv_handle_t* scheduler) -{ - natsLibuvEvents *nle = (natsLibuvEvents*) scheduler->data; - - uv_close((uv_handle_t*) nle->handle, finalCloseCb); -} - static void uvAsyncDetach(natsLibuvEvents *nle) { - uv_close((uv_handle_t*) nle->scheduler, closeSchedulerCb); + uv_close((uv_handle_t*) nle->scheduler, uvFinalCloseCb); } static void @@ -308,6 +304,10 @@ uvAsyncCb(uv_async_t *handle) case NATS_LIBUV_DETACH: { uvAsyncDetach(nle); + // We want to make sure that we will exit this loop since by now + // the `nle` structure may have been freed. Regardless, this is + // supposed to be the last event for this `nle` object. + more = false; break; } default: diff --git a/src/conn.c b/src/conn.c index 37fcc1790..91cea7a11 100644 --- a/src/conn.c +++ b/src/conn.c @@ -2146,9 +2146,21 @@ _evStopPolling(natsConnection *nc) nc->sockCtx.useEventLoop = false; nc->el.writeAdded = false; - s = nc->opts->evCbs.read(nc->el.data, NATS_EVENT_ACTION_REMOVE); + // The "write" event is added and removed as we write, however, we always + // have the "read" event added to the event loop. Removing it signals that + // the connection is closed and so the event loop adapter can then invoke + // natsConnection_ProcessCloseEvent() when the event loop is done polling + // the event. So we will remove "write" first, then finish with "read". + s = nc->opts->evCbs.write(nc->el.data, NATS_EVENT_ACTION_REMOVE); if (s == NATS_OK) - s = nc->opts->evCbs.write(nc->el.data, NATS_EVENT_ACTION_REMOVE); + s = nc->opts->evCbs.read(nc->el.data, NATS_EVENT_ACTION_REMOVE); + if (s == NATS_OK) + { + // We can't close the socket here, but we will mark as invalid and + // clear SSL object if applicable. + nc->sockCtx.fd = NATS_SOCK_INVALID; + _clearSSL(nc); + } return s; } @@ -2187,6 +2199,7 @@ _processOpError(natsConnection *nc, natsStatus s, bool initialConnect) SET_WRITE_DEADLINE(nc); natsConn_bufferFlush(nc); + // Shutdown the socket to stop any read/write operations. natsSock_Shutdown(nc->sockCtx.fd); nc->sockCtx.fdActive = false; } @@ -2195,12 +2208,10 @@ _processOpError(natsConnection *nc, natsStatus s, bool initialConnect) // on the socket since we are going to reconnect. if (nc->el.attached) { + // This will take care of invalidating the socket and clear SSL, + // but the actual socket close will be done from the event loop + // adapter by calling natsConnection_ProcessCloseEvent(). ls = _evStopPolling(nc); - natsSock_Close(nc->sockCtx.fd); - nc->sockCtx.fd = NATS_SOCK_INVALID; - - // We need to cleanup some things if the connection was SSL. - _clearSSL(nc); } // Fail pending flush requests. @@ -2579,13 +2590,20 @@ _close(natsConnection *nc, natsConnStatus status, bool fromPublicClose, bool doC { // If event loop attached, stop polling... if (nc->el.attached) + { + // This will take care of invalidating the socket and clear SSL, + // but the actual socket close will be done from the event loop + // adapter by calling natsConnection_ProcessCloseEvent(). _evStopPolling(nc); + } + else + { + natsSock_Close(nc->sockCtx.fd); + nc->sockCtx.fd = NATS_SOCK_INVALID; - natsSock_Close(nc->sockCtx.fd); - nc->sockCtx.fd = NATS_SOCK_INVALID; - - // We need to cleanup some things if the connection was SSL. - _clearSSL(nc); + // We need to cleanup some things if the connection was SSL. + _clearSSL(nc); + } } else { @@ -3411,6 +3429,7 @@ natsConnection_Reconnect(natsConnection *nc) natsSock_Shutdown(nc->sockCtx.fd); natsConn_Unlock(nc); + return NATS_OK; } @@ -4098,13 +4117,16 @@ natsConnection_ProcessReadEvent(natsConnection *nc) buffer = nc->el.buffer; size = nc->opts->ioBufSize; - natsConn_Unlock(nc); - // Do not try to read again here on success. If more than one connection // is attached to the same loop, and there is a constant stream of data // coming for the first connection, this would starve the second connection. // So return and we will be called back later by the event loop. + + // This needs to be protected by the connection lock. We are here because + // there is a read event, so we will gather some data in natsSock_Read() + // but not wait there. s = natsSock_Read(&(nc->sockCtx), buffer, size, &n); + natsConn_Unlock(nc); if (s == NATS_OK) s = natsParser_Parse(nc, buffer, n); @@ -4159,8 +4181,16 @@ natsConnection_ProcessWriteEvent(natsConnection *nc) if (s != NATS_OK) _processOpError(nc, s, false); +} + +void +natsConnection_ProcessCloseEvent(natsSock *socket) +{ + if ((socket == NULL) || (*socket == NATS_SOCK_INVALID)) + return; - (void) NATS_UPDATE_ERR_STACK(s); + natsSock_Close(*socket); + *socket = NATS_SOCK_INVALID; } natsStatus diff --git a/src/nats.h b/src/nats.h index 1948218ef..c3bfbb230 100644 --- a/src/nats.h +++ b/src/nats.h @@ -4194,6 +4194,20 @@ natsConnection_Reconnect(natsConnection *nc); NATS_EXTERN void natsConnection_ProcessReadEvent(natsConnection *nc); +/** \brief Process a socket close event when using external event loop. + * + * When using an external event loop, and the library wants to close + * the connection, the event loop adapter will ensure that the event + * loop library stops polling, and then will invoke this function + * so that the socket can be safely closed. + * + * @param socket the pointer to the #natsSock object. + * + * \warning This API is reserved for external event loop adapters. + */ +NATS_EXTERN void +natsConnection_ProcessCloseEvent(natsSock *socket); + /** \brief Process a write event when using external event loop. * * When using an external event loop, and the callback indicating that diff --git a/test/test.c b/test/test.c index c2792f282..96ede2143 100644 --- a/test/test.c +++ b/test/test.c @@ -160,7 +160,7 @@ struct threadArg int attached; int detached; - bool evStop; + bool doRead; bool doWrite; @@ -20230,6 +20230,11 @@ _evLoopRead(void *userData, bool add) natsMutex_Lock(arg->m); arg->doRead = add; + if (!add) + { + arg->closed = true; + arg->sock = NATS_SOCK_INVALID; + } natsCondition_Broadcast(arg->c); natsMutex_Unlock(arg->m); @@ -20266,30 +20271,48 @@ static void _eventLoop(void *closure) { struct threadArg *arg = (struct threadArg *) closure; - natsSock sock = NATS_SOCK_INVALID; natsConnection *nc = NULL; bool read = false; bool write= false; bool stop = false; + bool close= false; + natsSockCtx ctx; + + natsSock_Init(&ctx); while (!stop) { - nats_Sleep(100); natsMutex_Lock(arg->m); - while (!arg->evStop && ((sock = arg->sock) == NATS_SOCK_INVALID)) + while (!arg->done && !arg->closed && (arg->sock == NATS_SOCK_INVALID)) natsCondition_Wait(arg->c, arg->m); - stop = arg->evStop; + stop = arg->done; nc = arg->nc; read = arg->doRead; write = arg->doWrite; + close = arg->closed; + arg->closed = false; + if ((ctx.fd == NATS_SOCK_INVALID) && (arg->sock != NATS_SOCK_INVALID)) + ctx.fd = arg->sock; natsMutex_Unlock(arg->m); if (!stop) { if (read) - natsConnection_ProcessReadEvent(nc); + { + natsStatus s; + + natsSock_InitDeadline(&ctx, 50); + s = natsSock_WaitReady(WAIT_FOR_READ, &ctx); + if (s == NATS_OK) + natsConnection_ProcessReadEvent(nc); + } if (write) natsConnection_ProcessWriteEvent(nc); + if (close) + { + natsConnection_ProcessCloseEvent(&ctx.fd); + close = false; + } } } } @@ -20355,6 +20378,7 @@ void test_EventLoop(void) natsMutex_Lock(arg.m); while ((s != NATS_TIMEOUT) && !arg.reconnected) s = natsCondition_TimedWait(arg.c, arg.m, 2000); + arg.reconnected = false; natsMutex_Unlock(arg.m); testCond(s == NATS_OK); @@ -20366,6 +20390,29 @@ void test_EventLoop(void) s = natsSubscription_NextMsg(&msg, sub, 1000); testCond(s == NATS_OK); natsMsg_Destroy(msg); + msg = NULL; + + test("Explicit reconnect: "); + s = natsConnection_Reconnect(nc); + testCond(s == NATS_OK); + + test("Wait for reconnect: "); + natsMutex_Lock(arg.m); + while ((s != NATS_TIMEOUT) && !arg.reconnected) + s = natsCondition_TimedWait(arg.c, arg.m, 2000); + arg.reconnected = false; + natsMutex_Unlock(arg.m); + testCond(s == NATS_OK); + + test("Publish: "); + s = natsConnection_PublishString(nc, "foo", "bar"); + testCond(s == NATS_OK); + + test("Check msg received: "); + s = natsSubscription_NextMsg(&msg, sub, 1000); + testCond(s == NATS_OK); + natsMsg_Destroy(msg); + msg = NULL; test("Close and wait for close cb: "); natsConnection_Close(nc); @@ -20373,7 +20420,7 @@ void test_EventLoop(void) testCond(s == NATS_OK); natsMutex_Lock(arg.m); - arg.evStop = true; + arg.done = true; natsCondition_Broadcast(arg.c); natsMutex_Unlock(arg.m); @@ -20382,7 +20429,7 @@ void test_EventLoop(void) test("Check ev loop: "); natsMutex_Lock(arg.m); - if (arg.attached != 2 || !arg.detached) + if (arg.attached != 3 || !arg.detached || (arg.sock != NATS_SOCK_INVALID)) s = NATS_ERR; natsMutex_Unlock(arg.m); testCond(s == NATS_OK); @@ -20467,7 +20514,7 @@ void test_EventLoopRetryOnFailedConnect(void) testCond(s == NATS_OK); natsMutex_Lock(arg.m); - arg.evStop = true; + arg.done = true; natsCondition_Broadcast(arg.c); natsMutex_Unlock(arg.m); @@ -20511,7 +20558,7 @@ void test_EventLoopTLS(void) testCond(s == NATS_OK); test("Start server: "); - pid = _startServer("nats://127.0.0.1:4443", "-config tls.conf -DV", true); + pid = _startServer("nats://127.0.0.1:4443", "-config tls.conf", true); CHECK_SERVER_STARTED(pid); testCond(s == NATS_OK); @@ -20544,16 +20591,20 @@ void test_EventLoopTLS(void) natsMutex_Lock(arg.m); while ((s != NATS_TIMEOUT) && !arg.reconnected) s = natsCondition_TimedWait(arg.c, arg.m, 2000); + arg.reconnected = false; natsMutex_Unlock(arg.m); testCond(s == NATS_OK); - test("Shutdown evLoop: "); + test("Explicit reconnect: "); + s = natsConnection_Reconnect(nc); + testCond(s == NATS_OK); + + test("Wait for reconnect: "); natsMutex_Lock(arg.m); - arg.evStop = true; - natsCondition_Broadcast(arg.c); + while ((s != NATS_TIMEOUT) && !arg.reconnected) + s = natsCondition_TimedWait(arg.c, arg.m, 2000); + arg.reconnected = false; natsMutex_Unlock(arg.m); - natsThread_Join(arg.t); - natsThread_Destroy(arg.t); testCond(s == NATS_OK); test("Close and wait for close cb: "); @@ -20561,6 +20612,15 @@ void test_EventLoopTLS(void) s = _waitForConnClosed(&arg); testCond(s == NATS_OK); + test("Shutdown evLoop: "); + natsMutex_Lock(arg.m); + arg.done = true; + natsCondition_Broadcast(arg.c); + natsMutex_Unlock(arg.m); + natsThread_Join(arg.t); + natsThread_Destroy(arg.t); + testCond(s == NATS_OK); + natsConnection_Destroy(nc); natsOptions_Destroy(opts); @@ -21401,7 +21461,7 @@ void test_SSLServerNameIndication(void) arg.control = 3; s = _startMockupServer(&sock, "localhost", "4222"); - + // Start the thread that will try to connect to our server... IFOK(s, natsThread_Create(&t, _connectToMockupServer, (void*) &arg)); @@ -21411,7 +21471,7 @@ void test_SSLServerNameIndication(void) { s = NATS_SYS_ERROR; } - + testCond((s == NATS_OK) && (ctx.fd > 0)); test("Read ClientHello from client: "); @@ -34059,7 +34119,7 @@ void test_MicroQueueGroupForEndpoint(void) #define _testQueueGroup(_expected, _actual) \ (_expected) == NULL ? (_actual) == NULL : strcmp((_expected), (_actual)) == 0 - testCond((err == NULL) && + testCond((err == NULL) && (info != NULL) && (info->EndpointsLen == 3) && (stats != NULL) && (stats->EndpointsLen == 3) && (_testQueueGroup(tc.expectedServiceLevel, info->Endpoints[0].QueueGroup)) &&