Option to only show streams' intersection

The --stream-intersection option allows showing only the events that
fit in the intersection of all streams, thus filtering out the events
for which only some streams have the time range info. This is achieved
by using the packet header timestamp begin/end information available
within each stream.

Signed-off-by: Julien Desfossez <jdesfossez@efficios.com>
Signed-off-by: Jérémie Galarneau <jeremie.galarneau@efficios.com>

diff --git a/converter/babeltrace.c b/converter/babeltrace.c
index c521ac1516c459641b24ec476f4c425ad09fcfe4..58f534c81a9a3c3ba05ce60f7f6675c70835023b 100644 (file)
--- a/converter/babeltrace.c
+++ b/converter/babeltrace.c
@@ -70,6 +70,7 @@ static char *opt_input_format, *opt_output_format;
  */
 static GPtrArray *opt_input_paths;
 static char *opt_output_path;
  */
 static GPtrArray *opt_input_paths;
 static char *opt_output_path;

+int opt_stream_intersection;

 
 static struct bt_format *fmt_read;
 
 
 static struct bt_format *fmt_read;
 


@@ -101,6 +102,7 @@ enum {
        OPT_CLOCK_DATE,
        OPT_CLOCK_GMT,
        OPT_CLOCK_FORCE_CORRELATE,
        OPT_CLOCK_DATE,
        OPT_CLOCK_GMT,
        OPT_CLOCK_FORCE_CORRELATE,

+       OPT_STREAM_INTERSECTION,

 };
 
 /*
 };
 
 /*


@@ -130,6 +132,7 @@ static struct poptOption long_options[] = {
        { "clock-date", 0, POPT_ARG_NONE, NULL, OPT_CLOCK_DATE, NULL, NULL },
        { "clock-gmt", 0, POPT_ARG_NONE, NULL, OPT_CLOCK_GMT, NULL, NULL },
        { "clock-force-correlate", 0, POPT_ARG_NONE, NULL, OPT_CLOCK_FORCE_CORRELATE, NULL, NULL },
        { "clock-date", 0, POPT_ARG_NONE, NULL, OPT_CLOCK_DATE, NULL, NULL },
        { "clock-gmt", 0, POPT_ARG_NONE, NULL, OPT_CLOCK_GMT, NULL, NULL },
        { "clock-force-correlate", 0, POPT_ARG_NONE, NULL, OPT_CLOCK_FORCE_CORRELATE, NULL, NULL },

+       { "stream-intersection", 0, POPT_ARG_NONE, NULL, OPT_STREAM_INTERSECTION, NULL, NULL },

        { NULL, 0, 0, NULL, 0, NULL, NULL },
 };
 
        { NULL, 0, 0, NULL, 0, NULL, NULL },
 };
 


@@ -175,6 +178,7 @@ static void usage(FILE *fp)
        fprintf(fp, "      --clock-gmt                Print clock in GMT time zone (default: local time zone)\n");
        fprintf(fp, "      --clock-force-correlate    Assume that clocks are inherently correlated\n");
        fprintf(fp, "                                 across traces.\n");
        fprintf(fp, "      --clock-gmt                Print clock in GMT time zone (default: local time zone)\n");
        fprintf(fp, "      --clock-force-correlate    Assume that clocks are inherently correlated\n");
        fprintf(fp, "                                 across traces.\n");

+       fprintf(fp, "      --stream-intersection      Only print events when all streams are active.\n");

        list_formats(fp);
        fprintf(fp, "\n");
 }
        list_formats(fp);
        fprintf(fp, "\n");
 }


@@ -394,6 +398,9 @@ static int parse_options(int argc, char **argv)
                case OPT_CLOCK_FORCE_CORRELATE:
                        opt_clock_force_correlate = 1;
                        break;
                case OPT_CLOCK_FORCE_CORRELATE:
                        opt_clock_force_correlate = 1;
                        break;

+               case OPT_STREAM_INTERSECTION:
+                       opt_stream_intersection = 1;
+                       break;

 
                default:
                        ret = -EINVAL;
 
                default:
                        ret = -EINVAL;


@@ -615,6 +622,39 @@ end:
        return ret;
 }
 
        return ret;
 }
 

+static
+struct bt_ctf_iter *iter_create_intersect(struct bt_context *ctx,
+               struct bt_iter_pos *inter_begin_pos,
+               struct bt_iter_pos *inter_end_pos)
+{
+       uint64_t begin = 0, end = ULLONG_MAX;
+       /* Useless but needed for bt_iter_create_time_pos. */
+       struct bt_iter bt_iter;
+       int ret;
+
+       ret = ctf_find_packets_intersection(ctx, &begin, &end);
+       if (ret == 1) {
+               fprintf(stderr, "[error] No intersection found between trace files.\n");
+               ret = -1;
+               goto error;
+       } else if (ret != 0) {
+               goto error;
+       }
+       inter_begin_pos = bt_iter_create_time_pos(&bt_iter, begin);
+       if (!inter_begin_pos) {
+               goto error;
+       }
+       inter_end_pos = bt_iter_create_time_pos(&bt_iter, end);
+       if (!inter_end_pos) {
+               goto error;
+       }
+
+       return bt_ctf_iter_create(ctx, inter_begin_pos,
+                       inter_end_pos);
+error:
+       return NULL;
+}
+

 static
 int convert_trace(struct bt_trace_descriptor *td_write,
                  struct bt_context *ctx)
 static
 int convert_trace(struct bt_trace_descriptor *td_write,
                  struct bt_context *ctx)


@@ -622,17 +662,23 @@ int convert_trace(struct bt_trace_descriptor *td_write,
        struct bt_ctf_iter *iter;
        struct ctf_text_stream_pos *sout;
        struct bt_iter_pos begin_pos;
        struct bt_ctf_iter *iter;
        struct ctf_text_stream_pos *sout;
        struct bt_iter_pos begin_pos;

+       struct bt_iter_pos *inter_begin_pos = NULL, *inter_end_pos = NULL;

        struct bt_ctf_event *ctf_event;
        int ret;
 
        sout = container_of(td_write, struct ctf_text_stream_pos,
                        trace_descriptor);
 
        struct bt_ctf_event *ctf_event;
        int ret;
 
        sout = container_of(td_write, struct ctf_text_stream_pos,
                        trace_descriptor);
 

-       if (!sout->parent.event_cb)
+       if (!sout->parent.event_cb) {

                return 0;
                return 0;

+       }

 
 

-       begin_pos.type = BT_SEEK_BEGIN;
-       iter = bt_ctf_iter_create(ctx, &begin_pos, NULL);
+       if (opt_stream_intersection) {
+               iter = iter_create_intersect(ctx, inter_begin_pos, inter_end_pos);
+       } else {
+               begin_pos.type = BT_SEEK_BEGIN;
+               iter = bt_ctf_iter_create(ctx, &begin_pos, NULL);
+       }

        if (!iter) {
                ret = -1;
                goto error_iter;
        if (!iter) {
                ret = -1;
                goto error_iter;


@@ -644,14 +690,21 @@ int convert_trace(struct bt_trace_descriptor *td_write,
                        goto end;
                }
                ret = bt_iter_next(bt_ctf_get_iter(iter));
                        goto end;
                }
                ret = bt_iter_next(bt_ctf_get_iter(iter));

-               if (ret < 0)
+               if (ret < 0) {

                        goto end;
                        goto end;

+               }

        }
        ret = 0;
 
 end:
        bt_ctf_iter_destroy(iter);
 error_iter:
        }
        ret = 0;
 
 end:
        bt_ctf_iter_destroy(iter);
 error_iter:

+       if (inter_begin_pos) {
+               bt_iter_free_pos(inter_begin_pos);
+       }
+       if (inter_end_pos) {
+               bt_iter_free_pos(inter_end_pos);
+       }

        return ret;
 }
 
        return ret;
 }
 

diff --git a/formats/ctf/ctf.c b/formats/ctf/ctf.c
index da9047e37e817e96183b846c86790ba9dd7762d0..4dcec9032e413d10565beb64abad5993fdbade2f 100644 (file)
--- a/formats/ctf/ctf.c
+++ b/formats/ctf/ctf.c
@@ -869,6 +869,104 @@ void ctf_update_current_packet_index(struct ctf_stream_definition *stream,
        stream->packets_lost = packets_lost_diff;
 }
 
        stream->packets_lost = packets_lost_diff;
 }
 

+/*
+ * Find the timerange where all the streams in the trace are active
+ * simultaneously.
+ *
+ * Return 0 and update begin/end if necessary on success, return 1 for
+ * empty streams and return a negative value on error.
+ */
+static
+int ctf_intersect_trace(struct bt_trace_descriptor *td_read,
+               uint64_t *begin, uint64_t *end)
+{
+       struct ctf_trace *tin;
+       int stream_id, ret = 0;
+
+       tin = container_of(td_read, struct ctf_trace, parent);
+
+       for (stream_id = 0; stream_id < tin->streams->len;
+                       stream_id++) {
+               int filenr;
+               struct ctf_stream_declaration *stream_class;
+
+               stream_class = g_ptr_array_index(tin->streams, stream_id);
+               if (!stream_class) {
+                       continue;
+               }
+               for (filenr = 0; filenr < stream_class->streams->len; filenr++) {
+                       struct ctf_file_stream *file_stream;
+                       struct ctf_stream_pos *stream_pos;
+                       struct packet_index *index;
+
+                       file_stream = g_ptr_array_index(stream_class->streams,
+                                       filenr);
+                       if (!file_stream) {
+                               continue;
+                       }
+                       stream_pos = &file_stream->pos;
+                       if (!stream_pos->packet_index ||
+                                       stream_pos->packet_index->len <= 0) {
+                               ret = 1;
+                               goto end;
+                       }
+                       index = &g_array_index(stream_pos->packet_index,
+                                       struct packet_index, 0);
+                       if (index->ts_real.timestamp_begin > *begin) {
+                               *begin = index->ts_real.timestamp_begin;
+                       }
+                       index = &g_array_index(stream_pos->packet_index,
+                                       struct packet_index,
+                                       stream_pos->packet_index->len - 1);
+                       if (index->ts_real.timestamp_end < *end) {
+                               *end = index->ts_real.timestamp_end;
+                       }
+               }
+       }
+
+end:
+       return ret;
+}
+
+/*
+ * Find the timerange where all streams in the trace collection are active
+ * simultaneously.
+ *
+ * Return 0 on success.
+ * Return 1 if no intersections are found.
+ * Return a negative value on error.
+ */
+int ctf_find_packets_intersection(struct bt_context *ctx,
+               uint64_t *ts_begin, uint64_t *ts_end)
+{
+       int ret, i;
+
+       if (!ctx || !ctx->tc || !ctx->tc->array) {
+               ret = -EINVAL;
+               goto end;
+       }
+
+       for (i = 0; i < ctx->tc->array->len; i++) {
+               struct bt_trace_descriptor *td_read;
+
+               td_read = g_ptr_array_index(ctx->tc->array, i);
+               if (!td_read) {
+                       continue;
+               }
+               ret = ctf_intersect_trace(td_read, ts_begin, ts_end);
+               if (ret) {
+                       goto end;
+               }
+       }
+       if (*ts_end < *ts_begin) {
+               ret = 1;
+       } else {
+               ret = 0;
+       }
+end:
+       return ret;
+}
+

 /*
  * for SEEK_CUR: go to next packet.
  * for SEEK_SET: go to packet numer (index).
 /*
  * for SEEK_CUR: go to next packet.
  * for SEEK_SET: go to packet numer (index).

diff --git a/include/babeltrace/babeltrace-internal.h b/include/babeltrace/babeltrace-internal.h
index c9345509009d73eda04ab3df7ac4739980fa4d71..b879544de0056d943f6f3f36b5c10c554199b261 100644 (file)
--- a/include/babeltrace/babeltrace-internal.h
+++ b/include/babeltrace/babeltrace-internal.h
@@ -191,7 +191,8 @@ extern int opt_all_field_names,
        opt_clock_seconds,
        opt_clock_date,
        opt_clock_gmt,
        opt_clock_seconds,
        opt_clock_date,
        opt_clock_gmt,

-       opt_clock_force_correlate;
+       opt_clock_force_correlate,
+       opt_stream_intersection;

 
 extern uint64_t opt_clock_offset;
 extern uint64_t opt_clock_offset_ns;
 
 extern uint64_t opt_clock_offset;
 extern uint64_t opt_clock_offset_ns;

diff --git a/include/babeltrace/ctf/events-internal.h b/include/babeltrace/ctf/events-internal.h
index 48caa54429fd70fe0f9299f82b0e1930f5ee03c5..f1e6df811e3e74c615e47906a18f53240d7b6400 100644 (file)
--- a/include/babeltrace/ctf/events-internal.h
+++ b/include/babeltrace/ctf/events-internal.h
@@ -79,6 +79,8 @@ struct bt_ctf_iter {
        uint64_t events_lost;
 };
 
        uint64_t events_lost;
 };
 

+int ctf_find_packets_intersection(struct bt_context *ctx,
+               uint64_t *ts_begin, uint64_t *ts_end);

 void ctf_update_current_packet_index(struct ctf_stream_definition *stream,
                struct packet_index *prev_index,
                struct packet_index *cur_index);
 void ctf_update_current_packet_index(struct ctf_stream_definition *stream,
                struct packet_index *prev_index,
                struct packet_index *cur_index);