refactor: global be pool -> tls be pool

Author: joker-star-l

plugins/out_doris/doris.c

@@ -58,12 +58,17 @@ static inline void sync_fetch_and_add(size_t *dest, size_t value) {
 #endif
 }
 
+/* doris be connection pool, key: be_address, value: flb_upstream with single thread mode */
+FLB_TLS_DEFINE(struct flb_hash_table, doris_be_pool);
+
 static int cb_doris_init(struct flb_output_instance *ins,
                          struct flb_config *config, void *data)
 {
     struct flb_out_doris *ctx = NULL;
     (void) data;
 
+    FLB_TLS_INIT(be_pool);
+
     ctx = flb_doris_conf_create(ins, config);
     if (!ctx) {
         return -1;
@@ -81,6 +86,56 @@ static int cb_doris_init(struct flb_output_instance *ins,
     return 0;
 }
 
+static int cb_doris_worker_init(void *data, struct flb_config *config)
+{
+    struct flb_hash_table *be_pool;
+    struct flb_out_doris *ctx = data;
+
+    flb_plg_info(ctx->ins, "worker initializing");
+
+    be_pool = FLB_TLS_GET(doris_be_pool);
+    if (!be_pool) {
+        be_pool = flb_hash_table_create(FLB_HASH_TABLE_EVICT_NONE, 16, -1);
+        if (!be_pool) {
+            flb_errno();
+            return -1;
+        }
+        FLB_TLS_SET(doris_be_pool, be_pool);
+    }
+
+    return 0;
+}
+
+static int cb_doris_worker_exit(void *data, struct flb_config *config)
+{
+    struct flb_hash_table *be_pool;
+    struct flb_out_doris *ctx = data;
+
+    int i;
+    struct mk_list *tmp;
+    struct mk_list *head;
+    struct flb_hash_table_entry *entry;
+    struct flb_hash_table_chain *table;
+
+    flb_plg_info(ctx->ins, "worker exiting");
+
+    be_pool = FLB_TLS_GET(doris_be_pool);
+    if (be_pool) {
+        for (i = 0; i < be_pool->size; i++) {
+            table = &be_pool->table[i];
+            mk_list_foreach_safe(head, tmp, &table->chains) {
+                entry = mk_list_entry(head, struct flb_hash_table_entry, _head);
+                flb_upstream_destroy((struct flb_upstream*) entry->val);
+                entry->val = NULL;
+            }
+        }
+        flb_hash_table_destroy(be_pool);
+        FLB_TLS_SET(doris_be_pool, NULL);
+    }
+
+    return 0;
+}
+
 static int http_put(struct flb_out_doris *ctx,
                     const char *host, int port,
                     const void *body, size_t body_len,
@@ -102,6 +157,7 @@ static int http_put(struct flb_out_doris *ctx,
     struct flb_config_map_val *mv;
     struct flb_slist_entry *key = NULL;
     struct flb_slist_entry *val = NULL;
+    struct flb_hash_table *be_pool;
 
     int i;
     int root_type;
@@ -116,28 +172,24 @@ static int http_put(struct flb_out_doris *ctx,
     char address[1024] = {0};
     int len = 0;
 
+    be_pool = FLB_TLS_GET(doris_be_pool);
+
     /* Get upstream context and connection */
     if (strcmp(host, ctx->host) == 0 && port == ctx->port) { // address in config
         u = ctx->u;
     }
     else { // redirected address
         len = snprintf(address, sizeof(address), "%s:%i", host, port);
-        u = flb_hash_table_get_ptr(ctx->u_pool, address, len);
-        if (!u) { // first check
-            pthread_mutex_lock(&ctx->mutex); // lock
-            u = flb_hash_table_get_ptr(ctx->u_pool, address, len);
-            if (!u) { // second check
-                u = flb_upstream_create(ctx->u->base.config,
-                                        host,
-                                        port,
-                                        ctx->u->base.flags,
-                                        ctx->u->base.tls_context);
-                if (u) {
-                    flb_hash_table_add(ctx->u_pool, address, len, u, 0);
-                }
-            }
-            pthread_mutex_unlock(&ctx->mutex); // unlock
-            if (!u) {
+        u = flb_hash_table_get_ptr(be_pool, address, len);
+        if (!u) {
+            u = flb_upstream_create(ctx->u->base.config,
+                                    host,
+                                    port,
+                                    ctx->u->base.flags,
+                                    ctx->u->base.tls_context);
+            if (u) {
+                flb_hash_table_add(be_pool, address, len, u, 0);
+            } else {
                 flb_plg_error(ctx->ins, "no doris be connections available to %s:%i",
                               host, port);
                 return FLB_RETRY;
@@ -270,7 +322,7 @@ static int http_put(struct flb_out_doris *ctx,
 free_buf:
             flb_free(out_buf);
             msgpack_unpacked_destroy(&result);
-parse_done:
+parse_done:;
         }
         else {
             out_ret = FLB_RETRY;
@@ -475,13 +527,15 @@ static int cb_doris_format_test(struct flb_config *config,
 
 /* Plugin reference */
 struct flb_output_plugin out_doris_plugin = {
-    .name        = "doris",
-    .description = "Doris Output",
-    .cb_init     = cb_doris_init,
-    .cb_pre_run  = NULL,
-    .cb_flush    = cb_doris_flush,
-    .cb_exit     = cb_doris_exit,
-    .config_map  = config_map,
+    .name           = "doris",
+    .description    = "Doris Output",
+    .cb_init        = cb_doris_init,
+    .cb_pre_run     = NULL,
+    .cb_flush       = cb_doris_flush,
+    .cb_exit        = cb_doris_exit,
+    .cb_worker_init = cb_doris_worker_init,
+    .cb_worker_exit = cb_doris_worker_exit,
+    .config_map     = config_map,
 
     /* for testing */
     .test_formatter.callback = cb_doris_format_test,

plugins/out_doris/doris.h

@@ -68,11 +68,6 @@ struct flb_out_doris {
     /* Upstream connection to the backend server */
     struct flb_upstream *u;
 
-    /* doris be connection pool, key: string* be_address, value: flb_upstream* u */
-    struct flb_hash_table *u_pool;
-    pthread_mutex_t mutex;
-    int mutex_initialized;
-
     /* Plugin instance */
     struct flb_output_instance *ins;
 };

plugins/out_doris/doris_conf.c

@@ -95,7 +95,6 @@ struct flb_out_doris *flb_doris_conf_create(struct flb_output_instance *ins,
     int io_flags = 0;
     const char *tmp;
     struct flb_upstream *upstream;
-    struct flb_hash_table *u_pool;
     struct flb_out_doris *ctx = NULL;
     struct flb_doris_progress_reporter *reporter = NULL;
 
@@ -198,20 +197,6 @@ struct flb_out_doris *flb_doris_conf_create(struct flb_output_instance *ins,
     /* Set instance flags into upstream */
     flb_output_upstream_set(ctx->u, ins);
 
-    /* doris be connection pool */
-    u_pool = flb_hash_table_create(FLB_HASH_TABLE_EVICT_NONE, 16, -1);
-    if (!u_pool) {
-        flb_doris_conf_destroy(ctx);
-        return NULL;
-    }
-    ctx->u_pool = u_pool;
-    if (pthread_mutex_init(&ctx->mutex, NULL) == 0) {
-        ctx->mutex_initialized = 1;
-    } else {
-        flb_doris_conf_destroy(ctx);
-        return NULL;
-    }
-
     /* create and start the progress reporter */
     if (ctx->log_progress_interval > 0) {
         reporter = flb_calloc(1, sizeof(struct flb_doris_progress_reporter));
@@ -252,28 +237,5 @@ void flb_doris_conf_destroy(struct flb_out_doris *ctx)
         flb_upstream_destroy(ctx->u);
     }
 
-    if (ctx->u_pool) {
-        int i;
-        struct mk_list *tmp;
-        struct mk_list *head;
-        struct flb_hash_table_entry *entry;
-        struct flb_hash_table_chain *table;
-
-        for (i = 0; i < ctx->u_pool->size; i++) {
-            table = &ctx->u_pool->table[i];
-            mk_list_foreach_safe(head, tmp, &table->chains) {
-                entry = mk_list_entry(head, struct flb_hash_table_entry, _head);
-                flb_upstream_destroy((struct flb_upstream*) entry->val);
-                entry->val = NULL;
-            }
-        }
-
-        flb_hash_table_destroy(ctx->u_pool);
-    }
-
-    if (ctx->mutex_initialized) {
-        pthread_mutex_destroy(&ctx->mutex);
-    }
-
     flb_free(ctx);
 }