[ltt-dev] [PATCH 2/2] Add Support to lttv to analyse streaming trace.

[oussama.el-mfadli@polymtl.ca (Oussama El Mfadli)]

Hi,
As you suggest, I removed the lttv_process_traceset_middle_streaming.
Now the lttv_process_trace_update is called directly from the 
lttv_process_traceset_middle.
Also added a support if inotify is not available.

The patch is following:
-----------

Add Support to lttv to analyse streaming traces

---
  lttv/lttv/batchtest.c                              |    2 +-
  lttv/lttv/tracecontext.c                           |  184 
+++++++++++++++++++-
  lttv/lttv/tracecontext.h                           |    7 +-
  lttv/modules/gui/lttvwindow/lttvwindow/callbacks.c |    4 +-
  lttv/modules/text/batchAnalysis.c                  |   27 +++-
  5 files changed, 210 insertions(+), 14 deletions(-)

diff --git a/lttv/lttv/batchtest.c b/lttv/lttv/batchtest.c
index 08b892b..df15dcd 100644
--- a/lttv/lttv/batchtest.c
+++ b/lttv/lttv/batchtest.c
@@ -100,7 +100,7 @@ static void lttv_trace_option(void __UNUSED__ 
*hook_data)
  {
      LttTrace *trace;

-    trace = ltt_trace_open(a_trace);
+    trace = ltt_trace_open(a_trace, NO_FLAG);
      if(trace == NULL) {
          g_critical("cannot open trace %s", a_trace);
      } else {
diff --git a/lttv/lttv/tracecontext.c b/lttv/lttv/tracecontext.c
index 136b2ee..332ae46 100644
--- a/lttv/lttv/tracecontext.c
+++ b/lttv/lttv/tracecontext.c
@@ -28,6 +28,19 @@
  #include <lttv/filter.h>
  #include <errno.h>

+#include <limits.h>
+
+#include <linux/version.h>
+
+#if LINUX_VERSION_CODE >= KERNEL_VERSION(2,6,14)
+#include <sys/inotify.h>
+#define HAS_INOTIFY
+
+#else
+
+#undef HAS_INOTIFY
+#endif
+
  gint compare_tracefile(gconstpointer a, gconstpointer b)
  {
      gint comparison = 0;
@@ -690,12 +703,23 @@ guint 
lttv_process_traceset_middle(LttvTracesetContext *self,
      unsigned count = 0;

      guint read_ret;
-
+    guint i;
+    guint tracefile_counter = 0;
+
+    gboolean isStreaming = FALSE;
      //enum read_state last_read_state = LAST_NONE;

      gint last_ret = 0; /* return value of the last hook list called */
-
-    /* Get the next event from the pqueue, call its hooks,
+
+    guint num_trace = lttv_traceset_number(self->ts);
+    for(i=0; i< num_trace; i++)
+      if(self->traces[i]->t->streaming_is_enable == TRUE)
+      {
+        isStreaming = TRUE;
+        tracefile_counter = lttv_process_traceset_update(self);
+        break;
+      }
+      /* Get the next event from the pqueue, call its hooks,
         reinsert in the pqueue the following event from the same tracefile
         unless the tracefile is finished or the event is later than the
         end time. */
@@ -706,7 +730,10 @@ guint 
lttv_process_traceset_middle(LttvTracesetContext *self,
          /* End of traceset : tfc is NULL */
          if(unlikely(tfc == NULL))
          {
-            return count;
+            if(isStreaming)
+              return tracefile_counter == 0 ? 0 : -1;
+             else
+               return count;
          }

          /* Have we reached :
@@ -723,7 +750,10 @@ guint 
lttv_process_traceset_middle(LttvTracesetContext *self,
                          end_position) == 0)
                  || ltt_time_compare(end, tfc->timestamp) <= 0))
          {
-            return count;
+            if(isStreaming)
+              return tracefile_counter == 0 ? 0 : -1;
+             else
+               return count;
          }

          /* Get the tracefile with an event for the smallest time 
found. If two
@@ -814,6 +844,125 @@ void lttv_process_traceset_end(LttvTracesetContext 
*self,
              event_by_id_channel);
  }

+#ifdef HAS_INOTIFY
+/****************************************************************************
+ * lttv_process_trace_update
+ *
+ * process the changes that occur in the trace. Use Inotify System Call to
+ * monitor the tracefile.
+ *
+ * Return the number of file presently monitor. If 0, the current trace 
probably
+ * received all the data.
+ 
***************************************************************************/
+guint lttv_process_trace_update(LttvTraceContext *self)
+{
+
+    guint i, j, nb_tracefile;
+
+    gint ret;
+
+    LttvTracefileContext **tfc;
+
+    LttTrace *t = self->t;
+
+    if(t == NULL)
+      printf("the trace is NULL\n");
+
+    nb_tracefile = self->tracefiles->len;
+
+    GTree *pqueue = self->ts_context->pqueue;
+
+    for(i = 0 ; i < nb_tracefile ; i++) {
+        tfc = &g_array_index(self->tracefiles, LttvTracefileContext*, i);
+
+        g_tree_remove(pqueue, *tfc);
+
+    }
+
+    int offset = 0;
+    char buf[sizeof(struct inotify_event) + PATH_MAX];
+    int len = read(t->inotify_fd, buf, sizeof(struct inotify_event) + 
PATH_MAX);
+    while(offset < len) {
+        struct inotify_event *    ievent = (struct inotify_event 
*)&(buf[offset]);
+        for(i=0; i< t->inotify_watch_array.num; i++) {
+          if (t->inotify_watch_array.elem[i].wd == ievent->wd) {
+            if ( ievent->mask & IN_ISDIR )
+            {
+              open_tracefiles( t, g_quark_to_string(t->pathname), 
ievent->name);
+            }
+            else
+            {
+              if(ievent->mask & IN_OPEN)
+              {
+            t->open_tracefile_counter++;
+            break;
+              }
+              else if(ievent->mask & IN_CLOSE_WRITE)
+              {
+
+            t->open_tracefile_counter--;
+              }
+               else if(ievent->mask & IN_MODIFY)
+              {
+                (*tfc) = NULL;
+            for(j = 0; j < self->tracefiles->len; j++)
+            {
+              tfc = &g_array_index(self->tracefiles, 
LttvTracefileContext*, j);
+              if((*tfc)->tf->name == 
t->inotify_watch_array.elem[i].name && (*tfc)->tf->cpu_num == 
t->inotify_watch_array.elem[i].num)
+                break;
+            }
+            ret = ltt_tracefile_update((*tfc)->tf, t);
+            if(ret == 0)
+            {
+              (*tfc)->timestamp = 
ltt_event_time(ltt_tracefile_get_event((*tfc)->tf));
+              g_tree_insert(pqueue, (*tfc), (*tfc));
+            }
+             }
+            }
+            break;
+          }
+        }
+        offset += sizeof(*ievent) + ievent->len;
+    }
+    return t->open_tracefile_counter;
+}
+#else
+    
/****************************************************************************
+    * lttv_process_trace_update
+    *
+    * process the changes that occur in the trace. Use a regular file 
polling to
+    * monitor the tracefile.
+    *
+    * Return 1 if a tracefile has been updated, 0 otherwise
+    
***************************************************************************/
+    guint lttv_process_trace_update(LttvTraceContext *self)
+    {
+          guint i, nb_tracefile;
+
+    gint ret;
+
+    LttvTracefileContext **tfc;
+
+    nb_tracefile = self->tracefiles->len;
+
+    GTree *pqueue = self->ts_context->pqueue;
+
+    for(i = 0 ; i < nb_tracefile ; i++) {
+        tfc = &g_array_index(self->tracefiles, LttvTracefileContext*, i);
+
+        g_tree_remove(pqueue, *tfc);
+
+        ret = ltt_tracefile_update((*tfc)->tf);
+        if(ret == 1)
+        {
+          (*tfc)->timestamp = 
ltt_event_time(ltt_tracefile_get_event((*tfc)->tf));
+          g_tree_insert(pqueue, (*tfc), (*tfc));
+
+        }
+    }
+    return nb_tracefile;
+    }
+#endif
  /* Subtile modification :
   * if tracefile has no event at or after the time requested, it is not 
put in
   * the queue, as the next read would fail.
@@ -857,6 +1006,31 @@ void lttv_process_trace_seek_time(LttvTraceContext 
*self, LttTime start)
  }


+/****************************************************************************
+ * lttv_process_traceset_update
+ *
+ * process the changes that occur in the traceset.
+ *
+ * Return the number of file presently monitor(open for writting). If 
0, the
+ * current traceset probably received all the data.
+ *
+ * Author : Oussama El Mfadli
+ 
***************************************************************************/
+guint lttv_process_traceset_update(LttvTracesetContext *self)
+{
+    guint i, nb_trace, open_counter ;
+
+    LttvTraceContext *tc;
+
+    nb_trace = lttv_traceset_number(self->ts);
+    open_counter  = 0;
+    for(i = 0 ; i < nb_trace ; i++) {
+        tc = self->traces[i];
+        open_counter += lttv_process_trace_update(tc);
+    }
+    return open_counter;
+}
+
  void lttv_process_traceset_seek_time(LttvTracesetContext *self, 
LttTime start)
  {
      guint i, nb_trace;
diff --git a/lttv/lttv/tracecontext.h b/lttv/lttv/tracecontext.h
index acedeea..e36cec5 100644
--- a/lttv/lttv/tracecontext.h
+++ b/lttv/lttv/tracecontext.h
@@ -212,14 +212,15 @@ guint 
lttv_process_traceset_middle(LttvTracesetContext *self,
          LttTime end,
          gulong nb_events,
          const LttvTracesetContextPosition *end_position);
-
+
  void lttv_process_traceset_end(LttvTracesetContext *self,
          LttvHooks *after_traceset,
          LttvHooks *after_trace,
          LttvHooks *after_tracefile,
          LttvHooks *event,
          LttvHooksByIdChannelArray *event_by_id_channel);
-
+
+guint lttv_process_traceset_update(LttvTracesetContext *self);

  void lttv_process_traceset_seek_time(LttvTracesetContext *self, 
LttTime start);

@@ -229,6 +230,8 @@ void 
lttv_traceset_context_compute_time_span(LttvTracesetContext *self,
  gboolean lttv_process_traceset_seek_position(LttvTracesetContext *self,
          const LttvTracesetContextPosition *pos);

+guint lttv_process_trace_update(LttvTraceContext *self);
+
  void lttv_process_trace_seek_time(LttvTraceContext *self, LttTime start);

  void lttv_traceset_context_add_hooks(LttvTracesetContext *self,
diff --git a/lttv/modules/gui/lttvwindow/lttvwindow/callbacks.c 
b/lttv/modules/gui/lttvwindow/lttvwindow/callbacks.c
index d58c313..3db0749 100644
--- a/lttv/modules/gui/lttvwindow/lttvwindow/callbacks.c
+++ b/lttv/modules/gui/lttvwindow/lttvwindow/callbacks.c
@@ -1869,7 +1869,7 @@ void add_trace(GtkWidget * widget, gpointer user_data)
        get_absolute_pathname(dir, abs_path);
        trace_v = lttvwindowtraces_get_trace_by_name(abs_path);
        if(trace_v == NULL) {
-        trace = ltt_trace_open(abs_path);
+        trace = ltt_trace_open(abs_path, NO_FLAG);
          if(trace == NULL) {
            g_warning("cannot open trace %s", abs_path);

@@ -4969,7 +4969,7 @@ __EXPORT void 
create_main_window_with_trace_list(GSList *traces)
      get_absolute_pathname(path, abs_path);
      trace_v = lttvwindowtraces_get_trace_by_name(abs_path);
      if(trace_v == NULL) {
-      trace = ltt_trace_open(abs_path);
+      trace = ltt_trace_open(abs_path, NO_FLAG);
        if(trace == NULL) {
          g_warning("cannot open trace %s", abs_path);

diff --git a/lttv/modules/text/batchAnalysis.c 
b/lttv/modules/text/batchAnalysis.c
index 4b02f33..1647598 100644
--- a/lttv/modules/text/batchAnalysis.c
+++ b/lttv/modules/text/batchAnalysis.c
@@ -49,14 +49,17 @@ static LttvHooks
    *main_hooks;

  static char *a_trace;
-
+static gboolean a_streaming;
  static gboolean a_stats;

  void lttv_trace_option(void *hook_data)
  {
    LttTrace *trace;

-  trace = ltt_trace_open(a_trace);
+  unsigned int flag = NO_FLAG;
+  if( a_streaming )
+    flag |= TRACE_STREAMING;
+  trace = ltt_trace_open(a_trace, flag);
    if(trace == NULL) g_critical("cannot open trace %s", a_trace);
    lttv_traceset_add(traceset, lttv_trace_new(trace));
  }
@@ -138,8 +141,18 @@ static gboolean process_traceset(void *hook_data, 
void *call_data)
                                 end,
                                 G_MAXULONG,
                                 NULL);
-
-
+  if(a_streaming == TRUE)
+  {
+    gint tracefile_counter = -1;
+    while(tracefile_counter != 0)
+      {
+        tracefile_counter = lttv_process_traceset_middle(tc,
+                                   end,
+                                   G_MAXULONG,
+                                       NULL);
+
+     }
+  }
    //lttv_traceset_context_remove_hooks(tc,
    //before_traceset, after_traceset, NULL, before_trace, after_trace,
    //NULL, before_tracefile, after_tracefile, NULL, before_event, 
after_event);
@@ -186,6 +199,11 @@ static void init()
        "",
        LTTV_OPT_NONE, &a_stats, NULL, NULL);

+  a_streaming = FALSE;
+  lttv_option_add("streaming", 'S',
+      "define if the traceset is receiving streamed informations",
+      "",
+      LTTV_OPT_NONE, &a_streaming, NULL, NULL);

    traceset = lttv_traceset_new();

@@ -251,6 +269,7 @@ static void destroy()

    lttv_option_remove("trace");
    lttv_option_remove("stats");
+  lttv_option_remove("streaming");

    lttv_hooks_destroy(before_traceset);
    lttv_hooks_destroy(after_traceset);
-- 
1.7.0.4