Merge pull request syslog-ng#5148 from HofiOne/fix-threaded-source-stuck

logthrsource: fix sources getting stuck due to unclosed batches

Author: HofiOne

lib/filterx/filterx-eval.h

@@ -79,7 +79,6 @@ filterx_eval_prepare_for_fork(FilterXEvalContext *context, LogMessage **pmsg, co
   filterx_eval_sync_message(context, pmsg, path_options);
   if (context)
     filterx_scope_write_protect(context->scope);
-  log_msg_write_protect(*pmsg);
 }
 
 #endif

lib/logmpx.c

@@ -94,6 +94,7 @@ log_multiplexer_queue(LogPipe *s, LogMessage *msg, const LogPathOptions *path_op
        * data we still need */
 
       filterx_eval_prepare_for_fork(path_options->filterx_context, &msg, path_options);
+      log_msg_write_protect(msg);
     }
   for (fallback = 0; (fallback == 0) || (fallback == 1 && self->fallback_exists && !delivered); fallback++)
     {

lib/logthrsource/logthrsourcedrv.c

@@ -431,6 +431,10 @@ log_threaded_source_worker_blocking_post(LogThreadedSourceWorker *self, LogMessa
 {
   log_threaded_source_worker_post(self, msg);
 
+  /* unlocked, as only this thread can decrease the window size */
+  if (!self->control->auto_close_batches && !log_threaded_source_worker_free_to_send(self))
+    log_threaded_source_worker_close_batch(self);
+
   /*
    * The wakeup lock must be held before calling free_to_send() and suspend(),
    * otherwise g_cond_signal() might be called between free_to_send() and

lib/logthrsource/logthrsourcedrv.h

@@ -127,7 +127,7 @@ void log_threaded_source_worker_close_batch(LogThreadedSourceWorker *self);
 /* blocking API */
 void log_threaded_source_worker_blocking_post(LogThreadedSourceWorker *self, LogMessage *msg);
 
-/* non-blocking API, use it wisely (thread boundaries) */
+/* non-blocking API, use it wisely (thread boundaries); call close_batch() at least before suspending */
 void log_threaded_source_worker_post(LogThreadedSourceWorker *self, LogMessage *msg);
 gboolean log_threaded_source_worker_free_to_send(LogThreadedSourceWorker *self);
 

news/bugfix-5148.md

@@ -0,0 +1,2 @@
+
+`opentelemetry()` sources: fix source hang-up on flow-controlled paths