X-Git-Url: http://drtracing.org/?a=blobdiff_plain;f=src%2Fplugins%2Fctf%2Ffs-src%2Fdata-stream-file.c;h=7fe4e6a52f1d2a12e6a359a39cc9bb28a0341629;hb=127e2341231d859921e076a37a70166e85e59596;hp=64f7f2be57fbf2d5230ea69f8f4020934b12992d;hpb=18a1979b5eae4e700e0e6716d83ca909f508d3e0;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..7fe4e6a5 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,123 @@ 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`
+ *
+ * `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 +210,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 +221,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 +242,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,65 +284,14 @@ 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;
-	}
-
-	/*
-	 * Determine whether or not the destination is contained within the
-	 * current mapping.
-	 */
-	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;
-			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);
+	BT_ASSERT(offset >= 0);
+	BT_ASSERT(offset < ds_file->file->size);
 
-	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) {
-		goto end;
-	}
-
-test_end:
-	ds_file->end_reached = (offset == file_size);
-end:
-	return ret;
+	return ds_file_mmap(ds_file, offset);
 }
 
 BT_HIDDEN
@@ -272,6 +301,24 @@ struct ctf_msg_iter_medium_ops ctf_fs_ds_file_medops = {
 	.seek = medop_seek,
 };
 
+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 +331,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 +351,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 +424,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 +459,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 +508,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 +555,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 +605,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 +638,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 +669,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 +708,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 +722,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 +737,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 +753,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 +817,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)
 {