X-Git-Url: http://git.efficios.com/?a=blobdiff_plain;f=src%2Fplugins%2Fctf%2Ffs-src%2Ffs.c;h=0530f0152df1e542219310230068049731853006;hb=6d54260a276ca4284cafe05d45e84f6d6cc7bce4;hp=08d2acfb9581286724199b56c430a6a40c58bd63;hpb=c0e46a7c2b7bdd588eb716191940bf5b34f9fc5b;p=babeltrace.git

diff --git a/src/plugins/ctf/fs-src/fs.c b/src/plugins/ctf/fs-src/fs.c
index 08d2acfb..0530f015 100644
--- a/src/plugins/ctf/fs-src/fs.c
+++ b/src/plugins/ctf/fs-src/fs.c
@@ -45,6 +45,7 @@
 #include "../common/metadata/ctf-meta-configure-ir-trace.h"
 #include "../common/msg-iter/msg-iter.h"
 #include "query.h"
+#include "plugins/common/param-validation/param-validation.h"
 
 struct tracer_info {
 	const char *name;
@@ -65,11 +66,13 @@ int msg_iter_data_set_current_ds_file(struct ctf_fs_msg_iter_data *msg_iter_data
 		msg_iter_data->ds_file_group->ds_file_infos,
 		msg_iter_data->ds_file_info_index);
 
+	/* Destroy the previous ds file. */
 	ctf_fs_ds_file_destroy(msg_iter_data->ds_file);
+
+	/* Create the new ds file. */
 	msg_iter_data->ds_file = ctf_fs_ds_file_create(
 		msg_iter_data->ds_file_group->ctf_fs_trace,
-		msg_iter_data->pc_msg_iter,
-		msg_iter_data->msg_iter,
+		msg_iter_data->self_msg_iter,
 		msg_iter_data->ds_file_group->stream,
 		ds_file_info->path->str,
 		msg_iter_data->log_level);
@@ -77,6 +80,10 @@ int msg_iter_data_set_current_ds_file(struct ctf_fs_msg_iter_data *msg_iter_data
 		ret = -1;
 	}
 
+	/* Tell the ctf message iterator to iterate on the new ds file. */
+	ctf_msg_iter_set_medops_data(msg_iter_data->msg_iter,
+		msg_iter_data->ds_file);
+
 	return ret;
 }
 
@@ -91,7 +98,7 @@ void ctf_fs_msg_iter_data_destroy(
 	ctf_fs_ds_file_destroy(msg_iter_data->ds_file);
 
 	if (msg_iter_data->msg_iter) {
-		bt_msg_iter_destroy(msg_iter_data->msg_iter);
+		ctf_msg_iter_destroy(msg_iter_data->msg_iter);
 	}
 
 	g_free(msg_iter_data);
@@ -101,11 +108,11 @@ static
 void set_msg_iter_emits_stream_beginning_end_messages(
 		struct ctf_fs_msg_iter_data *msg_iter_data)
 {
-	bt_msg_iter_set_emit_stream_beginning_message(
-		msg_iter_data->ds_file->msg_iter,
+	ctf_msg_iter_set_emit_stream_beginning_message(
+		msg_iter_data->msg_iter,
 		msg_iter_data->ds_file_info_index == 0);
-	bt_msg_iter_set_emit_stream_end_message(
-		msg_iter_data->ds_file->msg_iter,
+	ctf_msg_iter_set_emit_stream_end_message(
+		msg_iter_data->msg_iter,
 		msg_iter_data->ds_file_info_index ==
 			msg_iter_data->ds_file_group->ds_file_infos->len - 1);
 }
@@ -117,12 +124,12 @@ bt_component_class_message_iterator_next_method_status ctf_fs_iterator_next_one(
 {
 	bt_component_class_message_iterator_next_method_status status;
 
-	BT_ASSERT(msg_iter_data->ds_file);
+	BT_ASSERT_DBG(msg_iter_data->ds_file);
 
 	while (true) {
 		bt_message *msg;
 
-		status = ctf_fs_ds_file_next(msg_iter_data->ds_file, &msg);
+		status = ctf_fs_ds_file_next(msg_iter_data->msg_iter, &msg);
 		switch (status) {
 		case BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK:
 			*out_msg = msg;
@@ -139,7 +146,7 @@ bt_component_class_message_iterator_next_method_status ctf_fs_iterator_next_one(
 			}
 
 			msg_iter_data->ds_file_info_index++;
-			bt_msg_iter_reset_for_next_stream_file(
+			ctf_msg_iter_reset_for_next_stream_file(
 				msg_iter_data->msg_iter);
 			set_msg_iter_emits_stream_beginning_end_messages(
 				msg_iter_data);
@@ -172,19 +179,29 @@ bt_component_class_message_iterator_next_method_status ctf_fs_iterator_next(
 		bt_message_array_const msgs, uint64_t capacity,
 		uint64_t *count)
 {
-	bt_component_class_message_iterator_next_method_status status =
-		BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK;
+	bt_component_class_message_iterator_next_method_status status;
 	struct ctf_fs_msg_iter_data *msg_iter_data =
 		bt_self_message_iterator_get_data(iterator);
 	uint64_t i = 0;
 
-	while (i < capacity &&
-			status == BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK) {
+	if (G_UNLIKELY(msg_iter_data->next_saved_error)) {
+		/*
+		 * Last time we were called, we hit an error but had some
+		 * messages to deliver, so we stashed the error here.  Return
+		 * it now.
+		 */
+		BT_CURRENT_THREAD_MOVE_ERROR_AND_RESET(msg_iter_data->next_saved_error);
+		status = msg_iter_data->next_saved_status;
+		goto end;
+	}
+
+	do {
 		status = ctf_fs_iterator_next_one(msg_iter_data, &msgs[i]);
 		if (status == BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK) {
 			i++;
 		}
-	}
+	} while (i < capacity &&
+			status == BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK);
 
 	if (i > 0) {
 		/*
@@ -198,10 +215,23 @@ bt_component_class_message_iterator_next_method_status ctf_fs_iterator_next(
 		 * called, possibly without any accumulated
 		 * message, in which case we'll return it.
 		 */
+		if (status < 0) {
+			/*
+			 * Save this error for the next _next call.  Assume that
+			 * this component always appends error causes when
+			 * returning an error status code, which will cause the
+			 * current thread error to be non-NULL.
+			 */
+			msg_iter_data->next_saved_error = bt_current_thread_take_error();
+			BT_ASSERT(msg_iter_data->next_saved_error);
+			msg_iter_data->next_saved_status = status;
+		}
+
 		*count = i;
 		status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK;
 	}
 
+end:
 	return status;
 }
 
@@ -216,7 +246,7 @@ int ctf_fs_iterator_reset(struct ctf_fs_msg_iter_data *msg_iter_data)
 		goto end;
 	}
 
-	bt_msg_iter_reset(msg_iter_data->msg_iter);
+	ctf_msg_iter_reset(msg_iter_data->msg_iter);
 	set_msg_iter_emits_stream_beginning_end_messages(msg_iter_data);
 
 end:
@@ -275,12 +305,12 @@ bt_component_class_message_iterator_initialize_method_status ctf_fs_iterator_ini
 
 	msg_iter_data->log_level = log_level;
 	msg_iter_data->self_comp = self_comp;
-	msg_iter_data->pc_msg_iter = self_msg_iter;
-	msg_iter_data->msg_iter = bt_msg_iter_create(
+	msg_iter_data->self_msg_iter = self_msg_iter;
+	msg_iter_data->msg_iter = ctf_msg_iter_create(
 		port_data->ds_file_group->ctf_fs_trace->metadata->tc,
 		bt_common_get_page_size(msg_iter_data->log_level) * 8,
 		ctf_fs_ds_file_medops, NULL, msg_iter_data->log_level,
-		self_comp);
+		self_comp, self_msg_iter);
 	if (!msg_iter_data->msg_iter) {
 		BT_COMP_LOGE_APPEND_CAUSE(self_comp, "Cannot create a CTF message iterator.");
 		ret = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_INITIALIZE_METHOD_STATUS_MEMORY_ERROR;
@@ -677,23 +707,86 @@ void ds_file_group_insert_ds_file_info_sorted(
 }
 
 static
-void ds_file_group_insert_ds_index_entry_sorted(
-	struct ctf_fs_ds_file_group *ds_file_group,
+bool ds_index_entries_equal(
+	const struct ctf_fs_ds_index_entry *left,
+	const struct ctf_fs_ds_index_entry *right)
+{
+	if (left->packet_size != right->packet_size) {
+		return false;
+	}
+
+	if (left->timestamp_begin != right->timestamp_begin) {
+		return false;
+	}
+
+	if (left->timestamp_end != right->timestamp_end) {
+		return false;
+	}
+
+	if (left->packet_seq_num != right->packet_seq_num) {
+		return false;
+	}
+
+	return true;
+}
+
+/*
+ * Insert `entry` into `index`, without duplication.
+ *
+ * The entry is inserted only if there isn't an identical entry already.
+ *
+ * In any case, the ownership of `entry` is transferred to this function.  So if
+ * the entry is not inserted, it is freed.
+ */
+
+static
+void ds_index_insert_ds_index_entry_sorted(
+	struct ctf_fs_ds_index *index,
 	struct ctf_fs_ds_index_entry *entry)
 {
 	guint i;
+	struct ctf_fs_ds_index_entry *other_entry;
 
 	/* Find the spot where to insert this index entry. */
-	for (i = 0; i < ds_file_group->index->entries->len; i++) {
-		struct ctf_fs_ds_index_entry *other_entry = g_ptr_array_index(
-			ds_file_group->index->entries, i);
+	for (i = 0; i < index->entries->len; i++) {
+		other_entry = g_ptr_array_index(index->entries, i);
 
-		if (entry->timestamp_begin_ns < other_entry->timestamp_begin_ns) {
+		if (entry->timestamp_begin_ns <= other_entry->timestamp_begin_ns) {
 			break;
 		}
 	}
 
-	array_insert(ds_file_group->index->entries, entry, i);
+	/*
+	 * Insert the entry only if a duplicate doesn't already exist.
+	 *
+	 * There can be duplicate packets if reading multiple overlapping
+	 * snapshots of the same trace.  We then want the index to contain
+	 * a reference to only one copy of that packet.
+	 */
+	if (i == index->entries->len ||
+			!ds_index_entries_equal(entry, other_entry)) {
+		array_insert(index->entries, entry, i);
+	} else {
+		g_free(entry);
+	}
+}
+
+static
+void merge_ctf_fs_ds_indexes(struct ctf_fs_ds_index *dest, struct ctf_fs_ds_index *src)
+{
+	guint i;
+
+	for (i = 0; i < src->entries->len; i++) {
+		struct ctf_fs_ds_index_entry *entry =
+			g_ptr_array_index(src->entries, i);
+
+		/*
+		* Ownership of the ctf_fs_ds_index_entry is transferred to
+		* ds_index_insert_ds_index_entry_sorted.
+		*/
+		g_ptr_array_index(src->entries, i) = NULL;
+		ds_index_insert_ds_index_entry_sorted(dest, entry);
+	}
 }
 
 static
@@ -709,29 +802,35 @@ int add_ds_file_to_ds_file_group(struct ctf_fs_trace *ctf_fs_trace,
 	struct ctf_fs_ds_file *ds_file = NULL;
 	struct ctf_fs_ds_file_info *ds_file_info = NULL;
 	struct ctf_fs_ds_index *index = NULL;
-	struct bt_msg_iter *msg_iter = NULL;
+	struct ctf_msg_iter *msg_iter = NULL;
 	struct ctf_stream_class *sc = NULL;
-	struct bt_msg_iter_packet_properties props;
+	struct ctf_msg_iter_packet_properties props;
 	bt_logging_level log_level = ctf_fs_trace->log_level;
 	bt_self_component *self_comp = ctf_fs_trace->self_comp;
+	bt_self_component_class *self_comp_class = ctf_fs_trace->self_comp_class;
 
-	msg_iter = bt_msg_iter_create(ctf_fs_trace->metadata->tc,
-		bt_common_get_page_size(log_level) * 8,
-		ctf_fs_ds_file_medops, NULL, log_level, self_comp);
-	if (!msg_iter) {
-		BT_COMP_LOGE_STR("Cannot create a CTF message iterator.");
+	/*
+	 * Create a temporary ds_file to read some properties about the data
+	 * stream file.
+	 */
+	ds_file = ctf_fs_ds_file_create(ctf_fs_trace, NULL, NULL, path,
+		log_level);
+	if (!ds_file) {
 		goto error;
 	}
 
-	ds_file = ctf_fs_ds_file_create(ctf_fs_trace, NULL, msg_iter,
-		NULL, path, log_level);
-	if (!ds_file) {
+	/* Create a temporary iterator to read the ds_file. */
+	msg_iter = ctf_msg_iter_create(ctf_fs_trace->metadata->tc,
+		bt_common_get_page_size(log_level) * 8,
+		ctf_fs_ds_file_medops, ds_file, log_level, self_comp, NULL);
+	if (!msg_iter) {
+		BT_COMP_LOGE_STR("Cannot create a CTF message iterator.");
 		goto error;
 	}
 
-	ret = bt_msg_iter_get_packet_properties(ds_file->msg_iter, &props);
+	ret = ctf_msg_iter_get_packet_properties(msg_iter, &props);
 	if (ret) {
-		BT_COMP_LOGE_APPEND_CAUSE(self_comp,
+		BT_COMP_OR_COMP_CLASS_LOGE_APPEND_CAUSE(self_comp, self_comp_class,
 			"Cannot get stream file's first packet's header and context fields (`%s`).",
 			path);
 		goto error;
@@ -750,7 +849,7 @@ int add_ds_file_to_ds_file_group(struct ctf_fs_trace *ctf_fs_trace,
 			sc->default_clock_class->offset_seconds,
 			sc->default_clock_class->offset_cycles, &begin_ns);
 		if (ret) {
-			BT_COMP_LOGE_APPEND_CAUSE(self_comp,
+			BT_COMP_OR_COMP_CLASS_LOGE_APPEND_CAUSE(self_comp, self_comp_class,
 				"Cannot convert clock cycles to nanoseconds from origin (`%s`).",
 				path);
 			goto error;
@@ -762,10 +861,13 @@ int add_ds_file_to_ds_file_group(struct ctf_fs_trace *ctf_fs_trace,
 		goto error;
 	}
 
-	index = ctf_fs_ds_file_build_index(ds_file, ds_file_info);
+	index = ctf_fs_ds_file_build_index(ds_file, ds_file_info, msg_iter);
 	if (!index) {
-		BT_COMP_LOGW("Failed to index CTF stream file \'%s\'",
+		BT_COMP_OR_COMP_CLASS_LOGE_APPEND_CAUSE(
+			self_comp, self_comp_class,
+			"Failed to index CTF stream file \'%s\'",
 			ds_file->file->path->str);
+		goto error;
 	}
 
 	if (begin_ns == -1) {
@@ -794,7 +896,8 @@ int add_ds_file_to_ds_file_group(struct ctf_fs_trace *ctf_fs_trace,
 			goto error;
 		}
 
-		ds_file_group_insert_ds_file_info_sorted(ds_file_group, ds_file_info);
+		ds_file_group_insert_ds_file_info_sorted(ds_file_group,
+			BT_MOVE_REF(ds_file_info));
 
 		add_group = true;
 		goto end;
@@ -827,9 +930,12 @@ int add_ds_file_to_ds_file_group(struct ctf_fs_trace *ctf_fs_trace,
 		}
 
 		add_group = true;
+	} else {
+		merge_ctf_fs_ds_indexes(ds_file_group->index, index);
 	}
 
-	ds_file_group_insert_ds_file_info_sorted(ds_file_group, ds_file_info);
+	ds_file_group_insert_ds_file_info_sorted(ds_file_group,
+		BT_MOVE_REF(ds_file_info));
 
 	goto end;
 
@@ -844,9 +950,10 @@ end:
 	}
 
 	ctf_fs_ds_file_destroy(ds_file);
+	ctf_fs_ds_file_info_destroy(ds_file_info);
 
 	if (msg_iter) {
-		bt_msg_iter_destroy(msg_iter);
+		ctf_msg_iter_destroy(msg_iter);
 	}
 
 	ctf_fs_ds_index_destroy(index);
@@ -862,11 +969,13 @@ int create_ds_file_groups(struct ctf_fs_trace *ctf_fs_trace)
 	GDir *dir = NULL;
 	bt_logging_level log_level = ctf_fs_trace->log_level;
 	bt_self_component *self_comp = ctf_fs_trace->self_comp;
+	bt_self_component_class *self_comp_class = ctf_fs_trace->self_comp_class;
 
 	/* Check each file in the path directory, except specific ones */
 	dir = g_dir_open(ctf_fs_trace->path->str, 0, &error);
 	if (!dir) {
-		BT_COMP_LOGE_APPEND_CAUSE(self_comp, "Cannot open directory `%s`: %s (code %d)",
+		BT_COMP_OR_COMP_CLASS_LOGE_APPEND_CAUSE(self_comp, self_comp_class,
+			"Cannot open directory `%s`: %s (code %d)",
 			ctf_fs_trace->path->str, error->message,
 			error->code);
 		goto error;
@@ -891,7 +1000,7 @@ int create_ds_file_groups(struct ctf_fs_trace *ctf_fs_trace)
 		/* Create the file. */
 		file = ctf_fs_file_create(log_level, self_comp);
 		if (!file) {
-			BT_COMP_LOGE_APPEND_CAUSE(self_comp,
+			BT_COMP_OR_COMP_CLASS_LOGE_APPEND_CAUSE(self_comp, self_comp_class,
 				"Cannot create stream file object for file `%s" G_DIR_SEPARATOR_S "%s`",
 				ctf_fs_trace->path->str, basename);
 			goto error;
@@ -910,7 +1019,8 @@ int create_ds_file_groups(struct ctf_fs_trace *ctf_fs_trace)
 
 		ret = ctf_fs_file_open(file, "rb");
 		if (ret) {
-			BT_COMP_LOGE_APPEND_CAUSE(self_comp, "Cannot open stream file `%s`",
+			BT_COMP_OR_COMP_CLASS_LOGE_APPEND_CAUSE(self_comp, self_comp_class,
+				"Cannot open stream file `%s`",
 				file->path->str);
 			goto error;
 		}
@@ -925,7 +1035,7 @@ int create_ds_file_groups(struct ctf_fs_trace *ctf_fs_trace)
 		ret = add_ds_file_to_ds_file_group(ctf_fs_trace,
 			file->path->str);
 		if (ret) {
-			BT_COMP_LOGE_APPEND_CAUSE(self_comp,
+			BT_COMP_OR_COMP_CLASS_LOGE_APPEND_CAUSE(self_comp, self_comp_class,
 				"Cannot add stream file `%s` to stream file group",
 				file->path->str);
 			ctf_fs_file_destroy(file);
@@ -1002,7 +1112,9 @@ end:
 }
 
 static
-struct ctf_fs_trace *ctf_fs_trace_create(bt_self_component *self_comp,
+struct ctf_fs_trace *ctf_fs_trace_create(
+		bt_self_component *self_comp,
+		bt_self_component_class *self_comp_class,
 		const char *path, const char *name,
 		struct ctf_fs_metadata_config *metadata_config,
 		bt_logging_level log_level)
@@ -1010,6 +1122,9 @@ struct ctf_fs_trace *ctf_fs_trace_create(bt_self_component *self_comp,
 	struct ctf_fs_trace *ctf_fs_trace;
 	int ret;
 
+	/* Only one of them must be set. */
+	BT_ASSERT(!self_comp != !self_comp_class);
+
 	ctf_fs_trace = g_new0(struct ctf_fs_trace, 1);
 	if (!ctf_fs_trace) {
 		goto end;
@@ -1017,6 +1132,7 @@ struct ctf_fs_trace *ctf_fs_trace_create(bt_self_component *self_comp,
 
 	ctf_fs_trace->log_level = log_level;
 	ctf_fs_trace->self_comp = self_comp;
+	ctf_fs_trace->self_comp_class = self_comp_class;
 	ctf_fs_trace->path = g_string_new(path);
 	if (!ctf_fs_trace->path) {
 		goto error;
@@ -1119,18 +1235,21 @@ int ctf_fs_component_create_ctf_fs_trace_one_path(
 	norm_path = bt_common_normalize_path(path_param, NULL);
 	if (!norm_path) {
 		BT_COMP_OR_COMP_CLASS_LOGE_APPEND_CAUSE(self_comp, self_comp_class,
-			"Failed to normalize path: `%s`.", path_param);
+			"Failed to normalize path: `%s`.",
+			path_param);
 		goto error;
 	}
 
 	ret = path_is_ctf_trace(norm_path->str);
 	if (ret < 0) {
 		BT_COMP_OR_COMP_CLASS_LOGE_APPEND_CAUSE(self_comp, self_comp_class,
-			"Failed to check if path is a CTF trace: path=%s", norm_path->str);
+			"Failed to check if path is a CTF trace: path=%s",
+			norm_path->str);
 		goto error;
 	} else if (ret == 0) {
 		BT_COMP_OR_COMP_CLASS_LOGE_APPEND_CAUSE(self_comp, self_comp_class,
-			"Path is not a CTF trace (does not contain a metadata file): `%s`.", norm_path->str);
+			"Path is not a CTF trace (does not contain a metadata file): `%s`.",
+			norm_path->str);
 		goto error;
 	}
 
@@ -1142,10 +1261,11 @@ int ctf_fs_component_create_ctf_fs_trace_one_path(
 		goto end;
 	}
 
-	ctf_fs_trace = ctf_fs_trace_create(self_comp, norm_path->str,
+	ctf_fs_trace = ctf_fs_trace_create(self_comp, self_comp_class, norm_path->str,
 		trace_name, &ctf_fs->metadata_config, log_level);
 	if (!ctf_fs_trace) {
-		BT_COMP_LOGE_APPEND_CAUSE(self_comp, "Cannot create trace for `%s`.",
+		BT_COMP_OR_COMP_CLASS_LOGE_APPEND_CAUSE(self_comp, self_comp_class,
+			"Cannot create trace for `%s`.",
 			norm_path->str);
 		goto error;
 	}
@@ -1211,18 +1331,7 @@ void merge_ctf_fs_ds_file_groups(struct ctf_fs_ds_file_group *dest, struct ctf_f
 	}
 
 	/* Merge both indexes. */
-	for (i = 0; i < src->index->entries->len; i++) {
-		struct ctf_fs_ds_index_entry *entry = g_ptr_array_index(
-			src->index->entries, i);
-
-		/*
-		 * Ownership of the ctf_fs_ds_index_entry is transferred to
-		 * dest.
-		 */
-		g_ptr_array_index(src->index->entries, i) = NULL;
-
-		ds_file_group_insert_ds_index_entry_sorted(dest, entry);
-	}
+	merge_ctf_fs_ds_indexes(dest->index, src->index);
 }
 /* Merge src_trace's data stream file groups into dest_trace's. */
 
@@ -1398,30 +1507,18 @@ int decode_clock_snapshot_after_event(struct ctf_fs_trace *ctf_fs_trace,
 	struct ctf_fs_ds_index_entry *index_entry,
 	enum target_event target_event, uint64_t *cs, int64_t *ts_ns)
 {
-	enum bt_msg_iter_status iter_status = BT_MSG_ITER_STATUS_OK;
+	enum ctf_msg_iter_status iter_status = CTF_MSG_ITER_STATUS_OK;
 	struct ctf_fs_ds_file *ds_file = NULL;
-	struct bt_msg_iter *msg_iter = NULL;
+	struct ctf_msg_iter *msg_iter = NULL;
 	bt_logging_level log_level = ctf_fs_trace->log_level;
 	bt_self_component *self_comp = ctf_fs_trace->self_comp;
 	int ret = 0;
 
 	BT_ASSERT(ctf_fs_trace);
-	BT_ASSERT(ctf_fs_trace->metadata);
-	BT_ASSERT(ctf_fs_trace->metadata->tc);
-
-	msg_iter = bt_msg_iter_create(ctf_fs_trace->metadata->tc,
-		bt_common_get_page_size(log_level) * 8, ctf_fs_ds_file_medops,
-		NULL, log_level, self_comp);
-	if (!msg_iter) {
-		/* bt_msg_iter_create() logs errors. */
-		ret = -1;
-		goto end;
-	}
-
 	BT_ASSERT(index_entry);
 	BT_ASSERT(index_entry->path);
 
-	ds_file = ctf_fs_ds_file_create(ctf_fs_trace, NULL, msg_iter,
+	ds_file = ctf_fs_ds_file_create(ctf_fs_trace, NULL,
 		NULL, index_entry->path, log_level);
 	if (!ds_file) {
 		BT_COMP_LOGE_APPEND_CAUSE(self_comp, "Failed to create a ctf_fs_ds_file");
@@ -1429,16 +1526,28 @@ int decode_clock_snapshot_after_event(struct ctf_fs_trace *ctf_fs_trace,
 		goto end;
 	}
 
+	BT_ASSERT(ctf_fs_trace->metadata);
+	BT_ASSERT(ctf_fs_trace->metadata->tc);
+
+	msg_iter = ctf_msg_iter_create(ctf_fs_trace->metadata->tc,
+		bt_common_get_page_size(log_level) * 8, ctf_fs_ds_file_medops,
+		ds_file, log_level, self_comp, NULL);
+	if (!msg_iter) {
+		/* ctf_msg_iter_create() logs errors. */
+		ret = -1;
+		goto end;
+	}
+
 	/*
 	 * Turn on dry run mode to prevent the creation and usage of Babeltrace
 	 * library objects (bt_field, bt_message_*, etc.).
 	 */
-	bt_msg_iter_set_dry_run(msg_iter, true);
+	ctf_msg_iter_set_dry_run(msg_iter, true);
 
 	/* Seek to the beginning of the target packet. */
-	iter_status = bt_msg_iter_seek(ds_file->msg_iter, index_entry->offset);
+	iter_status = ctf_msg_iter_seek(msg_iter, index_entry->offset);
 	if (iter_status) {
-		/* bt_msg_iter_seek() logs errors. */
+		/* ctf_msg_iter_seek() logs errors. */
 		ret = -1;
 		goto end;
 	}
@@ -1450,16 +1559,16 @@ int decode_clock_snapshot_after_event(struct ctf_fs_trace *ctf_fs_trace,
 		 * the first event. To extract the first event's clock
 		 * snapshot.
 		 */
-		iter_status = bt_msg_iter_curr_packet_first_event_clock_snapshot(
-			ds_file->msg_iter, cs);
+		iter_status = ctf_msg_iter_curr_packet_first_event_clock_snapshot(
+			msg_iter, cs);
 		break;
 	case LAST_EVENT:
 		/* Decode the packet to extract the last event's clock snapshot. */
-		iter_status = bt_msg_iter_curr_packet_last_event_clock_snapshot(
-			ds_file->msg_iter, cs);
+		iter_status = ctf_msg_iter_curr_packet_last_event_clock_snapshot(
+			msg_iter, cs);
 		break;
 	default:
-		abort();
+		bt_common_abort();
 	}
 	if (iter_status) {
 		ret = -1;
@@ -1481,7 +1590,7 @@ end:
 		ctf_fs_ds_file_destroy(ds_file);
 	}
 	if (msg_iter) {
-		bt_msg_iter_destroy(msg_iter);
+		ctf_msg_iter_destroy(msg_iter);
 	}
 
 	return ret;
@@ -1918,7 +2027,8 @@ bool is_tracer_affected_by_lttng_crash_quirk(
  */
 static
 int fix_packet_index_tracer_bugs(struct ctf_fs_component *ctf_fs,
-		bt_self_component *self_comp)
+		bt_self_component *self_comp,
+		bt_self_component_class *self_comp_class)
 {
 	int ret = 0;
 	struct tracer_info current_tracer_info;
@@ -1944,7 +2054,8 @@ int fix_packet_index_tracer_bugs(struct ctf_fs_component *ctf_fs,
 		BT_LOGI_STR("Trace may be affected by LTTng tracer packet timestamp bug. Fixing up.");
 		ret = fix_index_lttng_event_after_packet_bug(ctf_fs->trace);
 		if (ret) {
-			BT_COMP_LOGE_APPEND_CAUSE(self_comp,
+			BT_COMP_OR_COMP_CLASS_LOGE_APPEND_CAUSE(
+				self_comp, self_comp_class,
 				"Failed to fix LTTng event-after-packet bug.");
 			goto end;
 		}
@@ -1956,7 +2067,8 @@ int fix_packet_index_tracer_bugs(struct ctf_fs_component *ctf_fs,
 		BT_LOGI_STR("Trace may be affected by barectf tracer packet timestamp bug. Fixing up.");
 		ret = fix_index_barectf_event_before_packet_bug(ctf_fs->trace);
 		if (ret) {
-			BT_COMP_LOGE_APPEND_CAUSE(self_comp,
+			BT_COMP_OR_COMP_CLASS_LOGE_APPEND_CAUSE(
+				self_comp, self_comp_class,
 				"Failed to fix barectf event-before-packet bug.");
 			goto end;
 		}
@@ -1967,7 +2079,8 @@ int fix_packet_index_tracer_bugs(struct ctf_fs_component *ctf_fs,
 			&current_tracer_info)) {
 		ret = fix_index_lttng_crash_quirk(ctf_fs->trace);
 		if (ret) {
-			BT_COMP_LOGE_APPEND_CAUSE(self_comp,
+			BT_COMP_OR_COMP_CLASS_LOGE_APPEND_CAUSE(
+				self_comp, self_comp_class,
 				"Failed to fix lttng-crash timestamp quirks.");
 			goto end;
 		}
@@ -2084,7 +2197,7 @@ int ctf_fs_component_create_ctf_fs_trace(
 		traces->pdata[0] = NULL;
 	}
 
-	ret = fix_packet_index_tracer_bugs(ctf_fs, self_comp);
+	ret = fix_packet_index_tracer_bugs(ctf_fs, self_comp, self_comp_class);
 	if (ret) {
 		BT_COMP_OR_COMP_CLASS_LOGE_APPEND_CAUSE(self_comp, self_comp_class,
 			"Failed to fix packet index tracer bugs.");
@@ -2214,63 +2327,26 @@ end:
 	return ret;
 }
 
-/*
- * Validate the "paths" parameter passed to this component.  It must be
- * present, and it must be an array of strings.
- */
-
-static
-bool validate_inputs_parameter(struct ctf_fs_component *ctf_fs,
-		const bt_value *inputs, bt_self_component *self_comp,
-		bt_self_component_class *self_comp_class)
-{
-	bool ret;
-	bt_value_type type;
-	uint64_t i;
-	bt_logging_level log_level = ctf_fs->log_level;
-
-	if (!inputs) {
-		BT_COMP_OR_COMP_CLASS_LOGE_APPEND_CAUSE(self_comp,
-			self_comp_class, "missing \"inputs\" parameter");
-		goto error;
-	}
-
-	type = bt_value_get_type(inputs);
-	if (type != BT_VALUE_TYPE_ARRAY) {
-		BT_COMP_OR_COMP_CLASS_LOGE_APPEND_CAUSE(self_comp,
-			self_comp_class, "`inputs` parameter: expecting array value: type=%s",
-			bt_common_value_type_string(type));
-		goto error;
-	}
-
-	if (bt_value_array_is_empty(inputs)) {
-		BT_COMP_OR_COMP_CLASS_LOGE_APPEND_CAUSE(self_comp,
-			self_comp_class, "`inputs` parameter must not be empty");
-		goto error;
-	}
-
-	for (i = 0; i < bt_value_array_get_length(inputs); i++) {
-		const bt_value *elem;
+static const struct bt_param_validation_value_descr inputs_elem_descr = {
+	.type = BT_VALUE_TYPE_STRING,
+};
 
-		elem = bt_value_array_borrow_element_by_index_const(inputs, i);
-		type = bt_value_get_type(elem);
-		if (type != BT_VALUE_TYPE_STRING) {
-			BT_COMP_OR_COMP_CLASS_LOGE_APPEND_CAUSE(self_comp, self_comp_class,
-				"`inputs` parameter: expecting string value: index=%" PRIu64 ", type=%s",
-				i, bt_common_value_type_string(type));
-			goto error;
+static const struct bt_param_validation_map_value_entry_descr fs_params_entries_descr[] = {
+	{ "inputs", BT_PARAM_VALIDATION_MAP_VALUE_ENTRY_MANDATORY, {
+		BT_VALUE_TYPE_ARRAY,
+		.array = {
+			.min_length = 1,
+			.max_length = BT_PARAM_VALIDATION_INFINITE,
+			.element_type = &inputs_elem_descr,
 		}
-	}
-
-	ret = true;
-	goto end;
-
-error:
-	ret = false;
+	}},
+	{ "trace-name", BT_PARAM_VALIDATION_MAP_VALUE_ENTRY_OPTIONAL, { .type = BT_VALUE_TYPE_STRING } },
+	{ "clock-class-offset-s", BT_PARAM_VALIDATION_MAP_VALUE_ENTRY_OPTIONAL, { .type = BT_VALUE_TYPE_SIGNED_INTEGER } },
+	{ "clock-class-offset-ns", BT_PARAM_VALIDATION_MAP_VALUE_ENTRY_OPTIONAL, { .type = BT_VALUE_TYPE_SIGNED_INTEGER } },
+	{ "force-clock-class-origin-unix-epoch", BT_PARAM_VALIDATION_MAP_VALUE_ENTRY_OPTIONAL, { .type = BT_VALUE_TYPE_BOOL } },
+	BT_PARAM_VALIDATION_MAP_VALUE_ENTRY_END
+};
 
-end:
-	return ret;
-}
 
 bool read_src_fs_parameters(const bt_value *params,
 		const bt_value **inputs,
@@ -2281,22 +2357,25 @@ bool read_src_fs_parameters(const bt_value *params,
 	bool ret;
 	const bt_value *value;
 	bt_logging_level log_level = ctf_fs->log_level;
+	enum bt_param_validation_status validate_value_status;
+	gchar *error = NULL;
+
+	validate_value_status = bt_param_validation_validate(params,
+		fs_params_entries_descr, &error);
+	if (validate_value_status != BT_PARAM_VALIDATION_STATUS_OK) {
+		BT_COMP_OR_COMP_CLASS_LOGE_APPEND_CAUSE(self_comp, self_comp_class,
+			"%s", error);
+		ret = false;
+		goto end;
+	}
 
 	/* inputs parameter */
 	*inputs = bt_value_map_borrow_entry_value_const(params, "inputs");
-	if (!validate_inputs_parameter(ctf_fs, *inputs, self_comp, self_comp_class)) {
-		goto error;
-	}
 
 	/* clock-class-offset-s parameter */
 	value = bt_value_map_borrow_entry_value_const(params,
 		"clock-class-offset-s");
 	if (value) {
-		if (!bt_value_is_signed_integer(value)) {
-			BT_COMP_OR_COMP_CLASS_LOGE_APPEND_CAUSE(self_comp, self_comp_class,
-				"clock-class-offset-s must be an integer");
-			goto error;
-		}
 		ctf_fs->metadata_config.clock_class_offset_s =
 			bt_value_integer_signed_get(value);
 	}
@@ -2305,11 +2384,6 @@ bool read_src_fs_parameters(const bt_value *params,
 	value = bt_value_map_borrow_entry_value_const(params,
 		"clock-class-offset-ns");
 	if (value) {
-		if (!bt_value_is_signed_integer(value)) {
-			BT_COMP_OR_COMP_CLASS_LOGE_APPEND_CAUSE(self_comp, self_comp_class,
-				"clock-class-offset-ns must be an integer");
-			goto error;
-		}
 		ctf_fs->metadata_config.clock_class_offset_ns =
 			bt_value_integer_signed_get(value);
 	}
@@ -2318,41 +2392,24 @@ bool read_src_fs_parameters(const bt_value *params,
 	value = bt_value_map_borrow_entry_value_const(params,
 		"force-clock-class-origin-unix-epoch");
 	if (value) {
-		if (!bt_value_is_bool(value)) {
-			BT_COMP_OR_COMP_CLASS_LOGE_APPEND_CAUSE(self_comp, self_comp_class,
-				"force-clock-class-origin-unix-epoch must be a boolean");
-			goto error;
-		}
-
 		ctf_fs->metadata_config.force_clock_class_origin_unix_epoch =
 			bt_value_bool_get(value);
 	}
 
 	/* trace-name parameter */
 	*trace_name = bt_value_map_borrow_entry_value_const(params, "trace-name");
-	if (*trace_name) {
-		if (!bt_value_is_string(*trace_name)) {
-			BT_COMP_OR_COMP_CLASS_LOGE_APPEND_CAUSE(self_comp, self_comp_class,
-				"trace-name must be a string");
-			goto error;
-		}
-	}
 
 	ret = true;
-	goto end;
-
-error:
-	ret = false;
 
 end:
+	g_free(error);
 	return ret;
 }
 
 static
 struct ctf_fs_component *ctf_fs_create(
 	const bt_value *params,
-	bt_self_component_source *self_comp_src,
-	bt_self_component_class *self_comp_class)
+	bt_self_component_source *self_comp_src)
 {
 	struct ctf_fs_component *ctf_fs = NULL;
 	const bt_value *inputs_value;
@@ -2367,14 +2424,14 @@ struct ctf_fs_component *ctf_fs_create(
 	}
 
 	if (!read_src_fs_parameters(params, &inputs_value, &trace_name_value,
-			ctf_fs, self_comp, self_comp_class)) {
+			ctf_fs, self_comp, NULL)) {
 		goto error;
 	}
 
 	bt_self_component_set_data(self_comp, ctf_fs);
 
 	if (ctf_fs_component_create_ctf_fs_trace(ctf_fs, inputs_value,
-			trace_name_value, self_comp, self_comp_class)) {
+			trace_name_value, self_comp, NULL)) {
 		goto error;
 	}
 
@@ -2407,7 +2464,7 @@ bt_component_class_initialize_method_status ctf_fs_init(
 	bt_component_class_initialize_method_status ret =
 		BT_COMPONENT_CLASS_INITIALIZE_METHOD_STATUS_OK;
 
-	ctf_fs = ctf_fs_create(params, self_comp_src, NULL);
+	ctf_fs = ctf_fs_create(params, self_comp_src);
 	if (!ctf_fs) {
 		ret = BT_COMPONENT_CLASS_INITIALIZE_METHOD_STATUS_ERROR;
 	}