X-Git-Url: http://git.efficios.com/?a=blobdiff_plain;f=plugins%2Fctf%2Ffs-src%2Ffs.c;h=32121b7360ec51117e15b93c7941d802018df5c5;hb=8f7b565cb5c6fcbbcb4066c941efce27fd30c017;hp=07aa8cf36f9d724b0ab02dcc3b903533c501b4f4;hpb=a0a4aa52bea4880e5f98ec26ffc2aa122a53c217;p=babeltrace.git

diff --git a/plugins/ctf/fs-src/fs.c b/plugins/ctf/fs-src/fs.c
index 07aa8cf3..32121b73 100644
--- a/plugins/ctf/fs-src/fs.c
+++ b/plugins/ctf/fs-src/fs.c
@@ -47,6 +47,8 @@
 #include "data-stream-file.h"
 #include "file.h"
 #include "../common/metadata/decoder.h"
+#include "../common/notif-iter/notif-iter.h"
+#include "../common/utils/utils.h"
 #include "query.h"
 
 #define BT_LOG_TAG "PLUGIN-CTF-FS-SRC"
@@ -67,6 +69,7 @@ int notif_iter_data_set_current_ds_file(struct ctf_fs_notif_iter_data *notif_ite
 	ctf_fs_ds_file_destroy(notif_iter_data->ds_file);
 	notif_iter_data->ds_file = ctf_fs_ds_file_create(
 		notif_iter_data->ds_file_group->ctf_fs_trace,
+		notif_iter_data->notif_iter,
 		notif_iter_data->ds_file_group->stream,
 		ds_file_info->path->str);
 	if (!notif_iter_data->ds_file) {
@@ -85,6 +88,11 @@ void ctf_fs_notif_iter_data_destroy(
 	}
 
 	ctf_fs_ds_file_destroy(notif_iter_data->ds_file);
+
+	if (notif_iter_data->notif_iter) {
+		bt_ctf_notif_iter_destroy(notif_iter_data->notif_iter);
+	}
+
 	g_free(notif_iter_data);
 }
 
@@ -168,6 +176,16 @@ enum bt_notification_iterator_status ctf_fs_iterator_init(
 		goto error;
 	}
 
+	notif_iter_data->notif_iter = bt_ctf_notif_iter_create(
+		port_data->ds_file_group->ctf_fs_trace->metadata->trace,
+		bt_common_get_page_size() * 8,
+		ctf_fs_ds_file_medops, NULL);
+	if (!notif_iter_data->notif_iter) {
+		BT_LOGE_STR("Cannot create a CTF notification iterator.");
+		ret = BT_NOTIFICATION_ITERATOR_STATUS_NOMEM;
+		goto error;
+	}
+
 	notif_iter_data->ds_file_group = port_data->ds_file_group;
 	iret = notif_iter_data_set_current_ds_file(notif_iter_data);
 	if (iret) {
@@ -176,7 +194,7 @@ enum bt_notification_iterator_status ctf_fs_iterator_init(
 	}
 
 	ret = bt_private_notification_iterator_set_user_data(it, notif_iter_data);
-	if (ret) {
+	if (ret != BT_NOTIFICATION_ITERATOR_STATUS_OK) {
 		goto error;
 	}
 
@@ -186,10 +204,6 @@ enum bt_notification_iterator_status ctf_fs_iterator_init(
 error:
 	(void) bt_private_notification_iterator_set_user_data(it, NULL);
 
-	if (ret == BT_NOTIFICATION_ITERATOR_STATUS_OK) {
-		ret = BT_NOTIFICATION_ITERATOR_STATUS_ERROR;
-	}
-
 end:
 	ctf_fs_notif_iter_data_destroy(notif_iter_data);
 	return ret;
@@ -266,6 +280,31 @@ void port_data_destroy(void *data) {
 	g_free(port_data);
 }
 
+static
+GString *get_stream_instance_unique_name(
+		struct ctf_fs_ds_file_group *ds_file_group)
+{
+	GString *name;
+	struct ctf_fs_ds_file_info *ds_file_info;
+
+	name = g_string_new(NULL);
+	if (!name) {
+		goto end;
+	}
+
+	/*
+	 * If there's more than one stream file in the stream file
+	 * group, the first (earliest) stream file's path is used as
+	 * the stream's unique name.
+	 */
+	assert(ds_file_group->ds_file_infos->len > 0);
+	ds_file_info = g_ptr_array_index(ds_file_group->ds_file_infos, 0);
+	g_string_assign(name, ds_file_info->path->str);
+
+end:
+	return name;
+}
+
 static
 int create_one_port_for_trace(struct ctf_fs_component *ctf_fs,
 		struct ctf_fs_trace *ctf_fs_trace,
@@ -274,22 +313,12 @@ int create_one_port_for_trace(struct ctf_fs_component *ctf_fs,
 	int ret = 0;
 	struct ctf_fs_port_data *port_data = NULL;
 	GString *port_name = NULL;
-	struct ctf_fs_ds_file_info *ds_file_info =
-		g_ptr_array_index(ds_file_group->ds_file_infos, 0);
 
-	port_name = g_string_new(NULL);
+	port_name = get_stream_instance_unique_name(ds_file_group);
 	if (!port_name) {
 		goto error;
 	}
 
-	/*
-	 * Assign the name for the new output port. If there's more than
-	 * one stream file in the stream file group, the first
-	 * (earliest) stream file's path is used.
-	 */
-	assert(ds_file_group->ds_file_infos->len > 0);
-	ds_file_info = g_ptr_array_index(ds_file_group->ds_file_infos, 0);
-	g_string_assign(port_name, ds_file_info->path->str);
 	BT_LOGD("Creating one port named `%s`", port_name->str);
 
 	/* Create output port for this file */
@@ -345,6 +374,7 @@ end:
 	return ret;
 }
 
+static
 uint64_t get_packet_header_stream_instance_id(struct ctf_fs_trace *ctf_fs_trace,
 		struct bt_ctf_field *packet_header_field)
 {
@@ -374,51 +404,6 @@ end:
 	return stream_instance_id;
 }
 
-struct bt_ctf_stream_class *stream_class_from_packet_header(
-		struct ctf_fs_trace *ctf_fs_trace,
-		struct bt_ctf_field *packet_header_field)
-{
-	struct bt_ctf_field *stream_id_field = NULL;
-	struct bt_ctf_stream_class *stream_class = NULL;
-	uint64_t stream_id = -1ULL;
-	int ret;
-
-	if (!packet_header_field) {
-		goto single_stream_class;
-	}
-
-	stream_id_field = bt_ctf_field_structure_get_field_by_name(
-		packet_header_field, "stream_id");
-	if (!stream_id_field) {
-		goto single_stream_class;
-	}
-
-	ret = bt_ctf_field_unsigned_integer_get_value(stream_id_field,
-		&stream_id);
-	if (ret) {
-		stream_id = -1ULL;
-	}
-
-	if (stream_id == -1ULL) {
-single_stream_class:
-		/* Single stream class */
-		if (bt_ctf_trace_get_stream_class_count(
-				ctf_fs_trace->metadata->trace) == 0) {
-			goto end;
-		}
-
-		stream_class = bt_ctf_trace_get_stream_class_by_index(
-			ctf_fs_trace->metadata->trace, 0);
-	} else {
-		stream_class = bt_ctf_trace_get_stream_class_by_id(
-			ctf_fs_trace->metadata->trace, stream_id);
-	}
-
-end:
-	bt_put(stream_id_field);
-	return stream_class;
-}
-
 uint64_t get_packet_context_timestamp_begin_ns(
 		struct ctf_fs_trace *ctf_fs_trace,
 		struct bt_ctf_field *packet_context_field)
@@ -532,6 +517,7 @@ void ctf_fs_ds_file_group_destroy(struct ctf_fs_ds_file_group *ds_file_group)
 	}
 
 	bt_put(ds_file_group->stream);
+	bt_put(ds_file_group->stream_class);
 	g_free(ds_file_group);
 }
 
@@ -543,7 +529,6 @@ struct ctf_fs_ds_file_group *ctf_fs_ds_file_group_create(
 {
 	struct ctf_fs_ds_file_group *ds_file_group;
 
-	assert(stream_class);
 	ds_file_group = g_new0(struct ctf_fs_ds_file_group, 1);
 	if (!ds_file_group) {
 		goto error;
@@ -555,20 +540,10 @@ struct ctf_fs_ds_file_group *ctf_fs_ds_file_group_create(
 		goto error;
 	}
 
-	if (stream_instance_id == -1ULL) {
-		ds_file_group->stream = bt_ctf_stream_create(
-			stream_class, NULL);
-	} else {
-		ds_file_group->stream = bt_ctf_stream_create_with_id(
-			stream_class, NULL, stream_instance_id);
-	}
-
-	if (!ds_file_group->stream) {
-		goto error;
-	}
-
+	ds_file_group->stream_id = stream_instance_id;
+	assert(stream_class);
+	ds_file_group->stream_class = bt_get(stream_class);
 	ds_file_group->ctf_fs_trace = ctf_fs_trace;
-
 	goto end;
 
 error:
@@ -579,6 +554,26 @@ end:
 	return ds_file_group;
 }
 
+/* Replace by g_ptr_array_insert when we depend on glib >= 2.40. */
+static
+void array_insert(GPtrArray *array, gpointer element, size_t pos)
+{
+	size_t original_array_len = array->len;
+
+	/* Allocate an unused element at the end of the array. */
+	g_ptr_array_add(array, NULL);
+
+	/* If we are not inserting at the end, move the elements by one. */
+	if (pos < original_array_len) {
+		memmove(&(array->pdata[pos + 1]),
+			&(array->pdata[pos]),
+			(original_array_len - pos) * sizeof(gpointer));
+	}
+
+	/* Insert the value and bump the array len */
+	array->pdata[pos] = element;
+}
+
 static
 int ctf_fs_ds_file_group_add_ds_file_info(
 		struct ctf_fs_ds_file_group *ds_file_group,
@@ -606,12 +601,7 @@ int ctf_fs_ds_file_group_add_ds_file_info(
 		}
 	}
 
-	if (i == ds_file_group->ds_file_infos->len) {
-		/* Append instead */
-		i = -1;
-	}
-
-	g_ptr_array_insert(ds_file_group->ds_file_infos, i, ds_file_info);
+	array_insert(ds_file_group->ds_file_infos, ds_file_info, i);
 	ds_file_info = NULL;
 	goto end;
 
@@ -636,10 +626,18 @@ int add_ds_file_to_ds_file_group(struct ctf_fs_trace *ctf_fs_trace,
 	bool add_group = false;
 	int ret;
 	size_t i;
-	struct ctf_fs_ds_file *ds_file;
+	struct ctf_fs_ds_file *ds_file = NULL;
 	struct ctf_fs_ds_index *index = NULL;
+	struct bt_ctf_notif_iter *notif_iter = NULL;
 
-	ds_file = ctf_fs_ds_file_create(ctf_fs_trace, NULL, path);
+	notif_iter = bt_ctf_notif_iter_create(ctf_fs_trace->metadata->trace,
+		bt_common_get_page_size() * 8, ctf_fs_ds_file_medops, NULL);
+	if (!notif_iter) {
+		BT_LOGE_STR("Cannot create a CTF notification iterator.");
+		goto error;
+	}
+
+	ds_file = ctf_fs_ds_file_create(ctf_fs_trace, notif_iter, NULL, path);
 	if (!ds_file) {
 		goto error;
 	}
@@ -656,8 +654,8 @@ int add_ds_file_to_ds_file_group(struct ctf_fs_trace *ctf_fs_trace,
 		packet_header_field);
 	begin_ns = get_packet_context_timestamp_begin_ns(ctf_fs_trace,
 		packet_context_field);
-	stream_class = stream_class_from_packet_header(ctf_fs_trace,
-		packet_header_field);
+	stream_class = ctf_utils_stream_class_from_packet_header(
+		ctf_fs_trace->metadata->trace, packet_header_field);
 	if (!stream_class) {
 		goto error;
 	}
@@ -685,7 +683,6 @@ int add_ds_file_to_ds_file_group(struct ctf_fs_trace *ctf_fs_trace,
 		 * there's no timestamp to order the file within its
 		 * group.
 		 */
-
 		ds_file_group = ctf_fs_ds_file_group_create(ctf_fs_trace,
 			stream_class, stream_instance_id);
 		if (!ds_file_group) {
@@ -693,7 +690,7 @@ int add_ds_file_to_ds_file_group(struct ctf_fs_trace *ctf_fs_trace,
 		}
 
 		ret = ctf_fs_ds_file_group_add_ds_file_info(ds_file_group,
-				path, begin_ns, index);
+			path, begin_ns, index);
 		/* Ownership of index is transferred. */
 		index = NULL;
 		if (ret) {
@@ -709,20 +706,12 @@ int add_ds_file_to_ds_file_group(struct ctf_fs_trace *ctf_fs_trace,
 
 	/* Find an existing stream file group with this ID */
 	for (i = 0; i < ctf_fs_trace->ds_file_groups->len; i++) {
-		int64_t id;
-		struct bt_ctf_stream_class *cand_stream_class;
-
 		ds_file_group = g_ptr_array_index(
 			ctf_fs_trace->ds_file_groups, i);
-		id = bt_ctf_stream_get_id(ds_file_group->stream);
-		cand_stream_class = bt_ctf_stream_get_class(
-			ds_file_group->stream);
-
-		assert(cand_stream_class);
-		bt_put(cand_stream_class);
 
-		if (cand_stream_class == stream_class &&
-				(uint64_t) id == stream_instance_id) {
+		if (ds_file_group->stream_class == stream_class &&
+				ds_file_group->stream_id ==
+				stream_instance_id) {
 			break;
 		}
 
@@ -739,8 +728,8 @@ int add_ds_file_to_ds_file_group(struct ctf_fs_trace *ctf_fs_trace,
 		add_group = true;
 	}
 
-	ret = ctf_fs_ds_file_group_add_ds_file_info(ds_file_group,
-			path, begin_ns, index);
+	ret = ctf_fs_ds_file_group_add_ds_file_info(ds_file_group, path,
+		begin_ns, index);
 	index = NULL;
 	if (ret) {
 		goto error;
@@ -756,7 +745,13 @@ end:
 	if (add_group && ds_file_group) {
 		g_ptr_array_add(ctf_fs_trace->ds_file_groups, ds_file_group);
 	}
+
 	ctf_fs_ds_file_destroy(ds_file);
+
+	if (notif_iter) {
+		bt_ctf_notif_iter_destroy(notif_iter);
+	}
+
 	ctf_fs_ds_index_destroy(index);
 	bt_put(packet_header_field);
 	bt_put(packet_context_field);
@@ -771,6 +766,7 @@ int create_ds_file_groups(struct ctf_fs_trace *ctf_fs_trace)
 	const char *basename;
 	GError *error = NULL;
 	GDir *dir = NULL;
+	size_t i;
 
 	/* Check each file in the path directory, except specific ones */
 	dir = g_dir_open(ctf_fs_trace->path->str, 0, &error);
@@ -841,6 +837,43 @@ int create_ds_file_groups(struct ctf_fs_trace *ctf_fs_trace)
 		ctf_fs_file_destroy(file);
 	}
 
+	/*
+	 * At this point, DS file groupes are created, but their
+	 * associated stream objects do not exist yet. This is because
+	 * we need to name the created stream object with the data
+	 * stream file's path. We have everything we need here to do
+	 * this.
+	 */
+	for (i = 0; i < ctf_fs_trace->ds_file_groups->len; i++) {
+		struct ctf_fs_ds_file_group *ds_file_group =
+			g_ptr_array_index(ctf_fs_trace->ds_file_groups, i);
+		GString *name = get_stream_instance_unique_name(ds_file_group);
+
+		if (!name) {
+			goto error;
+		}
+
+		if (ds_file_group->stream_id == -1ULL) {
+			/* No stream ID */
+			ds_file_group->stream = bt_ctf_stream_create(
+				ds_file_group->stream_class, name->str);
+		} else {
+			/* Specific stream ID */
+			ds_file_group->stream = bt_ctf_stream_create_with_id(
+				ds_file_group->stream_class, name->str,
+				ds_file_group->stream_id);
+		}
+
+		g_string_free(name, TRUE);
+
+		if (!ds_file_group->stream) {
+			BT_LOGE("Cannot create stream for DS file group: "
+				"addr=%p, stream-name=\"%s\"",
+				ds_file_group, name->str);
+			goto error;
+		}
+	}
+
 	goto end;
 
 error:
@@ -1238,7 +1271,8 @@ struct ctf_fs_component *ctf_fs_create(struct bt_private_component *priv_comp,
 	struct ctf_fs_component *ctf_fs;
 	struct bt_value *value = NULL;
 	const char *path_param;
-	int ret;
+	enum bt_component_status ret;
+	enum bt_value_status value_ret;
 
 	ctf_fs = g_new0(struct ctf_fs_component, 1);
 	if (!ctf_fs) {
@@ -1246,7 +1280,7 @@ struct ctf_fs_component *ctf_fs_create(struct bt_private_component *priv_comp,
 	}
 
 	ret = bt_private_component_set_user_data(priv_comp, ctf_fs);
-	assert(ret == 0);
+	assert(ret == BT_COMPONENT_STATUS_OK);
 
 	/*
 	 * We don't need to get a new reference here because as long as
@@ -1259,8 +1293,8 @@ struct ctf_fs_component *ctf_fs_create(struct bt_private_component *priv_comp,
 		goto error;
 	}
 
-	ret = bt_value_string_get(value, &path_param);
-	assert(ret == 0);
+	value_ret = bt_value_string_get(value, &path_param);
+	assert(value_ret == BT_VALUE_STATUS_OK);
 	BT_PUT(value);
 	value = bt_value_map_get(params, "clock-class-offset-s");
 	if (value) {
@@ -1268,9 +1302,9 @@ struct ctf_fs_component *ctf_fs_create(struct bt_private_component *priv_comp,
 			BT_LOGE("clock-class-offset-s should be an integer");
 			goto error;
 		}
-		ret = bt_value_integer_get(value,
+		value_ret = bt_value_integer_get(value,
 			&ctf_fs->metadata_config.clock_class_offset_s);
-		assert(ret == 0);
+		assert(value_ret == BT_VALUE_STATUS_OK);
 		BT_PUT(value);
 	}
 
@@ -1280,9 +1314,9 @@ struct ctf_fs_component *ctf_fs_create(struct bt_private_component *priv_comp,
 			BT_LOGE("clock-class-offset-ns should be an integer");
 			goto error;
 		}
-		ret = bt_value_integer_get(value,
+		value_ret = bt_value_integer_get(value,
 			&ctf_fs->metadata_config.clock_class_offset_ns);
-		assert(ret == 0);
+		assert(value_ret == BT_VALUE_STATUS_OK);
 		BT_PUT(value);
 	}
 
@@ -1297,8 +1331,7 @@ struct ctf_fs_component *ctf_fs_create(struct bt_private_component *priv_comp,
 		goto error;
 	}
 
-	ret = create_ctf_fs_traces(ctf_fs, path_param);
-	if (ret) {
+	if (create_ctf_fs_traces(ctf_fs, path_param)) {
 		goto error;
 	}
 
@@ -1308,7 +1341,7 @@ error:
 	ctf_fs_destroy(ctf_fs);
 	ctf_fs = NULL;
 	ret = bt_private_component_set_user_data(priv_comp, NULL);
-	assert(ret == 0);
+	assert(ret == BT_COMPONENT_STATUS_OK);
 
 end:
 	bt_put(value);