filterx: defer filterx_scope_sync() to the latest point possible

Signed-off-by: Balazs Scheidler <[email protected]>

lib/filterx/filterx-eval.c

@@ -122,9 +122,3 @@ filterx_eval_exec_statements(FilterXScope *scope, GList *statements, LogMessage
   filterx_eval_set_context(NULL);
   return success;
 }
-
-void
-filterx_eval_sync_scope_and_message(FilterXScope *scope, LogMessage *msg)
-{
-  filterx_scope_sync(scope, msg);
-}

lib/filterx/filterx-pipe.c

@@ -53,31 +53,31 @@ log_filterx_pipe_queue(LogPipe *s, LogMessage *msg, const LogPathOptions *path_o
   gboolean res;
 
   path_options = log_path_options_chain(&local_path_options, path_options);
-  msg_trace(">>>>>> filterx rule evaluation begin",
-            evt_tag_str("rule", self->name),
-            log_pipe_location_tag(s),
-            evt_tag_msg_reference(msg));
 
   FilterXScope *scope = filterx_scope_ref(path_options->filterx_scope);
   if (!scope)
-    local_path_options.filterx_scope = scope = filterx_scope_new();
-
+    scope = filterx_scope_new();
   filterx_scope_make_writable(&scope);
 
+  msg_trace(">>>>>> filterx rule evaluation begin",
+            evt_tag_str("rule", self->name),
+            log_pipe_location_tag(s),
+            evt_tag_printf("path_scope", "%p", path_options->filterx_scope),
+            evt_tag_printf("scope", "%p", scope),
+            evt_tag_msg_reference(msg));
+
   NVTable *payload = nv_table_ref(msg->payload);
   res = filterx_eval_exec_statements(scope, self->stmts, msg);
-  if (res)
-    {
-      log_msg_make_writable(&msg, path_options);
-      filterx_eval_sync_scope_and_message(scope, msg);
-    }
 
   msg_trace("<<<<<< filterx rule evaluation result",
             evt_tag_str("result", res ? "matched" : "unmatched"),
             evt_tag_str("rule", self->name),
             log_pipe_location_tag(s),
+            evt_tag_printf("scope", "%p", scope),
+            evt_tag_int("dirty", filterx_scope_is_dirty(scope)),
             evt_tag_msg_reference(msg));
 
+  local_path_options.filterx_scope = scope;
   if (res)
     {
       log_pipe_forward_msg(s, msg, path_options);
@@ -120,7 +120,7 @@ log_filterx_pipe_new(GList *stmts, GlobalConfig *cfg)
   LogFilterXPipe *self = g_new0(LogFilterXPipe, 1);
 
   log_pipe_init_instance(&self->super, cfg);
-  self->super.flags |= PIF_CONFIG_RELATED;
+  self->super.flags = (self->super.flags | PIF_CONFIG_RELATED) & ~PIF_SYNC_SCOPE;
   self->super.init = log_filterx_pipe_init;
   self->super.queue = log_filterx_pipe_queue;
   self->super.free_fn = log_filterx_pipe_free;

lib/logmpx.c

@@ -88,9 +88,13 @@ log_multiplexer_queue(LogPipe *s, LogMessage *msg, const LogPathOptions *path_op
   log_path_options_push_junction(&local_path_options, &matched, path_options);
   if (_has_multiple_arcs(self))
     {
-      log_msg_write_protect(msg);
       if (path_options->filterx_scope)
-        filterx_scope_write_protect(path_options->filterx_scope);
+        {
+          log_msg_make_writable(&msg, path_options);
+          filterx_scope_sync(path_options->filterx_scope, msg);
+          filterx_scope_write_protect(path_options->filterx_scope);
+        }
+      log_msg_write_protect(msg);
     }
   for (fallback = 0; (fallback == 0) || (fallback == 1 && self->fallback_exists && !delivered); fallback++)
     {
@@ -216,6 +220,7 @@ log_multiplexer_new(GlobalConfig *cfg)
   LogMultiplexer *self = g_new0(LogMultiplexer, 1);
 
   log_pipe_init_instance(&self->super, cfg);
+  self->super.flags = self->super.flags & ~PIF_SYNC_SCOPE;
   self->super.init = log_multiplexer_init;
   self->super.deinit = log_multiplexer_deinit;
   self->super.queue = log_multiplexer_queue;

lib/logpipe.c

@@ -82,6 +82,7 @@ log_pipe_init_instance(LogPipe *self, GlobalConfig *cfg)
   self->queue = NULL;
   self->free_fn = log_pipe_free_method;
   self->arcs = _arcs;
+  self->flags = PIF_SYNC_SCOPE;
 }
 
 LogPipe *

lib/logpipe.h

@@ -72,6 +72,8 @@
 /* node created directly by the user */
 #define PIF_CONFIG_RELATED    0x0100
 
+#define PIF_SYNC_SCOPE        0x0200
+
 /* private flags range, to be used by other LogPipe instances for their own purposes */
 
 #define PIF_PRIVATE(x)       ((x) << 16)
@@ -456,6 +458,14 @@ log_pipe_queue(LogPipe *s, LogMessage *msg, const LogPathOptions *path_options)
         }
     }
 
+  if ((s->flags & PIF_SYNC_SCOPE) &&
+      path_options->filterx_scope &&
+      filterx_scope_is_dirty(path_options->filterx_scope))
+    {
+      log_msg_make_writable(&msg, path_options);
+      filterx_scope_sync(path_options->filterx_scope, msg);
+    }
+
   if (G_UNLIKELY(s->flags & (PIF_HARD_FLOW_CONTROL | PIF_JUNCTION_END | PIF_CONDITIONAL_MIDPOINT)))
     {
       path_options = log_path_options_chain(&local_path_options, path_options);