Merge pull request #1927 from eeff/v2.6

Pick to v2.6 fix(ekuiper): memory leak and data race

Author: fengzeroz

plugins/ekuiper/json_rw.c

@@ -185,7 +185,8 @@ int json_decode_write_req(char *buf, size_t len, json_write_req_t **result)
 
     json_obj = neu_json_decode_newb(buf, len);
     if (NULL == json_obj) {
-        goto decode_fail;
+        free(req);
+        return -1;
     }
 
     neu_json_elem_t req_elems[] = {
@@ -208,27 +209,20 @@ int json_decode_write_req(char *buf, size_t len, json_write_req_t **result)
     };
     ret = neu_json_decode_by_json(json_obj, NEU_JSON_ELEM_SIZE(req_elems),
                                   req_elems);
-    if (ret != 0) {
-        goto decode_fail;
-    }
-
     req->node_name  = req_elems[0].v.val_str;
     req->group_name = req_elems[1].v.val_str;
     req->tag_name   = req_elems[2].v.val_str;
     req->t          = req_elems[3].t;
     req->value      = req_elems[3].v;
 
-    *result = req;
-    goto decode_exit;
-
-decode_fail:
-    free(req);
-    ret = -1;
-
-decode_exit:
-    if (json_obj != NULL) {
-        neu_json_decode_free(json_obj);
+    if (0 == ret) {
+        *result = req;
+    } else {
+        json_decode_write_req_free(req);
+        ret = -1;
     }
+
+    neu_json_decode_free(json_obj);
     return ret;
 }
 
@@ -243,6 +237,8 @@ void json_decode_write_req_free(json_write_req_t *req)
     free(req->tag_name);
     if (req->t == NEU_JSON_STR) {
         free(req->value.val_str);
+    } else if (req->t == NEU_JSON_BYTES) {
+        free(req->value.val_bytes.bytes);
     }
 
     free(req);

plugins/ekuiper/plugin_ekuiper.c

@@ -114,10 +114,9 @@ static int ekuiper_plugin_uninit(neu_plugin_t *plugin)
     return rv;
 }
 
-static int ekuiper_plugin_start(neu_plugin_t *plugin)
+static inline int start(neu_plugin_t *plugin, const char *url)
 {
-    int   rv  = 0;
-    char *url = plugin->url ? plugin->url : EKUIPER_PLUGIN_URL; // default url
+    int rv = 0;
 
     rv = nng_pair0_open(&plugin->sock);
     if (rv != 0) {
@@ -143,16 +142,43 @@ static int ekuiper_plugin_start(neu_plugin_t *plugin)
         return rv;
     }
 
+    nng_mtx_lock(plugin->mtx);
+    while (plugin->receiving) {
+        nng_mtx_unlock(plugin->mtx);
+        nng_msleep(10);
+        nng_mtx_lock(plugin->mtx);
+    }
+    plugin->receiving = true;
     nng_recv_aio(plugin->sock, plugin->recv_aio);
+    nng_mtx_unlock(plugin->mtx);
+
+    return NEU_ERR_SUCCESS;
+}
+
+static int ekuiper_plugin_start(neu_plugin_t *plugin)
+{
+    int   rv  = 0;
+    char *url = plugin->url ? plugin->url : EKUIPER_PLUGIN_URL; // default url
+
+    rv = start(plugin, url);
+    if (rv != 0) {
+        return rv;
+    }
+
     plugin->started = true;
     plog_notice(plugin, "start successfully");
 
     return NEU_ERR_SUCCESS;
 }
 
-static int ekuiper_plugin_stop(neu_plugin_t *plugin)
+static inline void stop(neu_plugin_t *plugin)
 {
     nng_close(plugin->sock);
+}
+
+static int ekuiper_plugin_stop(neu_plugin_t *plugin)
+{
+    stop(plugin);
     plugin->started = false;
     plog_notice(plugin, "stop successfully");
     return NEU_ERR_SUCCESS;
@@ -201,41 +227,6 @@ static int parse_config(neu_plugin_t *plugin, const char *setting,
     return -1;
 }
 
-static inline int check_url_listenable(neu_plugin_t *plugin, const char *url,
-                                       const char *host, uint16_t port)
-{
-    if (NULL != plugin->host &&                  // already configured and
-        port == plugin->port &&                  // port is the same, then if
-        (0 == strcmp(plugin->host, host)         // 1. host is the same
-         || 0 == strcmp("0.0.0.0", host)         // 2. to bind to any address
-         || 0 == strcmp("0.0.0.0", plugin->host) // 3. bound to any address
-         )) {
-        // early return, no need check url is listenable
-        return 0;
-    }
-
-    nng_socket sock = NNG_SOCKET_INITIALIZER;
-    int        rv   = nng_pair0_open(&sock);
-    if (0 != rv) {
-        plog_error(plugin, "nng_pair0_open: %s", nng_strerror(rv));
-        return NEU_ERR_EINTERNAL;
-    }
-
-    if (0 != (rv = nng_listen(sock, url, NULL, 0))) {
-        plog_error(plugin, "nng_listen: %s", nng_strerror(rv));
-        if (NNG_EADDRINVAL == rv) {
-            rv = NEU_ERR_IP_ADDRESS_INVALID;
-        } else if (NNG_EADDRINUSE == rv) {
-            rv = NEU_ERR_IP_ADDRESS_IN_USE;
-        } else {
-            rv = NEU_ERR_EINTERNAL;
-        }
-    }
-
-    nng_close(sock);
-    return rv;
-}
-
 static int ekuiper_plugin_config(neu_plugin_t *plugin, const char *setting)
 {
     int      rv   = 0;
@@ -255,23 +246,31 @@ static int ekuiper_plugin_config(neu_plugin_t *plugin, const char *setting)
         goto error;
     }
 
-    if (0 != (rv = check_url_listenable(plugin, url, host, port))) {
+    if (plugin->started) {
+        stop(plugin);
+    }
+
+    // check we could start the plugin with the new setting
+    if (0 != (rv = start(plugin, url))) {
+        // recover with old setting
+        if (plugin->started && 0 != start(plugin, plugin->url)) {
+            plog_warn(plugin, "restart host:%s port:%" PRIu16 " fail",
+                      plugin->host, plugin->port);
+        }
         goto error;
     }
 
+    if (!plugin->started) {
+        stop(plugin);
+    }
+
     plog_notice(plugin, "config success");
 
     free(plugin->host);
+    free(plugin->url);
     plugin->host = host;
     plugin->port = port;
-    free(plugin->url);
-    plugin->url = url;
-
-    if (plugin->started) {
-        // restart service
-        ekuiper_plugin_stop(plugin);
-        ekuiper_plugin_start(plugin);
-    }
+    plugin->url  = url;
 
     return rv;
 
@@ -310,6 +309,9 @@ static int ekuiper_plugin_request(neu_plugin_t *      plugin,
         }
         break;
     }
+    case NEU_REQRESP_NODE_DELETED: {
+        break;
+    }
     case NEU_REQ_UPDATE_NODE: {
         break;
     }

plugins/ekuiper/plugin_ekuiper.h

@@ -37,6 +37,7 @@ struct neu_plugin {
     nng_socket          sock;
     nng_mtx *           mtx;
     bool                started;
+    bool                receiving;
     nng_aio *           recv_aio;
     char *              host;
     uint16_t            port;

plugins/ekuiper/read_write.c

@@ -86,6 +86,9 @@ void recv_data_callback(void *arg)
     rv = nng_aio_result(plugin->recv_aio);
     if (0 != rv) {
         plog_error(plugin, "nng_recv error: %s", nng_strerror(rv));
+        nng_mtx_lock(plugin->mtx);
+        plugin->receiving = false;
+        nng_mtx_unlock(plugin->mtx);
         return;
     }