Merge pull request syslog-ng#4943 from bazsi/filterx-refactor-eval-context-management

Filterx refactor eval context management

Author: alltilla

lib/driver.c

@@ -142,7 +142,7 @@ static void
 log_driver_init_instance(LogDriver *self, GlobalConfig *cfg)
 {
   log_pipe_init_instance(&self->super, cfg);
-  self->super.flags |= PIF_CONFIG_RELATED;
+  self->super.flags |= PIF_CONFIG_RELATED + PIF_SYNC_FILTERX;
   self->super.free_fn = log_driver_free;
   self->super.pre_init = log_driver_pre_init_method;
   self->super.init = log_driver_init_method;

lib/filter/filter-pipe.c

@@ -122,7 +122,7 @@ log_filter_pipe_new(FilterExprNode *expr, GlobalConfig *cfg)
   LogFilterPipe *self = g_new0(LogFilterPipe, 1);
 
   log_pipe_init_instance(&self->super, cfg);
-  self->super.flags |= PIF_CONFIG_RELATED;
+  self->super.flags |= PIF_CONFIG_RELATED + PIF_SYNC_FILTERX;
   self->super.init = log_filter_pipe_init;
   self->super.queue = log_filter_pipe_queue;
   self->super.free_fn = log_filter_pipe_free;

lib/filterx/filterx-eval.c

@@ -22,6 +22,7 @@
  */
 #include "filterx/filterx-eval.h"
 #include "filterx/filterx-expr.h"
+#include "logpipe.h"
 #include "scratch-buffers.h"
 #include "tls-support.h"
 
@@ -112,6 +113,33 @@ filterx_format_last_error(void)
                         buf ? buf->str : "");
 }
 
+
+/*
+ * This is not a real weakref implementation as we will never get rid off
+ * weak references until the very end of a scope.  If this wasn't the case
+ * we would have to:
+ *    1) run a proper GC
+ *    2) notify weak references once the object is detroyed
+ *
+ * None of that exists now and I doubt ever will (but never say never).
+ * Right now a weak ref is destroyed as a part of the scope finalization
+ * process at which point circular references will be broken so the rest can
+ * go too.
+ */
+void
+filterx_eval_store_weak_ref(FilterXObject *object)
+{
+  FilterXEvalContext *context = filterx_eval_get_context();
+
+  if (object && !object->weak_referenced)
+    {
+      /* avoid putting object to the list multiple times */
+      object->weak_referenced = TRUE;
+      g_ptr_array_add(context->weak_refs, filterx_object_ref(object));
+    }
+}
+
+
 static gboolean
 _evaluate_statement(FilterXExpr *expr)
 {
@@ -157,16 +185,10 @@ _evaluate_statement(FilterXExpr *expr)
 
 
 gboolean
-filterx_eval_exec_statements(FilterXScope *scope, GList *statements, LogMessage *msg)
+filterx_eval_exec_statements(FilterXEvalContext *context, GList *statements, LogMessage *msg)
 {
-  FilterXEvalContext local_context =
-  {
-    .msgs = &msg,
-    .num_msg = 1,
-    .template_eval_options = &DEFAULT_TEMPLATE_EVAL_OPTIONS,
-    .scope = scope,
-  };
-  filterx_eval_set_context(&local_context);
+  context->msgs = &msg;
+  context->num_msg = 1;
   gboolean success = FALSE;
   for (GList *l = statements; l; l = l->next)
     {
@@ -179,7 +201,39 @@ filterx_eval_exec_statements(FilterXScope *scope, GList *statements, LogMessage
   /* NOTE: we only store the results into the message if the entire evaluation was successful */
   success = TRUE;
 fail:
-  filterx_scope_set_dirty(scope);
-  filterx_eval_set_context(NULL);
+  filterx_scope_set_dirty(context->scope);
   return success;
 }
+
+void
+filterx_eval_init_context(FilterXEvalContext *context, FilterXEvalContext *previous_context)
+{
+  FilterXScope *scope;
+
+  if (previous_context)
+    scope = filterx_scope_ref(previous_context->scope);
+  else
+    scope = filterx_scope_new();
+  filterx_scope_make_writable(&scope);
+
+  memset(context, 0, sizeof(*context));
+  context->template_eval_options = &DEFAULT_TEMPLATE_EVAL_OPTIONS;
+  context->scope = scope;
+
+  if (previous_context)
+    context->weak_refs = previous_context->weak_refs;
+  else
+    context->weak_refs = g_ptr_array_new_with_free_func((GDestroyNotify) filterx_object_unref);
+  context->previous_context = previous_context;
+
+  filterx_eval_set_context(context);
+}
+
+void
+filterx_eval_deinit_context(FilterXEvalContext *context)
+{
+  if (!context->previous_context)
+    g_ptr_array_free(context->weak_refs, TRUE);
+  filterx_scope_unref(context->scope);
+  filterx_eval_set_context(context->previous_context);
+}

lib/filterx/filterx-eval.h

@@ -42,16 +42,44 @@ struct _FilterXEvalContext
   FilterXScope *scope;
   FilterXError error;
   LogTemplateEvalOptions *template_eval_options;
+  GPtrArray *weak_refs;
+  FilterXEvalContext *previous_context;
 };
 
 FilterXEvalContext *filterx_eval_get_context(void);
 FilterXScope *filterx_eval_get_scope(void);
 void filterx_eval_push_error(const gchar *message, FilterXExpr *expr, FilterXObject *object);
 void filterx_eval_set_context(FilterXEvalContext *context);
-gboolean filterx_eval_exec_statements(FilterXScope *scope, GList *statements, LogMessage *msg);
+gboolean filterx_eval_exec_statements(FilterXEvalContext *context, GList *statements, LogMessage *msg);
 void filterx_eval_sync_scope_and_message(FilterXScope *scope, LogMessage *msg);
 const gchar *filterx_eval_get_last_error(void);
 void filterx_eval_clear_errors(void);
 
+void filterx_eval_store_weak_ref(FilterXObject *object);
+
+void filterx_eval_init_context(FilterXEvalContext *context, FilterXEvalContext *previous_context);
+void filterx_eval_deinit_context(FilterXEvalContext *context);
+
+static inline void
+filterx_eval_sync_message(FilterXEvalContext *context, LogMessage **pmsg, const LogPathOptions *path_options)
+{
+  if (!context)
+    return;
+
+  if (!filterx_scope_is_dirty(context->scope))
+    return;
+
+  log_msg_make_writable(pmsg, path_options);
+  filterx_scope_sync(context->scope, *pmsg);
+}
+
+static inline void
+filterx_eval_prepare_for_fork(FilterXEvalContext *context, LogMessage **pmsg, const LogPathOptions *path_options)
+{
+  filterx_eval_sync_message(context, pmsg, path_options);
+  if (context)
+    filterx_scope_write_protect(context->scope);
+  log_msg_write_protect(*pmsg);
+}
 
 #endif

lib/filterx/filterx-object.h

@@ -87,7 +87,7 @@ struct _FilterXObject
    *                          propagates to the inner elements lazily
    *
    */
-  guint thread_index:16, modified_in_place:1, readonly:1;
+  guint thread_index:16, modified_in_place:1, readonly:1, weak_referenced:1;
   FilterXType *type;
 };
 

lib/filterx/filterx-pipe.c

@@ -49,35 +49,29 @@ static void
 log_filterx_pipe_queue(LogPipe *s, LogMessage *msg, const LogPathOptions *path_options)
 {
   LogFilterXPipe *self = (LogFilterXPipe *) s;
+  FilterXEvalContext eval_context;
   LogPathOptions local_path_options;
   gboolean res;
 
   path_options = log_path_options_chain(&local_path_options, path_options);
-
-  FilterXScope *scope = filterx_scope_ref(path_options->filterx_scope);
-  if (!scope)
-    scope = filterx_scope_new();
-  filterx_scope_make_writable(&scope);
+  filterx_eval_init_context(&eval_context, path_options->filterx_context);
 
   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);
+  res = filterx_eval_exec_statements(&eval_context, self->stmts, 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_int("dirty", filterx_scope_is_dirty(eval_context.scope)),
             evt_tag_msg_reference(msg));
 
-  local_path_options.filterx_scope = scope;
+  local_path_options.filterx_context = &eval_context;
   if (res)
     {
       log_pipe_forward_msg(s, msg, path_options);
@@ -89,7 +83,7 @@ log_filterx_pipe_queue(LogPipe *s, LogMessage *msg, const LogPathOptions *path_o
       log_msg_drop(msg, path_options, AT_PROCESSED);
     }
 
-  filterx_scope_unref(scope);
+  filterx_eval_deinit_context(&eval_context);
   nv_table_unref(payload);
 }
 
@@ -120,7 +114,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 = (self->super.flags | PIF_CONFIG_RELATED) & ~PIF_SYNC_SCOPE;
+  self->super.flags = (self->super.flags | PIF_CONFIG_RELATED);
   self->super.init = log_filterx_pipe_init;
   self->super.queue = log_filterx_pipe_queue;
   self->super.free_fn = log_filterx_pipe_free;

lib/filterx/filterx-scope.c

@@ -102,7 +102,6 @@ struct _FilterXScope
 {
   GAtomicCounter ref_cnt;
   GArray *variables;
-  GPtrArray *weak_refs;
   gboolean write_protected;
   gboolean dirty;
   gint generation;
@@ -215,16 +214,6 @@ filterx_scope_register_variable(FilterXScope *self,
   return &g_array_index(self->variables, FilterXVariable, v_index);
 }
 
-
-void
-filterx_scope_store_weak_ref(FilterXScope *self, FilterXObject *object)
-{
-  g_assert(self->write_protected == FALSE);
-
-  if (object)
-    g_ptr_array_add(self->weak_refs, filterx_object_ref(object));
-}
-
 /*
  * 1) sync objects to message
  * 2) drop undeclared objects
@@ -307,7 +296,6 @@ filterx_scope_new(void)
   g_atomic_counter_set(&self->ref_cnt, 1);
   self->variables = g_array_sized_new(FALSE, TRUE, sizeof(FilterXVariable), 16);
   g_array_set_clear_func(self->variables, (GDestroyNotify) _variable_free);
-  self->weak_refs = g_ptr_array_new_with_free_func((GDestroyNotify) filterx_object_unref);
   return self;
 }
 
@@ -371,7 +359,6 @@ static void
 _free(FilterXScope *self)
 {
   g_array_free(self->variables, TRUE);
-  g_ptr_array_free(self->weak_refs, TRUE);
   g_free(self);
 }
 

lib/filterx/filterx-scope.h

@@ -66,9 +66,6 @@ FilterXVariable *filterx_scope_register_variable(FilterXScope *self,
                                                  FilterXVariableHandle handle,
                                                  FilterXObject *initial_value);
 
-void filterx_scope_store_weak_ref(FilterXScope *self, FilterXObject *object);
-
-
 /* copy on write */
 void filterx_scope_write_protect(FilterXScope *self);
 FilterXScope *filterx_scope_make_writable(FilterXScope **pself);

lib/filterx/filterx-weakrefs.c

@@ -54,9 +54,7 @@
 void
 filterx_weakref_set(FilterXWeakRef *self, FilterXObject *object)
 {
-  FilterXScope *scope = filterx_eval_get_scope();
-  if (scope)
-    filterx_scope_store_weak_ref(scope, object);
+  filterx_eval_store_weak_ref(object);
   self->object = object;
 }
 

lib/filterx/object-json.c

@@ -131,9 +131,7 @@ filterx_json_convert_json_to_object_cached(FilterXObject *self, FilterXWeakRef *
 void
 filterx_json_associate_cached_object(struct json_object *jso, FilterXObject *filterx_obj)
 {
-  FilterXScope *scope = filterx_eval_get_scope();
-
-  filterx_scope_store_weak_ref(scope, filterx_obj);
+  filterx_eval_store_weak_ref(filterx_obj);
 
   /* we are not storing a reference in userdata to avoid circular
    * references.  That ref is retained by the filterx_scope_store_weak_ref()

lib/logmpx.c

@@ -88,13 +88,12 @@ 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))
     {
-      if (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);
+      /* if we are delivering to multiple branches, we need to sync the
+       * filterx state with our message and also need to make everything
+       * write protected so that changes in those branches don't overwrite
+       * data we still need */
+
+      filterx_eval_prepare_for_fork(path_options->filterx_context, &msg, path_options);
     }
   for (fallback = 0; (fallback == 0) || (fallback == 1 && self->fallback_exists && !delivered); fallback++)
     {
@@ -220,7 +219,6 @@ 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,7 +82,6 @@ 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

@@ -27,7 +27,7 @@
 
 #include "syslog-ng.h"
 #include "logmsg/logmsg.h"
-#include "filterx/filterx-scope.h"
+#include "filterx/filterx-eval.h"
 #include "cfg.h"
 #include "atomic.h"
 #include "messages.h"
@@ -72,7 +72,8 @@
 /* node created directly by the user */
 #define PIF_CONFIG_RELATED    0x0100
 
-#define PIF_SYNC_SCOPE        0x0200
+/* sync filterx state and message in right before calling queue() */
+#define PIF_SYNC_FILTERX      0x0200
 
 /* private flags range, to be used by other LogPipe instances for their own purposes */
 
@@ -218,7 +219,7 @@ struct _LogPathOptions
 
   gboolean *matched;
   const LogPathOptions *lpo_parent_junction;
-  FilterXScope *filterx_scope;
+  FilterXEvalContext *filterx_context;
 };
 
 #define LOG_PATH_OPTIONS_INIT { TRUE, FALSE, NULL, NULL }
@@ -458,13 +459,8 @@ 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 ((s->flags & PIF_SYNC_FILTERX))
+    filterx_eval_sync_message(path_options->filterx_context, &msg, path_options);
 
   if (G_UNLIKELY(s->flags & (PIF_HARD_FLOW_CONTROL | PIF_JUNCTION_END | PIF_CONDITIONAL_MIDPOINT)))
     {

lib/parser/parser-expr.c

@@ -182,7 +182,7 @@ void
 log_parser_init_instance(LogParser *self, GlobalConfig *cfg)
 {
   log_pipe_init_instance(&self->super, cfg);
-  self->super.flags |= PIF_CONFIG_RELATED;
+  self->super.flags |= PIF_CONFIG_RELATED + PIF_SYNC_FILTERX;
   self->super.init = log_parser_init_method;
   self->super.deinit = log_parser_deinit_method;
   self->super.free_fn = log_parser_free_method;