Merge pull request #1263 from fengzeroz/v2.4

version: 2.4.6

.github/workflows/ci.yml

@@ -8,8 +8,8 @@ on:
       - published
 
 env:
-  VERSION: 2.4.5
-  UI_VERSION: 2.4.3
+  VERSION: 2.4.6
+  UI_VERSION: 2.4.4
   BRANCH: v2.4
 
 jobs:

.gitignore

@@ -16,4 +16,7 @@ tmp
 *.deb
 *.rpm
 neuron-sdk
-.vscode
+.vscode
+log.html
+output.xml
+report.html

include/neuron/utils/protocol_buf.h

@@ -192,4 +192,16 @@ neu_protocol_unpack_buf_used_size(neu_protocol_unpack_buf_t *buf)
     return buf->offset;
 }
 
+inline static void
+neu_protocol_unpack_buf_use_all(neu_protocol_unpack_buf_t *buf)
+{
+    buf->offset = buf->size;
+}
+
+inline static uint16_t
+neu_protocol_unpack_buf_size(neu_protocol_unpack_buf_t *buf)
+{
+    return buf->size;
+}
+
 #endif

plugins/mqtt/mqtt_config.c

@@ -276,7 +276,7 @@ int mqtt_config_parse(neu_plugin_t *plugin, const char *setting,
     if (NULL == write_resp_topic.v.val_str &&
         0 > neu_asprintf(&write_resp_topic.v.val_str, "/neuron/%s/write/resp",
                          plugin->common.name)) {
-        plog_error(plugin, "setting write request topic error");
+        plog_error(plugin, "setting write response topic error");
         goto error;
     }
 
@@ -338,7 +338,7 @@ int mqtt_config_parse(neu_plugin_t *plugin, const char *setting,
     plog_info(plugin, "config format          : %s",
               mqtt_upload_format_str(config->format));
     plog_info(plugin, "config write-req-topic : %s", config->write_req_topic);
-    plog_info(plugin, "config write-resp-topic: %s", config->write_req_topic);
+    plog_info(plugin, "config write-resp-topic: %s", config->write_resp_topic);
     plog_info(plugin, "config cache           : %zu", config->cache);
     plog_info(plugin, "config cache-mem-size  : %zu", config->cache_mem_size);
     plog_info(plugin, "config cache-disk-size : %zu", config->cache_disk_size);

src/adapter/adapter.c

@@ -124,8 +124,8 @@ neu_adapter_t *neu_adapter_create(neu_adapter_info_t *info)
 
     rv = nng_pair1_open(&adapter->sock);
     assert(rv == 0);
-    nng_socket_set_int(adapter->sock, NNG_OPT_RECVBUF, 128);
-    nng_socket_set_int(adapter->sock, NNG_OPT_SENDBUF, 128);
+    nng_socket_set_int(adapter->sock, NNG_OPT_RECVBUF, 8192);
+    nng_socket_set_int(adapter->sock, NNG_OPT_SENDBUF, 8192);
     nng_socket_set_ms(adapter->sock, NNG_OPT_SENDTIMEO, 1000);
     nng_socket_get_int(adapter->sock, NNG_OPT_RECVFD, &adapter->recv_fd);
 

src/adapter/driver/driver.c

@@ -19,6 +19,7 @@
 #include <assert.h>
 #include <math.h>
 #include <netinet/in.h>
+#include <pthread.h>
 #include <stdlib.h>
 
 #include <nng/nng.h>
@@ -37,12 +38,20 @@
 #include "errcodes.h"
 #include "tag.h"
 
+typedef struct to_be_write_tag {
+    neu_datatag_t *tag;
+    neu_value_u    value;
+    void *         req;
+} to_be_write_tag_t;
+
 typedef struct group {
     char *name;
 
-    int64_t      timestamp;
-    neu_group_t *group;
-    UT_array *   static_tags;
+    int64_t         timestamp;
+    neu_group_t *   group;
+    UT_array *      static_tags;
+    UT_array *      wt_tags;
+    pthread_mutex_t wt_mtx;
 
     neu_event_timer_t *report;
     neu_event_timer_t *read;
@@ -74,6 +83,7 @@ static void update(neu_adapter_t *adapter, const char *group, const char *tag,
                    neu_dvalue_t value);
 static void write_response(neu_adapter_t *adapter, void *r, neu_error error);
 static group_t *find_group(neu_adapter_driver_t *driver, const char *name);
+static void     store_write_tag(group_t *group, to_be_write_tag_t *tag);
 
 static void write_response(neu_adapter_t *adapter, void *r, neu_error error)
 {
@@ -182,7 +192,13 @@ int neu_adapter_driver_uninit(neu_adapter_driver_t *driver)
         free(el->name);
         utarray_free(el->grp.tags);
 
+        utarray_foreach(el->wt_tags, to_be_write_tag_t *, tag)
+        {
+            neu_tag_free(tag->tag);
+        }
+
         utarray_free(el->static_tags);
+        utarray_free(el->wt_tags);
         neu_group_destroy(el->group);
         free(el);
     }
@@ -414,8 +430,12 @@ void neu_adapter_driver_write_tag(neu_adapter_driver_t *driver,
             driver->adapter.cb_funs.response(&driver->adapter, req, &error);
             free(req);
         } else {
-            driver->adapter.module->intf_funs->driver.write_tag(
-                driver->adapter.plugin, (void *) req, tag, cmd->value.value);
+            to_be_write_tag_t wtag = { 0 };
+            wtag.req               = (void *) req;
+            wtag.value             = cmd->value.value;
+            wtag.tag               = neu_tag_dup(tag);
+
+            store_write_tag(g, &wtag);
         }
 
         neu_tag_free(tag);
@@ -429,6 +449,7 @@ void neu_adapter_driver_write_tag(neu_adapter_driver_t *driver,
 int neu_adapter_driver_add_group(neu_adapter_driver_t *driver, const char *name,
                                  uint32_t interval)
 {
+    UT_icd   icd  = { sizeof(to_be_write_tag_t), NULL, NULL, NULL };
     group_t *find = NULL;
     int      ret  = NEU_ERR_GROUP_EXIST;
 
@@ -443,6 +464,10 @@ int neu_adapter_driver_add_group(neu_adapter_driver_t *driver, const char *name,
             .type        = NEU_EVENT_TIMER_NOBLOCK,
         };
 
+        pthread_mutex_init(&find->wt_mtx, NULL);
+
+        utarray_new(find->wt_tags, &icd);
+
         find->driver         = driver;
         find->name           = strdup(name);
         find->group          = neu_group_new(name, interval);
@@ -530,11 +555,18 @@ int neu_adapter_driver_del_group(neu_adapter_driver_t *driver, const char *name)
             neu_driver_cache_del(driver->cache, name, tag->name);
         }
 
+        utarray_foreach(find->wt_tags, to_be_write_tag_t *, tag)
+        {
+            neu_tag_free(tag->tag);
+        }
+
         utarray_free(find->static_tags);
         utarray_free(find->grp.tags);
+        utarray_free(find->wt_tags);
         neu_group_destroy(find->group);
         free(find);
 
+        pthread_mutex_destroy(&find->wt_mtx);
         neu_adapter_del_group_metrics(&driver->adapter, name);
         ret = NEU_ERR_SUCCESS;
     }
@@ -892,6 +924,17 @@ static int read_callback(void *usr_data)
         return 0;
     }
 
+    pthread_mutex_lock(&group->wt_mtx);
+    utarray_foreach(group->wt_tags, to_be_write_tag_t *, wtag)
+    {
+        group->driver->adapter.module->intf_funs->driver.write_tag(
+            group->driver->adapter.plugin, (void *) wtag->req, wtag->tag,
+            wtag->value);
+        neu_tag_free(wtag->tag);
+    }
+    utarray_clear(group->wt_tags);
+    pthread_mutex_unlock(&group->wt_mtx);
+
     if (neu_group_is_change(group->group, group->timestamp)) {
         neu_group_change_test(group->group, group->timestamp, (void *) group,
                               group_change);
@@ -1199,3 +1242,10 @@ static void read_group(int64_t timestamp, int64_t timeout,
         }
     }
 }
+
+static void store_write_tag(group_t *group, to_be_write_tag_t *tag)
+{
+    pthread_mutex_lock(&group->wt_mtx);
+    utarray_push_back(group->wt_tags, tag);
+    pthread_mutex_unlock(&group->wt_mtx);
+}

src/core/manager.c

@@ -97,8 +97,9 @@ neu_manager_t *neu_manager_create()
     rv = nng_listen(manager->socket, url, NULL, 0);
     assert(rv == 0);
 
-    nng_socket_set_int(manager->socket, NNG_OPT_RECVBUF, 2048);
-    nng_socket_set_int(manager->socket, NNG_OPT_SENDBUF, 2048);
+    nng_socket_set_int(manager->socket, NNG_OPT_RECVBUF, 8192);
+    nng_socket_set_int(manager->socket, NNG_OPT_SENDBUF, 8292);
+    nng_socket_set_ms(manager->socket, NNG_OPT_SENDTIMEO, 1000);
     nng_socket_get_int(manager->socket, NNG_OPT_RECVFD, &param.fd);
     manager->loop = neu_event_add_io(manager->events, param);
 
@@ -731,6 +732,7 @@ inline static void forward_msg_dup(neu_manager_t *manager, nng_msg *msg,
     if (nng_sendmsg(manager->socket, out_msg, 0) == 0) {
         nlog_debug("forward msg to pipe %d", pipe.id);
     } else {
+        nlog_warn("forward msg to pipe %d fail", pipe.id);
         nng_msg_free(out_msg);
     }
 }
@@ -746,6 +748,7 @@ inline static void forward_msg(neu_manager_t *manager, nng_msg *msg,
     if (nng_sendmsg(manager->socket, out_msg, 0) == 0) {
         nlog_debug("forward msg to %s", node);
     } else {
+        nlog_warn("forward msg to pipe (%s)%d fail", node, pipe.id);
         nng_msg_free(out_msg);
     }
 }

version

@@ -1 +1 @@
-2.4.5
+2.4.6