#include <inttypes.h>
#include <sys/mman.h>
#include <babeltrace/ctf-ir/stream.h>
-#include <babeltrace/plugin/notification/iterator.h>
+#include <babeltrace/graph/notification-iterator.h>
+#include <babeltrace/graph/notification-stream.h>
+#include <babeltrace/graph/notification-event.h>
+#include <babeltrace/graph/notification-packet.h>
#include "file.h"
#include "metadata.h"
#include "../common/notif-iter/notif-iter.h"
#define PRINT_PREFIX "ctf-fs-data-stream"
#include "print.h"
-static
+static inline
size_t remaining_mmap_bytes(struct ctf_fs_stream *stream)
{
return stream->mmap_valid_len - stream->request_offset;
}
static
-int mmap_next(struct ctf_fs_stream *stream)
+enum bt_ctf_notif_iter_medium_status mmap_next(struct ctf_fs_stream *stream)
{
- int ret = 0;
+ enum bt_ctf_notif_iter_medium_status ret =
+ BT_CTF_NOTIF_ITER_MEDIUM_STATUS_OK;
struct ctf_fs_component *ctf_fs = stream->file->ctf_fs;
/* Unmap old region */
stream->mmap_valid_len = MIN(stream->file->size - stream->mmap_offset,
stream->mmap_max_len);
+ if (stream->mmap_valid_len == 0) {
+ ret = BT_CTF_NOTIF_ITER_MEDIUM_STATUS_EOF;
+ goto end;
+ }
/* Round up to next page, assuming page size being a power of 2. */
stream->mmap_len = (stream->mmap_valid_len + ctf_fs->page_size - 1)
& ~(ctf_fs->page_size - 1);
/* Map new region */
+ assert(stream->mmap_len);
stream->mmap_addr = mmap((void *) 0, stream->mmap_len,
PROT_READ, MAP_PRIVATE, fileno(stream->file->fp),
stream->mmap_offset);
goto end;
error:
stream_munmap(stream);
- ret = -1;
+ ret = BT_CTF_NOTIF_ITER_MEDIUM_STATUS_ERROR;
end:
return ret;
}
/* Check if we have at least one memory-mapped byte left */
if (remaining_mmap_bytes(stream) == 0) {
/* Are we at the end of the file? */
- if (stream->request_offset >= stream->file->size) {
+ if (stream->mmap_offset >= stream->file->size) {
PDBG("Reached end of file \"%s\" (%p)\n",
stream->file->path->str, stream->file->fp);
status = BT_CTF_NOTIF_ITER_MEDIUM_STATUS_EOF;
goto end;
}
- if (mmap_next(stream)) {
+ status = mmap_next(stream);
+ switch (status) {
+ case BT_CTF_NOTIF_ITER_MEDIUM_STATUS_OK:
+ break;
+ case BT_CTF_NOTIF_ITER_MEDIUM_STATUS_EOF:
+ goto end;
+ default:
PERR("Cannot memory-map next region of file \"%s\" (%p)\n",
- stream->file->path->str, stream->file->fp);
+ stream->file->path->str,
+ stream->file->fp);
goto error;
}
}
index_file_path = g_build_filename(directory, "index",
index_basename->str, NULL);
mapped_file = g_mapped_file_new(index_file_path, FALSE, NULL);
+ if (!mapped_file) {
+ ret = -1;
+ goto end;
+ }
filesize = g_mapped_file_get_length(mapped_file);
if (filesize < sizeof(*header)) {
printf_error("Invalid LTTng trace index: file size < header size");
g_free(directory);
g_free(basename);
g_free(index_file_path);
- g_string_free(index_basename, TRUE);
- g_mapped_file_unref(mapped_file);
+ if (index_basename) {
+ g_string_free(index_basename, TRUE);
+ }
+ if (mapped_file) {
+ g_mapped_file_unref(mapped_file);
+ }
return ret;
invalid_index:
g_array_free(stream->index.entries, TRUE);
static
int build_index_from_stream(struct ctf_fs_stream *stream)
{
- int ret = 0;
-end:
- return ret;
+ return 0;
}
static
BT_HIDDEN
struct ctf_fs_stream *ctf_fs_stream_create(
- struct ctf_fs_component *ctf_fs, struct ctf_fs_file *file)
+ struct ctf_fs_component *ctf_fs, const char *path)
{
int ret;
struct ctf_fs_stream *stream = g_new0(struct ctf_fs_stream, 1);
goto error;
}
- stream->file = file;
+ stream->file = ctf_fs_file_create(ctf_fs);
+ if (!stream->file) {
+ goto error;
+ }
+
+ g_string_assign(stream->file->path, path);
+ ret = ctf_fs_file_open(ctf_fs, stream->file, "rb");
+ if (ret) {
+ goto error;
+ }
+
stream->notif_iter = bt_ctf_notif_iter_create(ctf_fs->metadata->trace,
- ctf_fs->page_size, medops, stream, ctf_fs->error_fp);
+ ctf_fs->cc_prio_map, ctf_fs->page_size, medops, stream,
+ ctf_fs->error_fp);
if (!stream->notif_iter) {
goto error;
}
goto end;
error:
/* Do not touch "borrowed" file. */
- stream->file = NULL;
ctf_fs_stream_destroy(stream);
stream = NULL;
end:
g_free(stream);
}
+
+BT_HIDDEN
+struct bt_notification_iterator_next_return ctf_fs_stream_next(
+ struct ctf_fs_stream *stream)
+{
+ enum bt_ctf_notif_iter_status notif_iter_status;
+ struct bt_notification_iterator_next_return ret = {
+ .status = BT_NOTIFICATION_ITERATOR_STATUS_ERROR,
+ .notification = NULL,
+ };
+
+ if (stream->end_reached) {
+ notif_iter_status = BT_CTF_NOTIF_ITER_STATUS_EOF;
+ goto translate_status;
+ }
+
+ notif_iter_status = bt_ctf_notif_iter_get_next_notification(stream->notif_iter,
+ &ret.notification);
+ if (notif_iter_status != BT_CTF_NOTIF_ITER_STATUS_OK &&
+ notif_iter_status != BT_CTF_NOTIF_ITER_STATUS_EOF) {
+ goto translate_status;
+ }
+
+ /* Should be handled in bt_ctf_notif_iter_get_next_notification. */
+ if (notif_iter_status == BT_CTF_NOTIF_ITER_STATUS_EOF) {
+ ret.notification = bt_notification_stream_end_create(
+ stream->stream);
+ if (!ret.notification) {
+ notif_iter_status = BT_CTF_NOTIF_ITER_STATUS_ERROR;
+ goto translate_status;
+ }
+
+ notif_iter_status = BT_CTF_NOTIF_ITER_STATUS_OK;
+ stream->end_reached = true;
+ }
+
+translate_status:
+ switch (notif_iter_status) {
+ case BT_CTF_NOTIF_ITER_STATUS_EOF:
+ ret.status = BT_NOTIFICATION_ITERATOR_STATUS_END;
+ break;
+ case BT_CTF_NOTIF_ITER_STATUS_OK:
+ ret.status = BT_NOTIFICATION_ITERATOR_STATUS_OK;
+ break;
+ case BT_CTF_NOTIF_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.
+ */
+ assert(false);
+ case BT_CTF_NOTIF_ITER_STATUS_INVAL:
+ case BT_CTF_NOTIF_ITER_STATUS_ERROR:
+ default:
+ ret.status = BT_NOTIFICATION_ITERATOR_STATUS_ERROR;
+ break;
+ }
+
+ return ret;
+}