X-Git-Url: http://git.lttng.org/?a=blobdiff_plain;f=ltt%2Fbranches%2Fpoly%2Flttd%2Flttd.c;h=27240882189de464882d2f2ab1d64fa674fc8f29;hb=cdad978721add90e7f10411e68795099bf87946e;hp=02724ce5285f6d10bdedfff992510f2a6b0dd93b;hpb=ce2b5176f95e595c37ca7c21d31b51865fa58a87;p=ltt-control.git diff --git a/ltt/branches/poly/lttd/lttd.c b/ltt/branches/poly/lttd/lttd.c index 02724ce..2724088 100644 --- a/ltt/branches/poly/lttd/lttd.c +++ b/ltt/branches/poly/lttd/lttd.c @@ -14,7 +14,9 @@ #include #endif +#define _REENTRANT #define _GNU_SOURCE +#include #include #include #include @@ -27,6 +29,7 @@ #include #include #include +#include /* Relayfs IOCTL */ #include @@ -35,7 +38,7 @@ /* Get the next sub buffer that can be read. */ #define RELAYFS_GET_SUBBUF _IOR(0xF4, 0x00,__u32) /* Release the oldest reserved (by "get") sub buffer. */ -#define RELAYFS_PUT_SUBBUF _IO(0xF4, 0x01) +#define RELAYFS_PUT_SUBBUF _IOW(0xF4, 0x01,__u32) /* returns the number of sub buffers in the per cpu channel. */ #define RELAYFS_GET_N_SUBBUFS _IOR(0xF4, 0x02,__u32) /* returns the size of the sub buffers. */ @@ -56,6 +59,7 @@ struct fd_pair { unsigned int n_subbufs; unsigned int subbuf_size; void *mmap; + pthread_mutex_t mutex; }; struct channel_trace_fd { @@ -67,6 +71,7 @@ static char *trace_name = NULL; static char *channel_name = NULL; static int daemon_mode = 0; static int append_mode = 0; +static unsigned long num_threads = 1; volatile static int quit_program = 0; /* For signal handler */ /* Args : @@ -86,6 +91,7 @@ void show_arguments(void) printf("-c directory Root directory of the relayfs trace channels.\n"); printf("-d Run in background (daemon).\n"); printf("-a Append to an possibly existing trace.\n"); + printf("-N Number of threads to start.\n"); printf("\n"); } @@ -131,6 +137,12 @@ int parse_arguments(int argc, char **argv) case 'a': append_mode = 1; break; + case 'N': + if(argn+1 < argc) { + num_threads = strtoul(argv[argn+1], NULL, 0); + argn++; + } + break; default: printf("Invalid argument '%s'.\n", argv[argn]); printf("\n"); @@ -297,35 +309,41 @@ end: int read_subbuffer(struct fd_pair *pair) { - unsigned int subbuf_index; - int err, ret; + unsigned int consumed_old; + int err, ret=0; err = ioctl(pair->channel, RELAYFS_GET_SUBBUF, - &subbuf_index); - printf("index : %u\n", subbuf_index); + &consumed_old); + printf("cookie : %u\n", consumed_old); if(err != 0) { - perror("Error in reserving sub buffer"); - ret = -EPERM; + ret = errno; + perror("Reserving sub buffer failed (everything is normal)"); goto get_error; } err = TEMP_FAILURE_RETRY(write(pair->trace, - pair->mmap + (subbuf_index * pair->subbuf_size), + pair->mmap + + (consumed_old & ((pair->n_subbufs * pair->subbuf_size)-1)), pair->subbuf_size)); if(err < 0) { + ret = errno; perror("Error in writing to file"); - ret = err; goto write_error; } write_error: - err = ioctl(pair->channel, RELAYFS_PUT_SUBBUF); + err = ioctl(pair->channel, RELAYFS_PUT_SUBBUF, &consumed_old); if(err != 0) { - perror("Error in unreserving sub buffer"); - ret = -EPERM; + ret = errno; + if(errno == -EFAULT) { + perror("Error in unreserving sub buffer\n"); + } else if(errno == -EIO) { + perror("Reader has been pushed by the writer, last subbuffer corrupted."); + /* FIXME : we may delete the last written buffer if we wish. */ + } goto get_error; } @@ -334,29 +352,11 @@ get_error: } -/* read_channels - * - * Read the realyfs channels and write them in the paired tracefiles. - * - * @fd_pairs : paired channels and trace files. - * - * returns 0 on success, -1 on error. - * - * Note that the high priority polled channels are consumed first. We then poll - * again to see if these channels are still in priority. Only when no - * high priority channel is left, we start reading low priority channels. - * - * Note that a channel is considered high priority when the buffer is almost - * full. - */ -int read_channels(struct channel_trace_fd *fd_pairs) +int map_channels(struct channel_trace_fd *fd_pairs) { - struct pollfd *pollfd; int i,j; - int num_rdy, num_hup; - int high_prio; - int ret; + int ret=0; if(fd_pairs->num_pairs <= 0) { printf("No channel to read\n"); @@ -380,6 +380,11 @@ int read_channels(struct channel_trace_fd *fd_pairs) perror("Error in getting the size of the subbuffers"); goto end; } + ret = pthread_mutex_init(&pair->mutex, NULL); /* Fast mutex */ + if(ret != 0) { + perror("Error in mutex init"); + goto end; + } } /* Mmap each FD */ @@ -394,6 +399,82 @@ int read_channels(struct channel_trace_fd *fd_pairs) } } + goto end; /* success */ + + /* Error handling */ + /* munmap only the successfully mmapped indexes */ +munmap: + /* Munmap each FD */ + for(j=0;jpair[j]; + int err_ret; + + err_ret = munmap(pair->mmap, pair->subbuf_size * pair->n_subbufs); + if(err_ret != 0) { + perror("Error in munmap"); + } + ret |= err_ret; + } + +end: + return ret; + + +} + + +int unmap_channels(struct channel_trace_fd *fd_pairs) +{ + int j; + int ret=0; + + /* Munmap each FD */ + for(j=0;jnum_pairs;j++) { + struct fd_pair *pair = &fd_pairs->pair[j]; + int err_ret; + + err_ret = munmap(pair->mmap, pair->subbuf_size * pair->n_subbufs); + if(err_ret != 0) { + perror("Error in munmap"); + } + ret |= err_ret; + err_ret = pthread_mutex_destroy(&pair->mutex); + if(err_ret != 0) { + perror("Error in mutex destroy"); + } + ret |= err_ret; + } + + return ret; +} + + +/* read_channels + * + * Thread worker. + * + * Read the relayfs channels and write them in the paired tracefiles. + * + * @fd_pairs : paired channels and trace files. + * + * returns (void*)0 on success, (void*)-1 on error. + * + * Note that the high priority polled channels are consumed first. We then poll + * again to see if these channels are still in priority. Only when no + * high priority channel is left, we start reading low priority channels. + * + * Note that a channel is considered high priority when the buffer is almost + * full. + */ + +void * read_channels(void *arg) +{ + struct pollfd *pollfd; + int i,j; + int num_rdy, num_hup; + int high_prio; + int ret = 0; + struct channel_trace_fd *fd_pairs = (struct channel_trace_fd *)arg; /* Start polling the FD */ @@ -441,10 +522,18 @@ int read_channels(struct channel_trace_fd *fd_pairs) num_hup++; break; case POLLPRI: - printf("Urgent read on fd %d\n", pollfd[i].fd); - /* Take care of high priority channels first. */ - high_prio = 1; - ret |= read_subbuffer(&fd_pairs->pair[i]); + if(pthread_mutex_trylock(&fd_pairs->pair[i].mutex) == 0) { + printf("Urgent read on fd %d\n", pollfd[i].fd); + /* Take care of high priority channels first. */ + high_prio = 1; + /* it's ok to have an unavailable subbuffer */ + ret = read_subbuffer(&fd_pairs->pair[i]); + if(ret == -EAGAIN) ret = 0; + + ret = pthread_mutex_unlock(&fd_pairs->pair[i].mutex); + if(ret) + printf("Error in mutex unlock : %s\n", strerror(ret)); + } break; } } @@ -455,9 +544,17 @@ int read_channels(struct channel_trace_fd *fd_pairs) for(i=0;inum_pairs;i++) { switch(pollfd[i].revents) { case POLLIN: - /* Take care of low priority channels. */ - printf("Normal read on fd %d\n", pollfd[i].fd); - ret |= read_subbuffer(&fd_pairs->pair[i]); + if(pthread_mutex_trylock(&fd_pairs->pair[i].mutex) == 0) { + /* Take care of low priority channels. */ + printf("Normal read on fd %d\n", pollfd[i].fd); + /* it's ok to have an unavailable subbuffer */ + ret = read_subbuffer(&fd_pairs->pair[i]); + if(ret == -EAGAIN) ret = 0; + + ret = pthread_mutex_unlock(&fd_pairs->pair[i].mutex); + if(ret) + printf("Error in mutex unlock : %s\n", strerror(ret)); + } break; } } @@ -468,23 +565,8 @@ int read_channels(struct channel_trace_fd *fd_pairs) free_fd: free(pollfd); - /* munmap only the successfully mmapped indexes */ - i = fd_pairs->num_pairs; -munmap: - /* Munmap each FD */ - for(j=0;jpair[j]; - int err_ret; - - err_ret = munmap(pair->mmap, pair->subbuf_size * pair->n_subbufs); - if(err_ret != 0) { - perror("Error in munmap"); - } - ret |= err_ret; - } - end: - return ret; + return (void*)ret; } @@ -504,9 +586,12 @@ void close_channel_trace_pairs(struct channel_trace_fd *fd_pairs) int main(int argc, char ** argv) { - int ret; + int ret = 0; struct channel_trace_fd fd_pairs = { NULL, 0 }; struct sigaction act; + pthread_t *tids; + unsigned int i; + void *tret; ret = parse_arguments(argc, argv); @@ -536,11 +621,36 @@ int main(int argc, char ** argv) sigaction(SIGQUIT, &act, NULL); sigaction(SIGINT, &act, NULL); - //return 0; + if(ret = open_channel_trace_pairs(channel_name, trace_name, &fd_pairs)) goto close_channel; - ret = read_channels(&fd_pairs); + if(ret = map_channels(&fd_pairs)) + goto close_channel; + + tids = malloc(sizeof(pthread_t) * num_threads); + for(i=0; i