logpipe: add optimize() method

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

lib/cfg-source.c

@@ -173,7 +173,7 @@ _extract_source_from_file_location(GString *result, const gchar *filename, const
   FILE *f;
   gint lineno = 0;
   gint buflen = 65520;
-  gchar *line = g_malloc(buflen);
+  gboolean res = FALSE;
 
   if (yylloc->first_column < 1 || yylloc->last_column < 1 ||
       yylloc->first_column > buflen-1 || yylloc->last_column > buflen-1)
@@ -183,6 +183,7 @@ _extract_source_from_file_location(GString *result, const gchar *filename, const
   if (!f)
     return FALSE;
 
+  gchar *line = g_malloc(buflen);
   while (fgets(line, buflen, f))
     {
       lineno++;
@@ -216,13 +217,13 @@ _extract_source_from_file_location(GString *result, const gchar *filename, const
         }
     }
   fclose(f);
+  res = TRUE;
 
   /* NOTE: do we have the appropriate number of lines? */
   if (lineno <= yylloc->first_line)
-    return FALSE;
-
+    res = FALSE;
   g_free(line);
-  return TRUE;
+  return res;
 }
 
 static gboolean

lib/cfg-tree.c

@@ -654,6 +654,7 @@ log_expr_node_lookup_flag(const gchar *flag)
   return 0;
 }
 
+/* stores the ref to @pipe */
 static LogPipe *
 cfg_tree_assoc_pipe(CfgTree *self, LogExprNode *node, LogPipe *pipe, const gchar *info)
 {
@@ -1619,6 +1620,40 @@ cfg_tree_stop(CfgTree *self)
   return success;
 }
 
+gboolean
+cfg_tree_optimize(CfgTree *self)
+{
+  gint i;
+
+  g_assert(self->compiled);
+
+  for (i = 0; i < self->initialized_pipes->len; i++)
+    {
+      LogPipe **ppipe = (LogPipe **) &g_ptr_array_index(self->initialized_pipes, i);
+      LogPipe *pipe = *ppipe;
+
+      log_pipe_optimize(&pipe);
+      if (pipe != *ppipe)
+        {
+//          log_pipe_unref(*ppipe);
+          *ppipe = log_pipe_ref(pipe);
+        }
+    }
+
+  for (i = 0; i < self->initialized_pipes->len;)
+    {
+      if (g_ptr_array_index(self->initialized_pipes, i) == NULL)
+        {
+          g_ptr_array_remove_index(self->initialized_pipes, i);
+        }
+      else
+        i++;
+    }
+
+
+  return TRUE;
+}
+
 gboolean
 cfg_tree_pre_config_init(CfgTree *self)
 {
@@ -1667,7 +1702,7 @@ void
 cfg_tree_init_instance(CfgTree *self, GlobalConfig *cfg)
 {
   memset(self, 0, sizeof(*self));
-  self->initialized_pipes = g_ptr_array_new();
+  self->initialized_pipes = g_ptr_array_new_with_free_func((GDestroyNotify) log_pipe_unref);
   self->objects = g_hash_table_new_full(cfg_tree_objects_hash, cfg_tree_objects_equal, NULL,
                                         (GDestroyNotify) log_expr_node_unref);
   self->templates = g_hash_table_new_full(g_str_hash, g_str_equal, NULL, (GDestroyNotify) log_template_unref);
@@ -1679,7 +1714,6 @@ cfg_tree_init_instance(CfgTree *self, GlobalConfig *cfg)
 void
 cfg_tree_free_instance(CfgTree *self)
 {
-  g_ptr_array_foreach(self->initialized_pipes, (GFunc) log_pipe_unref, NULL);
   g_ptr_array_free(self->initialized_pipes, TRUE);
 
   g_ptr_array_foreach(self->rules, (GFunc) log_expr_node_unref, NULL);

lib/cfg-tree.h

@@ -188,6 +188,7 @@ gchar *cfg_tree_get_rule_name(CfgTree *self, gint content, LogExprNode *node);
 gchar *cfg_tree_get_child_id(CfgTree *self, gint content, LogExprNode *node);
 
 gboolean cfg_tree_compile(CfgTree *self);
+gboolean cfg_tree_optimize(CfgTree *self);
 gboolean cfg_tree_start(CfgTree *self);
 gboolean cfg_tree_stop(CfgTree *self);
 gboolean cfg_tree_pre_config_init(CfgTree *self);

lib/cfg.c

@@ -301,6 +301,8 @@ cfg_init(GlobalConfig *cfg)
     return FALSE;
   if (!cfg_tree_compile(&cfg->tree))
     return FALSE;
+  if (!cfg_tree_optimize(&cfg->tree))
+    return FALSE;
   app_config_pre_pre_init();
   if (!cfg_tree_pre_config_init(&cfg->tree))
     return FALSE;

lib/logmpx.c

@@ -212,7 +212,17 @@ _arcs(LogPipe *s)
   if (s->pipe_next)
     list = g_list_append(list, arc_new((LogPipe *)self, s->pipe_next, ARC_TYPE_PIPE_NEXT));
   return list;
-};
+}
+
+static LogPipe *
+_optimize(LogPipe *s)
+{
+  LogMultiplexer *self = (LogMultiplexer *)s;
+
+  for (gint i = 0; i < self->next_hops->len; i++)
+    log_pipe_optimize((LogPipe **) &g_ptr_array_index(self->next_hops, i));
+  return log_pipe_optimize_method(s);
+}
 
 LogMultiplexer *
 log_multiplexer_new(GlobalConfig *cfg)
@@ -226,6 +236,7 @@ log_multiplexer_new(GlobalConfig *cfg)
   self->super.free_fn = log_multiplexer_free;
   self->next_hops = g_ptr_array_new();
   self->super.arcs = _arcs;
+  self->super.optimize = _optimize;
   self->delivery_propagation = TRUE;
   log_pipe_add_info(&self->super, "multiplexer");
   return self;

lib/logpipe.c

@@ -26,6 +26,7 @@
 #include "cfg-tree.h"
 #include "cfg-walker.h"
 #include "perf/perf.h"
+#include "messages.h"
 
 gboolean (*pipe_single_step_hook)(LogPipe *pipe, LogMessage *msg, const LogPathOptions *path_options);
 
@@ -90,6 +91,19 @@ log_pipe_queue(LogPipe *self, LogMessage *msg, const LogPathOptions *path_option
 {
   g_assert((self->flags & PIF_INITIALIZED) != 0);
 
+#if 0
+  if (self->flags & PIF_NOOP)
+    {
+      gchar buf[256];
+      msg_warning("noop pipe traversed!!!!",
+                evt_tag_printf("pipe", "%p", self),
+                evt_tag_str("location", log_expr_node_format_location(self->expr_node, buf, sizeof(buf))),
+                evt_tag_str("expr", self->expr_node->expr_text));
+    }
+  else
+    msg_warning("other pipe traversed!!!!");
+#endif
+
   if (G_UNLIKELY((self->flags & PIF_CONFIG_RELATED) != 0 && pipe_single_step_hook))
     {
       if (!pipe_single_step_hook(self, msg, path_options))
@@ -165,6 +179,43 @@ log_pipe_post_config_init_method(LogPipe *self)
   return TRUE;
 }
 
+void
+log_pipe_optimize(LogPipe **pself)
+{
+  LogPipe *self = *pself;
+
+  if (!self)
+    return;
+  if ((self->flags & PIF_OPTIMIZED) == 0)
+    {
+//      self->flags |= PIF_OPTIMIZED;
+
+      LogPipe *new_pipe = self->optimize(self);
+      if (new_pipe != self)
+        *pself = new_pipe;
+    }
+}
+
+LogPipe *
+log_pipe_optimize_method(LogPipe *self)
+{
+  log_pipe_optimize(&self->pipe_next);
+
+  if (self->flags & PIF_NOOP)
+    {
+      gchar buf[256];
+
+      msg_warning("getting rid of logpipe",
+                evt_tag_printf("pipe", "%p", self),
+                evt_tag_printf("replacement", "%p", self->pipe_next),
+                evt_tag_str("location", log_expr_node_format_location(self->expr_node, buf, sizeof(buf))),
+                evt_tag_str("expr", self->expr_node->expr_text));
+      return self->pipe_next;
+    }
+
+  return self;
+}
+
 void
 log_pipe_init_instance(LogPipe *self, GlobalConfig *cfg)
 {
@@ -178,6 +229,7 @@ log_pipe_init_instance(LogPipe *self, GlobalConfig *cfg)
   self->queue = log_pipe_forward_msg;
   self->free_fn = log_pipe_free_method;
   self->arcs = _arcs;
+  self->optimize = log_pipe_optimize_method;
 }
 
 LogPipe *
@@ -186,6 +238,7 @@ log_pipe_new(GlobalConfig *cfg)
   LogPipe *self = g_new0(LogPipe, 1);
 
   log_pipe_init_instance(self, cfg);
+  self->flags |= PIF_NOOP;
   return self;
 }
 

lib/logpipe.h

@@ -74,6 +74,9 @@
 /* sync filterx state to message in right before calling queue() */
 #define PIF_SYNC_FILTERX_TO_MSG      0x0200
 
+#define PIF_NOOP		0x0400
+#define PIF_OPTIMIZED		0x0800
+
 /* private flags range, to be used by other LogPipe instances for their own purposes */
 
 #define PIF_PRIVATE(x)       ((x) << 16)
@@ -322,6 +325,7 @@ struct _LogPipe
 
   const gchar *(*generate_persist_name)(const LogPipe *self);
   GList *(*arcs)(LogPipe *self);
+  LogPipe *(*optimize)(LogPipe *self);
 
   /* clone this pipe when used in multiple locations in the processing
    * pipe-line. If it contains state, it should behave as if it was
@@ -353,6 +357,8 @@ gboolean log_pipe_pre_config_init_method(LogPipe *self);
 gboolean log_pipe_post_config_init_method(LogPipe *self);
 void log_pipe_init_instance(LogPipe *self, GlobalConfig *cfg);
 void log_pipe_clone_method(LogPipe *dst, const LogPipe *src);
+void log_pipe_optimize(LogPipe **s);
+LogPipe *log_pipe_optimize_method(LogPipe *s);
 void log_pipe_forward_notify(LogPipe *self, gint notify_code, gpointer user_data);
 EVTTAG *log_pipe_location_tag(LogPipe *pipe);
 void log_pipe_attach_expr_node(LogPipe *self, LogExprNode *expr_node);