X-Git-Url: http://git.efficios.com/?a=blobdiff_plain;f=src%2Fplugins%2Fctf%2Ffs-src%2Fdata-stream-file.c;h=fe14f8cc3622d99fb0688ee894ff15d74d936b47;hb=148f02cb367caf23f134c1cb805f669706448649;hp=64f7f2be57fbf2d5230ea69f8f4020934b12992d;hpb=5d452e1fef736bf4e80bc48d9d60c08c949ced36;p=babeltrace.git

diff --git a/src/plugins/ctf/fs-src/data-stream-file.c b/src/plugins/ctf/fs-src/data-stream-file.c
index 64f7f2be..fe14f8cc 100644
--- a/src/plugins/ctf/fs-src/data-stream-file.c
+++ b/src/plugins/ctf/fs-src/data-stream-file.c
@@ -22,8 +22,8 @@
  * SOFTWARE.
  */
 
-#define BT_COMP_LOG_SELF_COMP (ds_file->self_comp)
-#define BT_LOG_OUTPUT_LEVEL (ds_file->log_level)
+#define BT_COMP_LOG_SELF_COMP (self_comp)
+#define BT_LOG_OUTPUT_LEVEL (log_level)
 #define BT_LOG_TAG "PLUGIN/SRC.CTF.FS/DS"
 #include "logging/comp-logging.h"
 
@@ -47,16 +47,33 @@
 static inline
 size_t remaining_mmap_bytes(struct ctf_fs_ds_file *ds_file)
 {
-	BT_ASSERT_DBG(ds_file->mmap_len >= ds_file->request_offset);
-	return ds_file->mmap_len - ds_file->request_offset;
+	BT_ASSERT_DBG(ds_file->mmap_len >= ds_file->request_offset_in_mapping);
+	return ds_file->mmap_len - ds_file->request_offset_in_mapping;
 }
 
+/*
+ * Return true if `offset_in_file` is in the current mapping.
+ */
+
 static
-int ds_file_munmap(struct ctf_fs_ds_file *ds_file)
+bool offset_ist_mapped(struct ctf_fs_ds_file *ds_file, off_t offset_in_file)
 {
-	int ret = 0;
+	return offset_in_file >= ds_file->mmap_offset_in_file &&
+		offset_in_file < (ds_file->mmap_offset_in_file + ds_file->mmap_len);
+}
+
+static
+enum ctf_msg_iter_medium_status ds_file_munmap(
+		struct ctf_fs_ds_file *ds_file)
+{
+	enum ctf_msg_iter_medium_status status;
+	bt_self_component *self_comp = ds_file->self_comp;
+	bt_logging_level log_level = ds_file->log_level;
 
-	if (!ds_file || !ds_file->mmap_addr) {
+	BT_ASSERT(ds_file);
+
+	if (!ds_file->mmap_addr) {
+		status = CTF_MSG_ITER_MEDIUM_STATUS_OK;
 		goto end;
 	}
 
@@ -66,63 +83,125 @@ int ds_file_munmap(struct ctf_fs_ds_file *ds_file)
 			ds_file->mmap_addr, ds_file->mmap_len,
 			ds_file->file ? ds_file->file->path->str : "NULL",
 			ds_file->file ? ds_file->file->fp : NULL);
-		ret = -1;
+		status = CTF_MSG_ITER_MEDIUM_STATUS_ERROR;
 		goto end;
 	}
 
 	ds_file->mmap_addr = NULL;
 
+	status = CTF_MSG_ITER_MEDIUM_STATUS_OK;
 end:
-	return ret;
+	return status;
 }
 
+/*
+ * mmap a region of `ds_file` such that `requested_offset_in_file` is in the
+ * mapping.  If the currently mmap-ed region already contains
+ * `requested_offset_in_file`, the mapping is kept.
+ *
+ * Set `ds_file->requested_offset_in_mapping` based on `request_offset_in_file`,
+ * such that the next call to `request_bytes` will return bytes starting at that
+ * position.
+ *
+ * `requested_offset_in_file` must be a valid offset in the file.
+ */
 static
-enum ctf_msg_iter_medium_status ds_file_mmap_next(
-		struct ctf_fs_ds_file *ds_file)
+enum ctf_msg_iter_medium_status ds_file_mmap(
+		struct ctf_fs_ds_file *ds_file, off_t requested_offset_in_file)
 {
-	enum ctf_msg_iter_medium_status ret =
-			CTF_MSG_ITER_MEDIUM_STATUS_OK;
+	enum ctf_msg_iter_medium_status status;
+	bt_self_component *self_comp = ds_file->self_comp;
+	bt_logging_level log_level = ds_file->log_level;
 
-	/* Unmap old region */
-	if (ds_file->mmap_addr) {
-		if (ds_file_munmap(ds_file)) {
-			goto error;
-		}
+	/* Ensure the requested offset is in the file range. */
+	BT_ASSERT(requested_offset_in_file >= 0);
+	BT_ASSERT(requested_offset_in_file < ds_file->file->size);
 
-		/*
-		 * mmap_len is guaranteed to be page-aligned except on the
-		 * last mapping where it may not be possible (since the file's
-		 * size itself may not be a page multiple).
-		 */
-		ds_file->mmap_offset += ds_file->mmap_len;
-		ds_file->request_offset = 0;
+	/*
+	 * If the mapping already contains the requested offset, just adjust
+	 * requested_offset_in_mapping.
+	 */
+	if (offset_ist_mapped(ds_file, requested_offset_in_file)) {
+		ds_file->request_offset_in_mapping =
+			requested_offset_in_file - ds_file->mmap_offset_in_file;
+		status = CTF_MSG_ITER_MEDIUM_STATUS_OK;
+		goto end;
 	}
 
-	ds_file->mmap_len = MIN(ds_file->file->size - ds_file->mmap_offset,
-			ds_file->mmap_max_len);
-	if (ds_file->mmap_len == 0) {
-		ret = CTF_MSG_ITER_MEDIUM_STATUS_EOF;
+	/* Unmap old region */
+	status = ds_file_munmap(ds_file);
+	if (status != CTF_MSG_ITER_MEDIUM_STATUS_OK) {
 		goto end;
 	}
-	/* Map new region */
-	BT_ASSERT(ds_file->mmap_len);
+
+	/*
+	 * Compute a mapping that has the required alignment properties and
+	 * contains `requested_offset_in_file`.
+	 */
+	ds_file->request_offset_in_mapping =
+		requested_offset_in_file % bt_mmap_get_offset_align_size(ds_file->log_level);
+	ds_file->mmap_offset_in_file =
+		requested_offset_in_file - ds_file->request_offset_in_mapping;
+	ds_file->mmap_len = MIN(ds_file->file->size - ds_file->mmap_offset_in_file,
+		ds_file->mmap_max_len);
+
+	BT_ASSERT(ds_file->mmap_len > 0);
+
 	ds_file->mmap_addr = bt_mmap((void *) 0, ds_file->mmap_len,
 			PROT_READ, MAP_PRIVATE, fileno(ds_file->file->fp),
-			ds_file->mmap_offset, ds_file->log_level);
+			ds_file->mmap_offset_in_file, ds_file->log_level);
 	if (ds_file->mmap_addr == MAP_FAILED) {
 		BT_COMP_LOGE("Cannot memory-map address (size %zu) of file \"%s\" (%p) at offset %jd: %s",
 				ds_file->mmap_len, ds_file->file->path->str,
-				ds_file->file->fp, (intmax_t) ds_file->mmap_offset,
+				ds_file->file->fp, (intmax_t) ds_file->mmap_offset_in_file,
 				strerror(errno));
-		goto error;
+		status = CTF_MSG_ITER_MEDIUM_STATUS_ERROR;
+		goto end;
 	}
 
-	goto end;
-error:
-	ds_file_munmap(ds_file);
-	ret = CTF_MSG_ITER_MEDIUM_STATUS_ERROR;
+	status = CTF_MSG_ITER_MEDIUM_STATUS_OK;
+
 end:
-	return ret;
+	return status;
+}
+
+/*
+ * Change the mapping of the file to read the region that follows the current
+ * mapping.
+ *
+ * If the file hasn't been mapped yet, then everything (mmap_offset_in_file,
+ * mmap_len, request_offset_in_mapping) should have the value 0, which will
+ * result in the beginning of the file getting mapped.
+ *
+ * return _EOF if the current mapping is the end of the file.
+ */
+
+static
+enum ctf_msg_iter_medium_status ds_file_mmap_next(
+		struct ctf_fs_ds_file *ds_file)
+{
+	enum ctf_msg_iter_medium_status status;
+
+	/*
+	 * If we're called, it's because more bytes are requested but we have
+	 * given all the bytes of the current mapping.
+	 */
+	BT_ASSERT(ds_file->request_offset_in_mapping == ds_file->mmap_len);
+
+	/*
+	 * If the current mapping coincides with the end of the file, there is
+	 * no next mapping.
+	 */
+	if (ds_file->mmap_offset_in_file + ds_file->mmap_len == ds_file->file->size) {
+		status = CTF_MSG_ITER_MEDIUM_STATUS_EOF;
+		goto end;
+	}
+
+	status = ds_file_mmap(ds_file,
+		ds_file->mmap_offset_in_file + ds_file->mmap_len);
+
+end:
+	return status;
 }
 
 static
@@ -133,10 +212,10 @@ enum ctf_msg_iter_medium_status medop_request_bytes(
 	enum ctf_msg_iter_medium_status status =
 		CTF_MSG_ITER_MEDIUM_STATUS_OK;
 	struct ctf_fs_ds_file *ds_file = data;
+	bt_self_component *self_comp = ds_file->self_comp;
+	bt_logging_level log_level = ds_file->log_level;
 
-	if (request_sz == 0) {
-		goto end;
-	}
+	BT_ASSERT(request_sz > 0);
 
 	/*
 	 * Check if we have at least one memory-mapped byte left. If we don't,
@@ -144,7 +223,7 @@ enum ctf_msg_iter_medium_status medop_request_bytes(
 	 */
 	if (remaining_mmap_bytes(ds_file) == 0) {
 		/* Are we at the end of the file? */
-		if (ds_file->mmap_offset >= ds_file->file->size) {
+		if (ds_file->mmap_offset_in_file >= ds_file->file->size) {
 			BT_COMP_LOGD("Reached end of file \"%s\" (%p)",
 				ds_file->file->path->str, ds_file->file->fp);
 			status = CTF_MSG_ITER_MEDIUM_STATUS_EOF;
@@ -165,10 +244,13 @@ enum ctf_msg_iter_medium_status medop_request_bytes(
 		}
 	}
 
+	BT_ASSERT(remaining_mmap_bytes(ds_file) > 0);
 	*buffer_sz = MIN(remaining_mmap_bytes(ds_file), request_sz);
+
 	BT_ASSERT(ds_file->mmap_addr);
-	*buffer_addr = ((uint8_t *) ds_file->mmap_addr) + ds_file->request_offset;
-	ds_file->request_offset += *buffer_sz;
+	*buffer_addr = ((uint8_t *) ds_file->mmap_addr) + ds_file->request_offset_in_mapping;
+
+	ds_file->request_offset_in_mapping += *buffer_sz;
 	goto end;
 
 error:
@@ -204,74 +286,252 @@ end:
 }
 
 static
-enum ctf_msg_iter_medium_status medop_seek(enum ctf_msg_iter_seek_whence whence,
-		off_t offset, void *data)
+enum ctf_msg_iter_medium_status medop_seek(off_t offset, void *data)
 {
-	enum ctf_msg_iter_medium_status ret =
-			CTF_MSG_ITER_MEDIUM_STATUS_OK;
 	struct ctf_fs_ds_file *ds_file = data;
-	off_t offset_in_mapping, file_size = ds_file->file->size;
-
-	if (whence != CTF_MSG_ITER_SEEK_WHENCE_SET ||
-		offset < 0 || offset > file_size) {
-		BT_COMP_LOGE("Invalid medium seek request: whence=%d, offset=%jd, "
-				"file-size=%jd", (int) whence, (intmax_t) offset,
-				(intmax_t) file_size);
-		ret = CTF_MSG_ITER_MEDIUM_STATUS_INVAL;
-		goto end;
-	}
 
-	/* If there is no current mapping, map the right file directly. */
-	if (!ds_file->mmap_addr) {
-		goto map_requested_offset;
-	}
+	BT_ASSERT(offset >= 0);
+	BT_ASSERT(offset < ds_file->file->size);
+
+	return ds_file_mmap(ds_file, offset);
+}
+
+BT_HIDDEN
+struct ctf_msg_iter_medium_ops ctf_fs_ds_file_medops = {
+	.request_bytes = medop_request_bytes,
+	.borrow_stream = medop_borrow_stream,
+	.seek = medop_seek,
+};
+
+struct ctf_fs_ds_group_medops_data {
+	/* Weak, set once at creation time. */
+	struct ctf_fs_ds_file_group *ds_file_group;
 
 	/*
-	 * Determine whether or not the destination is contained within the
-	 * current mapping.
+	 * Index (as in element rank) of the index entry of ds_file_groups'
+	 * index we will read next (so, the one after the one we are reading
+	 * right now).
 	 */
-	if (offset < ds_file->mmap_offset ||
-			offset >= ds_file->mmap_offset + ds_file->mmap_len) {
-		int unmap_ret;
-		BT_COMP_LOGD("Medium seek request cannot be accomodated by the current "
-				"file mapping: offset=%jd, mmap-offset=%jd, "
-				"mmap-len=%zu", (intmax_t) offset, (intmax_t) ds_file->mmap_offset,
-				ds_file->mmap_len);
-		unmap_ret = ds_file_munmap(ds_file);
-		if (unmap_ret) {
-			ret = CTF_MSG_ITER_MEDIUM_STATUS_ERROR;
+	guint next_index_entry_index;
+
+	/*
+	 * File we are currently reading.  Changes whenever we switch to
+	 * reading another data file.
+	 *
+	 * Owned by this.
+	 */
+	struct ctf_fs_ds_file *file;
+
+	/* Weak, for context / logging / appending causes. */
+	bt_self_message_iterator *self_msg_iter;
+	bt_logging_level log_level;
+};
+
+static
+enum ctf_msg_iter_medium_status medop_group_request_bytes(
+		size_t request_sz,
+		uint8_t **buffer_addr,
+		size_t *buffer_sz,
+		void *void_data)
+{
+	struct ctf_fs_ds_group_medops_data *data = void_data;
+
+	/* Return bytes from the current file. */
+	return medop_request_bytes(request_sz, buffer_addr, buffer_sz, data->file);
+}
+
+static
+bt_stream *medop_group_borrow_stream(
+		bt_stream_class *stream_class,
+		int64_t stream_id,
+		void *void_data)
+{
+	struct ctf_fs_ds_group_medops_data *data = void_data;
+
+	return medop_borrow_stream(stream_class, stream_id, data->file);
+}
+
+/*
+ * Set `data->file` to prepare it to read the packet described
+ * by `index_entry`.
+ */
+
+static
+enum ctf_msg_iter_medium_status ctf_fs_ds_group_medops_set_file(
+		struct ctf_fs_ds_group_medops_data *data,
+		struct ctf_fs_ds_index_entry *index_entry,
+		bt_self_message_iterator *self_msg_iter,
+		bt_logging_level log_level)
+{
+	enum ctf_msg_iter_medium_status status;
+
+	BT_ASSERT(data);
+	BT_ASSERT(index_entry);
+
+	/* Check if that file is already the one mapped. */
+	if (!data->file || strcmp(index_entry->path, data->file->file->path->str) != 0) {
+		/* Destroy the previously used file. */
+		ctf_fs_ds_file_destroy(data->file);
+
+		/* Create the new file. */
+		data->file = ctf_fs_ds_file_create(
+			data->ds_file_group->ctf_fs_trace,
+			self_msg_iter,
+			data->ds_file_group->stream,
+			index_entry->path,
+			log_level);
+		if (!data->file) {
+			BT_MSG_ITER_LOGE_APPEND_CAUSE(self_msg_iter,
+				"failed to create ctf_fs_ds_file.");
+			status = CTF_MSG_ITER_MEDIUM_STATUS_ERROR;
 			goto end;
 		}
-		goto map_requested_offset;
-	} else {
-		ds_file->request_offset = offset - ds_file->mmap_offset;
-		goto test_end;
 	}
 
-map_requested_offset:
-	offset_in_mapping = offset %
-		bt_mmap_get_offset_align_size(ds_file->log_level);
+	/*
+	 * Ensure the right portion of the file will be returned on the next
+	 * request_bytes call.
+	 */
+	status = ds_file_mmap(data->file, index_entry->offset);
+	if (status != CTF_MSG_ITER_MEDIUM_STATUS_OK) {
+		goto end;
+	}
+
+	status = CTF_MSG_ITER_MEDIUM_STATUS_OK;
+
+end:
+	return status;
+}
 
-	ds_file->mmap_offset = offset - offset_in_mapping;
-	ds_file->request_offset = offset_in_mapping;
-	ret = ds_file_mmap_next(ds_file);
-	if (ret != CTF_MSG_ITER_MEDIUM_STATUS_OK) {
+static
+enum ctf_msg_iter_medium_status medop_group_switch_packet(void *void_data)
+{
+	struct ctf_fs_ds_group_medops_data *data = void_data;
+	struct ctf_fs_ds_index_entry *index_entry;
+	enum ctf_msg_iter_medium_status status;
+
+	/* If we have gone through all index entries, we are done. */
+	if (data->next_index_entry_index >=
+		data->ds_file_group->index->entries->len) {
+		status = CTF_MSG_ITER_MEDIUM_STATUS_EOF;
 		goto end;
 	}
 
-test_end:
-	ds_file->end_reached = (offset == file_size);
+	/*
+	 * Otherwise, look up the next index entry / packet and prepare it
+	 *  for reading.
+	 */
+	index_entry = g_ptr_array_index(
+		data->ds_file_group->index->entries,
+		data->next_index_entry_index);
+
+	status = ctf_fs_ds_group_medops_set_file(
+		data, index_entry, data->self_msg_iter, data->log_level);
+	if (status != CTF_MSG_ITER_MEDIUM_STATUS_OK) {
+		goto end;
+	}
+
+	data->next_index_entry_index++;
+
+	status = CTF_MSG_ITER_MEDIUM_STATUS_OK;
 end:
-	return ret;
+	return status;
 }
 
 BT_HIDDEN
-struct ctf_msg_iter_medium_ops ctf_fs_ds_file_medops = {
-	.request_bytes = medop_request_bytes,
-	.borrow_stream = medop_borrow_stream,
-	.seek = medop_seek,
+void ctf_fs_ds_group_medops_data_destroy(
+		struct ctf_fs_ds_group_medops_data *data)
+{
+	if (!data) {
+		goto end;
+	}
+
+	ctf_fs_ds_file_destroy(data->file);
+
+	g_free(data);
+
+end:
+	return;
+}
+
+enum ctf_msg_iter_medium_status ctf_fs_ds_group_medops_data_create(
+		struct ctf_fs_ds_file_group *ds_file_group,
+		bt_self_message_iterator *self_msg_iter,
+		bt_logging_level log_level,
+		struct ctf_fs_ds_group_medops_data **out)
+{
+	struct ctf_fs_ds_group_medops_data *data;
+	enum ctf_msg_iter_medium_status status;
+
+	BT_ASSERT(self_msg_iter);
+	BT_ASSERT(ds_file_group);
+	BT_ASSERT(ds_file_group->index);
+	BT_ASSERT(ds_file_group->index->entries->len > 0);
+
+	data = g_new0(struct ctf_fs_ds_group_medops_data, 1);
+	if (!data) {
+		BT_MSG_ITER_LOGE_APPEND_CAUSE(self_msg_iter,
+			"Failed to allocate a struct ctf_fs_ds_group_medops_data");
+		status = CTF_MSG_ITER_MEDIUM_STATUS_MEMORY_ERROR;
+		goto error;
+	}
+
+	data->ds_file_group = ds_file_group;
+	data->self_msg_iter = self_msg_iter;
+	data->log_level = log_level;
+
+	/*
+	 * No need to prepare the first file.  ctf_msg_iter will call
+	 * switch_packet before reading the first packet, it will be
+	 * done then.
+	 */
+
+	*out = data;
+	status = CTF_MSG_ITER_MEDIUM_STATUS_OK;
+	goto end;
+
+error:
+	ctf_fs_ds_group_medops_data_destroy(data);
+
+end:
+	return status;
+}
+
+void ctf_fs_ds_group_medops_data_reset(struct ctf_fs_ds_group_medops_data *data)
+{
+	data->next_index_entry_index = 0;
+}
+
+struct ctf_msg_iter_medium_ops ctf_fs_ds_group_medops = {
+	.request_bytes = medop_group_request_bytes,
+	.borrow_stream = medop_group_borrow_stream,
+	.switch_packet = medop_group_switch_packet,
+
+	/*
+	 * We don't support seeking using this medops.  It would probably be
+	 * possible, but it's not needed at the moment.
+	 */
+	.seek = NULL,
 };
 
+static
+struct ctf_fs_ds_index_entry *ctf_fs_ds_index_entry_create(
+		bt_self_component *self_comp, bt_logging_level log_level)
+{
+	struct ctf_fs_ds_index_entry *entry;
+
+	entry = g_new0(struct ctf_fs_ds_index_entry, 1);
+	if (!entry) {
+		BT_COMP_LOGE_APPEND_CAUSE(self_comp, "Failed to allocate a ctf_fs_ds_index_entry.");
+		goto end;
+	}
+
+	entry->packet_seq_num = UINT64_MAX;
+
+end:
+	return entry;
+}
+
 static
 int convert_cycles_to_ns(struct ctf_clock_class *clock_class,
 		uint64_t cycles, int64_t *ns)
@@ -284,7 +544,8 @@ int convert_cycles_to_ns(struct ctf_clock_class *clock_class,
 static
 struct ctf_fs_ds_index *build_index_from_idx_file(
 		struct ctf_fs_ds_file *ds_file,
-		struct ctf_fs_ds_file_info *file_info)
+		struct ctf_fs_ds_file_info *file_info,
+		struct ctf_msg_iter *msg_iter)
 {
 	int ret;
 	gchar *directory = NULL;
@@ -303,10 +564,13 @@ struct ctf_fs_ds_index *build_index_from_idx_file(
 	size_t i;
 	struct ctf_stream_class *sc;
 	struct ctf_msg_iter_packet_properties props;
+	uint32_t version_major, version_minor;
+	bt_self_component *self_comp = ds_file->self_comp;
+	bt_logging_level log_level = ds_file->log_level;
 
 	BT_COMP_LOGI("Building index from .idx file of stream file %s",
 			ds_file->file->path->str);
-	ret = ctf_msg_iter_get_packet_properties(ds_file->msg_iter, &props);
+	ret = ctf_msg_iter_get_packet_properties(msg_iter, &props);
 	if (ret) {
 		BT_COMP_LOGI_STR("Cannot read first packet's header and context fields.");
 		goto error;
@@ -373,6 +637,16 @@ struct ctf_fs_ds_index *build_index_from_idx_file(
 		goto error;
 	}
 
+	version_major = be32toh(header->index_major);
+	version_minor = be32toh(header->index_minor);
+	if (version_major != 1) {
+		BT_COMP_LOGW(
+			"Unknown LTTng trace index version: "
+			"major=%" PRIu32 ", minor=%" PRIu32,
+			version_major, version_minor);
+		goto error;
+	}
+
 	file_index_entry_size = be32toh(header->packet_index_len);
 	file_entry_count = (filesize - sizeof(*header)) / file_index_entry_size;
 	if ((filesize - sizeof(*header)) % file_index_entry_size) {
@@ -398,8 +672,11 @@ struct ctf_fs_ds_index *build_index_from_idx_file(
 			goto error;
 		}
 
-		index_entry = g_new0(struct ctf_fs_ds_index_entry, 1);
+		index_entry = ctf_fs_ds_index_entry_create(
+			ds_file->self_comp, ds_file->log_level);
 		if (!index_entry) {
+			BT_COMP_LOGE_APPEND_CAUSE(ds_file->self_comp,
+				"Failed to create a ctf_fs_ds_index_entry.");
 			goto error;
 		}
 
@@ -444,6 +721,10 @@ struct ctf_fs_ds_index *build_index_from_idx_file(
 			goto error;
 		}
 
+		if (version_minor >= 1) {
+			index_entry->packet_seq_num = be64toh(file_index->packet_seq_num);
+		}
+
 		total_packets_size += packet_size;
 		file_pos += file_index_entry_size;
 
@@ -487,6 +768,8 @@ int init_index_entry(struct ctf_fs_ds_index_entry *entry,
 {
 	int ret = 0;
 	struct ctf_stream_class *sc;
+	bt_self_component *self_comp = ds_file->self_comp;
+	bt_logging_level log_level = ds_file->log_level;
 
 	sc = ctf_trace_class_borrow_stream_class_by_id(ds_file->metadata->tc,
 		props->stream_class_id);
@@ -535,12 +818,15 @@ end:
 static
 struct ctf_fs_ds_index *build_index_from_stream_file(
 		struct ctf_fs_ds_file *ds_file,
-		struct ctf_fs_ds_file_info *file_info)
+		struct ctf_fs_ds_file_info *file_info,
+		struct ctf_msg_iter *msg_iter)
 {
 	int ret;
 	struct ctf_fs_ds_index *index = NULL;
 	enum ctf_msg_iter_status iter_status = CTF_MSG_ITER_STATUS_OK;
 	off_t current_packet_offset_bytes = 0;
+	bt_self_component *self_comp = ds_file->self_comp;
+	bt_logging_level log_level = ds_file->log_level;
 
 	BT_COMP_LOGI("Indexing stream file %s", ds_file->file->path->str);
 
@@ -565,14 +851,14 @@ struct ctf_fs_ds_index *build_index_from_stream_file(
 			break;
 		}
 
-		iter_status = ctf_msg_iter_seek(ds_file->msg_iter,
+		iter_status = ctf_msg_iter_seek(msg_iter,
 				current_packet_offset_bytes);
 		if (iter_status != CTF_MSG_ITER_STATUS_OK) {
 			goto error;
 		}
 
 		iter_status = ctf_msg_iter_get_packet_properties(
-			ds_file->msg_iter, &props);
+			msg_iter, &props);
 		if (iter_status != CTF_MSG_ITER_STATUS_OK) {
 			goto error;
 		}
@@ -596,9 +882,11 @@ struct ctf_fs_ds_index *build_index_from_stream_file(
 			goto error;
 		}
 
-		index_entry = g_new0(struct ctf_fs_ds_index_entry, 1);
+		index_entry = ctf_fs_ds_index_entry_create(
+			ds_file->self_comp, ds_file->log_level);
 		if (!index_entry) {
-			BT_COMP_LOGE_STR("Failed to allocate a new index entry.");
+			BT_COMP_LOGE_APPEND_CAUSE(ds_file->self_comp,
+				"Failed to create a ctf_fs_ds_index_entry.");
 			goto error;
 		}
 
@@ -633,8 +921,7 @@ error:
 BT_HIDDEN
 struct ctf_fs_ds_file *ctf_fs_ds_file_create(
 		struct ctf_fs_trace *ctf_fs_trace,
-		bt_self_message_iterator *pc_msg_iter,
-		struct ctf_msg_iter *msg_iter,
+		bt_self_message_iterator *self_msg_iter,
 		bt_stream *stream, const char *path,
 		bt_logging_level log_level)
 {
@@ -648,7 +935,7 @@ struct ctf_fs_ds_file *ctf_fs_ds_file_create(
 
 	ds_file->log_level = log_level;
 	ds_file->self_comp = ctf_fs_trace->self_comp;
-	ds_file->self_msg_iter = pc_msg_iter;
+	ds_file->self_msg_iter = self_msg_iter;
 	ds_file->file = ctf_fs_file_create(log_level, ds_file->self_comp);
 	if (!ds_file->file) {
 		goto error;
@@ -663,12 +950,6 @@ struct ctf_fs_ds_file *ctf_fs_ds_file_create(
 		goto error;
 	}
 
-	ds_file->msg_iter = msg_iter;
-	ctf_msg_iter_set_medops_data(ds_file->msg_iter, ds_file);
-	if (!ds_file->msg_iter) {
-		goto error;
-	}
-
 	ds_file->mmap_max_len = offset_align * 2048;
 
 	goto end;
@@ -685,18 +966,21 @@ end:
 BT_HIDDEN
 struct ctf_fs_ds_index *ctf_fs_ds_file_build_index(
 		struct ctf_fs_ds_file *ds_file,
-		struct ctf_fs_ds_file_info *file_info)
+		struct ctf_fs_ds_file_info *file_info,
+		struct ctf_msg_iter *msg_iter)
 {
 	struct ctf_fs_ds_index *index;
+	bt_self_component *self_comp = ds_file->self_comp;
+	bt_logging_level log_level = ds_file->log_level;
 
-	index = build_index_from_idx_file(ds_file, file_info);
+	index = build_index_from_idx_file(ds_file, file_info, msg_iter);
 	if (index) {
 		goto end;
 	}
 
 	BT_COMP_LOGI("Failed to build index from .index file; "
 		"falling back to stream indexing.");
-	index = build_index_from_stream_file(ds_file, file_info);
+	index = build_index_from_stream_file(ds_file, file_info, msg_iter);
 end:
 	return index;
 }
@@ -746,40 +1030,6 @@ void ctf_fs_ds_file_destroy(struct ctf_fs_ds_file *ds_file)
 	g_free(ds_file);
 }
 
-BT_HIDDEN
-bt_component_class_message_iterator_next_method_status ctf_fs_ds_file_next(
-		struct ctf_fs_ds_file *ds_file,
-		bt_message **msg)
-{
-	enum ctf_msg_iter_status msg_iter_status;
-	bt_component_class_message_iterator_next_method_status status;
-
-	msg_iter_status = ctf_msg_iter_get_next_message(
-		ds_file->msg_iter, ds_file->self_msg_iter, msg);
-
-	switch (msg_iter_status) {
-	case CTF_MSG_ITER_STATUS_EOF:
-		status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_END;
-		break;
-	case CTF_MSG_ITER_STATUS_OK:
-		status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK;
-		break;
-	case CTF_MSG_ITER_STATUS_AGAIN:
-		/*
-		 * Should not make it this far as this is
-		 * medium-specific; there is nothing for the user to do
-		 * and it should have been handled upstream.
-		 */
-		bt_common_abort();
-	case CTF_MSG_ITER_STATUS_INVAL:
-	case CTF_MSG_ITER_STATUS_ERROR:
-	default:
-		status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_ERROR;
-		break;
-	}
-	return status;
-}
-
 BT_HIDDEN
 void ctf_fs_ds_index_destroy(struct ctf_fs_ds_index *index)
 {