X-Git-Url: http://git.lttng.org/?a=blobdiff_plain;f=ustd%2Fustd.c;h=bfa6352e1ccf2b4063660ff279b007f29ec3702a;hb=aa08b4413291fabcbd1b1144377d37034ad361de;hp=a4c7e1f3f0c8a6dbdd17ac2fbf349a3fe7cb6f9d;hpb=8cefc1450e2d929de5d2878518e7b8b2b77ff755;p=ust.git diff --git a/ustd/ustd.c b/ustd/ustd.c index a4c7e1f..bfa6352 100644 --- a/ustd/ustd.c +++ b/ustd/ustd.c @@ -1,64 +1,41 @@ -/* - * Copyright (C) 2009 Pierre-Marc Fournier +/* Copyright (C) 2009 Pierre-Marc Fournier * - * This program is free software: you can redistribute it and/or modify - * it under the terms of the GNU General Public License as published by - * the Free Software Foundation, either version 3 of the License, or - * (at your option) any later version. + * This library is free software; you can redistribute it and/or + * modify it under the terms of the GNU Lesser General Public + * License as published by the Free Software Foundation; either + * version 2.1 of the License, or (at your option) any later version. * - * This program is distributed in the hope that it will be useful, + * This library is distributed in the hope that it will be useful, * but WITHOUT ANY WARRANTY; without even the implied warranty of - * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the - * GNU General Public License for more details. + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. * - * You should have received a copy of the GNU General Public License - * along with this program. If not, see . + * You should have received a copy of the GNU Lesser General Public + * License along with this library; if not, write to the Free Software + * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA */ #define _GNU_SOURCE #include +#include #include #include #include #include +#include #include #include #include +#include +#include +#include -#include "localerr.h" +#include "ustd.h" +#include "usterr.h" #include "ustcomm.h" -struct list_head buffers = LIST_HEAD_INIT(buffers); - -struct buffer_info { - char *name; - pid_t pid; - struct ustcomm_connection conn; - - int shmid; - int bufstruct_shmid; - - /* the buffer memory */ - void *mem; - /* buffer size */ - int memlen; - /* number of subbuffers in buffer */ - int n_subbufs; - /* size of each subbuffer */ - int subbuf_size; - - /* the buffer information struct */ - void *bufstruct_mem; - - int file_fd; /* output file */ - - struct list_head list; - - long consumed_old; -}; - /* return value: 0 = subbuffer is finished, it won't produce data anymore * 1 = got subbuffer successfully * <0 = error @@ -68,30 +45,47 @@ struct buffer_info { #define GET_SUBBUF_DONE 0 #define GET_SUBBUF_DIED 2 +#define PUT_SUBBUF_OK 1 +#define PUT_SUBBUF_DIED 0 +#define PUT_SUBBUF_PUSHED 2 + +char *sock_path=NULL; +char *trace_path=NULL; +int daemon_mode = 0; +char *pidfile = NULL; + +/* Number of active buffers and the mutex to protect it. */ +int active_buffers = 0; +pthread_mutex_t active_buffers_mutex = PTHREAD_MUTEX_INITIALIZER; +/* Whether a request to end the program was received. */ +sig_atomic_t terminate_req = 0; + int get_subbuffer(struct buffer_info *buf) { - char *send_msg; - char *received_msg; - char *rep_code; + char *send_msg=NULL; + char *received_msg=NULL; + char *rep_code=NULL; int retval; int result; asprintf(&send_msg, "get_subbuffer %s", buf->name); result = ustcomm_send_request(&buf->conn, send_msg, &received_msg); - free(send_msg); - if(result < 0) { - ERR("get_subbuffer: ustcomm_send_request failed"); - return -1; - } - else if(result == 0) { + if((result == -1 && errno == EPIPE) || result == 0) { DBG("app died while being traced"); - return GET_SUBBUF_DIED; + retval = GET_SUBBUF_DIED; + goto end; + } + else if(result < 0) { + ERR("get_subbuffer: ustcomm_send_request failed"); + retval = -1; + goto end; } result = sscanf(received_msg, "%as %ld", &rep_code, &buf->consumed_old); if(result != 2 && result != 1) { ERR("unable to parse response to get_subbuffer"); - return -1; + retval = -1; + goto end_rep; } DBG("received msg is %s", received_msg); @@ -101,7 +95,8 @@ int get_subbuffer(struct buffer_info *buf) retval = GET_SUBBUF_OK; } else if(nth_token_is(received_msg, "END", 0) == 1) { - return GET_SUBBUF_DONE; + retval = GET_SUBBUF_DONE; + goto end_rep; } else { DBG("error getting subbuffer %s", buf->name); @@ -109,171 +104,108 @@ int get_subbuffer(struct buffer_info *buf) } /* FIMXE: free correctly the stuff */ - free(received_msg); - free(rep_code); +end_rep: + if(rep_code) + free(rep_code); +end: + if(send_msg) + free(send_msg); + if(received_msg) + free(received_msg); + return retval; } int put_subbuffer(struct buffer_info *buf) { - char *send_msg; - char *received_msg; - char *rep_code; + char *send_msg=NULL; + char *received_msg=NULL; + char *rep_code=NULL; int retval; int result; asprintf(&send_msg, "put_subbuffer %s %ld", buf->name, buf->consumed_old); result = ustcomm_send_request(&buf->conn, send_msg, &received_msg); + if(result < 0 && errno == ECONNRESET) { + retval = PUT_SUBBUF_DIED; + goto end; + } if(result < 0) { ERR("put_subbuffer: send_message failed"); - return -1; + retval = -1; + goto end; } - free(send_msg); result = sscanf(received_msg, "%as", &rep_code); if(result != 1) { ERR("unable to parse response to put_subbuffer"); - return -1; + retval = -1; + goto end_rep; } - free(received_msg); if(!strcmp(rep_code, "OK")) { DBG("subbuffer put %s", buf->name); - retval = 1; + retval = PUT_SUBBUF_OK; } else { - ERR("invalid response to put_subbuffer"); + DBG("put_subbuffer: received error, we were pushed"); + retval = PUT_SUBBUF_PUSHED; + goto end_rep; } - free(rep_code); +end_rep: + if(rep_code) + free(rep_code); + +end: + if(send_msg) + free(send_msg); + if(received_msg) + free(received_msg); + return retval; } -ssize_t patient_write(int fd, const void *buf, size_t count) +void decrement_active_buffers(void *arg) { - const char *bufc = (const char *) buf; - int result; - - for(;;) { - result = write(fd, bufc, count); - if(result <= 0) { - return result; - } - count -= result; - bufc += result; + pthread_mutex_lock(&active_buffers_mutex); + active_buffers--; + pthread_mutex_unlock(&active_buffers_mutex); +} - if(count == 0) { - break; +int create_dir_if_needed(char *dir) +{ + int result; + result = mkdir(dir, 0777); + if(result == -1) { + if(errno != EEXIST) { + PERROR("mkdir"); + return -1; } } - return bufc-(const char *)buf; -} - -int get_subbuffer_died(struct buffer_info *buf) -{ return 0; } -//int ltt_do_get_subbuf(struct rchan_buf *buf, struct ltt_channel_buf_struct *ltt_buf, long *pconsumed_old) -//{ -// struct ltt_channel_buf_struct *ltt_buf = buf->bufstruct_mem; -// -////ust// struct ltt_channel_struct *ltt_channel = (struct ltt_channel_struct *)buf->chan->private_data; -// long consumed_old, consumed_idx, commit_count, write_offset; -// consumed_old = atomic_long_read(<t_buf->consumed); -// consumed_idx = SUBBUF_INDEX(consumed_old, buf->chan); -// commit_count = local_read(<t_buf->commit_count[consumed_idx]); -// /* -// * Make sure we read the commit count before reading the buffer -// * data and the write offset. Correct consumed offset ordering -// * wrt commit count is insured by the use of cmpxchg to update -// * the consumed offset. -// */ -// smp_rmb(); -// write_offset = local_read(<t_buf->offset); -// /* -// * Check that the subbuffer we are trying to consume has been -// * already fully committed. -// */ -// if (((commit_count - buf->chan->subbuf_size) -// & ltt_channel->commit_count_mask) -// - (BUFFER_TRUNC(consumed_old, buf->chan) -// >> ltt_channel->n_subbufs_order) -// != 0) { -// return -EAGAIN; -// } -// /* -// * Check that we are not about to read the same subbuffer in -// * which the writer head is. -// */ -// if ((SUBBUF_TRUNC(write_offset, buf->chan) -// - SUBBUF_TRUNC(consumed_old, buf->chan)) -// == 0) { -// return -EAGAIN; -// } -// -// *pconsumed_old = consumed_old; -// return 0; -//} - -void *consumer_thread(void *arg) +int is_directory(const char *dir) { - struct buffer_info *buf = (struct buffer_info *) arg; int result; - int died = 0; + struct stat st; - for(;;) { - /* get the subbuffer */ - if(died == 0) { - result = get_subbuffer(buf); - if(result == -1) { - ERR("error getting subbuffer"); - continue; - } - else if(result == GET_SUBBUF_DONE) { - /* this is done */ - break; - } - else if(result == GET_SUBBUF_DIED) { - died = 1; - } - } - if(died == 1) { - result = get_subbuffer_died(buf); - if(result <= 0) { - break; - } - } - - /* write data to file */ - result = patient_write(buf->file_fd, buf->mem + (buf->consumed_old & (buf->n_subbufs * buf->subbuf_size-1)), buf->subbuf_size); - if(result == -1) { - PERROR("write"); - /* FIXME: maybe drop this trace */ - } - - /* put the subbuffer */ - if(died == 0) { - result = put_subbuffer(buf); - if(result == -1) { - ERR("error putting subbuffer"); - break; - } - } - else { -// result = put_subbuffer_died(buf); - } + result = stat(dir, &st); + if(result == -1) { + PERROR("stat"); + return 0; } - DBG("thread for buffer %s is stopping", buf->name); - - /* FIXME: destroy, unalloc... */ + if(!S_ISDIR(st.st_mode)) { + return 0; + } - return NULL; + return 1; } -int add_buffer(pid_t pid, char *bufname) +struct buffer_info *connect_buffer(pid_t pid, const char *bufname) { struct buffer_info *buf; char *send_msg; @@ -281,12 +213,12 @@ int add_buffer(pid_t pid, char *bufname) int result; char *tmp; int fd; - pthread_t thr; + struct shmid_ds shmds; buf = (struct buffer_info *) malloc(sizeof(struct buffer_info)); if(buf == NULL) { ERR("add_buffer: insufficient memory"); - return -1; + return NULL; } buf->name = bufname; @@ -295,33 +227,57 @@ int add_buffer(pid_t pid, char *bufname) /* connect to app */ result = ustcomm_connect_app(buf->pid, &buf->conn); if(result) { - ERR("unable to connect to process"); - return -1; + WARN("unable to connect to process, it probably died before we were able to connect"); + return NULL; } + /* get pidunique */ + asprintf(&send_msg, "get_pidunique"); + result = ustcomm_send_request(&buf->conn, send_msg, &received_msg); + free(send_msg); + if(result == -1) { + ERR("problem in ustcomm_send_request(get_pidunique)"); + return NULL; + } + + result = sscanf(received_msg, "%lld", &buf->pidunique); + if(result != 1) { + ERR("unable to parse response to get_pidunique"); + return NULL; + } + free(received_msg); + DBG("got pidunique %lld", buf->pidunique); + /* get shmid */ asprintf(&send_msg, "get_shmid %s", buf->name); - ustcomm_send_request(&buf->conn, send_msg, &received_msg); + result = ustcomm_send_request(&buf->conn, send_msg, &received_msg); free(send_msg); - DBG("got buffer name %s", buf->name); + if(result == -1) { + ERR("problem in ustcomm_send_request(get_shmid)"); + return NULL; + } result = sscanf(received_msg, "%d %d", &buf->shmid, &buf->bufstruct_shmid); if(result != 2) { ERR("unable to parse response to get_shmid"); - return -1; + return NULL; } free(received_msg); DBG("got shmids %d %d", buf->shmid, buf->bufstruct_shmid); /* get n_subbufs */ asprintf(&send_msg, "get_n_subbufs %s", buf->name); - ustcomm_send_request(&buf->conn, send_msg, &received_msg); + result = ustcomm_send_request(&buf->conn, send_msg, &received_msg); free(send_msg); + if(result == -1) { + ERR("problem in ustcomm_send_request(g_n_subbufs)"); + return NULL; + } result = sscanf(received_msg, "%d", &buf->n_subbufs); if(result != 1) { ERR("unable to parse response to get_n_subbufs"); - return -1; + return NULL; } free(received_msg); DBG("got n_subbufs %d", buf->n_subbufs); @@ -334,7 +290,7 @@ int add_buffer(pid_t pid, char *bufname) result = sscanf(received_msg, "%d", &buf->subbuf_size); if(result != 1) { ERR("unable to parse response to get_subbuf_size"); - return -1; + return NULL; } free(received_msg); DBG("got subbuf_size %d", buf->subbuf_size); @@ -342,46 +298,338 @@ int add_buffer(pid_t pid, char *bufname) /* attach memory */ buf->mem = shmat(buf->shmid, NULL, 0); if(buf->mem == (void *) 0) { - perror("shmat"); - return -1; + PERROR("shmat"); + return NULL; } DBG("successfully attached buffer memory"); buf->bufstruct_mem = shmat(buf->bufstruct_shmid, NULL, 0); if(buf->bufstruct_mem == (void *) 0) { - perror("shmat"); - return -1; + PERROR("shmat"); + return NULL; } DBG("successfully attached buffer bufstruct memory"); + /* obtain info on the memory segment */ + result = shmctl(buf->shmid, IPC_STAT, &shmds); + if(result == -1) { + PERROR("shmctl"); + return NULL; + } + buf->memlen = shmds.shm_segsz; + /* open file for output */ - asprintf(&tmp, "/tmp/trace/%s_0", buf->name); - result = fd = open(tmp, O_WRONLY | O_CREAT | O_TRUNC, 00600); + if(!trace_path) { + /* Only create the directory if using the default path, because + * of the risk of typo when using trace path override. We don't + * want to risk creating plenty of useless directories in that case. + */ + result = create_dir_if_needed(USTD_DEFAULT_TRACE_PATH); + if(result == -1) { + ERR("could not create directory %s", USTD_DEFAULT_TRACE_PATH); + return NULL; + } + + trace_path = USTD_DEFAULT_TRACE_PATH; + } + + asprintf(&tmp, "%s/%u_%lld", trace_path, buf->pid, buf->pidunique); + result = create_dir_if_needed(tmp); + if(result == -1) { + ERR("could not create directory %s", tmp); + free(tmp); + return NULL; + } + free(tmp); + + asprintf(&tmp, "%s/%u_%lld/%s_0", trace_path, buf->pid, buf->pidunique, buf->name); + result = fd = open(tmp, O_WRONLY | O_CREAT | O_TRUNC | O_EXCL, 00600); if(result == -1) { PERROR("open"); - return -1; + ERR("failed opening trace file %s", tmp); + return NULL; } buf->file_fd = fd; free(tmp); - //list_add(&buf->list, &buffers); + pthread_mutex_lock(&active_buffers_mutex); + active_buffers++; + pthread_mutex_unlock(&active_buffers_mutex); + + return buf; +} + +int consumer_loop(struct buffer_info *buf) +{ + int result; + + pthread_cleanup_push(decrement_active_buffers, NULL); + + for(;;) { + /* get the subbuffer */ + result = get_subbuffer(buf); + if(result == -1) { + ERR("error getting subbuffer"); + continue; + } + else if(result == GET_SUBBUF_DONE) { + /* this is done */ + break; + } + else if(result == GET_SUBBUF_DIED) { + finish_consuming_dead_subbuffer(buf); + break; + } + + /* write data to file */ + result = patient_write(buf->file_fd, buf->mem + (buf->consumed_old & (buf->n_subbufs * buf->subbuf_size-1)), buf->subbuf_size); + if(result == -1) { + PERROR("write"); + /* FIXME: maybe drop this trace */ + } + + /* put the subbuffer */ + /* FIXME: we actually should unput the buffer before consuming... */ + result = put_subbuffer(buf); + if(result == -1) { + ERR("unknown error putting subbuffer (channel=%s)", buf->name); + break; + } + else if(result == PUT_SUBBUF_PUSHED) { + ERR("Buffer overflow (channel=%s), reader pushed. This channel will not be usable passed this point.", buf->name); + break; + } + else if(result == PUT_SUBBUF_DIED) { + WARN("application died while putting subbuffer"); + /* FIXME: probably need to skip the first subbuffer in finish_consuming_dead_subbuffer */ + finish_consuming_dead_subbuffer(buf); + break; + } + else if(result == PUT_SUBBUF_OK) { + } + } + + DBG("thread for buffer %s is stopping", buf->name); - pthread_create(&thr, NULL, consumer_thread, buf); + /* FIXME: destroy, unalloc... */ + + pthread_cleanup_pop(1); return 0; } -int main(int argc, char **argv) +void free_buffer(struct buffer_info *buf) +{ +} + +struct consumer_thread_args { + pid_t pid; + const char *bufname; +}; + +void *consumer_thread(void *arg) +{ + struct buffer_info *buf = (struct buffer_info *) arg; + struct consumer_thread_args *args = (struct consumer_thread_args *) arg; + + DBG("GOT ARGS: pid %d bufname %s", args->pid, args->bufname); + + buf = connect_buffer(args->pid, args->bufname); + if(buf == NULL) { + ERR("failed to connect to buffer"); + goto end; + } + + consumer_loop(buf); + + free_buffer(buf); + + end: + /* bufname is free'd in free_buffer() */ + free(args); + return NULL; +} + +int start_consuming_buffer(pid_t pid, const char *bufname) +{ + pthread_t thr; + struct consumer_thread_args *args; + + DBG("beginning of start_consuming_buffer: args: pid %d bufname %s", pid, bufname); + + args = (struct consumer_thread_args *) malloc(sizeof(struct consumer_thread_args)); + + args->pid = pid; + args->bufname = strdup(bufname); + DBG("beginning2 of start_consuming_buffer: args: pid %d bufname %s", args->pid, args->bufname); + + pthread_create(&thr, NULL, consumer_thread, args); + DBG("end of start_consuming_buffer: args: pid %d bufname %s", args->pid, args->bufname); + + return 0; +} + +void usage(void) +{ + fprintf(stderr, "Usage:\nustd OPTIONS\n\nOptions:\n" + "\t-h\t\tDisplay this usage.\n" + "\t-o DIR\t\tSpecify the directory where to output the traces.\n" + "\t-s PATH\t\tSpecify the path to use for the daemon socket.\n" + "\t-d\t\tStart as a daemon.\n" + "\t--pidfile FILE\tWrite the PID in this file (when using -d).\n"); +} + +int parse_args(int argc, char **argv) +{ + int c; + + while (1) { + int option_index = 0; + static struct option long_options[] = { + {"pidfile", 1, 0, 'p'}, + {"help", 0, 0, 'h'}, + {"version", 0, 0, 'V'}, + {0, 0, 0, 0} + }; + + c = getopt_long(argc, argv, "hs:o:d", long_options, &option_index); + if (c == -1) + break; + + switch (c) { + case 0: + printf("option %s", long_options[option_index].name); + if (optarg) + printf(" with arg %s", optarg); + printf("\n"); + break; + case 's': + sock_path = optarg; + break; + case 'o': + trace_path = optarg; + if(!is_directory(trace_path)) { + ERR("Not a valid directory. (%s)", trace_path); + return -1; + } + break; + case 'd': + daemon_mode = 1; + break; + case 'p': + pidfile = strdup(optarg); + break; + case 'h': + usage(); + exit(0); + case 'V': + printf("Version 0.0\n"); + break; + + default: + /* unknown option or other error; error is + printed by getopt, just return */ + return -1; + } + } + + return 0; +} + +void sigterm_handler(int sig) +{ + terminate_req = 1; +} + +static int write_pidfile(const char *file_name, pid_t pid) +{ + FILE *pidfp; + + pidfp = fopen(file_name, "w+"); + if(!pidfp) { + PERROR("fopen (%s)", pidfile); + WARN("killing child process"); + return -1; + } + + fprintf(pidfp, "%d\n", pid); + + fclose(pidfp); + + return 0; +} + +int start_ustd(int fd) { struct ustcomm_ustd ustd; int result; + sigset_t sigset; + struct sigaction sa; - result = ustcomm_init_ustd(&ustd); + result = sigemptyset(&sigset); + if(result == -1) { + PERROR("sigemptyset"); + return 1; + } + sa.sa_handler = sigterm_handler; + sa.sa_mask = sigset; + sa.sa_flags = SA_RESTART; + result = sigaction(SIGTERM, &sa, NULL); + if(result == -1) { + PERROR("sigaction"); + return 1; + } + + result = ustcomm_init_ustd(&ustd, sock_path); if(result == -1) { ERR("failed to initialize socket"); return 1; } + /* setup handler for SIGPIPE */ + result = sigemptyset(&sigset); + if(result == -1) { + PERROR("sigemptyset"); + return 1; + } + result = sigaddset(&sigset, SIGPIPE); + if(result == -1) { + PERROR("sigaddset"); + return 1; + } + result = sigprocmask(SIG_BLOCK, &sigset, NULL); + if(result == -1) { + PERROR("sigprocmask"); + return 1; + } + + /* Write pidfile */ + if(pidfile) { + result = write_pidfile(pidfile, getpid()); + if(result == -1) { + ERR("failed to write pidfile"); + return 1; + } + } + + /* Notify parent that we are successfully started. */ + if(fd != -1) { + /* write any one character */ + result = write(fd, "!", 1); + if(result == -1) { + PERROR("write"); + return -1; + } + if(result != 1) { + ERR("Problem sending confirmation of daemon start to parent"); + return -1; + } + result = close(fd); + if(result == -1) { + PERROR("close"); + } + } + /* app loop */ for(;;) { char *recvbuf; @@ -390,7 +638,7 @@ int main(int argc, char **argv) result = ustcomm_ustd_recv_message(&ustd, &recvbuf, NULL, 100); if(result == -1) { ERR("error in ustcomm_ustd_recv_message"); - continue; + goto loop_end; } if(result > 0) { if(!strncmp(recvbuf, "collect", 7)) { @@ -400,19 +648,97 @@ int main(int argc, char **argv) result = sscanf(recvbuf, "%*s %d %50as", &pid, &bufname); if(result != 2) { - fprintf(stderr, "parsing error: %s\n", recvbuf); + ERR("parsing error: %s", recvbuf); + goto free_bufname; } - result = add_buffer(pid, bufname); + result = start_consuming_buffer(pid, bufname); if(result < 0) { ERR("error in add_buffer"); - continue; + goto free_bufname; } + + free_bufname: + free(bufname); } free(recvbuf); } + + loop_end: + + if(terminate_req) { + pthread_mutex_lock(&active_buffers_mutex); + if(active_buffers == 0) { + pthread_mutex_unlock(&active_buffers_mutex); + break; + } + pthread_mutex_unlock(&active_buffers_mutex); + } + } + + return 0; +} + +int start_ustd_daemon() +{ + int result; + int fd[2]; + pid_t child_pid; + + result = pipe(fd); + + result = child_pid = fork(); + if(result == -1) { + PERROR("fork"); + return -1; + } + else if(result == 0) { + return start_ustd(fd[1]); + } + else { + char buf; + FILE *pidfp; + + result = read(fd[0], &buf, 1); + if(result == -1) { + PERROR("read"); + return -1; + } + if(result != 1) { + ERR("did not receive valid confirmation that the daemon is started"); + return -1; + } + + result = close(fd[0]); + if(result == -1) { + PERROR("close"); + } + + DBG("The daemon is now successfully started"); } + /* Wait for confirmation that the server is ready. */ + + return 0; } + +int main(int argc, char **argv) +{ + int result; + + result = parse_args(argc, argv); + if(result == -1) { + exit(1); + } + + if(daemon_mode) { + result = start_ustd_daemon(); + } + else { + result = start_ustd(-1); + } + + return result; +}