X-Git-Url: http://git.lttng.org./?a=blobdiff_plain;f=ustd%2Fustd.c;h=24087d324a54311b1bb780b3e26eb4d8db3622d7;hb=00c475d60474005a643313f8f80f23ac908d8ae4;hp=a70f2f70a209f3436864ed50efac25275ab20000;hpb=c97d443730716166051e0b6b95d26ab3faadf8dc;p=lttng-ust.git diff --git a/ustd/ustd.c b/ustd/ustd.c index a70f2f70..24087d32 100644 --- a/ustd/ustd.c +++ b/ustd/ustd.c @@ -18,6 +18,7 @@ #define _GNU_SOURCE #include +#include #include #include #include @@ -29,10 +30,12 @@ #include #include #include +#include #include "ustd.h" #include "localerr.h" #include "ustcomm.h" +#include "share.h" /* return value: 0 = subbuffer is finished, it won't produce data anymore * 1 = got subbuffer successfully @@ -48,69 +51,40 @@ #define PUT_SUBBUF_PUSHED 2 char *sock_path=NULL; - -int test_sigpipe(void) -{ - sigset_t sigset; - int result; +char *trace_path=NULL; - result = sigemptyset(&sigset); - if(result == -1) { - perror("sigemptyset"); - return -1; - } - result = sigaddset(&sigset, SIGPIPE); - if(result == -1) { - perror("sigaddset"); - return -1; - } - - result = sigtimedwait(&sigset, NULL, &(struct timespec){0,0}); - if(result == -1 && errno == EAGAIN) { - /* no signal received */ - return 0; - } - else if(result == -1) { - perror("sigtimedwait"); - return -1; - } - else if(result == SIGPIPE) { - /* received sigpipe */ - return 1; - } - else { - assert(0); - } -} +/* 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(test_sigpipe()) { - WARN("process %d destroyed before we could connect to it", buf->pid); - return GET_SUBBUF_DONE; + if((result == -1 && errno == EPIPE) || result == 0) { + DBG("app died while being traced"); + retval = GET_SUBBUF_DIED; + goto end; } else if(result < 0) { ERR("get_subbuffer: ustcomm_send_request failed"); - return -1; - } - else if(result == 0) { - DBG("app died while being traced"); - return GET_SUBBUF_DIED; + 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); @@ -120,7 +94,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); @@ -128,33 +103,44 @@ 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); @@ -162,35 +148,28 @@ int put_subbuffer(struct buffer_info *buf) } else { DBG("put_subbuffer: received error, we were pushed"); - return PUT_SUBBUF_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; } -/* This write is patient because it restarts if it was incomplete. - */ - -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; - - if(count == 0) { - break; - } - } - - return bufc-(const char *)buf; + pthread_mutex_lock(&active_buffers_mutex); + active_buffers--; + pthread_mutex_unlock(&active_buffers_mutex); } void *consumer_thread(void *arg) @@ -198,6 +177,8 @@ void *consumer_thread(void *arg) struct buffer_info *buf = (struct buffer_info *) arg; int result; + pthread_cleanup_push(decrement_active_buffers, NULL); + for(;;) { /* get the subbuffer */ result = get_subbuffer(buf); @@ -222,6 +203,7 @@ void *consumer_thread(void *arg) } /* 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); @@ -235,6 +217,7 @@ void *consumer_thread(void *arg) 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) { } @@ -244,6 +227,8 @@ void *consumer_thread(void *arg) /* FIXME: destroy, unalloc... */ + pthread_cleanup_pop(1); + return NULL; } @@ -253,7 +238,7 @@ int create_dir_if_needed(char *dir) result = mkdir(dir, 0777); if(result == -1) { if(errno != EEXIST) { - perror("mkdir"); + PERROR("mkdir"); return -1; } } @@ -261,6 +246,24 @@ int create_dir_if_needed(char *dir) return 0; } +int is_directory(const char *dir) +{ + int result; + struct stat st; + + result = stat(dir, &st); + if(result == -1) { + PERROR("stat"); + return 0; + } + + if(!S_ISDIR(st.st_mode)) { + return 0; + } + + return 1; +} + int add_buffer(pid_t pid, char *bufname) { struct buffer_info *buf; @@ -288,6 +291,23 @@ int add_buffer(pid_t pid, char *bufname) return -1; } + /* 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 -1; + } + + result = sscanf(received_msg, "%lld", &buf->pidunique); + if(result != 1) { + ERR("unable to parse response to get_pidunique"); + return -1; + } + free(received_msg); + DBG("got pidunique %lld", buf->pidunique); + /* get shmid */ asprintf(&send_msg, "get_shmid %s", buf->name); result = ustcomm_send_request(&buf->conn, send_msg, &received_msg); @@ -338,14 +358,14 @@ int add_buffer(pid_t pid, char *bufname) /* attach memory */ buf->mem = shmat(buf->shmid, NULL, 0); if(buf->mem == (void *) 0) { - perror("shmat"); + PERROR("shmat"); return -1; } DBG("successfully attached buffer memory"); buf->bufstruct_mem = shmat(buf->bufstruct_shmid, NULL, 0); if(buf->bufstruct_mem == (void *) 0) { - perror("shmat"); + PERROR("shmat"); return -1; } DBG("successfully attached buffer bufstruct memory"); @@ -353,19 +373,27 @@ int add_buffer(pid_t pid, char *bufname) /* obtain info on the memory segment */ result = shmctl(buf->shmid, IPC_STAT, &shmds); if(result == -1) { - perror("shmctl"); + PERROR("shmctl"); return -1; } buf->memlen = shmds.shm_segsz; /* open file for output */ - result = create_dir_if_needed(USTD_DEFAULT_TRACE_PATH); - if(result == -1) { - ERR("could not create directory %s", USTD_DEFAULT_TRACE_PATH); - return -1; + 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 -1; + } + + trace_path = USTD_DEFAULT_TRACE_PATH; } - asprintf(&tmp, "%s/%u", USTD_DEFAULT_TRACE_PATH, buf->pid); + 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); @@ -374,8 +402,8 @@ int add_buffer(pid_t pid, char *bufname) } free(tmp); - asprintf(&tmp, "%s/%u/%s_0", USTD_DEFAULT_TRACE_PATH, buf->pid, buf->name); - result = fd = open(tmp, O_WRONLY | O_CREAT | O_TRUNC, 00600); + 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"); ERR("failed opening trace file %s", tmp); @@ -384,36 +412,124 @@ int add_buffer(pid_t pid, char *bufname) buf->file_fd = fd; free(tmp); + pthread_mutex_lock(&active_buffers_mutex); + active_buffers++; + pthread_mutex_unlock(&active_buffers_mutex); + pthread_create(&thr, NULL, consumer_thread, buf); 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"); +} + +int parse_args(int argc, char **argv) +{ + int c; + + while (1) { + int option_index = 0; + static struct option long_options[] = { + {"help", 0, 0, 'h'}, + {"version", 0, 0, 'V'}, + {0, 0, 0, 0} + }; + + c = getopt_long(argc, argv, "hs:o:", 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 '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; +} + int main(int argc, char **argv) { 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 = parse_args(argc, argv); + if(result == -1) { + exit(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"); + PERROR("sigemptyset"); return 1; } result = sigaddset(&sigset, SIGPIPE); if(result == -1) { - perror("sigaddset"); + PERROR("sigaddset"); return 1; } result = sigprocmask(SIG_BLOCK, &sigset, NULL); if(result == -1) { - perror("sigprocmask"); + PERROR("sigprocmask"); return 1; } @@ -447,6 +563,15 @@ int main(int argc, char **argv) free(recvbuf); } + + 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;