fix(restful): avoid proxy handler data race

plugins/restful/proxy.c

@@ -16,10 +16,11 @@
  * 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 "proxy.h"
 #include "errcodes.h"
 #include "http.h"
+#include "proxy.h"
 #include "utils/log.h"
 #include "utils/utlist.h"
 #include "json/neu_json_fn.h"
@@ -27,6 +28,7 @@
 struct rest_proxy_ctx;
 
 typedef struct {
+    pthread_mutex_t        mtx;
     size_t                 src_uri_len; // source uri length
     size_t                 dst_uri_len; // destination uri length
     size_t                 uri_buf_len; // request uri buffer length
@@ -60,41 +62,53 @@ static int rest_proxy_alloc(rest_proxy_t **proxy, const char *src_url,
 
     rest_proxy_t *p = calloc(1, sizeof(rest_proxy_t));
     if (NULL == p) {
+        nlog_error("alloc rest proxy fail");
         return -1;
     }
 
+    if (0 != pthread_mutex_init(&p->mtx, NULL)) {
+        nlog_error("init rest proxy mutex fail");
+        ret = -1;
+        goto err_mutex_init;
+    }
+
     ret = nng_url_parse(&url, dst_url);
     if (0 != ret) {
-        free(p);
         nlog_error("parse url: `%s` fail", dst_url);
-        return ret;
+        goto err_url_parse;
     }
 
     p->dst_uri_len = strlen(url->u_requri);
     p->uri_buf_len = p->dst_uri_len + 1;
     p->uri_buf     = calloc(p->uri_buf_len, sizeof(char));
     if (NULL == p->uri_buf) {
-        nng_url_free(url);
-        free(p);
         nlog_error("alloc uri buffer for `%s` fail", dst_url);
-        return -1;
+        ret = -1;
+        goto err_uri_buf;
     }
     // copy destination uri to head of buffer
     strcpy(p->uri_buf, url->u_requri);
 
     ret = nng_http_client_alloc(&p->client, url);
     if (0 != ret) {
-        free(p->uri_buf);
-        nng_url_free(url);
-        free(p);
         nlog_error("alloc http client for `%s` fail", dst_url);
-        return ret;
+        goto err_http_client;
     }
 
     p->src_uri_len = strlen(src_url);
     *proxy         = p;
     nng_url_free(url);
     return ret;
+
+err_http_client:
+    free(p->uri_buf);
+err_uri_buf:
+    nng_url_free(url);
+err_url_parse:
+    pthread_mutex_destroy(&p->mtx);
+err_mutex_init:
+    free(p);
+    return ret;
 }
 
 static void rest_proxy_free(rest_proxy_t *proxy)
@@ -118,6 +132,7 @@ static void rest_proxy_free(rest_proxy_t *proxy)
         free(el);
     }
 
+    pthread_mutex_destroy(&proxy->mtx);
     free(proxy);
 }
 
@@ -127,11 +142,14 @@ static inline int rest_proxy_set_uri(rest_proxy_t *proxy, nng_http_req *req)
     size_t      len     = strlen(uri);
     size_t      buf_len = len - proxy->src_uri_len + proxy->dst_uri_len + 1;
 
+    // TODO: use per context buffer if a single lock cause too much contention
+    pthread_mutex_lock(&proxy->mtx);
     if (buf_len > proxy->uri_buf_len) {
         proxy->uri_buf_len = buf_len;
         proxy->uri_buf     = realloc(proxy->uri_buf, buf_len);
         if (NULL == proxy->uri_buf) {
             nlog_error("realloc uri buffer fail");
+            pthread_mutex_unlock(&proxy->mtx);
             return -1;
         }
     }
@@ -142,8 +160,10 @@ static inline int rest_proxy_set_uri(rest_proxy_t *proxy, nng_http_req *req)
 
     if (0 != nng_http_req_set_uri(req, proxy->uri_buf)) {
         nlog_error("proxy fail set req uri: `%s`", proxy->uri_buf);
+        pthread_mutex_unlock(&proxy->mtx);
         return -1;
     }
+    pthread_mutex_unlock(&proxy->mtx);
 
     return 0;
 }
@@ -176,16 +196,22 @@ static void rest_proxy_cb(void *arg)
 
     ctx->user_aio = NULL;
     ctx->res      = NULL;
+
+    pthread_mutex_lock(&ctx->proxy->mtx);
     DL_APPEND(ctx->proxy->free_list, ctx); // put to free list
+    pthread_mutex_unlock(&ctx->proxy->mtx);
 }
 
 struct rest_proxy_ctx *rest_proxy_get_ctx(rest_proxy_t *proxy)
 {
+    pthread_mutex_lock(&proxy->mtx);
     struct rest_proxy_ctx *ctx = proxy->free_list;
     if (NULL != ctx) {
         DL_DELETE(proxy->free_list, ctx);
+        pthread_mutex_unlock(&proxy->mtx);
         return ctx;
     }
+    pthread_mutex_unlock(&proxy->mtx);
 
     ctx = calloc(1, sizeof(*ctx));
     if (NULL == ctx) {