Test and fix packet-based metadata parsing
[babeltrace.git] / formats / ctf / ctf.c
index a418a6d507d4f1f33051495bdd3c95ada513f5ae..7c354bf0a40710a9eed836226d6bb273f94b2148 100644 (file)
 #include <babeltrace/ctf/metadata.h>
 #include <babeltrace/babeltrace.h>
 #include <inttypes.h>
+#include <stdio.h>
 #include <uuid/uuid.h>
 #include <sys/mman.h>
 #include <errno.h>
+#include <endian.h>
 #include <sys/types.h>
 #include <sys/stat.h>
 #include <fcntl.h>
 #define WRITE_PACKET_LEN       (getpagesize() * 8 * CHAR_BIT)
 #define UUID_LEN 16    /* uuid by value len */
 
+#ifndef min
+#define min(a, b)      (((a) < (b)) ? (a) : (b))
+#endif
+
 extern int yydebug;
 
 struct trace_descriptor *ctf_open_trace(const char *path, int flags);
@@ -79,6 +85,162 @@ struct format ctf_format = {
        .close_trace = ctf_close_trace,
 };
 
+static
+int ctf_read_event(struct stream_pos *ppos, struct ctf_stream *stream)
+{
+       struct ctf_stream_pos *pos =
+               container_of(ppos, struct ctf_stream_pos, parent);
+       struct ctf_stream_class *stream_class = stream->stream_class;
+       struct ctf_event *event_class;
+       uint64_t id = 0;
+       int len_index;
+       int ret;
+
+       if (pos->offset == EOF)
+               return EOF;
+
+       /* Read event header */
+       if (stream_class->event_header) {
+               ret = generic_rw(ppos, &stream_class->event_header->p);
+               if (ret)
+                       goto error;
+               /* lookup event id */
+               len_index = struct_declaration_lookup_field_index(stream_class->event_header_decl,
+                               g_quark_from_static_string("id"));
+               if (len_index >= 0) {
+                       struct definition_integer *defint;
+                       struct definition *field;
+
+                       field = struct_definition_get_field_from_index(stream_class->event_header, len_index);
+                       assert(field->declaration->id == CTF_TYPE_INTEGER);
+                       defint = container_of(field, struct definition_integer, p);
+                       assert(defint->declaration->signedness == FALSE);
+                       id = defint->value._unsigned;   /* set id */
+               }
+
+               /* lookup timestamp */
+               len_index = struct_declaration_lookup_field_index(stream_class->event_header_decl,
+                               g_quark_from_static_string("timestamp"));
+               if (len_index >= 0) {
+                       struct definition_integer *defint;
+                       struct definition *field;
+
+                       field = struct_definition_get_field_from_index(stream_class->event_header, len_index);
+                       assert(field->declaration->id == CTF_TYPE_INTEGER);
+                       defint = container_of(field, struct definition_integer, p);
+                       assert(defint->declaration->signedness == FALSE);
+                       /* update timestamp */
+                       stream->timestamp = defint->value._unsigned;
+               }
+
+       }
+
+       /* Read stream-declared event context */
+       if (stream_class->event_context) {
+               ret = generic_rw(ppos, &stream_class->event_context->p);
+               if (ret)
+                       goto error;
+       }
+
+       if (id >= stream_class->events_by_id->len) {
+               fprintf(stdout, "[error] Event id %" PRIu64 " is outside range.\n", id);
+               return -EINVAL;
+       }
+       event_class = g_ptr_array_index(stream_class->events_by_id, id);
+       if (!event_class) {
+               fprintf(stdout, "[error] Event id %" PRIu64 " is unknown.\n", id);
+               return -EINVAL;
+       }
+
+       /* Read event-declared event context */
+       if (event_class->context) {
+               ret = generic_rw(ppos, &event_class->context->p);
+               if (ret)
+                       goto error;
+       }
+
+       /* Read event payload */
+       if (event_class->fields) {
+               ret = generic_rw(ppos, &event_class->fields->p);
+               if (ret)
+                       goto error;
+       }
+
+       return 0;
+
+error:
+       fprintf(stdout, "[error] Unexpected end of stream. Either the trace data stream is corrupted or metadata description does not match data layout.\n");
+       return ret;
+}
+
+static
+int ctf_write_event(struct stream_pos *pos, struct ctf_stream *stream)
+{
+       struct ctf_stream_class *stream_class = stream->stream_class;
+       struct ctf_event *event_class;
+       uint64_t id = 0;
+       int len_index;
+       int ret;
+
+       /* print event header */
+       if (stream_class->event_header) {
+               /* lookup event id */
+               len_index = struct_declaration_lookup_field_index(stream_class->event_header_decl,
+                               g_quark_from_static_string("id"));
+               if (len_index >= 0) {
+                       struct definition_integer *defint;
+                       struct definition *field;
+
+                       field = struct_definition_get_field_from_index(stream_class->event_header, len_index);
+                       assert(field->declaration->id == CTF_TYPE_INTEGER);
+                       defint = container_of(field, struct definition_integer, p);
+                       assert(defint->declaration->signedness == FALSE);
+                       id = defint->value._unsigned;   /* set id */
+               }
+
+               ret = generic_rw(pos, &stream_class->event_header->p);
+               if (ret)
+                       goto error;
+       }
+
+       /* print stream-declared event context */
+       if (stream_class->event_context) {
+               ret = generic_rw(pos, &stream_class->event_context->p);
+               if (ret)
+                       goto error;
+       }
+
+       if (id >= stream_class->events_by_id->len) {
+               fprintf(stdout, "[error] Event id %" PRIu64 " is outside range.\n", id);
+               return -EINVAL;
+       }
+       event_class = g_ptr_array_index(stream_class->events_by_id, id);
+       if (!event_class) {
+               fprintf(stdout, "[error] Event id %" PRIu64 " is unknown.\n", id);
+               return -EINVAL;
+       }
+
+       /* print event-declared event context */
+       if (event_class->context) {
+               ret = generic_rw(pos, &event_class->context->p);
+               if (ret)
+                       goto error;
+       }
+
+       /* Read and print event payload */
+       if (event_class->fields) {
+               ret = generic_rw(pos, &event_class->fields->p);
+               if (ret)
+                       goto error;
+       }
+
+       return 0;
+
+error:
+       fprintf(stdout, "[error] Unexpected end of stream. Either the trace data stream is corrupted or metadata description does not match data layout.\n");
+       return ret;
+}
+
 void ctf_init_pos(struct ctf_stream_pos *pos, int fd, int open_flags)
 {
        pos->fd = fd;
@@ -100,12 +262,13 @@ void ctf_init_pos(struct ctf_stream_pos *pos, int fd, int open_flags)
                pos->prot = PROT_READ;
                pos->flags = MAP_PRIVATE;
                pos->parent.rw_table = read_dispatch_table;
+               pos->parent.event_cb = ctf_read_event;
                break;
-       case O_WRONLY:
        case O_RDWR:
                pos->prot = PROT_WRITE; /* Write has priority */
                pos->flags = MAP_SHARED;
                pos->parent.rw_table = write_dispatch_table;
+               pos->parent.event_cb = ctf_write_event;
                if (fd >= 0)
                        ctf_move_pos_slow(pos, 0, SEEK_SET);    /* position for write */
                break;
@@ -138,9 +301,6 @@ void ctf_move_pos_slow(struct ctf_stream_pos *pos, size_t offset, int whence)
        off_t off;
        struct packet_index *index;
 
-       /* Only allow random seek in read mode */
-       assert(pos->prot != PROT_WRITE || whence == SEEK_CUR);
-
        if (pos->prot == PROT_WRITE && pos->content_size_loc)
                *pos->content_size_loc = pos->offset;
 
@@ -160,18 +320,23 @@ void ctf_move_pos_slow(struct ctf_stream_pos *pos, size_t offset, int whence)
         * except to get exactly at the beginning of the next packet.
         */
        if (pos->prot == PROT_WRITE) {
-               /* The writer will add padding */
-               assert(pos->offset + offset == pos->packet_size);
-
-               /*
-                * Don't increment for initial stream move (only condition where
-                * pos->offset can be 0.
-                */
-               if (pos->offset)
+               switch (whence) {
+               case SEEK_CUR:
+                       /* The writer will add padding */
+                       assert(pos->offset + offset == pos->packet_size);
                        pos->mmap_offset += WRITE_PACKET_LEN / CHAR_BIT;
+                       break;
+               case SEEK_SET:
+                       assert(offset == 0);    /* only seek supported for now */
+                       pos->cur_index = 0;
+                       break;
+               default:
+                       assert(0);
+               }
                pos->content_size = -1U;        /* Unknown at this point */
                pos->packet_size = WRITE_PACKET_LEN;
-               off = posix_fallocate(pos->fd, pos->mmap_offset, pos->packet_size / CHAR_BIT);
+               off = posix_fallocate(pos->fd, pos->mmap_offset,
+                                     pos->packet_size / CHAR_BIT);
                assert(off >= 0);
                pos->offset = 0;
        } else {
@@ -189,7 +354,7 @@ void ctf_move_pos_slow(struct ctf_stream_pos *pos, size_t offset, int whence)
                        assert(0);
                }
                if (pos->cur_index >= pos->packet_index->len) {
-                       pos->offset = -EOF;
+                       pos->offset = EOF;
                        return;
                }
                index = &g_array_index(pos->packet_index, struct packet_index,
@@ -211,15 +376,141 @@ void ctf_move_pos_slow(struct ctf_stream_pos *pos, size_t offset, int whence)
        }
 }
 
-/*
- * TODO: for now, we treat the metadata file as a simple text file
- * (without any header nor packets nor padding).
- */
+static
+int packet_metadata(struct ctf_trace *td, FILE *fp)
+{
+       uint32_t magic;
+       size_t len;
+       int ret = 0;
+
+       len = fread(&magic, sizeof(magic), 1, fp);
+       if (len != 1) {
+               goto end;
+       }
+       if (magic == TSDL_MAGIC) {
+               ret = 1;
+               td->byte_order = BYTE_ORDER;
+       } else if (magic == GUINT32_SWAP_LE_BE(TSDL_MAGIC)) {
+               ret = 1;
+               td->byte_order = (BYTE_ORDER == BIG_ENDIAN) ?
+                                       LITTLE_ENDIAN : BIG_ENDIAN;
+       }
+       CTF_TRACE_SET_FIELD(td, byte_order);
+end:
+       rewind(fp);
+       return ret;
+}
+
+static
+int ctf_open_trace_metadata_packet_read(struct ctf_trace *td, FILE *in,
+                                       FILE *out)
+{
+       struct metadata_packet_header header;
+       size_t readlen, writelen, toread;
+       char buf[4096];
+       int ret = 0;
+
+       readlen = fread(&header, sizeof(header), 1, in);
+       if (readlen < 1)
+               return -EINVAL;
+
+       if (td->byte_order != BYTE_ORDER) {
+               header.magic = GUINT32_SWAP_LE_BE(header.magic);
+               header.checksum = GUINT32_SWAP_LE_BE(header.checksum);
+               header.content_size = GUINT32_SWAP_LE_BE(header.content_size);
+               header.packet_size = GUINT32_SWAP_LE_BE(header.packet_size);
+       }
+       if (header.checksum)
+               fprintf(stdout, "[warning] checksum verification not supported yet.\n");
+       if (header.compression_scheme) {
+               fprintf(stdout, "[error] compression (%u) not supported yet.\n",
+                       header.compression_scheme);
+               return -EINVAL;
+       }
+       if (header.encryption_scheme) {
+               fprintf(stdout, "[error] encryption (%u) not supported yet.\n",
+                       header.encryption_scheme);
+               return -EINVAL;
+       }
+       if (header.checksum_scheme) {
+               fprintf(stdout, "[error] checksum (%u) not supported yet.\n",
+                       header.checksum_scheme);
+               return -EINVAL;
+       }
+       if (!CTF_TRACE_FIELD_IS_SET(td, uuid)) {
+               memcpy(td->uuid, header.uuid, sizeof(header.uuid));
+               CTF_TRACE_SET_FIELD(td, uuid);
+       } else {
+               if (uuid_compare(header.uuid, td->uuid))
+                       return -EINVAL;
+       }
+
+       toread = header.content_size / CHAR_BIT;
+
+       for (;;) {
+               readlen = fread(buf, sizeof(char), min(sizeof(buf), toread), in);
+               if (ferror(in)) {
+                       ret = -EINVAL;
+                       break;
+               }
+               printf("read %s\n", buf);
+               writelen = fwrite(buf, sizeof(char), readlen, out);
+               if (writelen < readlen) {
+                       ret = -EIO;
+                       break;
+               }
+               if (ferror(out)) {
+                       ret = -EINVAL;
+                       break;
+               }
+               toread -= readlen;
+               if (!toread) {
+                       ret = -EOF;
+                       break;
+               }
+               if (feof(in)) {
+                       ret = -EINVAL;
+                       break;
+               }
+       }
+       return ret;
+}
+
+static
+int ctf_open_trace_metadata_stream_read(struct ctf_trace *td, FILE **fp,
+                                       char **buf)
+{
+       FILE *in, *out;
+       size_t size;
+       int ret;
+
+       in = *fp;
+       out = open_memstream(buf, &size);
+       if (out == NULL)
+               return -errno;
+
+       for (;;) {
+               ret = ctf_open_trace_metadata_packet_read(td, in, out);
+               if (ret == -EOF) {
+                       ret = 0;
+                       break;
+               } else if (ret) {
+                       break;
+               }
+       }
+       fclose(out);    /* flush the buffer */
+       fclose(in);
+       /* open for reading */
+       *fp = fmemopen(*buf, size, "rb");
+       return 0;
+}
+
 static
 int ctf_open_trace_metadata_read(struct ctf_trace *td)
 {
        struct ctf_scanner *scanner;
        FILE *fp;
+       char *buf = NULL;
        int ret = 0;
 
        td->metadata.pos.fd = openat(td->dirfd, "metadata", O_RDONLY);
@@ -238,6 +529,12 @@ int ctf_open_trace_metadata_read(struct ctf_trace *td)
                goto end_stream;
        }
 
+       if (packet_metadata(td, fp)) {
+               ret = ctf_open_trace_metadata_stream_read(td, &fp, &buf);
+               if (ret)
+                       goto end_packet_read;
+       }
+
        scanner = ctf_scanner_alloc(fp);
        if (!scanner) {
                fprintf(stdout, "[error] Error allocating scanner\n");
@@ -272,7 +569,9 @@ int ctf_open_trace_metadata_read(struct ctf_trace *td)
 end:
        ctf_scanner_free(scanner);
 end_scanner_alloc:
+end_packet_read:
        fclose(fp);
+       free(buf);
 end_stream:
        close(td->metadata.pos.fd);
        return ret;
@@ -283,7 +582,7 @@ static
 int create_stream_packet_index(struct ctf_trace *td,
                               struct ctf_file_stream *file_stream)
 {
-       struct ctf_stream *stream;
+       struct ctf_stream_class *stream;
        int len_index;
        struct ctf_stream_pos *pos;
        struct stat filestats;
@@ -330,14 +629,14 @@ int create_stream_packet_index(struct ctf_trace *td,
                        len_index = struct_declaration_lookup_field_index(td->packet_header->declaration, g_quark_from_static_string("magic"));
                        if (len_index >= 0) {
                                struct definition_integer *defint;
-                               struct field *field;
+                               struct definition *field;
 
                                field = struct_definition_get_field_from_index(td->packet_header, len_index);
-                               assert(field->definition->declaration->id == CTF_TYPE_INTEGER);
-                               defint = container_of(field->definition, struct definition_integer, p);
+                               assert(field->declaration->id == CTF_TYPE_INTEGER);
+                               defint = container_of(field, struct definition_integer, p);
                                assert(defint->declaration->signedness == FALSE);
                                if (defint->value._unsigned != CTF_MAGIC) {
-                                       fprintf(stdout, "[error] Invalid magic number %" PRIX64 " at packet %u (file offset %zd).\n",
+                                       fprintf(stdout, "[error] Invalid magic number 0x%" PRIX64 " at packet %u (file offset %zd).\n",
                                                        defint->value._unsigned,
                                                        file_stream->pos.packet_index->len,
                                                        (ssize_t) pos->mmap_offset);
@@ -346,16 +645,16 @@ int create_stream_packet_index(struct ctf_trace *td,
                        }
 
                        /* check uuid */
-                       len_index = struct_declaration_lookup_field_index(td->packet_header->declaration, g_quark_from_static_string("trace_uuid"));
+                       len_index = struct_declaration_lookup_field_index(td->packet_header->declaration, g_quark_from_static_string("uuid"));
                        if (len_index >= 0) {
                                struct definition_array *defarray;
-                               struct field *field;
+                               struct definition *field;
                                uint64_t i;
                                uint8_t uuidval[UUID_LEN];
 
                                field = struct_definition_get_field_from_index(td->packet_header, len_index);
-                               assert(field->definition->declaration->id == CTF_TYPE_ARRAY);
-                               defarray = container_of(field->definition, struct definition_array, p);
+                               assert(field->declaration->id == CTF_TYPE_ARRAY);
+                               defarray = container_of(field, struct definition_array, p);
                                assert(array_len(defarray) == UUID_LEN);
                                assert(defarray->declaration->elem->id == CTF_TYPE_INTEGER);
 
@@ -379,11 +678,11 @@ int create_stream_packet_index(struct ctf_trace *td,
                        len_index = struct_declaration_lookup_field_index(td->packet_header->declaration, g_quark_from_static_string("stream_id"));
                        if (len_index >= 0) {
                                struct definition_integer *defint;
-                               struct field *field;
+                               struct definition *field;
 
                                field = struct_definition_get_field_from_index(td->packet_header, len_index);
-                               assert(field->definition->declaration->id == CTF_TYPE_INTEGER);
-                               defint = container_of(field->definition, struct definition_integer, p);
+                               assert(field->declaration->id == CTF_TYPE_INTEGER);
+                               defint = container_of(field, struct definition_integer, p);
                                assert(defint->declaration->signedness == FALSE);
                                stream_id = defint->value._unsigned;
                        }
@@ -404,7 +703,7 @@ int create_stream_packet_index(struct ctf_trace *td,
                                fprintf(stdout, "[error] Stream %" PRIu64 " is not declared in metadata.\n", stream_id);
                                return -EINVAL;
                        }
-                       file_stream->stream = stream;
+                       file_stream->stream.stream_class = stream;
                }
                first_packet = 0;
 
@@ -417,11 +716,11 @@ int create_stream_packet_index(struct ctf_trace *td,
                        len_index = struct_declaration_lookup_field_index(stream->packet_context->declaration, g_quark_from_static_string("content_size"));
                        if (len_index >= 0) {
                                struct definition_integer *defint;
-                               struct field *field;
+                               struct definition *field;
 
                                field = struct_definition_get_field_from_index(stream->packet_context, len_index);
-                               assert(field->definition->declaration->id == CTF_TYPE_INTEGER);
-                               defint = container_of(field->definition, struct definition_integer, p);
+                               assert(field->declaration->id == CTF_TYPE_INTEGER);
+                               defint = container_of(field, struct definition_integer, p);
                                assert(defint->declaration->signedness == FALSE);
                                packet_index.content_size = defint->value._unsigned;
                        } else {
@@ -433,11 +732,11 @@ int create_stream_packet_index(struct ctf_trace *td,
                        len_index = struct_declaration_lookup_field_index(stream->packet_context->declaration, g_quark_from_static_string("packet_size"));
                        if (len_index >= 0) {
                                struct definition_integer *defint;
-                               struct field *field;
+                               struct definition *field;
 
                                field = struct_definition_get_field_from_index(stream->packet_context, len_index);
-                               assert(field->definition->declaration->id == CTF_TYPE_INTEGER);
-                               defint = container_of(field->definition, struct definition_integer, p);
+                               assert(field->declaration->id == CTF_TYPE_INTEGER);
+                               defint = container_of(field, struct definition_integer, p);
                                assert(defint->declaration->signedness == FALSE);
                                packet_index.packet_size = defint->value._unsigned;
                        } else {
@@ -498,7 +797,7 @@ int ctf_open_file_stream_read(struct ctf_trace *td, const char *path, int flags)
        if (ret)
                goto error_index;
        /* Add stream file to stream class */
-       g_ptr_array_add(file_stream->stream->files, file_stream);
+       g_ptr_array_add(file_stream->stream.stream_class->files, file_stream);
        return 0;
 
 error_index:
@@ -534,8 +833,6 @@ int ctf_open_trace_read(struct ctf_trace *td, const char *path, int flags)
                goto error_dirfd;
        }
 
-       td->streams = g_ptr_array_new();
-
        /*
         * Keep the metadata file separate.
         */
@@ -564,7 +861,8 @@ int ctf_open_trace_read(struct ctf_trace *td, const char *path, int flags)
                }
                if (!diriter)
                        break;
-               if (!strcmp(diriter->d_name, ".")
+               /* Ignore hidden files, ., .. and metadata. */
+               if (!strncmp(diriter->d_name, ".", 1)
                                || !strcmp(diriter->d_name, "..")
                                || !strcmp(diriter->d_name, "metadata"))
                        continue;
@@ -581,7 +879,6 @@ int ctf_open_trace_read(struct ctf_trace *td, const char *path, int flags)
 readdir_error:
        free(dirent);
 error_metadata:
-       g_ptr_array_free(td->streams, TRUE);
        close(td->dirfd);
 error_dirfd:
        closedir(td->dir);
@@ -606,7 +903,7 @@ struct trace_descriptor *ctf_open_trace(const char *path, int flags)
                if (ret)
                        goto error;
                break;
-       case O_WRONLY:
+       case O_RDWR:
                fprintf(stdout, "[error] Opening CTF traces for output is not supported yet.\n");
                goto error;
        default:
@@ -634,7 +931,7 @@ void ctf_close_trace(struct trace_descriptor *tdp)
 
        if (td->streams) {
                for (i = 0; i < td->streams->len; i++) {
-                       struct ctf_stream *stream;
+                       struct ctf_stream_class *stream;
                        int j;
                        stream = g_ptr_array_index(td->streams, i);
                        for (j = 0; j < stream->files->len; j++) {
This page took 0.03124 seconds and 4 git commands to generate.