input: send resume signal to the input thread event loop if plugin is threaded

Signed-off-by: Daniel Lenar <[email protected]>

Author: danlenar

include/fluent-bit/flb_input_thread.h

@@ -93,6 +93,7 @@ int flb_input_thread_instance_init(struct flb_config *config,
 int flb_input_thread_instance_pre_run(struct flb_config *config, struct flb_input_instance *ins);
 
 int flb_input_thread_instance_pause(struct flb_input_instance *ins);
+int flb_input_thread_instance_resume(struct flb_input_instance *ins);
 int flb_input_thread_instance_exit(struct flb_input_instance *ins);
 
 int flb_input_thread_collectors_signal_start(struct flb_input_instance *ins);

src/flb_input.c

@@ -1695,7 +1695,13 @@ int flb_input_pause(struct flb_input_instance *ins)
 int flb_input_resume(struct flb_input_instance *ins)
 {
     if (ins->p->cb_resume) {
-        ins->p->cb_resume(ins->context, ins->config);
+        if (flb_input_is_threaded(ins)) {
+            /* signal the thread event loop about the 'resume' operation */
+            flb_input_thread_instance_resume(ins);
+        }
+        else {
+            ins->p->cb_resume(ins->context, ins->config);
+        }
     }
 
     return 0;

src/flb_input_chunk.c

@@ -1185,7 +1185,7 @@ size_t flb_input_chunk_set_limits(struct flb_input_instance *in)
         in->mem_buf_status == FLB_INPUT_PAUSED) {
         in->mem_buf_status = FLB_INPUT_RUNNING;
         if (in->p->cb_resume) {
-            in->p->cb_resume(in->context, in->config);
+            flb_input_resume(in);
             flb_info("[input] %s resume (mem buf overlimit)",
                       in->name);
         }
@@ -1196,7 +1196,7 @@ size_t flb_input_chunk_set_limits(struct flb_input_instance *in)
         in->storage_buf_status == FLB_INPUT_PAUSED) {
         in->storage_buf_status = FLB_INPUT_RUNNING;
         if (in->p->cb_resume) {
-            in->p->cb_resume(in->context, in->config);
+            flb_input_resume(in);
             flb_info("[input] %s resume (storage buf overlimit %zu/%zu)",
                       in->name,
                       ((struct flb_storage_input *)in->storage)->cio->total_chunks_up,

src/flb_input_thread.c

@@ -78,6 +78,11 @@ static inline int handle_input_event(flb_pipefd_t fd, struct flb_input_instance
                 ins->p->cb_pause(ins->context, ins->config);
             }
         }
+        else if (operation == FLB_INPUT_THREAD_RESUME) {
+            if (ins->p->cb_resume) {
+                ins->p->cb_resume(ins->context, ins->config);
+            }
+        }
         else if (operation == FLB_INPUT_THREAD_EXIT) {
             return FLB_INPUT_THREAD_EXIT;
         }
@@ -482,6 +487,31 @@ int flb_input_thread_instance_pause(struct flb_input_instance *ins)
     return 0;
 }
 
+/*
+ * Signal the thread event loop to resume the running plugin instance. This function
+ * must be called only from the main thread/pipeline.
+ */
+int flb_input_thread_instance_resume(struct flb_input_instance *ins)
+{
+    int ret;
+    uint64_t val;
+    struct flb_input_thread_instance *thi = ins->thi;
+
+    flb_plg_debug(ins, "thread resume instance");
+
+    /* compose message to resume the thread */
+    val = FLB_BITS_U64_SET(FLB_INPUT_THREAD_TO_THREAD,
+                           FLB_INPUT_THREAD_RESUME);
+
+    ret = flb_pipe_w(thi->ch_parent_events[1], &val, sizeof(val));
+    if (ret <= 0) {
+        flb_errno();
+        return -1;
+    }
+
+    return 0;
+}
+
 int flb_input_thread_instance_exit(struct flb_input_instance *ins)
 {
     int ret;