Merge pull request #2506 from moruoxian/main

Fix adapter thread exit crash issue to enable graceful adapter shutdown.

src/adapter/adapter.c

@@ -44,6 +44,8 @@
 #include "plugin.h"
 #include "storage.h"
 
+void adapter_msg_q_exit(adapter_msg_q_t *q);
+
 static void *adapter_consumer(void *arg);
 static int   adapter_trans_data(enum neu_event_io_type type, int fd,
                                 void *usr_data);
@@ -116,8 +118,12 @@ static void *adapter_consumer(void *arg)
     neu_adapter_t *adapter = (neu_adapter_t *) arg;
 
     while (1) {
-        neu_msg_t *         msg    = NULL;
-        uint32_t            n      = adapter_msg_q_pop(adapter->msg_q, &msg);
+        neu_msg_t *msg = NULL;
+        uint32_t   n   = adapter_msg_q_pop(adapter->msg_q, &msg);
+        if (msg == NULL) {
+            // safe quit
+            break;
+        }
         neu_reqresp_head_t *header = neu_msg_get_header(msg);
 
         nlog_debug("adapter(%s) recv msg from: %s %p, type: %s, %u",
@@ -1710,6 +1716,16 @@ void neu_adapter_destroy(neu_adapter_t *adapter)
     close(adapter->control_fd);
     close(adapter->trans_data_fd);
 
+    // First set the message queue exit flag to let threads exit naturally
+    if (adapter->msg_q != NULL) {
+        adapter_msg_q_exit(adapter->msg_q);
+    }
+
+    // Wait for the consumer thread to exit
+    if (adapter->consumer_tid != 0) {
+        pthread_join(adapter->consumer_tid, NULL);
+    }
+
     adapter->module->intf_funs->close(adapter->plugin);
 
     if (NULL != adapter->metrics) {
@@ -1717,9 +1733,6 @@ void neu_adapter_destroy(neu_adapter_t *adapter)
         neu_node_metrics_free(adapter->metrics);
     }
 
-    if (adapter->consumer_tid != 0) {
-        pthread_cancel(adapter->consumer_tid);
-    }
     if (adapter->msg_q != NULL) {
         adapter_msg_q_free(adapter->msg_q);
     }

src/adapter/msg_q.c

@@ -36,8 +36,11 @@ struct adapter_msg_q {
 
     pthread_mutex_t mtx;
     pthread_cond_t  cond;
+    volatile bool   exit_flag;
 };
 
+void adapter_msg_q_exit(adapter_msg_q_t *q);
+
 adapter_msg_q_t *adapter_msg_q_new(const char *name, uint32_t size)
 {
     struct adapter_msg_q *q = calloc(1, sizeof(struct adapter_msg_q));
@@ -53,11 +56,11 @@ adapter_msg_q_t *adapter_msg_q_new(const char *name, uint32_t size)
 
 void adapter_msg_q_free(adapter_msg_q_t *q)
 {
+    adapter_msg_q_exit(q);
     struct item *tmp = NULL, *elt = NULL;
     nlog_warn("app: %s, drop %u msg", q->name, q->current);
     pthread_mutex_destroy(&q->mtx);
     pthread_cond_destroy(&q->cond);
-
     DL_FOREACH_SAFE(q->list, elt, tmp)
     {
         DL_DELETE(q->list, elt);
@@ -70,6 +73,14 @@ void adapter_msg_q_free(adapter_msg_q_t *q)
     free(q);
 }
 
+void adapter_msg_q_exit(adapter_msg_q_t *q)
+{
+    pthread_mutex_lock(&q->mtx);
+    q->exit_flag = true;
+    pthread_cond_broadcast(&q->cond);
+    pthread_mutex_unlock(&q->mtx);
+}
+
 int adapter_msg_q_push(adapter_msg_q_t *q, neu_msg_t *msg)
 {
     int ret = -1;
@@ -96,13 +107,16 @@ int adapter_msg_q_push(adapter_msg_q_t *q, neu_msg_t *msg)
 uint32_t adapter_msg_q_pop(adapter_msg_q_t *q, neu_msg_t **p_data)
 {
     uint32_t ret = 0;
-
     pthread_mutex_lock(&q->mtx);
-    while (q->current == 0) {
+    while (q->current == 0 && !q->exit_flag) {
         pthread_cond_wait(&q->cond, &q->mtx);
     }
+    if (q->exit_flag) {
+        pthread_mutex_unlock(&q->mtx);
+        *p_data = NULL;
+        return 0;
+    }
     struct item *elt = DL_LAST(q->list);
-
     if (elt != NULL) {
         DL_DELETE(q->list, elt);
         *p_data = elt->msg;
@@ -110,7 +124,6 @@ uint32_t adapter_msg_q_pop(adapter_msg_q_t *q, neu_msg_t **p_data)
         q->current -= 1;
         ret = q->current;
     }
-
     pthread_mutex_unlock(&q->mtx);
     return ret;
 }