From: Mathieu Desnoyers Date: Fri, 6 May 2011 01:02:53 +0000 (-0400) Subject: CTF: open trace read X-Git-Tag: v0.1~107 X-Git-Url: http://git.efficios.com/?p=babeltrace.git;a=commitdiff_plain;h=8c572eba034b95716a4bbd019dfbe0827605ba90 CTF: open trace read Signed-off-by: Mathieu Desnoyers --- diff --git a/formats/ctf/ctf.c b/formats/ctf/ctf.c index 54aaed7e..6488ffc9 100644 --- a/formats/ctf/ctf.c +++ b/formats/ctf/ctf.c @@ -38,9 +38,10 @@ /* * We currently simply map a page to read the packet header and packet - * context to get the packet length and content length. + * context to get the packet length and content length. (in bits) */ -#define MAX_PACKET_HEADER_LEN getpagesize() +#define MAX_PACKET_HEADER_LEN (getpagesize() * CHAR_BIT) +#define WRITE_PACKET_LEN (getpagesize() * 8 * CHAR_BIT) #define UUID_LEN 16 /* uuid by value len */ extern int yydebug; @@ -80,19 +81,71 @@ static struct format ctf_format = { .close_trace = ctf_close_trace, }; +void init_pos(struct stream_pos *pos, int fd) +{ + pos->fd = fd; + pos->mmap_offset = 0; + pos->packet_size = 0; + pos->content_size = 0; + pos->content_size_loc = NULL; + pos->base = NULL; + pos->offset = 0; + pos->dummy = false; + pos->packet_index = g_array_new(FALSE, TRUE, + sizeof(struct packet_index)); + pos->cur_index = 0; + if (fd >= 0) { + int flags = fcntl(fd, F_GETFL, 0); + + switch (flags & O_ACCMODE) { + case O_RDONLY: + pos->prot = PROT_READ; + pos->flags = MAP_PRIVATE; + break; + case O_WRONLY: + case O_RDWR: + pos->prot = PROT_WRITE; /* Write has priority */ + pos->flags = MAP_SHARED; + move_pos_slow(pos, 0); /* position for write */ + break; + default: + assert(0); + } + + } +} + +void fini_pos(struct stream_pos *pos) +{ + int ret; + + if (pos->prot == PROT_WRITE && pos->content_size_loc) + *pos->content_size_loc = pos->offset; + if (pos->base) { + /* unmap old base */ + ret = munmap(pos->base, pos->packet_size / CHAR_BIT); + if (ret) { + fprintf(stdout, "Unable to unmap old base: %s.\n", + strerror(errno)); + assert(0); + } + } + (void) g_array_free(pos->packet_index, TRUE); +} + void move_pos_slow(struct stream_pos *pos, size_t offset) { int ret; + off_t off; + struct packet_index *index; - /* - * The caller should never ask for move_pos across packets, - * except to get exactly at the beginning of the next packet. - */ - assert(pos->offset + offset == pos->content_size); + + if (pos->prot == PROT_WRITE && pos->content_size_loc) + *pos->content_size_loc = pos->offset; if (pos->base) { /* unmap old base */ - ret = munmap(pos->base, pos->packet_size); + ret = munmap(pos->base, pos->packet_size / CHAR_BIT); if (ret) { fprintf(stdout, "Unable to unmap old base: %s.\n", strerror(errno)); @@ -100,13 +153,46 @@ void move_pos_slow(struct stream_pos *pos, size_t offset) } } - pos->mmap_offset += pos->packet_size / CHAR_BIT; + /* + * The caller should never ask for move_pos across packets, + * 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) + pos->mmap_offset += WRITE_PACKET_LEN / CHAR_BIT; + 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); + assert(off >= 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) + ++pos->cur_index; + index = &g_array_index(pos->packet_index, struct packet_index, + pos->cur_index); + pos->mmap_offset = index->offset; + + /* Lookup context/packet size in index */ + pos->content_size = index->content_size; + pos->packet_size = index->packet_size; + } /* map new base. Need mapping length from header. */ - pos->base = mmap(NULL, MAX_PACKET_HEADER_LEN, PROT_READ, - MAP_PRIVATE, pos->fd, pos->mmap_offset); - pos->content_size = 0; /* Unknown at this point */ - pos->packet_size = 0; /* Unknown at this point */ - + pos->base = mmap(NULL, pos->packet_size / CHAR_BIT, pos->prot, + pos->flags, pos->fd, pos->mmap_offset); + pos->offset = 0; } /* @@ -201,20 +287,25 @@ int create_stream_packet_index(struct trace_descriptor *td, if (pos->base) { /* unmap old base */ - ret = munmap(pos->base, pos->packet_size); + ret = munmap(pos->base, pos->packet_size / CHAR_BIT); if (ret) { fprintf(stdout, "Unable to unmap old base: %s.\n", strerror(errno)); return ret; } + pos->base = NULL; } /* map new base. Need mapping length from header. */ - pos->base = mmap(NULL, MAX_PACKET_HEADER_LEN, PROT_READ, + pos->base = mmap(NULL, MAX_PACKET_HEADER_LEN / CHAR_BIT, PROT_READ, MAP_PRIVATE, pos->fd, pos->mmap_offset); pos->content_size = MAX_PACKET_HEADER_LEN; /* Unknown at this point */ pos->packet_size = MAX_PACKET_HEADER_LEN; /* Unknown at this point */ pos->offset = 0; /* Position of the packet header */ + packet_index.offset = pos->mmap_offset; + packet_index.content_size = 0; + packet_index.packet_size = 0; + /* read and check header, set stream id (and check) */ if (td->ctf_trace.packet_header) { /* Read packet header */ @@ -231,8 +322,10 @@ int create_stream_packet_index(struct trace_descriptor *td, defint = container_of(field->definition, struct definition_integer, p); assert(defint->declaration->signedness == FALSE); if (defint->value._unsigned != CTF_MAGIC) { - fprintf(stdout, "[error] Invalid magic number %" PRIX64 ".\n", - defint->value._unsigned); + fprintf(stdout, "[error] Invalid magic number %" PRIX64 " at packet %u (file offset %zd).\n", + defint->value._unsigned, + file_stream->pos.packet_index->len, + (ssize_t) pos->mmap_offset); return -EINVAL; } } @@ -245,7 +338,6 @@ int create_stream_packet_index(struct trace_descriptor *td, uint64_t i; uint8_t uuidval[UUID_LEN]; - field = struct_definition_get_field_from_index(td->ctf_trace.packet_header, len_index); assert(field->definition->declaration->id == CTF_TYPE_ARRAY); defarray = container_of(field->definition, struct definition_array, p); @@ -316,10 +408,10 @@ int create_stream_packet_index(struct trace_descriptor *td, assert(field->definition->declaration->id == CTF_TYPE_INTEGER); defint = container_of(field->definition, struct definition_integer, p); assert(defint->declaration->signedness == FALSE); - pos->content_size = defint->value._unsigned; + packet_index.content_size = defint->value._unsigned; } else { /* Use file size for packet size */ - pos->content_size = filestats.st_size * CHAR_BIT; + packet_index.content_size = filestats.st_size * CHAR_BIT; } /* read packet size from header */ @@ -332,25 +424,22 @@ int create_stream_packet_index(struct trace_descriptor *td, assert(field->definition->declaration->id == CTF_TYPE_INTEGER); defint = container_of(field->definition, struct definition_integer, p); assert(defint->declaration->signedness == FALSE); - pos->packet_size = defint->value._unsigned; + packet_index.packet_size = defint->value._unsigned; } else { /* Use content size if non-zero, else file size */ - pos->packet_size = pos->content_size ? : filestats.st_size * CHAR_BIT; + packet_index.packet_size = packet_index.content_size ? : filestats.st_size * CHAR_BIT; } } else { /* Use file size for packet size */ - pos->content_size = filestats.st_size * CHAR_BIT; + packet_index.content_size = filestats.st_size * CHAR_BIT; /* Use content size if non-zero, else file size */ - pos->packet_size = pos->content_size ? : filestats.st_size * CHAR_BIT; + packet_index.packet_size = packet_index.content_size ? : filestats.st_size * CHAR_BIT; } - packet_index.offset = pos->mmap_offset; - packet_index.content_size = pos->content_size; - packet_index.packet_size = pos->packet_size; /* add index to packet array */ g_array_append_val(file_stream->pos.packet_index, packet_index); - pos->mmap_offset += pos->packet_size / CHAR_BIT; + pos->mmap_offset += packet_index.packet_size / CHAR_BIT; } return 0; @@ -370,9 +459,7 @@ int ctf_open_file_stream_read(struct trace_descriptor *td, const char *path, int if (ret < 0) goto error; file_stream = g_new0(struct ctf_file_stream, 1); - file_stream->pos.fd = ret; - file_stream->pos.packet_index = g_array_new(FALSE, TRUE, - sizeof(struct packet_index)); + init_pos(&file_stream->pos, ret); ret = create_stream_packet_index(td, file_stream); if (ret) goto error_index; @@ -381,7 +468,7 @@ int ctf_open_file_stream_read(struct trace_descriptor *td, const char *path, int return 0; error_index: - (void) g_array_free(file_stream->pos.packet_index, TRUE); + fini_pos(&file_stream->pos); close(file_stream->pos.fd); g_free(file_stream); error: @@ -499,7 +586,7 @@ struct trace_descriptor *ctf_open_trace(const char *path, int flags) td = g_new0(struct trace_descriptor, 1); - switch (flags) { + switch (flags & O_ACCMODE) { case O_RDONLY: ret = ctf_open_trace_read(td, path, flags); if (ret) @@ -524,7 +611,7 @@ error: static void ctf_close_file_stream(struct ctf_file_stream *file_stream) { - (void) g_array_free(file_stream->pos.packet_index, TRUE); + fini_pos(&file_stream->pos); close(file_stream->pos.fd); } diff --git a/formats/ctf/metadata/ctf-visitor-generate-io-struct.c b/formats/ctf/metadata/ctf-visitor-generate-io-struct.c index f5430711..1eb763de 100644 --- a/formats/ctf/metadata/ctf-visitor-generate-io-struct.c +++ b/formats/ctf/metadata/ctf-visitor-generate-io-struct.c @@ -1442,13 +1442,6 @@ int ctf_event_visit(FILE *fd, int depth, struct ctf_node *node, fprintf(fd, "[error] %s: missing name field in event declaration\n", __func__); goto error; } - /* Allow only one event without id per stream */ - if (!CTF_EVENT_FIELD_IS_SET(event, id) - && event->stream->events_by_id->len != 0) { - ret = -EPERM; - fprintf(fd, "[error] %s: missing id field in event declaration\n", __func__); - goto error; - } if (!CTF_EVENT_FIELD_IS_SET(event, stream_id)) { /* Allow missing stream_id if there is only a single stream */ if (trace->streams->len == 1) { @@ -1460,6 +1453,13 @@ int ctf_event_visit(FILE *fd, int depth, struct ctf_node *node, goto error; } } + /* Allow only one event without id per stream */ + if (!CTF_EVENT_FIELD_IS_SET(event, id) + && event->stream->events_by_id->len != 0) { + ret = -EPERM; + fprintf(fd, "[error] %s: missing id field in event declaration\n", __func__); + goto error; + } if (event->stream->events_by_id->len <= event->id) g_ptr_array_set_size(event->stream->events_by_id, event->id + 1); g_ptr_array_index(event->stream->events_by_id, event->id) = event; @@ -1646,14 +1646,14 @@ int ctf_stream_visit(FILE *fd, int depth, struct ctf_node *node, fprintf(fd, "[error] %s: missing stream_id field in packet header declaration, but stream_id attribute is declared for stream.\n", __func__); goto error; } - } - - /* Allow only one id-less stream */ - if (!CTF_STREAM_FIELD_IS_SET(stream, stream_id) - && trace->streams->len != 0) { - ret = -EPERM; - fprintf(fd, "[error] %s: missing id field in stream declaration\n", __func__); - goto error; + } else { + /* Allow only one id-less stream */ + if (trace->streams->len != 0) { + ret = -EPERM; + fprintf(fd, "[error] %s: missing id field in stream declaration\n", __func__); + goto error; + } + stream->stream_id = 0; } if (trace->streams->len <= stream->stream_id) g_ptr_array_set_size(trace->streams, stream->stream_id + 1); diff --git a/include/babeltrace/ctf/types.h b/include/babeltrace/ctf/types.h index 8c98ff2c..8f9bf992 100644 --- a/include/babeltrace/ctf/types.h +++ b/include/babeltrace/ctf/types.h @@ -20,6 +20,8 @@ */ #include +#include +#include #include #include #include @@ -36,13 +38,17 @@ struct packet_index { struct stream_pos { int fd; /* backing file fd. -1 if unset. */ GArray *packet_index; /* contains struct packet_index */ + int prot; /* mmap protection */ + int flags; /* mmap flags */ /* Current position */ off_t mmap_offset; /* mmap offset in the file, in bytes */ size_t packet_size; /* current packet size, in bits */ size_t content_size; /* current content size, in bits */ + uint32_t *content_size_loc; /* pointer to current content size */ char *base; /* mmap base address */ size_t offset; /* offset from base, in bits */ + size_t cur_index; /* current index in packet index */ int dummy; /* dummy position, for length calculation */ }; @@ -113,17 +119,8 @@ void ctf_sequence_end(struct stream_pos *pos, void move_pos_slow(struct stream_pos *pos, size_t offset); -static inline -void init_pos(struct stream_pos *pos, int fd) -{ - pos->fd = fd; - pos->mmap_offset = 0; - pos->packet_size = 0; - pos->content_size = 0; - pos->base = NULL; - pos->offset = 0; - pos->dummy = false; -} +void init_pos(struct stream_pos *pos, int fd); +void fini_pos(struct stream_pos *pos); /* * move_pos - move position of a relative bit offset @@ -131,12 +128,18 @@ void init_pos(struct stream_pos *pos, int fd) * TODO: allow larger files by updating base too. */ static inline -void move_pos(struct stream_pos *pos, size_t offset) +void move_pos(struct stream_pos *pos, size_t bit_offset) { - if (pos->fd >= 0 && (pos->offset + offset >= pos->content_size)) - move_pos_slow(pos, offset); - else - pos->offset += offset; + if (pos->fd >= 0) { + if (((pos->prot == PROT_READ) + && (pos->offset + bit_offset >= pos->content_size)) + || ((pos->prot == PROT_WRITE) + && (pos->offset + bit_offset >= pos->packet_size))) { + move_pos_slow(pos, bit_offset); + return; + } + } + pos->offset += bit_offset; } /* @@ -145,15 +148,9 @@ void move_pos(struct stream_pos *pos, size_t offset) * TODO: allow larger files by updating base too. */ static inline -void align_pos(struct stream_pos *pos, size_t offset) +void align_pos(struct stream_pos *pos, size_t bit_offset) { - pos->offset += offset_align(pos->offset, offset); -} - -static inline -void copy_pos(struct stream_pos *dest, struct stream_pos *src) -{ - memcpy(dest, src, sizeof(struct stream_pos)); + move_pos(pos, offset_align(pos->offset, bit_offset)); } static inline @@ -164,4 +161,30 @@ char *get_pos_addr(struct stream_pos *pos) return pos->base + (pos->offset / CHAR_BIT); } +static inline +void dummy_pos(struct stream_pos *pos, struct stream_pos *dummy) +{ + memcpy(dummy, pos, sizeof(struct stream_pos)); + dummy->dummy = 1; + dummy->fd = -1; +} + +/* + * Check if current packet can hold data. + * Returns 0 for success, negative error otherwise. + */ +static inline +int pos_packet(struct stream_pos *dummy) +{ + if (dummy->offset > dummy->packet_size) + return -ENOSPC; + return 0; +} + +static inline +void pos_pad_packet(struct stream_pos *pos) +{ + move_pos(pos, pos->packet_size - pos->offset); +} + #endif /* _BABELTRACE_CTF_TYPES_H */ diff --git a/tests/Makefile.am b/tests/Makefile.am index 1038535e..583f7ac6 100644 --- a/tests/Makefile.am +++ b/tests/Makefile.am @@ -5,3 +5,8 @@ noinst_PROGRAMS = test-bitfield test-dummytrace test_bitfield_SOURCES = test-bitfield.c test_dummytrace_SOURCES = test-dummytrace.c + + +test_dummytrace_LDADD = \ + $(top_builddir)/formats/libbabeltrace_registry.la \ + $(top_builddir)/formats/ctf/libctf.la diff --git a/tests/test-dummytrace.c b/tests/test-dummytrace.c index 81e97b98..ed00f34b 100644 --- a/tests/test-dummytrace.c +++ b/tests/test-dummytrace.c @@ -1,3 +1,23 @@ +/* + * text-to-ctf.c + * + * BabelTrace - Convert Text to CTF + * + * Copyright 2010, 2011 - Mathieu Desnoyers + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in + * all copies or substantial portions of the Software. + * + * Depends on glibc 2.10 for getline(). + */ + #include #include #include @@ -9,73 +29,142 @@ #include #include -#define FILE_LEN (4 * 4096) +#include +#include + +int babeltrace_debug = 0; + +static uuid_t s_uuid; + +static +void write_packet_header(struct stream_pos *pos, uuid_t uuid) +{ + struct stream_pos dummy; + + /* magic */ + dummy_pos(pos, &dummy); + align_pos(&dummy, sizeof(uint32_t) * CHAR_BIT); + move_pos(&dummy, sizeof(uint32_t) * CHAR_BIT); + assert(!pos_packet(&dummy)); + + align_pos(pos, sizeof(uint32_t) * CHAR_BIT); + *(uint32_t *) get_pos_addr(pos) = 0xC1FC1FC1; + move_pos(pos, sizeof(uint32_t) * CHAR_BIT); + + /* trace_uuid */ + dummy_pos(pos, &dummy); + align_pos(&dummy, sizeof(uint8_t) * CHAR_BIT); + move_pos(&dummy, 16 * CHAR_BIT); + assert(!pos_packet(&dummy)); + + align_pos(pos, sizeof(uint8_t) * CHAR_BIT); + memcpy(get_pos_addr(pos), uuid, 16); + move_pos(pos, 16 * CHAR_BIT); +} + +static +void write_packet_context(struct stream_pos *pos) +{ + struct stream_pos dummy; + + /* content_size */ + dummy_pos(pos, &dummy); + align_pos(&dummy, sizeof(uint32_t) * CHAR_BIT); + move_pos(&dummy, sizeof(uint32_t) * CHAR_BIT); + assert(!pos_packet(&dummy)); + + align_pos(pos, sizeof(uint32_t) * CHAR_BIT); + *(uint32_t *) get_pos_addr(pos) = -1U; /* Not known yet */ + pos->content_size_loc = (uint32_t *) get_pos_addr(pos); + move_pos(pos, sizeof(uint32_t) * CHAR_BIT); + + /* packet_size */ + dummy_pos(pos, &dummy); + align_pos(&dummy, sizeof(uint32_t) * CHAR_BIT); + move_pos(&dummy, sizeof(uint32_t) * CHAR_BIT); + assert(!pos_packet(&dummy)); + + align_pos(pos, sizeof(uint32_t) * CHAR_BIT); + *(uint32_t *) get_pos_addr(pos) = pos->packet_size; + move_pos(pos, sizeof(uint32_t) * CHAR_BIT); +} + +static +void trace_string(char *line, struct stream_pos *pos, size_t len) +{ + struct stream_pos dummy; + int attempt = 0; + + printf_debug("read: %s\n", line); +retry: + dummy_pos(pos, &dummy); + align_pos(&dummy, sizeof(uint8_t) * CHAR_BIT); + move_pos(&dummy, len * CHAR_BIT); + if (pos_packet(&dummy)) { + /* TODO write content size */ + pos_pad_packet(pos); + write_packet_header(pos, s_uuid); + write_packet_context(pos); + if (attempt++ == 1) { + fprintf(stdout, "[Error] Line too large for packet size (%zukB) (discarded)\n", + pos->packet_size / CHAR_BIT / 1024); + return; + } + goto retry; + } + + align_pos(pos, sizeof(uint8_t) * CHAR_BIT); + memcpy(get_pos_addr(pos), line, len); + move_pos(pos, len * CHAR_BIT); +} + +static +void trace_text(FILE *input, int output) +{ + struct stream_pos pos; + ssize_t len; + char *line = NULL, *nl; + size_t linesize; + + init_pos(&pos, output); + + write_packet_header(&pos, s_uuid); + write_packet_context(&pos); + for (;;) { + len = getline(&line, &linesize, input); + if (len < 0) + break; + nl = strrchr(line, '\n'); + if (nl) + *nl = '\0'; + trace_string(line, &pos, nl - line + 1); + } + fini_pos(&pos); +} int main(int argc, char **argv) { - char *base, *pos; int fd, ret; - off_t off; - uuid_t uuid; + ret = unlink("dummystream"); + if (ret < 0) { + perror("unlink"); + return -1; + } fd = open("dummystream", O_RDWR|O_CREAT, S_IRUSR|S_IWUSR); if (fd < 0) { perror("open"); return -1; } - off = posix_fallocate(fd, 0, FILE_LEN); - if (off < 0) { - printf("Error in fallocate\n"); - return -1; - } -#if 0 - { - ssize_t len; - - off = lseek(fd, FILE_LEN - 1, SEEK_CUR); - if (off < 0) { - perror("lseek"); - return -1; - } - len = write(fd, "", 1); - if (len < 0) { - perror("write"); - return -1; - } - } -#endif - base = mmap(NULL, FILE_LEN, PROT_READ|PROT_WRITE, - MAP_SHARED, fd, 0); - if (!base) { - perror("mmap"); - return -1; - } - pos = base; - - /* magic */ - *(uint32_t *) pos = 0xC1FC1FC1; - pos += sizeof(uint32_t); - /* trace_uuid */ - ret = uuid_parse("2a6422d0-6cee-11e0-8c08-cb07d7b3a564", uuid); + ret = uuid_parse("2a6422d0-6cee-11e0-8c08-cb07d7b3a564", s_uuid); if (ret) { printf("uuid parse error\n"); + close(fd); return -1; } - memcpy(pos, uuid, 16); - pos += 16; - - /* stream_id */ - *(uint32_t *) pos = 0; - pos += sizeof(uint32_t); - ret = munmap(base, FILE_LEN); - if (ret) { - perror("munmap"); - return -1; - } + trace_text(stdin, fd); close(fd); return 0; } - -