X-Git-Url: http://git.efficios.com/?a=blobdiff_plain;f=src%2Fplugins%2Fctf%2Ffs-src%2Fdata-stream-file.c;h=b8091f660755c6524715fc92c71ed45a09a31ac5;hb=cad707e2548d7c9ea8501f12aec7ed28eaaf7975;hp=6588c13b19b55d44e81a94868c9c83d77f54156e;hpb=c4f23e30bf67d2523163614bc9461d84cbe1ae80;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 6588c13b..b8091f66 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"
 
@@ -55,6 +55,8 @@ static
 int ds_file_munmap(struct ctf_fs_ds_file *ds_file)
 {
 	int ret = 0;
+	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) {
 		goto end;
@@ -77,11 +79,13 @@ end:
 }
 
 static
-enum bt_msg_iter_medium_status ds_file_mmap_next(
+enum ctf_msg_iter_medium_status ds_file_mmap_next(
 		struct ctf_fs_ds_file *ds_file)
 {
-	enum bt_msg_iter_medium_status ret =
-			BT_MSG_ITER_MEDIUM_STATUS_OK;
+	enum ctf_msg_iter_medium_status ret =
+			CTF_MSG_ITER_MEDIUM_STATUS_OK;
+	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) {
@@ -101,7 +105,7 @@ enum bt_msg_iter_medium_status ds_file_mmap_next(
 	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 = BT_MSG_ITER_MEDIUM_STATUS_EOF;
+		ret = CTF_MSG_ITER_MEDIUM_STATUS_EOF;
 		goto end;
 	}
 	/* Map new region */
@@ -120,19 +124,21 @@ enum bt_msg_iter_medium_status ds_file_mmap_next(
 	goto end;
 error:
 	ds_file_munmap(ds_file);
-	ret = BT_MSG_ITER_MEDIUM_STATUS_ERROR;
+	ret = CTF_MSG_ITER_MEDIUM_STATUS_ERROR;
 end:
 	return ret;
 }
 
 static
-enum bt_msg_iter_medium_status medop_request_bytes(
+enum ctf_msg_iter_medium_status medop_request_bytes(
 		size_t request_sz, uint8_t **buffer_addr,
 		size_t *buffer_sz, void *data)
 {
-	enum bt_msg_iter_medium_status status =
-		BT_MSG_ITER_MEDIUM_STATUS_OK;
+	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;
@@ -147,15 +153,15 @@ enum bt_msg_iter_medium_status medop_request_bytes(
 		if (ds_file->mmap_offset >= ds_file->file->size) {
 			BT_COMP_LOGD("Reached end of file \"%s\" (%p)",
 				ds_file->file->path->str, ds_file->file->fp);
-			status = BT_MSG_ITER_MEDIUM_STATUS_EOF;
+			status = CTF_MSG_ITER_MEDIUM_STATUS_EOF;
 			goto end;
 		}
 
 		status = ds_file_mmap_next(ds_file);
 		switch (status) {
-		case BT_MSG_ITER_MEDIUM_STATUS_OK:
+		case CTF_MSG_ITER_MEDIUM_STATUS_OK:
 			break;
-		case BT_MSG_ITER_MEDIUM_STATUS_EOF:
+		case CTF_MSG_ITER_MEDIUM_STATUS_EOF:
 			goto end;
 		default:
 			BT_COMP_LOGE("Cannot memory-map next region of file \"%s\" (%p)",
@@ -172,7 +178,7 @@ enum bt_msg_iter_medium_status medop_request_bytes(
 	goto end;
 
 error:
-	status = BT_MSG_ITER_MEDIUM_STATUS_ERROR;
+	status = CTF_MSG_ITER_MEDIUM_STATUS_ERROR;
 
 end:
 	return status;
@@ -204,20 +210,22 @@ end:
 }
 
 static
-enum bt_msg_iter_medium_status medop_seek(enum bt_msg_iter_seek_whence whence,
+enum ctf_msg_iter_medium_status medop_seek(enum ctf_msg_iter_seek_whence whence,
 		off_t offset, void *data)
 {
-	enum bt_msg_iter_medium_status ret =
-			BT_MSG_ITER_MEDIUM_STATUS_OK;
+	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;
+	bt_self_component *self_comp = ds_file->self_comp;
+	bt_logging_level log_level = ds_file->log_level;
 
-	if (whence != BT_MSG_ITER_SEEK_WHENCE_SET ||
+	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 = BT_MSG_ITER_MEDIUM_STATUS_INVAL;
+		ret = CTF_MSG_ITER_MEDIUM_STATUS_INVAL;
 		goto end;
 	}
 
@@ -239,7 +247,7 @@ enum bt_msg_iter_medium_status medop_seek(enum bt_msg_iter_seek_whence whence,
 				ds_file->mmap_len);
 		unmap_ret = ds_file_munmap(ds_file);
 		if (unmap_ret) {
-			ret = BT_MSG_ITER_MEDIUM_STATUS_ERROR;
+			ret = CTF_MSG_ITER_MEDIUM_STATUS_ERROR;
 			goto end;
 		}
 		goto map_requested_offset;
@@ -255,7 +263,7 @@ map_requested_offset:
 	ds_file->mmap_offset = offset - offset_in_mapping;
 	ds_file->request_offset = offset_in_mapping;
 	ret = ds_file_mmap_next(ds_file);
-	if (ret != BT_MSG_ITER_MEDIUM_STATUS_OK) {
+	if (ret != CTF_MSG_ITER_MEDIUM_STATUS_OK) {
 		goto end;
 	}
 
@@ -266,12 +274,30 @@ end:
 }
 
 BT_HIDDEN
-struct bt_msg_iter_medium_ops ctf_fs_ds_file_medops = {
+struct ctf_msg_iter_medium_ops ctf_fs_ds_file_medops = {
 	.request_bytes = medop_request_bytes,
 	.borrow_stream = medop_borrow_stream,
 	.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 +310,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;
@@ -302,11 +329,14 @@ struct ctf_fs_ds_index *build_index_from_idx_file(
 	size_t file_entry_count;
 	size_t i;
 	struct ctf_stream_class *sc;
-	struct bt_msg_iter_packet_properties props;
+	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 = 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_LOGI_STR("Cannot read first packet's header and context fields.");
 		goto error;
@@ -373,6 +403,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 +438,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 +487,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;
 
@@ -482,11 +529,13 @@ error:
 static
 int init_index_entry(struct ctf_fs_ds_index_entry *entry,
 		struct ctf_fs_ds_file *ds_file,
-		struct bt_msg_iter_packet_properties *props,
+		struct ctf_msg_iter_packet_properties *props,
 		off_t packet_size, off_t packet_offset)
 {
 	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 +584,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 bt_msg_iter_status iter_status = BT_MSG_ITER_STATUS_OK;
+	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);
 
@@ -552,7 +604,7 @@ struct ctf_fs_ds_index *build_index_from_stream_file(
 	while (true) {
 		off_t current_packet_size_bytes;
 		struct ctf_fs_ds_index_entry *index_entry;
-		struct bt_msg_iter_packet_properties props;
+		struct ctf_msg_iter_packet_properties props;
 
 		if (current_packet_offset_bytes < 0) {
 			BT_COMP_LOGE_STR("Cannot get the current packet's offset.");
@@ -565,15 +617,15 @@ struct ctf_fs_ds_index *build_index_from_stream_file(
 			break;
 		}
 
-		iter_status = bt_msg_iter_seek(ds_file->msg_iter,
+		iter_status = ctf_msg_iter_seek(msg_iter,
 				current_packet_offset_bytes);
-		if (iter_status != BT_MSG_ITER_STATUS_OK) {
+		if (iter_status != CTF_MSG_ITER_STATUS_OK) {
 			goto error;
 		}
 
-		iter_status = bt_msg_iter_get_packet_properties(
-			ds_file->msg_iter, &props);
-		if (iter_status != BT_MSG_ITER_STATUS_OK) {
+		iter_status = ctf_msg_iter_get_packet_properties(
+			msg_iter, &props);
+		if (iter_status != CTF_MSG_ITER_STATUS_OK) {
 			goto error;
 		}
 
@@ -596,9 +648,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 +687,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 bt_msg_iter *msg_iter,
+		bt_self_message_iterator *self_msg_iter,
 		bt_stream *stream, const char *path,
 		bt_logging_level log_level)
 {
@@ -648,7 +701,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->pc_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 +716,6 @@ struct ctf_fs_ds_file *ctf_fs_ds_file_create(
 		goto error;
 	}
 
-	ds_file->msg_iter = msg_iter;
-	bt_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 +732,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;
 }
@@ -748,31 +798,30 @@ void ctf_fs_ds_file_destroy(struct ctf_fs_ds_file *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)
+		struct ctf_msg_iter *msg_iter,
+		const bt_message **msg)
 {
-	enum bt_msg_iter_status msg_iter_status;
+	enum ctf_msg_iter_status msg_iter_status;
 	bt_component_class_message_iterator_next_method_status status;
 
-	msg_iter_status = bt_msg_iter_get_next_message(
-		ds_file->msg_iter, ds_file->pc_msg_iter, msg);
+	msg_iter_status = ctf_msg_iter_get_next_message(msg_iter, msg);
 
 	switch (msg_iter_status) {
-	case BT_MSG_ITER_STATUS_EOF:
+	case CTF_MSG_ITER_STATUS_EOF:
 		status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_END;
 		break;
-	case BT_MSG_ITER_STATUS_OK:
+	case CTF_MSG_ITER_STATUS_OK:
 		status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK;
 		break;
-	case BT_MSG_ITER_STATUS_AGAIN:
+	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.
 		 */
-		abort();
-	case BT_MSG_ITER_STATUS_INVAL:
-	case BT_MSG_ITER_STATUS_ERROR:
+		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;