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

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

 Add Support to lttv to analyse streaming trace.
 Currently, only the textDump module support the streaming traces, but all the tools
 are in place for the other modules.

---
 lttv/lttv/tracecontext.c          |  136 +++++++++++++++++++++++++++++++++++++
 lttv/lttv/tracecontext.h          |   14 +++-
 lttv/modules/text/batchAnalysis.c |   24 ++++++-
 3 files changed, 168 insertions(+), 6 deletions(-)

diff --git a/lttv/lttv/tracecontext.c b/lttv/lttv/tracecontext.c
index 136b2ee..9f50859 100644
--- a/lttv/lttv/tracecontext.c
+++ b/lttv/lttv/tracecontext.c
@@ -28,6 +28,11 @@
 #include <lttv/filter.h>
 #include <errno.h>
 
+#include <linux/limits.h>
+#include <sys/inotify.h>
+#include <stdlib.h>
+#include <unistd.h>
+
 gint compare_tracefile(gconstpointer a, gconstpointer b)
 {
 	gint comparison = 0;
@@ -675,6 +680,28 @@ void lttv_process_traceset_begin(LttvTracesetContext *self,
 
 //enum read_state { LAST_NONE, LAST_OK, LAST_EMPTY };
 
+/****************************************************************************
+ * lttv_process_traceset_middle_streaming
+ *
+ * This function should be called periodacly to retreive the new informations
+ * that were added in the tracefile since the last call.
+ * It will also process them.
+ * 
+ * Return the number of file presently monitor. If 0, the current trace probably 
+ * received all the data.
+ *
+ * Author : Oussama El Mfadli
+ ***************************************************************************/
+guint lttv_process_traceset_middle_streaming(LttvTracesetContext *self,
+		LttTime end,
+		gulong nb_events,
+		const LttvTracesetContextPosition *end_position)
+{
+	guint tracefile_counter = lttv_process_traceset_update(self);
+	lttv_process_traceset_middle(self, end, nb_events, end_position);
+	return tracefile_counter;
+}
+
 /* Note : a _middle must be preceded from a _seek or another middle */
 guint lttv_process_traceset_middle(LttvTracesetContext *self,
 		LttTime end,
@@ -814,6 +841,90 @@ void lttv_process_traceset_end(LttvTracesetContext *self,
 			event_by_id_channel);
 }
 
+/****************************************************************************
+ * 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.
+ *
+ * Author : Oussama El Mfadli
+ ***************************************************************************/
+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 )
+		    {
+		      ltt_tracefile_add_streaming( 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;
+}
+
 /* 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 +968,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..57b85e7 100644
--- a/lttv/lttv/tracecontext.h
+++ b/lttv/lttv/tracecontext.h
@@ -212,14 +212,22 @@ guint lttv_process_traceset_middle(LttvTracesetContext *self,
 		LttTime end,
 		gulong nb_events,
 		const LttvTracesetContextPosition *end_position);
-
+		
+		
+guint lttv_process_traceset_middle_streaming(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 +237,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/text/batchAnalysis.c b/lttv/modules/text/batchAnalysis.c
index 4b02f33..39448f1 100644
--- a/lttv/modules/text/batchAnalysis.c
+++ b/lttv/modules/text/batchAnalysis.c
@@ -49,14 +49,14 @@ 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);
+  trace = ltt_trace_open_streaming(a_trace, a_streaming);
   if(trace == NULL) g_critical("cannot open trace %s", a_trace);
   lttv_traceset_add(traceset, lttv_trace_new(trace));
 }
@@ -138,8 +138,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_streaming(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 +196,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 +266,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