[HTTP_DECODER]bugfix for multi-thread

5 files changed, 70 insertions, 52 deletions

diff --git a/.gitlab-ci.yml b/.gitlab-ci.yml
index 8b8a2ef..7a008ac 100644
--- a/.gitlab-ci.yml
+++ b/.gitlab-ci.yml
@@ -6,8 +6,8 @@ variables:
   INSTALL_DEPENDENCY_LIBRARY: sapp sapp-devel framework_env libMESA_prof_load-devel
                               libMESA_htable-devel libMESA_jump_layer
                               libMESA_handle_logger-devel libMESA_field_stat2-devel
-                              libfieldstat3-devel libfieldstat4-devel libbreakpad_mini-devel libcjson-devel
-                              zlib-devel brotli-devel
+                              libfieldstat3-devel libfieldstat4-devel libbreakpad_mini-devel
+                              libcjson-devel zlib-devel brotli-devel
   SYMBOL_TARGET: stellar-c
   TEST_NAME: gtest_stellar-c
   INSTALL_PREFIX: "/opt/tsg/"
diff --git a/src/http_decoder/http_decoder.c b/src/http_decoder/http_decoder.c
index c282a2a..523253b 100644
--- a/src/http_decoder/http_decoder.c
+++ b/src/http_decoder/http_decoder.c
@@ -48,12 +48,6 @@ struct http_decoder_exdata {
     struct http_decoder *decoder;
 };
 
-struct http_event_context {
-    int topic_id;
-    struct session *ref_session;
-    struct http_decoder_exdata *ref_exdata;
-};
-
 struct http_decoder_context {
     int plugin_id;
 	int topic_id;
@@ -63,17 +57,15 @@ struct http_decoder_context {
     int fs_incoming_trans_id;
     struct fieldstat_easy *fse;
     struct stellar *st;
-    struct http_event_context *http_ev_ctx;
 };
 
 static void http_event_handler(enum http_event event,
                                struct http_decoder_half_data **data,
-                               void *http_ev_ctx)
+                               struct http_event_context *ev_ctx)
 {
-    struct http_event_context *ctx = (struct http_event_context *)http_ev_ctx;
-    assert(ctx);
+    assert(ev_ctx);
 
-    struct http_decoder_result_queue *queue = ctx->ref_exdata->queue;
+    struct http_decoder_result_queue *queue = ev_ctx->ref_queue;
     struct http_message *msg = NULL;
     struct http_decoder_half_data *half_data = NULL;
     int ret = 0;
@@ -95,7 +87,7 @@ static void http_event_handler(enum http_event event,
         half_data = http_decoder_half_data_new();
         ret = http_decoder_result_queue_push_req(queue, half_data);
         if (ret < 0) {
-            fprintf(stderr, "http_decoder_result_queue_push failed.");
+            fprintf(stderr, "http_decoder_result_queue_push req failed.");
             http_decoder_half_data_free(half_data);
             half_data = NULL;
         }
@@ -105,13 +97,13 @@ static void http_event_handler(enum http_event event,
         msg = CALLOC(struct http_message, 1);
 	    msg->type = HTTP_MESSAGE_REQ_LINE;
         msg->data = *data;
-        session_mq_publish_message(ctx->ref_session, ctx->topic_id, msg);
+        session_mq_publish_message(ev_ctx->ref_session, ev_ctx->topic_id, msg);
         break;
     case HTTP_EVENT_REQ_HDR_END:
         msg = CALLOC(struct http_message, 1);
 	    msg->type = HTTP_MESSAGE_REQ_HEADER;
         msg->data = *data;
-        session_mq_publish_message(ctx->ref_session, ctx->topic_id, msg);
+        session_mq_publish_message(ev_ctx->ref_session, ev_ctx->topic_id, msg);
         break;
     case HTTP_EVENT_REQ_BODY_BEGIN:
         break;
@@ -119,7 +111,7 @@ static void http_event_handler(enum http_event event,
         msg = CALLOC(struct http_message, 1);
 	    msg->type = HTTP_MESSAGE_REQ_BODY;
         msg->data = *data;
-        session_mq_publish_message(ctx->ref_session, ctx->topic_id, msg);
+        session_mq_publish_message(ev_ctx->ref_session, ev_ctx->topic_id, msg);
         break;
     case HTTP_EVENT_REQ_BODY_END:
         break;
@@ -147,7 +139,7 @@ static void http_event_handler(enum http_event event,
         half_data = http_decoder_half_data_new();
         ret = http_decoder_result_queue_push_res(queue, half_data);
         if (ret < 0) {
-            fprintf(stderr, "http_decoder_result_queue_push failed.");
+            fprintf(stderr, "http_decoder_result_queue_push res failed.");
             http_decoder_half_data_free(half_data);
             half_data = NULL;
         }
@@ -157,7 +149,7 @@ static void http_event_handler(enum http_event event,
         msg = CALLOC(struct http_message, 1);
 	    msg->type = HTTP_MESSAGE_RES_LINE;
         msg->data = *data;
-        session_mq_publish_message(ctx->ref_session, ctx->topic_id, msg);
+        session_mq_publish_message(ev_ctx->ref_session, ev_ctx->topic_id, msg);
         break;
     case HTTP_EVENT_RES_HDR:
         break;
@@ -165,7 +157,7 @@ static void http_event_handler(enum http_event event,
         msg = CALLOC(struct http_message, 1);
 	    msg->type = HTTP_MESSAGE_RES_HEADER;
         msg->data = *data;
-        session_mq_publish_message(ctx->ref_session, ctx->topic_id, msg); 
+        session_mq_publish_message(ev_ctx->ref_session, ev_ctx->topic_id, msg); 
         break;
     case HTTP_EVENT_RES_BODY_BEGIN:
         break;
@@ -173,7 +165,7 @@ static void http_event_handler(enum http_event event,
         msg = CALLOC(struct http_message, 1);
 	    msg->type = HTTP_MESSAGE_RES_BODY;
         msg->data = *data;
-        session_mq_publish_message(ctx->ref_session, ctx->topic_id, msg); 
+        session_mq_publish_message(ev_ctx->ref_session, ev_ctx->topic_id, msg); 
         break;
     case HTTP_EVENT_RES_BODY_END:
         break;
@@ -271,7 +263,7 @@ static int http_protocol_identify(const char *data, size_t data_len)
 int http_decoder_entry(struct session *sess, int events,
                        const struct packet *pkt, void *cb_arg)
 {
-    struct http_decoder_context *ctx = (struct http_decoder_context *)cb_arg;
+    struct http_decoder_context *decoder_ctx = (struct http_decoder_context *)cb_arg;
     size_t payload_len = 0;
     uint64_t inner_flag = 0;
 
@@ -280,21 +272,19 @@ int http_decoder_entry(struct session *sess, int events,
         return 0;
     }
 
-    struct http_decoder_exdata *ex_data = session_get_ex_data(sess, ctx->ex_data_idx);
+    struct http_decoder_exdata *ex_data =
+        session_get_ex_data(sess, decoder_ctx->ex_data_idx);
 
     if (events & SESS_EV_CLOSING) {
         if (ex_data != NULL) {
             http_decoder_exdata_free(ex_data);
-            session_set_ex_data(sess, ctx->ex_data_idx, NULL);
+            session_set_ex_data(sess, decoder_ctx->ex_data_idx, NULL);
         }
 
         return 0;
     }
 
     const char *payload = session_get0_current_payload(sess, &payload_len);
-    // const char *addr_readable = session_get0_readable_addr(sess);
-    // enum session_addr_type addr_type;
-    // struct session_addr *addr = session_get0_addr(sess, &addr_type);
     
     if (events & SESS_EV_OPENING) {
         assert(ex_data == NULL);
@@ -309,15 +299,16 @@ int http_decoder_entry(struct session *sess, int events,
             if (ret < 0) {
                 // ignore this session's event
                 struct session_event *s_event =
-                    session_get_intrinsic_event(sess, ctx->plugin_id);
+                    session_get_intrinsic_event(sess, decoder_ctx->plugin_id);
 
-                session_event_assign(s_event, ctx->st, sess, 0, http_decoder_entry, ctx);
+                session_event_assign(s_event, decoder_ctx->st, sess, 0,
+                                     http_decoder_entry, decoder_ctx);
                 return 0;
             }
         }
 
         ex_data = http_decoder_exdata_new(HD_RESULT_QUEUE_SIZE);
-        session_set_ex_data(sess, ctx->ex_data_idx, ex_data);
+        session_set_ex_data(sess, decoder_ctx->ex_data_idx, ex_data);
     }
 
     if (0 == payload_len || NULL == ex_data) {
@@ -329,7 +320,6 @@ int http_decoder_entry(struct session *sess, int events,
         return -1;
     }
 
-    // printf("session:%u dir:%d readable:%s\n", addr->ipv4.saddr, dir, addr_readable);
     struct http_decoder_half *cur_half = NULL;
     if (dir == PACKET_DIRECTION_C2S) {
         cur_half = ex_data->decoder->c2s_half;
@@ -337,19 +327,22 @@ int http_decoder_entry(struct session *sess, int events,
         cur_half = ex_data->decoder->s2c_half;
     }
 
-    ctx->http_ev_ctx->topic_id = ctx->topic_id;
-    ctx->http_ev_ctx->ref_exdata = ex_data;
-    ctx->http_ev_ctx->ref_session = sess;
+    http_decoder_half_reinit(cur_half, decoder_ctx->topic_id, ex_data->queue, sess);
+    http_decoder_half_parse(cur_half, payload, payload_len);
 
-    http_decoder_half_parse(cur_half, ctx->http_ev_ctx, payload, payload_len);
+    int thread_id = session_get_current_thread_id(sess);
 
-    long long trans_cnt = http_decoder_half_trans_count(cur_half);
-
-    fieldstat_easy_counter_incrby(ctx->fse, 0, ctx->fs_incoming_bytes_id,
+    fieldstat_easy_counter_incrby(decoder_ctx->fse, thread_id,
+                                  decoder_ctx->fs_incoming_bytes_id,
                                   NULL, 0, payload_len);
-    fieldstat_easy_counter_incrby(ctx->fse, 0, ctx->fs_incoming_pkts_id,
+
+    fieldstat_easy_counter_incrby(decoder_ctx->fse, thread_id,
+                                  decoder_ctx->fs_incoming_pkts_id,
                                   NULL, 0, 1);
-    fieldstat_easy_counter_incrby(ctx->fse, 0, ctx->fs_incoming_trans_id,
+
+    long long trans_cnt = http_decoder_half_trans_count(cur_half);
+    fieldstat_easy_counter_incrby(decoder_ctx->fse, thread_id,
+                                  decoder_ctx->fs_incoming_trans_id,
                                   NULL, 0, trans_cnt);
 
     return 0;
@@ -373,10 +366,6 @@ static void _http_decoder_context_free(struct http_decoder_context *ctx)
         return;
     }
 
-    if (ctx->http_ev_ctx != NULL) {
-        FREE(ctx->http_ev_ctx);
-    }
-
     if (ctx->fse != NULL) {
         fieldstat_easy_free(ctx->fse);
         ctx->fse = NULL;
@@ -406,7 +395,6 @@ void *http_decoder_init(struct stellar *st)
 {
     struct http_decoder_context *ctx = CALLOC(struct http_decoder_context, 1);
 
-    ctx->http_ev_ctx = CALLOC(struct http_event_context, 1);
     ctx->st = st;
     ctx->ex_data_idx = stellar_session_get_ex_new_index(st, "HTTP_DECODER",
 														http_decoder_ex_data_free,
@@ -425,7 +413,9 @@ void *http_decoder_init(struct stellar *st)
     }
     ctx->topic_id = topic_id;
 
-    ctx->fse = fieldstat_easy_new(1, "http_decoder_statistics", NULL, 0);
+    int thread_num = stellar_get_worker_thread_num(st);
+
+    ctx->fse = fieldstat_easy_new(thread_num, "http_decoder_statistics", NULL, 0);
     if (NULL == ctx->fse) {
         fprintf(stderr, "fieldstat_easy_new failed.");
         goto failed;
diff --git a/src/http_decoder/http_decoder_half.c b/src/http_decoder/http_decoder_half.c
index 17c49a1..40e5d01 100644
--- a/src/http_decoder/http_decoder_half.c
+++ b/src/http_decoder/http_decoder_half.c
@@ -40,11 +40,11 @@ struct http_decoder_half {
     struct http_decoder_half_data *ref_data;
     enum http_event event;
     http_event_cb *http_ev_cb;
-    void *http_ev_ctx;
+    struct http_event_context *http_ev_ctx;
     long long trans_counter;
 };
 
-// #define HTTP_DECODER_DEBUG
+//#define HTTP_DECODER_DEBUG
 #ifdef HTTP_DECODER_DEBUG
 static void printf_debug_info(const char *desc, const char *at, size_t length)
 {
@@ -600,7 +600,9 @@ http_decoder_half_new(http_event_cb *ev_cb, int is_cache_body, int type)
     struct http_decoder_half *half = CALLOC(struct http_decoder_half, 1);
     assert(half);
 
+    half->http_ev_ctx = CALLOC(struct http_event_context, 1);
     http_decoder_half_init(half, ev_cb, is_cache_body, type);
+
     return half;
 }
 
@@ -610,17 +612,31 @@ void http_decoder_half_free(struct http_decoder_half *half)
         return;
     }
 
+    if (half->http_ev_ctx != NULL) {
+        FREE(half->http_ev_ctx);
+    }
+
     FREE(half);
 }
 
-int http_decoder_half_parse(struct http_decoder_half *half, void *http_ev_ctx,
+void http_decoder_half_reinit(struct http_decoder_half *half, int topic_id,
+                              struct http_decoder_result_queue *queue,
+                              struct session *sess)
+{
+    assert(half != NULL);
+
+    half->http_ev_ctx->topic_id = topic_id;
+    half->http_ev_ctx->ref_session = sess;
+    half->http_ev_ctx->ref_queue = queue;
+}
+
+int http_decoder_half_parse(struct http_decoder_half *half,
                             const char *data, size_t data_len)
 {
     if (NULL == half || NULL == data || 0 == data_len) {
         return -1;
     }
 
-    half->http_ev_ctx = http_ev_ctx;
     half->error = llhttp_execute(&half->parser, data, data_len);
 
     int ret = 0;
diff --git a/src/http_decoder/http_decoder_half.h b/src/http_decoder/http_decoder_half.h
index a92708d..99cc260 100644
--- a/src/http_decoder/http_decoder_half.h
+++ b/src/http_decoder/http_decoder_half.h
@@ -19,6 +19,7 @@ extern "C"
 
 #include <stddef.h>
 
+#include "stellar/session.h"
 #include "http_decoder.h"
 #include "http_content_decompress.h"
 #include "http_decoder_result_queue.h"
@@ -44,18 +45,28 @@ enum http_event {
     HTTP_EVENT_RES_END = 1 << 16,
 };
 
+struct http_event_context {
+    int topic_id;
+    struct session *ref_session;
+    struct http_decoder_result_queue *ref_queue;
+};
+
 struct http_decoder_half;
 struct http_decoder_half_data;
 
 typedef void http_event_cb(enum http_event event, struct http_decoder_half_data **data,
-                           void *cb_args);
+                           struct http_event_context *ev_ctx);
 
 struct http_decoder_half *
 http_decoder_half_new(http_event_cb *event_cb, int is_cache_body, int type);
 
 void http_decoder_half_free(struct http_decoder_half *half);
 
-int http_decoder_half_parse(struct http_decoder_half *half, void *ev_ctx,
+void http_decoder_half_reinit(struct http_decoder_half *half, int topic_id,
+                              struct http_decoder_result_queue *queue,
+                              struct session *sess);
+
+int http_decoder_half_parse(struct http_decoder_half *half,
                             const char *data, size_t len);
 
 long long http_decoder_half_trans_count(struct http_decoder_half *half);
diff --git a/test/http_decoder/CMakeLists.txt b/test/http_decoder/CMakeLists.txt
index 75209b1..9bfc7ee 100644
--- a/test/http_decoder/CMakeLists.txt
+++ b/test/http_decoder/CMakeLists.txt
@@ -106,6 +106,7 @@ add_test(NAME HTTP_GET_REQ_PIPELINE_TEST COMMAND ${TEST_MAIN} ${CMAKE_CURRENT_SO
 add_test(NAME HTTP_TRANS_PIPELINE_TEST COMMAND ${TEST_MAIN} ${CMAKE_CURRENT_SOURCE_DIR}/test_result_json/http_trans_pipeline.json
          -f "find ${CMAKE_CURRENT_SOURCE_DIR}/http_pcap/ -name http_trans_pipeline.pcap|sort -V" WORKING_DIRECTORY ${TEST_RUN_DIR})
 
+
 set_tests_properties(HTTP_GET_SINGLE_TRANS_TEST
                      HTTP_GET_MULTI_TRANS_TEST
                      HTTP_GET_LONG_COOKIE_TEST