Merge pull request #1801 from fengzeroz/v2.6

version: 2.6.4

.github/workflows/ci.yml

@@ -8,7 +8,7 @@ on:
       - published
 
 env:
-  VERSION: 2.6.3
+  VERSION: 2.6.4
   UI_VERSION: 2.6.3
   BRANCH: v2.6
 

CMakeLists.txt

@@ -123,6 +123,7 @@ set(NEURON_SOURCES
     src/core/node_manager.c
     src/core/template_manager.c
     src/core/storage.c
+    src/adapter/msg_q.c
     src/adapter/storage.c
     src/adapter/adapter.c
     src/adapter/driver/cache.c

plugins/mqtt/mqtt_config.c

@@ -401,6 +401,7 @@ int mqtt_config_parse(neu_plugin_t *plugin, const char *setting,
 
 error:
     free(err_param);
+    free(client_id.v.val_str);
     free(write_req_topic.v.val_str);
     free(write_resp_topic.v.val_str);
     free(host.v.val_str);

plugins/mqtt/mqtt_handle.c

@@ -152,6 +152,38 @@ static inline int send_read_req(neu_plugin_t *plugin, neu_json_mqtt_t *mqtt,
     return 0;
 }
 
+static int json_value_to_tag_value(union neu_json_value *req,
+                                   enum neu_json_type t, neu_dvalue_t *value)
+{
+    switch (t) {
+    case NEU_JSON_INT:
+        value->type      = NEU_TYPE_INT64;
+        value->value.u64 = req->val_int;
+        break;
+    case NEU_JSON_STR:
+        value->type = NEU_TYPE_STRING;
+        strncpy(value->value.str, req->val_str, sizeof(value->value.str));
+        break;
+    case NEU_JSON_DOUBLE:
+        value->type      = NEU_TYPE_DOUBLE;
+        value->value.d64 = req->val_double;
+        break;
+    case NEU_JSON_BOOL:
+        value->type          = NEU_TYPE_BOOL;
+        value->value.boolean = req->val_bool;
+        break;
+    case NEU_JSON_BYTES:
+        value->type               = NEU_TYPE_BYTES;
+        value->value.bytes.length = req->val_bytes.length;
+        memcpy(value->value.bytes.bytes, req->val_bytes.bytes,
+               req->val_bytes.length);
+        break;
+    default:
+        return -1;
+    }
+    return 0;
+}
+
 static int send_write_tag_req(neu_plugin_t *plugin, neu_json_mqtt_t *mqtt,
                               neu_json_write_req_t *req)
 {
@@ -168,30 +200,7 @@ static int send_write_tag_req(neu_plugin_t *plugin, neu_json_mqtt_t *mqtt,
     cmd.group  = req->group;
     cmd.tag    = req->tag;
 
-    switch (req->t) {
-    case NEU_JSON_INT:
-        cmd.value.type      = NEU_TYPE_INT64;
-        cmd.value.value.u64 = req->value.val_int;
-        break;
-    case NEU_JSON_STR:
-        cmd.value.type = NEU_TYPE_STRING;
-        strcpy(cmd.value.value.str, req->value.val_str);
-        break;
-    case NEU_JSON_DOUBLE:
-        cmd.value.type      = NEU_TYPE_DOUBLE;
-        cmd.value.value.d64 = req->value.val_double;
-        break;
-    case NEU_JSON_BOOL:
-        cmd.value.type          = NEU_TYPE_BOOL;
-        cmd.value.value.boolean = req->value.val_bool;
-        break;
-    case NEU_JSON_BYTES:
-        cmd.value.type               = NEU_TYPE_BYTES;
-        cmd.value.value.bytes.length = req->value.val_bytes.length;
-        memcpy(cmd.value.value.bytes.bytes, req->value.val_bytes.bytes,
-               req->value.val_bytes.length);
-        break;
-    default:
+    if (0 != json_value_to_tag_value(&req->value, req->t, &cmd.value)) {
         plog_error(plugin, "invalid tag value type: %d", req->t);
         return -1;
     }
@@ -237,32 +246,9 @@ static int send_write_tags_req(neu_plugin_t *plugin, neu_json_mqtt_t *mqtt,
 
     for (int i = 0; i < cmd.n_tag; i++) {
         strcpy(cmd.tags[i].tag, req->tags[i].tag);
-        switch (req->tags[i].t) {
-        case NEU_JSON_INT:
-            cmd.tags[i].value.type      = NEU_TYPE_INT64;
-            cmd.tags[i].value.value.u64 = req->tags[i].value.val_int;
-            break;
-        case NEU_JSON_STR:
-            cmd.tags[i].value.type = NEU_TYPE_STRING;
-            strcpy(cmd.tags[i].value.value.str, req->tags[i].value.val_str);
-            break;
-        case NEU_JSON_DOUBLE:
-            cmd.tags[i].value.type      = NEU_TYPE_DOUBLE;
-            cmd.tags[i].value.value.d64 = req->tags[i].value.val_double;
-            break;
-        case NEU_JSON_BOOL:
-            cmd.tags[i].value.type          = NEU_TYPE_BOOL;
-            cmd.tags[i].value.value.boolean = req->tags[i].value.val_bool;
-            break;
-        case NEU_JSON_BYTES:
-            cmd.tags[i].value.type = NEU_TYPE_BYTES;
-            cmd.tags[i].value.value.bytes.length =
-                req->tags[i].value.val_bytes.length;
-            memcpy(cmd.tags[i].value.value.bytes.bytes,
-                   req->tags[i].value.val_bytes.bytes,
-                   req->tags[i].value.val_bytes.length);
-            break;
-        default:
+        if (0 !=
+            json_value_to_tag_value(&req->tags[i].value, req->tags[i].t,
+                                    &cmd.tags[i].value)) {
             plog_error(plugin, "invalid tag value type: %d", req->tags[i].t);
             free(cmd.tags);
             return -1;
@@ -490,7 +476,6 @@ int handle_read_response(neu_plugin_t *plugin, neu_json_mqtt_t *mqtt_json,
 
 end:
     neu_json_decode_mqtt_req_free(mqtt_json);
-    free(data->tags);
     return rv;
 }
 

plugins/restful/rw_handle.c

@@ -174,7 +174,7 @@ void handle_write_tags(nng_aio *aio)
                 switch (req->tags[i].t) {
                 case NEU_JSON_INT:
                     cmd.tags[i].value.type      = NEU_TYPE_INT64;
-                    cmd.tags[i].value.value.u64 = req->tags[i].value.val_int;
+                    cmd.tags[i].value.value.i64 = req->tags[i].value.val_int;
                     break;
                 case NEU_JSON_STR:
                     cmd.tags[i].value.type = NEU_TYPE_STRING;

src/adapter/adapter.c

@@ -38,11 +38,12 @@
 #include "plugin.h"
 #include "storage.h"
 
-static int adapter_trans_data(enum neu_event_io_type type, int fd,
-                              void *usr_data);
-static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data);
-static int adapter_command(neu_adapter_t *adapter, neu_reqresp_head_t header,
-                           void *data);
+static void *adapter_consumer(void *arg);
+static int   adapter_trans_data(enum neu_event_io_type type, int fd,
+                                void *usr_data);
+static int   adapter_loop(enum neu_event_io_type type, int fd, void *usr_data);
+static int   adapter_command(neu_adapter_t *adapter, neu_reqresp_head_t header,
+                             void *data);
 static int adapter_response(neu_adapter_t *adapter, neu_reqresp_head_t *header,
                             void *data);
 static int adapter_responseto(neu_adapter_t *     adapter,
@@ -104,6 +105,36 @@ void neu_adapter_set_error(int error)
     create_adapter_error = error;
 }
 
+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_reqresp_head_t *header = neu_msg_get_header(msg);
+
+        nlog_debug("adapter(%s) recv msg from: %s %p, type: %s, %u",
+                   adapter->name, header->sender, header->ctx,
+                   neu_reqresp_type_string(header->type), n);
+        adapter->module->intf_funs->request(
+            adapter->plugin, (neu_reqresp_head_t *) header, &header[1]);
+        neu_trans_data_free((neu_reqresp_trans_data_t *) &header[1]);
+        neu_msg_free(msg);
+    }
+
+    return NULL;
+}
+
+static inline zlog_category_t *get_log_category(const char *node)
+{
+    char name[NEU_NODE_NAME_LEN] = { 0 };
+    // replace path separators
+    for (int i = 0; node[i]; ++i) {
+        name[i] = ('/' == node[i] || '\\' == node[i]) ? '_' : node[i];
+    }
+    return zlog_get_category(name);
+}
+
 neu_adapter_t *neu_adapter_create(neu_adapter_info_t *info, bool load)
 {
     int                  rv      = 0;
@@ -167,6 +198,9 @@ neu_adapter_t *neu_adapter_create(neu_adapter_info_t *info, bool load)
         break;
     case NEU_NA_TYPE_NDRIVER:
     case NEU_NA_TYPE_APP: {
+        adapter->msg_q = adapter_msg_q_new(adapter->name, 1024);
+        pthread_create(&adapter->consumer_tid, NULL, adapter_consumer,
+                       (void *) adapter);
         while (true) {
             uint16_t           port  = neu_manager_get_port();
             struct sockaddr_in local = {
@@ -202,7 +236,7 @@ neu_adapter_t *neu_adapter_create(neu_adapter_info_t *info, bool load)
     common->adapter             = adapter;
     common->adapter_callbacks   = &adapter->cb_funs;
     common->link_state          = NEU_NODE_LINK_STATE_DISCONNECTED;
-    common->log                 = zlog_get_category(adapter->name);
+    common->log                 = get_log_category(adapter->name);
     strcpy(common->name, adapter->name);
 
     zlog_level_switch(common->log, default_log_level);
@@ -263,7 +297,7 @@ int neu_adapter_rename(neu_adapter_t *adapter, const char *new_name)
         return NEU_ERR_EINTERNAL;
     }
 
-    zlog_category_t *log = zlog_get_category(name);
+    zlog_category_t *log = get_log_category(name);
     if (NULL == log) {
         free(name);
         free(old_name);
@@ -584,11 +618,18 @@ static int adapter_trans_data(enum neu_event_io_type type, int fd,
         return 0;
     }
 
+    if (header->type == NEU_REQRESP_TRANS_DATA) {
+        if (adapter_msg_q_push(adapter->msg_q, msg) < 0) {
+            nlog_warn("adapter: %s trans data msg q is full, drop msg",
+                      adapter->name);
+            neu_trans_data_free((neu_reqresp_trans_data_t *) &header[1]);
+            neu_msg_free(msg);
+        }
+        return 0;
+    }
+
     adapter->module->intf_funs->request(
         adapter->plugin, (neu_reqresp_head_t *) header, &header[1]);
-    if (header->type == NEU_REQRESP_TRANS_DATA) {
-        neu_trans_data_free((neu_reqresp_trans_data_t *) &header[1]);
-    }
     neu_msg_free(msg);
     return 0;
 }
@@ -1211,6 +1252,13 @@ 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);
+    }
+
     char *setting = NULL;
     if (adapter_load_setting(adapter->name, &setting) != 0) {
         remove_logs(adapter->name);

src/adapter/adapter_internal.h

@@ -26,6 +26,7 @@
 
 #include "adapter_info.h"
 #include "core/manager.h"
+#include "msg_q.h"
 
 struct neu_adapter {
     char *name;
@@ -45,6 +46,9 @@ struct neu_adapter {
     int control_fd;
     int trans_data_fd;
 
+    adapter_msg_q_t *msg_q;
+    pthread_t        consumer_tid;
+
     uint16_t trans_data_port;
 
     neu_events_t *events;

src/adapter/driver/driver.c

@@ -562,20 +562,62 @@ static void cal_decimal(neu_type_e tag_type, neu_type_e value_type,
 {
     switch (tag_type) {
     case NEU_TYPE_INT8:
-    case NEU_TYPE_UINT8:
+        if (value_type == NEU_TYPE_DOUBLE) {
+            value->i8 = (int8_t) round(value->d64 / decimal);
+        } else {
+            value->i8 = (int8_t) round(value->i64 / decimal);
+        }
+        break;
     case NEU_TYPE_INT16:
-    case NEU_TYPE_UINT16:
+        if (value_type == NEU_TYPE_DOUBLE) {
+            value->i16 = (int16_t) round(value->d64 / decimal);
+        } else {
+            value->i16 = (int16_t) round(value->i64 / decimal);
+        }
+        break;
     case NEU_TYPE_INT32:
-    case NEU_TYPE_UINT32:
+        if (value_type == NEU_TYPE_DOUBLE) {
+            value->i32 = (int32_t) round(value->d64 / decimal);
+        } else {
+            value->i32 = (int32_t) round(value->i64 / decimal);
+        }
+        break;
     case NEU_TYPE_INT64:
-    case NEU_TYPE_UINT64:
+        if (value_type == NEU_TYPE_DOUBLE) {
+            value->i64 = (int64_t) round(value->d64 / decimal);
+        } else {
+            value->i64 = (int64_t) round(value->i64 / decimal);
+        }
+        break;
+    case NEU_TYPE_UINT8:
+        if (value_type == NEU_TYPE_DOUBLE) {
+            value->u8 = (uint8_t) round(value->d64 / decimal);
+        } else {
+            value->u8 = (uint8_t) round(value->u64 / decimal);
+        }
+        break;
+    case NEU_TYPE_UINT16:
     case NEU_TYPE_WORD:
+        if (value_type == NEU_TYPE_DOUBLE) {
+            value->u16 = (uint16_t) round(value->d64 / decimal);
+        } else {
+            value->u16 = (uint16_t) round(value->u64 / decimal);
+        }
+        break;
+    case NEU_TYPE_UINT32:
     case NEU_TYPE_LWORD:
+        if (value_type == NEU_TYPE_DOUBLE) {
+            value->u32 = (uint32_t) round(value->d64 / decimal);
+        } else {
+            value->u32 = (uint32_t) round(value->u64 / decimal);
+        }
+        break;
+    case NEU_TYPE_UINT64:
     case NEU_TYPE_DWORD:
-        if (value_type == NEU_TYPE_INT64 || value_type == NEU_TYPE_LWORD) {
-            value->u64 = value->u64 / decimal;
-        } else if (value_type == NEU_TYPE_DOUBLE) {
+        if (value_type == NEU_TYPE_DOUBLE) {
             value->u64 = (uint64_t) round(value->d64 / decimal);
+        } else {
+            value->u64 = (uint64_t) round(value->u64 / decimal);
         }
         break;
     case NEU_TYPE_FLOAT:
@@ -622,16 +664,13 @@ void neu_adapter_driver_write_tags(neu_adapter_driver_t *driver,
     UT_icd    icd  = { sizeof(neu_plugin_tag_value_t), NULL, NULL, NULL };
     utarray_new(tags, &icd);
     for (int i = 0; i < cmd->n_tag; i++) {
-        neu_plugin_tag_value_t tv = { 0 };
-
         neu_datatag_t *tag = neu_group_find_tag(g->group, cmd->tags[i].tag);
         if (tag != NULL && neu_tag_attribute_test(tag, NEU_ATTRIBUTE_WRITE) &&
             neu_tag_attribute_test(tag, NEU_ATTRIBUTE_STATIC) == false) {
-            tv.tag = neu_tag_dup(tag);
-
             if (tag->type == NEU_TYPE_FLOAT || tag->type == NEU_TYPE_DOUBLE) {
                 if (cmd->tags[i].value.type == NEU_TYPE_INT64) {
-                    tv.value.d64 = (double) tv.value.i64;
+                    cmd->tags[i].value.value.d64 =
+                        (double) cmd->tags[i].value.value.i64;
                 }
             }
             if (tag->decimal != 0) {
@@ -640,7 +679,10 @@ void neu_adapter_driver_write_tags(neu_adapter_driver_t *driver,
             }
             fix_value(tag, cmd->tags[i].value.type, &cmd->tags[i].value);
 
-            tv.value = cmd->tags[i].value.value;
+            neu_plugin_tag_value_t tv = {
+                .tag   = neu_tag_dup(tag),
+                .value = cmd->tags[i].value.value,
+            };
             utarray_push_back(tags, &tv);
         }
         if (tag != NULL) {
@@ -707,20 +749,16 @@ void neu_adapter_driver_write_gtags(neu_adapter_driver_t *driver,
         group_t *g = find_group(driver, cmd->groups[i].group);
 
         for (int k = 0; k < cmd->groups[i].n_tag; k++) {
-            neu_plugin_tag_value_t tv = { 0 };
-
             neu_datatag_t *tag =
                 neu_group_find_tag(g->group, cmd->groups[i].tags[k].tag);
-
             if (tag != NULL &&
                 neu_tag_attribute_test(tag, NEU_ATTRIBUTE_WRITE) &&
                 neu_tag_attribute_test(tag, NEU_ATTRIBUTE_STATIC) == false) {
-                tv.tag = neu_tag_dup(tag);
-
                 if (tag->type == NEU_TYPE_FLOAT ||
                     tag->type == NEU_TYPE_DOUBLE) {
                     if (cmd->groups[i].tags[k].value.type == NEU_TYPE_INT64) {
-                        tv.value.d64 = (double) tv.value.u64;
+                        cmd->groups[i].tags[k].value.value.d64 =
+                            (double) cmd->groups[i].tags[k].value.value.i64;
                     }
                 }
                 if (tag->decimal != 0) {
@@ -731,7 +769,10 @@ void neu_adapter_driver_write_gtags(neu_adapter_driver_t *driver,
                 fix_value(tag, cmd->groups[i].tags[k].value.type,
                           &cmd->groups[i].tags[k].value);
 
-                tv.value = cmd->groups[i].tags[k].value.value;
+                neu_plugin_tag_value_t tv = {
+                    tv.tag   = neu_tag_dup(tag),
+                    tv.value = cmd->groups[i].tags[k].value.value,
+                };
                 utarray_push_back(tags, &tv);
             }
             if (tag != NULL) {
@@ -1363,11 +1404,9 @@ UT_array *neu_adapter_driver_get_ptag(neu_adapter_driver_t *driver,
 static void report_to_app(neu_adapter_driver_t *driver, group_t *group,
                           struct sockaddr_in dst)
 {
-    static __thread uint8_t buf[1024] = { 0 };
-    neu_reqresp_head_t *    header    = (neu_reqresp_head_t *) buf;
-
-    memset(buf, 0, sizeof(buf));
-    header->type = NEU_REQRESP_TRANS_DATA;
+    neu_reqresp_head_t header = {
+        .type = NEU_REQRESP_TRANS_DATA,
+    };
 
     UT_array *tags =
         neu_adapter_driver_get_read_tag(group->driver, group->name);
@@ -1395,8 +1434,8 @@ static void report_to_app(neu_adapter_driver_t *driver, group_t *group,
 
         pthread_mutex_init(&data->ctx->mtx, NULL);
 
-        if (driver->adapter.cb_funs.responseto(&driver->adapter, header, data,
-                                               dst) <= 0) {
+        if (driver->adapter.cb_funs.responseto(&driver->adapter, &header, data,
+                                               dst) != 0) {
             neu_trans_data_free(data);
         }
 

src/adapter/msg_q.c

@@ -0,0 +1,116 @@
+/**
+ * NEURON IIoT System for Industry 4.0
+ * Copyright (C) 2020-2024 EMQ Technologies Co., Ltd All rights reserved.
+ *
+ * This program is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License as published by the Free Software Foundation; either
+ * version 3 of the License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public License
+ * along with this program; if not, write to the Free Software Foundation,
+ * Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
+ **/
+#include <pthread.h>
+
+#include "utils/log.h"
+#include "utils/utextend.h"
+
+#include "msg_q.h"
+
+struct item {
+    neu_msg_t *  msg;
+    struct item *prev, *next;
+};
+
+struct adapter_msg_q {
+    struct item *list;
+    uint32_t     max;
+    uint32_t     current;
+    char *       name;
+
+    pthread_mutex_t mtx;
+    pthread_cond_t  cond;
+};
+
+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));
+
+    pthread_mutex_init(&q->mtx, NULL);
+    pthread_cond_init(&q->cond, NULL);
+    q->max     = size;
+    q->name    = strdup(name);
+    q->current = 0;
+
+    return q;
+}
+
+void adapter_msg_q_free(adapter_msg_q_t *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);
+        neu_reqresp_head_t *header = neu_msg_get_header(elt->msg);
+        neu_trans_data_free((neu_reqresp_trans_data_t *) &header[1]);
+        neu_msg_free(elt->msg);
+        free(elt);
+    }
+    free(q->name);
+    free(q);
+}
+
+int adapter_msg_q_push(adapter_msg_q_t *q, neu_msg_t *msg)
+{
+    int ret = -1;
+    pthread_mutex_lock(&q->mtx);
+    if (q->current < q->max) {
+        q->current += 1;
+        struct item *elt = calloc(1, sizeof(struct item));
+        elt->msg         = msg;
+        DL_APPEND(q->list, elt);
+        ret = 0;
+    }
+    pthread_mutex_unlock(&q->mtx);
+
+    if (ret == -1) {
+        nlog_warn("app: %s, msg q is full, %u(%u)", q->name, q->current,
+                  q->max);
+    } else {
+        pthread_cond_signal(&q->cond);
+    }
+
+    return ret;
+}
+
+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) {
+        pthread_cond_wait(&q->cond, &q->mtx);
+    }
+    struct item *elt = DL_LAST(q->list);
+
+    if (elt != NULL) {
+        DL_DELETE(q->list, elt);
+        *p_data = elt->msg;
+        free(elt);
+        q->current -= 1;
+        ret = q->current;
+    }
+
+    pthread_mutex_unlock(&q->mtx);
+    return ret;
+}

src/adapter/msg_q.h

@@ -0,0 +1,36 @@
+/**
+ * NEURON IIoT System for Industry 4.0
+ * Copyright (C) 2020-2024 EMQ Technologies Co., Ltd All rights reserved.
+ *
+ * This program is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License as published by the Free Software Foundation; either
+ * version 3 of the License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public License
+ * along with this program; if not, write to the Free Software Foundation,
+ * Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
+ **/
+
+#ifndef ADAPTER_MSG_Q_H
+#define ADAPTER_MSG_Q_H
+
+#include <stdint.h>
+
+#include "base/msg_internal.h"
+#include "msg.h"
+
+typedef struct adapter_msg_q adapter_msg_q_t;
+
+adapter_msg_q_t *adapter_msg_q_new(const char *name, uint32_t size);
+void             adapter_msg_q_free(adapter_msg_q_t *q);
+
+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);
+
+#endif

src/base/group.c

@@ -16,6 +16,7 @@
  * along with this program; if not, write to the Free Software Foundation,
  * Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
  **/
+#include <pthread.h>
 #include <string.h>
 #include <sys/time.h>
 
@@ -38,7 +39,8 @@ struct neu_group {
     tag_elem_t *tags;
     uint32_t    interval;
 
-    int64_t timestamp;
+    int64_t         timestamp;
+    pthread_mutex_t mtx;
 };
 
 static UT_array *to_array(tag_elem_t *tags);
@@ -53,6 +55,7 @@ neu_group_t *neu_group_new(const char *name, uint32_t interval)
 
     group->name     = strdup(name);
     group->interval = interval;
+    pthread_mutex_init(&group->mtx, NULL);
 
     return group;
 }
@@ -61,6 +64,7 @@ void neu_group_destroy(neu_group_t *group)
 {
     tag_elem_t *el = NULL, *tmp = NULL;
 
+    pthread_mutex_lock(&group->mtx);
     HASH_ITER(hh, group->tags, el, tmp)
     {
         HASH_DEL(group->tags, el);
@@ -68,7 +72,9 @@ void neu_group_destroy(neu_group_t *group)
         neu_tag_free(el->tag);
         free(el);
     }
+    pthread_mutex_unlock(&group->mtx);
 
+    pthread_mutex_destroy(&group->mtx);
     free(group->name);
     free(group);
 }
@@ -118,8 +124,10 @@ int neu_group_add_tag(neu_group_t *group, const neu_datatag_t *tag)
 {
     tag_elem_t *el = NULL;
 
+    pthread_mutex_lock(&group->mtx);
     HASH_FIND_STR(group->tags, tag->name, el);
     if (el != NULL) {
+        pthread_mutex_unlock(&group->mtx);
         return NEU_ERR_TAG_NAME_CONFLICT;
     }
 
@@ -129,6 +137,7 @@ int neu_group_add_tag(neu_group_t *group, const neu_datatag_t *tag)
 
     HASH_ADD_STR(group->tags, name, el);
     update_timestamp(group);
+    pthread_mutex_unlock(&group->mtx);
 
     return 0;
 }
@@ -138,6 +147,7 @@ int neu_group_update_tag(neu_group_t *group, const neu_datatag_t *tag)
     tag_elem_t *el  = NULL;
     int         ret = NEU_ERR_TAG_NOT_EXIST;
 
+    pthread_mutex_lock(&group->mtx);
     HASH_FIND_STR(group->tags, tag->name, el);
     if (el != NULL) {
         neu_tag_copy(el->tag, tag);
@@ -145,6 +155,7 @@ int neu_group_update_tag(neu_group_t *group, const neu_datatag_t *tag)
         update_timestamp(group);
         ret = NEU_ERR_SUCCESS;
     }
+    pthread_mutex_unlock(&group->mtx);
 
     return ret;
 }
@@ -154,6 +165,7 @@ int neu_group_del_tag(neu_group_t *group, const char *tag_name)
     tag_elem_t *el  = NULL;
     int         ret = NEU_ERR_TAG_NOT_EXIST;
 
+    pthread_mutex_lock(&group->mtx);
     HASH_FIND_STR(group->tags, tag_name, el);
     if (el != NULL) {
         HASH_DEL(group->tags, el);
@@ -164,15 +176,18 @@ int neu_group_del_tag(neu_group_t *group, const char *tag_name)
         update_timestamp(group);
         ret = NEU_ERR_SUCCESS;
     }
+    pthread_mutex_unlock(&group->mtx);
 
     return ret;
 }
 
-UT_array *neu_group_get_tag(const neu_group_t *group)
+UT_array *neu_group_get_tag(neu_group_t *group)
 {
     UT_array *array = NULL;
 
+    pthread_mutex_lock(&group->mtx);
     array = to_array(group->tags);
+    pthread_mutex_unlock(&group->mtx);
 
     return array;
 }
@@ -182,6 +197,7 @@ UT_array *neu_group_query_tag(neu_group_t *group, const char *name)
     tag_elem_t *el = NULL, *tmp = NULL;
     UT_array *  array = NULL;
 
+    pthread_mutex_lock(&group->mtx);
     utarray_new(array, neu_tag_get_icd());
     HASH_ITER(hh, group->tags, el, tmp)
     {
@@ -189,6 +205,7 @@ UT_array *neu_group_query_tag(neu_group_t *group, const char *name)
             utarray_push_back(array, el->tag);
         }
     }
+    pthread_mutex_unlock(&group->mtx);
 
     return array;
 }
@@ -197,7 +214,9 @@ UT_array *neu_group_get_read_tag(neu_group_t *group)
 {
     UT_array *array = NULL;
 
+    pthread_mutex_lock(&group->mtx);
     array = to_read_array(group->tags);
+    pthread_mutex_unlock(&group->mtx);
 
     return array;
 }
@@ -216,10 +235,12 @@ neu_datatag_t *neu_group_find_tag(neu_group_t *group, const char *tag)
     tag_elem_t *   find   = NULL;
     neu_datatag_t *result = NULL;
 
+    pthread_mutex_lock(&group->mtx);
     HASH_FIND_STR(group->tags, tag, find);
     if (find != NULL) {
         result = neu_tag_dup(find->tag);
     }
+    pthread_mutex_unlock(&group->mtx);
 
     return result;
 }

src/base/group.h

@@ -38,7 +38,7 @@ int          neu_group_update(neu_group_t *group, uint32_t interval);
 int          neu_group_add_tag(neu_group_t *group, const neu_datatag_t *tag);
 int          neu_group_update_tag(neu_group_t *group, const neu_datatag_t *tag);
 int          neu_group_del_tag(neu_group_t *group, const char *tag_name);
-UT_array *   neu_group_get_tag(const neu_group_t *group);
+UT_array *   neu_group_get_tag(neu_group_t *group);
 UT_array *   neu_group_query_tag(neu_group_t *group, const char *name);
 UT_array *   neu_group_get_read_tag(neu_group_t *group);
 uint16_t     neu_group_tag_size(const neu_group_t *group);

src/connection/connection.c

@@ -347,8 +347,8 @@ ssize_t neu_conn_send(neu_conn_t *conn, uint8_t *buf, ssize_t len)
                     break;
                 }
             } else {
-                if (rc == -1 && errno == EAGAIN) {
-                    if (retry > 50) {
+                if (!conn->block && rc == -1 && errno == EAGAIN) {
+                    if (retry > 10) {
                         zlog_error(conn->param.log,
                                    "conn fd: %d, send buf len: %zd, ret: %zd, "
                                    "errno: %s(%d)",
@@ -830,7 +830,14 @@ static void conn_connect(neu_conn_t *conn)
             return;
         }
 
-        if (ret != 0 && errno != EINPROGRESS) {
+        if ((conn->block && ret == 0) ||
+            (!conn->block && ret != 0 && errno == EINPROGRESS)) {
+            zlog_notice(conn->param.log, "connect %s:%d success",
+                        conn->param.params.tcp_client.ip,
+                        conn->param.params.tcp_client.port);
+            conn->is_connected = true;
+            conn->fd           = fd;
+        } else {
             close(fd);
             zlog_error(conn->param.log, "connect %s:%d error: %s(%d)",
                        conn->param.params.tcp_client.ip,
@@ -838,13 +845,8 @@ static void conn_connect(neu_conn_t *conn)
                        errno);
             conn->is_connected = false;
             return;
-        } else {
-            zlog_notice(conn->param.log, "connect %s:%d success",
-                        conn->param.params.tcp_client.ip,
-                        conn->param.params.tcp_client.port);
-            conn->is_connected = true;
-            conn->fd           = fd;
         }
+
         break;
     }
     case NEU_CONN_UDP: {

src/core/manager.c

@@ -600,8 +600,25 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data)
 
         manager_storage_del_node(manager, cmd->node);
         if (neu_adapter_get_type(adapter) == NEU_NA_TYPE_APP) {
+            UT_array *subscriptions = neu_subscribe_manager_get(
+                manager->subscribe_manager, cmd->node);
             neu_subscribe_manager_unsub_all(manager->subscribe_manager,
                                             cmd->node);
+
+            utarray_foreach(subscriptions, neu_resp_subscribe_info_t *, sub)
+            {
+                // NOTE: neu_req_unsubscribe_t and neu_resp_subscribe_info_t
+                //       have compatible memory layout
+                msg = neu_msg_new(NEU_REQ_UNSUBSCRIBE_GROUP, NULL, sub);
+                if (NULL == msg) {
+                    break;
+                }
+                neu_reqresp_head_t *hd = neu_msg_get_header(msg);
+                strcpy(hd->receiver, sub->driver);
+                strcpy(hd->sender, "manager");
+                forward_msg(manager, hd, hd->receiver);
+            }
+            utarray_free(subscriptions);
         }
         notify_monitors(manager, NEU_REQ_DEL_NODE_EVENT, cmd);
         neu_reqresp_node_deleted_t resp = { 0 };

src/event/event_linux.c

@@ -66,7 +66,7 @@ struct event_data {
     bool  use;
 };
 
-#define EVENT_SIZE 1024
+#define EVENT_SIZE 1400
 
 struct neu_events {
     int       epoll_fd;
@@ -182,8 +182,12 @@ neu_events_t *neu_event_new(void)
     neu_events_t *events = calloc(1, sizeof(struct neu_events));
 
     events->epoll_fd = epoll_create(1);
-    events->stop     = false;
-    events->n_event  = 0;
+
+    nlog_notice("create epoll: %d(%d)", events->epoll_fd, errno);
+    assert(events->epoll_fd > 0);
+
+    events->stop    = false;
+    events->n_event = 0;
     pthread_mutex_init(&events->mtx, NULL);
 
     pthread_create(&events->thread, NULL, event_loop, events);
@@ -235,6 +239,7 @@ neu_event_timer_t *neu_event_add_timer(neu_events_t *          events,
     timer_ctx->event_data->usr_data       = timer.usr_data;
     timer_ctx->event_data->callback.timer = timer.cb;
     timer_ctx->event_data->ctx.timer = events->event_datas[index].ctx.timer;
+    timer_ctx->event_data->index     = index;
 
     timer_ctx->value = value;
     timer_ctx->fd    = timer_fd;
@@ -247,17 +252,17 @@ neu_event_timer_t *neu_event_add_timer(neu_events_t *          events,
     zlog_notice(neuron,
                 "add timer, second: %" PRId64 ", millisecond: %" PRId64
                 ", timer: %d in epoll %d, "
-                "ret: %d",
+                "ret: %d, index: %d",
                 timer.second, timer.millisecond, timer_fd, events->epoll_fd,
-                ret);
+                ret, index);
 
     return timer_ctx;
 }
 
 int neu_event_del_timer(neu_events_t *events, neu_event_timer_t *timer)
 {
-    zlog_notice(neuron, "del timer: %d from epoll: %d", timer->fd,
-                events->epoll_fd);
+    zlog_notice(neuron, "del timer: %d from epoll: %d, index: %d", timer->fd,
+                events->epoll_fd, timer->event_data->index);
 
     timer->stop = true;
     epoll_ctl(events->epoll_fd, EPOLL_CTL_DEL, timer->fd, NULL);
@@ -273,8 +278,13 @@ int neu_event_del_timer(neu_events_t *events, neu_event_timer_t *timer)
 
 neu_event_io_t *neu_event_add_io(neu_events_t *events, neu_event_io_param_t io)
 {
-    int             ret      = 0;
-    int             index    = get_free_event(events);
+    int ret   = 0;
+    int index = get_free_event(events);
+
+    nlog_notice("add io, fd: %d, epoll: %d, index: %d", io.fd, events->epoll_fd,
+                index);
+    assert(index >= 0);
+
     neu_event_io_t *io_ctx   = &events->event_datas[index].ctx.io;
     io_ctx->event_data       = &events->event_datas[index];
     struct epoll_event event = {
@@ -287,13 +297,15 @@ neu_event_io_t *neu_event_add_io(neu_events_t *events, neu_event_io_param_t io)
     io_ctx->event_data->usr_data    = io.usr_data;
     io_ctx->event_data->callback.io = io.cb;
     io_ctx->event_data->ctx.io      = events->event_datas[index].ctx.io;
+    io_ctx->event_data->index       = index;
 
     io_ctx->fd = io.fd;
 
     ret = epoll_ctl(events->epoll_fd, EPOLL_CTL_ADD, io.fd, &event);
 
-    nlog_notice("add io, fd: %d, epoll: %d, ret: %d", io.fd, events->epoll_fd,
-                ret);
+    nlog_notice("add io, fd: %d, epoll: %d, ret: %d(%d), index: %d", io.fd,
+                events->epoll_fd, ret, errno, index);
+    assert(ret == 0);
 
     return io_ctx;
 }
@@ -304,7 +316,8 @@ int neu_event_del_io(neu_events_t *events, neu_event_io_t *io)
         return 0;
     }
 
-    zlog_notice(neuron, "del io: %d from epoll: %d", io->fd, events->epoll_fd);
+    zlog_notice(neuron, "del io: %d from epoll: %d, index: %d", io->fd,
+                events->epoll_fd, io->event_data->index);
 
     epoll_ctl(events->epoll_fd, EPOLL_CTL_DEL, io->fd, NULL);
     free_event(events, io->event_data->index);

src/main.c

@@ -125,9 +125,10 @@ static int neuron_run(const neu_cli_args_t *args)
     int           rv = 0;
 
     signal(SIGINT, sig_handler);
-    signal(SIGTERM, sig_handler);
     signal(SIGABRT, sig_handler);
     signal(SIGSEGV, sig_handler);
+    signal(SIGTERM, sig_handler);
+    signal(SIGKILL, sig_handler);
 
     // try to enable core dump
     rl.rlim_cur = rl.rlim_max = RLIM_INFINITY;

version

@@ -1 +1 @@
-2.6.3
+2.6.4