move struct ctf_stream -> struct ctf_stream_class
[babeltrace.git] / formats / ctf / ctf.c
index d214d1aae4f25c0bd4d0c9b027187aa9b1113abc..441d2bb491d2567e6eb2a54d5e9d394fa89d469c 100644 (file)
@@ -49,7 +49,8 @@ extern int yydebug;
 struct trace_descriptor *ctf_open_trace(const char *path, int flags);
 void ctf_close_trace(struct trace_descriptor *descriptor);
 
-static rw_dispatch read_dispatch_table[] = {
+static
+rw_dispatch read_dispatch_table[] = {
        [ CTF_TYPE_INTEGER ] = ctf_integer_read,
        [ CTF_TYPE_FLOAT ] = ctf_float_read,
        [ CTF_TYPE_ENUM ] = ctf_enum_read,
@@ -60,7 +61,8 @@ static rw_dispatch read_dispatch_table[] = {
        [ CTF_TYPE_SEQUENCE ] = ctf_sequence_rw,
 };
 
-static rw_dispatch write_dispatch_table[] = {
+static
+rw_dispatch write_dispatch_table[] = {
        [ CTF_TYPE_INTEGER ] = ctf_integer_write,
        [ CTF_TYPE_FLOAT ] = ctf_float_write,
        [ CTF_TYPE_ENUM ] = ctf_enum_write,
@@ -71,11 +73,150 @@ static rw_dispatch write_dispatch_table[] = {
        [ CTF_TYPE_SEQUENCE ] = ctf_sequence_rw,
 };
 
+static
 struct format ctf_format = {
        .open_trace = ctf_open_trace,
        .close_trace = ctf_close_trace,
 };
 
+static
+int ctf_read_event(struct stream_pos *ppos, struct ctf_stream_class *stream_class)
+{
+       struct ctf_stream_pos *pos =
+               container_of(ppos, struct ctf_stream_pos, parent);
+       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 */
+               }
+       }
+
+       /* 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_class *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;
@@ -92,20 +233,20 @@ void ctf_init_pos(struct ctf_stream_pos *pos, int fd, int open_flags)
                                                sizeof(struct packet_index));
        else
                pos->packet_index = NULL;
-
        switch (open_flags & O_ACCMODE) {
        case O_RDONLY:
                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);      /* position for write */
+                       ctf_move_pos_slow(pos, 0, SEEK_SET);    /* position for write */
                break;
        default:
                assert(0);
@@ -130,13 +271,12 @@ void ctf_fini_pos(struct ctf_stream_pos *pos)
        (void) g_array_free(pos->packet_index, TRUE);
 }
 
-void ctf_move_pos_slow(struct ctf_stream_pos *pos, size_t offset)
+void ctf_move_pos_slow(struct ctf_stream_pos *pos, size_t offset, int whence)
 {
        int ret;
        off_t off;
        struct packet_index *index;
 
-
        if (pos->prot == PROT_WRITE && pos->content_size_loc)
                *pos->content_size_loc = pos->offset;
 
@@ -148,6 +288,7 @@ void ctf_move_pos_slow(struct ctf_stream_pos *pos, size_t offset)
                                strerror(errno));
                        assert(0);
                }
+               pos->base = NULL;
        }
 
        /*
@@ -155,29 +296,43 @@ void ctf_move_pos_slow(struct ctf_stream_pos *pos, size_t offset)
         * 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 {
-               /* The reader will expect us to skip padding */
-               assert(pos->offset + offset == pos->content_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 reader will expect us to skip padding */
+                       assert(pos->offset + offset == pos->content_size);
                        ++pos->cur_index;
+                       break;
+               case SEEK_SET:
+                       assert(offset == 0);    /* only seek supported for now */
+                       pos->cur_index = 0;
+                       break;
+               default:
+                       assert(0);
+               }
+               if (pos->cur_index >= pos->packet_index->len) {
+                       pos->offset = EOF;
+                       return;
+               }
                index = &g_array_index(pos->packet_index, struct packet_index,
                                       pos->cur_index);
                pos->mmap_offset = index->offset;
@@ -185,11 +340,16 @@ void ctf_move_pos_slow(struct ctf_stream_pos *pos, size_t offset)
                /* Lookup context/packet size in index */
                pos->content_size = index->content_size;
                pos->packet_size = index->packet_size;
+               pos->offset = index->data_offset;
        }
        /* map new base. Need mapping length from header. */
        pos->base = mmap(NULL, pos->packet_size / CHAR_BIT, pos->prot,
                         pos->flags, pos->fd, pos->mmap_offset);
-       pos->offset = 0;
+       if (pos->base == MAP_FAILED) {
+               fprintf(stdout, "[error] mmap error %s.\n",
+                       strerror(errno));
+               assert(0);
+       }
 }
 
 /*
@@ -264,7 +424,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;
@@ -305,19 +465,20 @@ int create_stream_packet_index(struct ctf_trace *td,
                /* read and check header, set stream id (and check) */
                if (td->packet_header) {
                        /* Read packet header */
-                       generic_rw(&pos->parent, &td->packet_header->p);
-
+                       ret = generic_rw(&pos->parent, &td->packet_header->p);
+                       if (ret)
+                               return ret;
                        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);
@@ -329,13 +490,13 @@ 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("trace_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);
 
@@ -359,11 +520,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;
                        }
@@ -390,17 +551,18 @@ int create_stream_packet_index(struct ctf_trace *td,
 
                if (stream->packet_context) {
                        /* Read packet context */
-                       generic_rw(&pos->parent, &stream->packet_context->p);
-
+                       ret = generic_rw(&pos->parent, &stream->packet_context->p);
+                       if (ret)
+                               return ret;
                        /* read content size from header */
                        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 {
@@ -412,11 +574,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 {
@@ -430,12 +592,31 @@ int create_stream_packet_index(struct ctf_trace *td,
                        packet_index.packet_size = packet_index.content_size ? : filestats.st_size * CHAR_BIT;
                }
 
+               /* Validate content size and packet size values */
+               if (packet_index.content_size > packet_index.packet_size) {
+                       fprintf(stdout, "[error] Content size (%zu bits) is larger than packet size (%zu bits).\n",
+                               packet_index.content_size, packet_index.packet_size);
+                       return -EINVAL;
+               }
+
+               if (packet_index.packet_size > (filestats.st_size - packet_index.offset) * CHAR_BIT) {
+                       fprintf(stdout, "[error] Packet size (%zu bits) is larger than remaining file size (%zu bits).\n",
+                               packet_index.content_size, (filestats.st_size - packet_index.offset) * CHAR_BIT);
+                       return -EINVAL;
+               }
+
+               /* Save position after header and context */
+               packet_index.data_offset = pos->offset;
+
                /* add index to packet array */
                g_array_append_val(file_stream->pos.packet_index, packet_index);
 
                pos->mmap_offset += packet_index.packet_size / CHAR_BIT;
        }
 
+       /* Move pos back to beginning of file */
+       ctf_move_pos_slow(pos, 0, SEEK_SET);    /* position for write */
+
        return 0;
 }
 
@@ -524,7 +705,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;
@@ -549,30 +731,6 @@ error:
        return ret;
 }
 
-static
-int ctf_open_trace_write(struct ctf_trace *td, const char *path, int flags)
-{
-       int ret;
-
-       ret = mkdir(path, S_IRWXU|S_IRWXG);
-       if (ret)
-               return ret;
-
-       /* Open trace directory */
-       td->dir = opendir(path);
-       if (!td->dir) {
-               fprintf(stdout, "[error] Unable to open trace directory.\n");
-               ret = -ENOENT;
-               goto error;
-       }
-       
-
-       return 0;
-
-error:
-       return ret;
-}
-
 struct trace_descriptor *ctf_open_trace(const char *path, int flags)
 {
        struct ctf_trace *td;
@@ -582,19 +740,17 @@ struct trace_descriptor *ctf_open_trace(const char *path, int flags)
 
        switch (flags & O_ACCMODE) {
        case O_RDONLY:
+               if (!path) {
+                       fprintf(stdout, "[error] Path missing for input CTF trace.\n");
+                       goto error;
+               }
                ret = ctf_open_trace_read(td, path, 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;
-#if 0
-               ret = ctf_open_trace_write(td, path, flags);
-               if (ret)
-                       goto error;
-#endif //0
-               break;
        default:
                fprintf(stdout, "[error] Incorrect open flags.\n");
                goto error;
@@ -620,13 +776,12 @@ 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++) {
                                struct ctf_file_stream *file_stream;
-                               file_stream = g_ptr_array_index(td->streams, j);
+                               file_stream = g_ptr_array_index(stream->files, j);
                                ctf_close_file_stream(file_stream);
                        }
 
This page took 0.030001 seconds and 4 git commands to generate.