proxy: added multithreaded worker support for proxy plugins

This patch adds worker support to golang output plugins, without which if golang plugins were becoming unresponsive it was blocking other i/p and o/p plugins as well
This fixes issue fluent/fluent-bit-go#45

Signed-off-by: Gautam Punhani <[email protected]>

src/flb_output.c

@@ -856,6 +856,15 @@ int flb_output_init_all(struct flb_config *config)
                 flb_output_instance_destroy(ins);
                 return -1;
             }
+
+            /* Multi-threading enabled if configured */
+            ret = flb_output_enable_multi_threading(ins, config);
+            if (ret == -1) {
+                flb_error("[output] could not start thread pool for '%s' plugin",
+                          p->name);
+                return -1;
+            }
+
             continue;
         }
 #endif
@@ -958,18 +967,28 @@ int flb_output_init_all(struct flb_config *config)
             return -1;
         }
 
-        /* Multi-threading enabled ? (through 'workers' property) */
-        if (ins->tp_workers > 0) {
-            ret = flb_output_thread_pool_create(config, ins);
-            if (ret == -1) {
-                flb_error("[output] could not start thread pool for '%s' plugin",
-                          p->name);
-                flb_output_instance_destroy(ins);
-                return -1;
-            }
+        /* Multi-threading enabled if configured */
+        ret = flb_output_enable_multi_threading(ins, config);
+        if (ret == -1) {
+            flb_error("[output] could not start thread pool for '%s' plugin",
+                      p->name);
+            return -1;
+        }
+    }
+
+    return 0;
+}
 
-            flb_output_thread_pool_start(ins);
+/* Add thread pool for output plugin if configured with workers */
+int flb_output_enable_multi_threading(struct flb_output_instance *ins, struct flb_config *config)
+{
+    /* Multi-threading enabled ? (through 'workers' property) */
+    if (ins->tp_workers > 0) {
+        if(flb_output_thread_pool_create(config, ins) != 0) {
+            flb_output_instance_destroy(ins);
+            return -1;
         }
+        flb_output_thread_pool_start(ins);
     }
 
     return 0;