Merge pull request #918 from nats-io/fix_event_loop_attach_detach

[FIXED] EventLoop: Handling of possible failure on initial attach

Author: kozlovic

src/adapters/libevent.h

@@ -83,6 +83,32 @@ keepAliveCb(evutil_socket_t fd, short flags, void * arg)
     // do nothing...
 }
 
+static void
+natsLibeventEvents_free(natsLibeventEvents *nle, bool processDetachedEvents)
+{
+    if (nle->read != NULL)
+        event_free(nle->read);
+    if (nle->write != NULL)
+        event_free(nle->write);
+    if (nle->keepActive != NULL)
+    {
+        event_active(nle->keepActive, 0, 0);
+        event_free(nle->keepActive);
+    }
+    if (processDetachedEvents)
+        natsConnection_ProcessDetachedEvent(nle->nc);
+    free(nle);
+}
+
+// This callback is invoked from the event loop thread and will free the
+// `natsLibeventEvents` object.
+static void
+_freeCb(evutil_socket_t ignoredSocket, short ignoredEvent, void *arg)
+{
+    natsLibeventEvents *nle = (natsLibeventEvents*) arg;
+    natsLibeventEvents_free(nle, true);
+}
+
 /** \brief Attach a connection to the given event loop.
  *
  * This callback is invoked after `NATS` library has connected, or reconnected.
@@ -101,6 +127,7 @@ natsLibevent_Attach(void **userData, void *loop, natsConnection *nc, natsSock so
     struct event_base   *libeventLoop = (struct event_base*) loop;
     natsLibeventEvents  *nle          = (natsLibeventEvents*) (*userData);
     natsStatus          s             = NATS_OK;
+    bool                created       = false;
 
     // This is the first attach (when reconnecting, nle will be non-NULL).
     if (nle == NULL)
@@ -109,6 +136,8 @@ natsLibevent_Attach(void **userData, void *loop, natsConnection *nc, natsSock so
         if (nle == NULL)
             return NATS_NO_MEMORY;
 
+        // Indicate that we have created the object here (in case we get a failure).
+        created   = true;
         nle->nc   = nc;
         nle->loop = libeventLoop;
 
@@ -153,8 +182,8 @@ natsLibevent_Attach(void **userData, void *loop, natsConnection *nc, natsSock so
 
     if (s == NATS_OK)
         *userData = (void*) nle;
-    else
-        natsLibevent_Detach((void*) nle);
+    else if (created)
+        natsLibeventEvents_free(nle, false);
 
     return s;
 }
@@ -224,27 +253,6 @@ natsLibevent_Write(void *userData, bool add)
     return (res == 0 ? NATS_OK : NATS_ERR);
 }
 
-// This callback is invoked from the event loop thread and will free the
-// `natsLibeventEvents` object.
-static void
-_freeCb(evutil_socket_t ignoredSocket, short ignoredEvent, void *arg)
-{
-    natsLibeventEvents  *nle = (natsLibeventEvents*) arg;
-
-    if (nle->read != NULL)
-        event_free(nle->read);
-    if (nle->write != NULL)
-        event_free(nle->write);
-    if (nle->keepActive != NULL)
-    {
-        event_active(nle->keepActive, 0, 0);
-        event_free(nle->keepActive);
-    }
-    // This will release the connection that is retained by the library on the first attach.
-    natsConnection_Destroy(nle->nc);
-    free(nle);
-}
-
 /** \brief The connection is closed, it can be safely detached.
  *
  * When a connection is closed (not disconnected, pending a reconnect), this

src/adapters/libuv.h

@@ -237,24 +237,33 @@ uvAsyncAttach(natsLibuvEvents *nle)
 }
 
 static void
-uvFinalCloseCb(uv_handle_t* handle)
+natsLibuvEvents_free(natsLibuvEvents *nle, bool processDetachedEvent)
 {
-    natsLibuvEvents *nle = (natsLibuvEvents*) handle->data;
-    natsLibuvEvent  *event;
+    natsLibuvEvent *event;
 
     while ((event = nle->head) != NULL)
     {
         nle->head = event->next;
         free(event);
     }
     free(nle->scheduler);
-    uv_mutex_destroy(nle->lock);
-    free(nle->lock);
-    // This will release the connection that is retained by the library on the first attach.
-    natsConnection_Destroy(nle->nc);
+    if (nle->lock != NULL)
+    {
+        uv_mutex_destroy(nle->lock);
+        free(nle->lock);
+    }
+    if (processDetachedEvent)
+        natsConnection_ProcessDetachedEvent(nle->nc);
     free(nle);
 }
 
+static void
+uvFinalCloseCb(uv_handle_t* handle)
+{
+    natsLibuvEvents *nle = (natsLibuvEvents*) handle->data;
+    natsLibuvEvents_free(nle, true);
+}
+
 static void
 uvAsyncDetach(natsLibuvEvents *nle)
 {
@@ -348,6 +357,7 @@ natsLibuv_Attach(void **userData, void *loop, natsConnection *nc, natsSock socke
     bool            sched   = false;
     natsLibuvEvents *nle    = (natsLibuvEvents*) (*userData);
     natsStatus      s       = NATS_OK;
+    bool            created = false;
 
     sched = ((uv_key_get(&uvLoopThreadKey) != loop) ? true : false);
 
@@ -362,6 +372,9 @@ natsLibuv_Attach(void **userData, void *loop, natsConnection *nc, natsSock socke
         if (nle == NULL)
             return NATS_NO_MEMORY;
 
+        // Indicate that we have created the object here (in case we get a failure).
+        created = true;
+
         nle->lock = (uv_mutex_t*) malloc(sizeof(uv_mutex_t));
         if (nle->lock == NULL)
             s = NATS_NO_MEMORY;
@@ -402,8 +415,8 @@ natsLibuv_Attach(void **userData, void *loop, natsConnection *nc, natsSock socke
 
     if (s == NATS_OK)
         *userData = (void*) nle;
-    else
-        natsLibuv_Detach((void*) nle);
+    else if (created)
+        natsLibuvEvents_free(nle, false);
 
     return s;
 }

src/conn.c

@@ -4367,6 +4367,12 @@ natsConnection_ProcessCloseEvent(natsSock *socket)
     *socket = NATS_SOCK_INVALID;
 }
 
+void
+natsConnection_ProcessDetachedEvent(natsConnection *nc)
+{
+    natsConn_release(nc);
+}
+
 natsStatus
 natsConnection_GetClientID(natsConnection *nc, uint64_t *cid)
 {

src/nats.h

@@ -5085,6 +5085,19 @@ natsConnection_ProcessCloseEvent(natsSock *socket);
 NATS_EXTERN void
 natsConnection_ProcessWriteEvent(natsConnection *nc);
 
+/** \brief Process a detach event when using external event loop.
+ *
+ * When a connection is closed, the library will invoke the adapter's
+ * #natsEvLoop_Detach callback. But the code in the adapter may run
+ * asynchronously. The adapter will invoke this function when the
+ * adapter has fully detached the NATS connection from the event loop,
+ * so that resources held by the library for this connection can be released.
+ *
+ * @param nc the pointer to the #natsConnection object.
+ */
+NATS_EXTERN void
+natsConnection_ProcessDetachedEvent(natsConnection *nc);
+
 /** \brief Connects to a `NATS Server` using any of the URL from the given list.
  *
  * Attempts to connect to a `NATS Server`.