X-Git-Url: http://git.lttng.org./?a=blobdiff_plain;f=ustd%2Fustd.c;h=39ad7be3122bb19f67a8ba304570e0f8e01c7cf7;hb=a3cdd4a7fac4691fc61fcd3c9e6ad4a783fa6113;hp=a4c7e1f3f0c8a6dbdd17ac2fbf349a3fe7cb6f9d;hpb=8cefc1450e2d929de5d2878518e7b8b2b77ff755;p=lttng-ust.git diff --git a/ustd/ustd.c b/ustd/ustd.c index a4c7e1f3..39ad7be3 100644 --- a/ustd/ustd.c +++ b/ustd/ustd.c @@ -22,43 +22,18 @@ #include #include #include +#include #include #include #include +#include +#include +#include "ustd.h" #include "localerr.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,6 +43,44 @@ 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 + +int test_sigpipe(void) +{ + sigset_t sigset; + int result; + + 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); + } +} + int get_subbuffer(struct buffer_info *buf) { char *send_msg; @@ -79,7 +92,11 @@ int get_subbuffer(struct buffer_info *buf) asprintf(&send_msg, "get_subbuffer %s", buf->name); result = ustcomm_send_request(&buf->conn, send_msg, &received_msg); free(send_msg); - if(result < 0) { + if(test_sigpipe()) { + WARN("process %d destroyed before we could connect to it", buf->pid); + return GET_SUBBUF_DONE; + } + else if(result < 0) { ERR("get_subbuffer: ustcomm_send_request failed"); return -1; } @@ -139,16 +156,20 @@ int put_subbuffer(struct buffer_info *buf) 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"); + return PUT_SUBBUF_PUSHED; } free(rep_code); 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) { const char *bufc = (const char *) buf; @@ -170,80 +191,25 @@ ssize_t patient_write(int fd, const void *buf, size_t count) 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) { struct buffer_info *buf = (struct buffer_info *) arg; int result; - int died = 0; 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; - } + result = get_subbuffer(buf); + if(result == -1) { + ERR("error getting subbuffer"); + continue; } - if(died == 1) { - result = get_subbuffer_died(buf); - if(result <= 0) { - break; - } + 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 */ @@ -254,15 +220,21 @@ void *consumer_thread(void *arg) } /* put the subbuffer */ - if(died == 0) { - result = put_subbuffer(buf); - if(result == -1) { - ERR("error putting subbuffer"); - break; - } + 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 { -// result = put_subbuffer_died(buf); + 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); + } + else if(result == PUT_SUBBUF_OK) { } } @@ -282,6 +254,7 @@ int add_buffer(pid_t pid, char *bufname) char *tmp; int fd; pthread_t thr; + struct shmid_ds shmds; buf = (struct buffer_info *) malloc(sizeof(struct buffer_info)); if(buf == NULL) { @@ -295,15 +268,18 @@ 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"); + WARN("unable to connect to process, it probably died before we were able to connect"); return -1; } /* 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 -1; + } result = sscanf(received_msg, "%d %d", &buf->shmid, &buf->bufstruct_shmid); if(result != 2) { @@ -315,8 +291,12 @@ int add_buffer(pid_t pid, char *bufname) /* 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 -1; + } result = sscanf(received_msg, "%d", &buf->n_subbufs); if(result != 1) { @@ -354,18 +334,25 @@ int add_buffer(pid_t pid, char *bufname) } 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 -1; + } + 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(result == -1) { PERROR("open"); + ERR("failed opening trace file %s", tmp); return -1; } buf->file_fd = fd; free(tmp); - //list_add(&buf->list, &buffers); - pthread_create(&thr, NULL, consumer_thread, buf); return 0; @@ -375,6 +362,7 @@ int main(int argc, char **argv) { struct ustcomm_ustd ustd; int result; + sigset_t sigset; result = ustcomm_init_ustd(&ustd); if(result == -1) { @@ -382,6 +370,22 @@ int main(int argc, char **argv) return 1; } + 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; + } + /* app loop */ for(;;) { char *recvbuf;