block all signals in listener thread
[ust.git] / libust / tracectl.c
index 5439eed5cd0b3c2970fe56292d16b456fe77ec75..7a5cf9562282ce7fe90cc3d5e4c405d3bf44b8c6 100644 (file)
 
 #define _GNU_SOURCE
 #include <stdio.h>
+#include <stdlib.h>
 #include <stdint.h>
+#include <pthread.h>
 #include <signal.h>
 #include <sys/types.h>
 #include <sys/socket.h>
 #include <sys/un.h>
-#include <sched.h>
 #include <fcntl.h>
 #include <poll.h>
 #include <regex.h>
+#include <urcu/uatomic_arch.h>
 
-#include <urcu.h>
-
-#include "marker.h"
+#include <ust/marker.h>
+#include <ust/tracectl.h>
 #include "tracer.h"
-#include "localerr.h"
+#include "usterr.h"
 #include "ustcomm.h"
-#include "relay.h" /* FIXME: remove */
+#include "buffers.h"
 #include "marker-control.h"
-
-//#define USE_CLONE
+#include "multipoll.h"
 
 #define USTSIGNAL SIGIO
 
 #define MSG_NOTIF 1
 #define MSG_REGISTER_NOTIF 2
 
-char consumer_stack[10000];
+/* This should only be accessed by the constructor, before the creation
+ * of the listener, and then only by the listener.
+ */
+s64 pidunique = -1LL;
 
 struct list_head blocked_consumers = LIST_HEAD_INIT(blocked_consumers);
 
@@ -55,9 +58,8 @@ struct tracecmd { /* no padding */
        uint16_t command;
 };
 
-//struct listener_arg {
-//     int pipe_fd;
-//};
+/* volatile because shared between the listener and the main thread */
+int buffers_to_export = 0;
 
 struct trctl_msg {
        /* size: the size of all the fields except size itself */
@@ -83,13 +85,26 @@ struct blocked_consumer {
        struct ustcomm_server server;
        struct ustcomm_source src;
 
-       /* args to ltt_do_get_subbuf */
-       struct rchan_buf *rbuf;
-       struct ltt_channel_buf_struct *lttbuf;
+       /* args to ust_buffers_get_subbuf */
+       struct ust_buffer *buf;
 
        struct list_head list;
 };
 
+static long long make_pidunique(void)
+{
+       s64 retval;
+       struct timeval tv;
+       
+       gettimeofday(&tv, NULL);
+
+       retval = tv.tv_sec;
+       retval <<= 32;
+       retval |= tv.tv_usec;
+
+       return retval;
+}
+
 static void print_markers(FILE *fp)
 {
        struct marker_iter iter;
@@ -99,7 +114,7 @@ static void print_markers(FILE *fp)
        marker_iter_start(&iter);
 
        while(iter.marker) {
-               fprintf(fp, "marker: %s_%s %d \"%s\"\n", iter.marker->channel, iter.marker->name, (int)imv_read(iter.marker->state), iter.marker->format);
+               fprintf(fp, "marker: %s/%s %d \"%s\" %p\n", iter.marker->channel, iter.marker->name, (int)imv_read(iter.marker->state), iter.marker->format, iter.marker->location);
                marker_iter_next(&iter);
        }
        unlock_markers();
@@ -107,37 +122,6 @@ static void print_markers(FILE *fp)
 
 static int init_socket(void);
 
-/* This needs to be called whenever a new thread is created. It notifies
- * liburcu of the new thread.
- */
-
-void ust_register_thread(void)
-{
-       rcu_register_thread();
-}
-
-int fd_notif = -1;
-void notif_cb(void)
-{
-       int result;
-       struct trctl_msg msg;
-
-       /* FIXME: fd_notif should probably be protected by a spinlock */
-
-       if(fd_notif == -1)
-               return;
-
-       msg.type = MSG_NOTIF;
-       msg.size = sizeof(msg.type);
-
-       /* FIXME: don't block here */
-       result = write(fd_notif, &msg, msg.size+sizeof(msg.size));
-       if(result == -1) {
-               PERROR("write");
-               return;
-       }
-}
-
 /* Ask the daemon to collect a trace called trace_name and being
  * produced by this pid.
  *
@@ -147,8 +131,8 @@ void notif_cb(void)
 
 static void inform_consumer_daemon(const char *trace_name)
 {
-       int i;
-       struct ltt_trace_struct *trace;
+       int i,j;
+       struct ust_trace *trace;
        pid_t pid = getpid();
        int result;
 
@@ -161,10 +145,19 @@ static void inform_consumer_daemon(const char *trace_name)
        }
 
        for(i=0; i < trace->nr_channels; i++) {
-               result = ustcomm_request_consumer(pid, trace->channels[i].channel_name);
-               if(result == -1) {
-                       WARN("Failed to request collection for channel %s. Is the daemon available?", trace->channels[i].channel_name);
-                       /* continue even if fail */
+               if(trace->channels[i].request_collection) {
+                       /* iterate on all cpus */
+                       for(j=0; j<trace->channels[i].n_cpus; j++) {
+                               char *buf;
+                               asprintf(&buf, "%s_%d", trace->channels[i].channel_name, j);
+                               result = ustcomm_request_consumer(pid, buf);
+                               if(result == -1) {
+                                       WARN("Failed to request collection for channel %s. Is the daemon available?", trace->channels[i].channel_name);
+                                       /* continue even if fail */
+                               }
+                               free(buf);
+                               STORE_SHARED(buffers_to_export, LOAD_SHARED(buffers_to_export)+1);
+                       }
                }
        }
 
@@ -172,667 +165,911 @@ static void inform_consumer_daemon(const char *trace_name)
        ltt_unlock_traces();
 }
 
-void process_blocked_consumers(void)
+int process_blkd_consumer_act(void *priv, int fd, short events)
 {
-       int n_fds = 0;
-       struct pollfd *fds;
-       struct blocked_consumer *bc;
-       int idx = 0;
-       char inbuf;
        int result;
+       long consumed_old = 0;
+       char *reply;
+       struct blocked_consumer *bc = (struct blocked_consumer *) priv;
+       char inbuf;
 
-       list_for_each_entry(bc, &blocked_consumers, list) {
-               n_fds++;
+       result = read(bc->fd_producer, &inbuf, 1);
+       if(result == -1) {
+               PERROR("read");
+               return -1;
        }
+       if(result == 0) {
+               int res;
+               DBG("listener: got messsage that a buffer ended");
 
-       fds = (struct pollfd *) malloc(n_fds * sizeof(struct pollfd));
-       if(fds == NULL) {
-               ERR("malloc returned NULL");
-               return;
+               res = close(bc->fd_producer);
+               if(res == -1) {
+                       PERROR("close");
+               }
+
+               list_del(&bc->list);
+
+               result = ustcomm_send_reply(&bc->server, "END", &bc->src);
+               if(result < 0) {
+                       ERR("ustcomm_send_reply failed");
+                       return -1;
+               }
+
+               return 0;
+       }
+
+       result = ust_buffers_get_subbuf(bc->buf, &consumed_old);
+       if(result == -EAGAIN) {
+               WARN("missed buffer?");
+               return 0;
        }
+       else if(result < 0) {
+               ERR("ust_buffers_get_subbuf: error: %s", strerror(-result));
+       }
+       asprintf(&reply, "%s %ld", "OK", consumed_old);
+       result = ustcomm_send_reply(&bc->server, reply, &bc->src);
+       if(result < 0) {
+               ERR("ustcomm_send_reply failed");
+               free(reply);
+               return -1;
+       }
+       free(reply);
+
+       list_del(&bc->list);
+
+       return 0;
+}
+
+void blocked_consumers_add_to_mp(struct mpentries *ent)
+{
+       struct blocked_consumer *bc;
 
        list_for_each_entry(bc, &blocked_consumers, list) {
-               fds[idx].fd = bc->fd_producer;
-               fds[idx].events = POLLIN;
-               bc->tmp_poll_idx = idx;
-               idx++;
+               multipoll_add(ent, bc->fd_producer, POLLIN, process_blkd_consumer_act, bc, NULL);
        }
 
-       while((result = poll(fds, n_fds, 0)) == -1 && errno == EINTR)
-               /* nothing */;
-       if(result == -1) {
-               PERROR("poll");
-               return;
+}
+
+void seperate_channel_cpu(const char *channel_and_cpu, char **channel, int *cpu)
+{
+       const char *sep;
+
+       sep = rindex(channel_and_cpu, '_');
+       if(sep == NULL) {
+               *cpu = -1;
+               sep = channel_and_cpu + strlen(channel_and_cpu);
+       }
+       else {
+               *cpu = atoi(sep+1);
        }
 
-       list_for_each_entry(bc, &blocked_consumers, list) {
-               if(fds[bc->tmp_poll_idx].revents) {
-                       long consumed_old = 0;
-                       char *reply;
+       asprintf(channel, "%.*s", (int)(sep-channel_and_cpu), channel_and_cpu);
+}
 
-                       result = read(bc->fd_producer, &inbuf, 1);
-                       if(result == -1) {
-                               PERROR("read");
-                               continue;
-                       }
-                       if(result == 0) {
-                               DBG("PRODUCER END");
+static int do_cmd_get_shmid(const char *recvbuf, struct ustcomm_source *src)
+{
+       int retval = 0;
+       struct ust_trace *trace;
+       char trace_name[] = "auto";
+       int i;
+       char *channel_and_cpu;
+       int found = 0;
+       int result;
+       char *ch_name;
+       int ch_cpu;
 
-                               close(bc->fd_producer);
+       DBG("get_shmid");
 
-                               list_del(&bc->list);
+       channel_and_cpu = nth_token(recvbuf, 1);
+       if(channel_and_cpu == NULL) {
+               ERR("cannot parse channel");
+               goto end;
+       }
 
-                               result = ustcomm_send_reply(&bc->server, "END", &bc->src);
-                               if(result < 0) {
-                                       ERR("ustcomm_send_reply failed");
-                                       continue;
-                               }
+       seperate_channel_cpu(channel_and_cpu, &ch_name, &ch_cpu);
+       if(ch_cpu == -1) {
+               ERR("problem parsing channel name");
+               goto free_short_chan_name;
+       }
 
-                               continue;
-                       }
+       ltt_lock_traces();
+       trace = _ltt_trace_find(trace_name);
+       ltt_unlock_traces();
 
-                       result = ltt_do_get_subbuf(bc->rbuf, bc->lttbuf, &consumed_old);
-                       if(result == -EAGAIN) {
-                               WARN("missed buffer?");
-                               continue;
-                       }
-                       else if(result < 0) {
-                               DBG("ltt_do_get_subbuf: error: %s", strerror(-result));
-                       }
-                       asprintf(&reply, "%s %ld", "OK", consumed_old);
-                       result = ustcomm_send_reply(&bc->server, reply, &bc->src);
-                       if(result < 0) {
+       if(trace == NULL) {
+               ERR("cannot find trace!");
+               retval = -1;
+               goto free_short_chan_name;
+       }
+
+       for(i=0; i<trace->nr_channels; i++) {
+               struct ust_channel *channel = &trace->channels[i];
+               struct ust_buffer *buf = channel->buf[ch_cpu];
+
+               if(!strcmp(trace->channels[i].channel_name, ch_name)) {
+                       char *reply;
+
+//                     DBG("the shmid for the requested channel is %d", buf->shmid);
+//                     DBG("the shmid for its buffer structure is %d", channel->buf_struct_shmids);
+                       asprintf(&reply, "%d %d", buf->shmid, channel->buf_struct_shmids[ch_cpu]);
+
+                       result = ustcomm_send_reply(&ustcomm_app.server, reply, src);
+                       if(result) {
                                ERR("ustcomm_send_reply failed");
                                free(reply);
-                               continue;
+                               retval = -1;
+                               goto free_short_chan_name;
                        }
+
                        free(reply);
 
-                       list_del(&bc->list);
+                       found = 1;
+                       break;
                }
        }
 
+       if(!found) {
+               ERR("channel not found (%s)", channel_and_cpu);
+       }
+
+       free_short_chan_name:
+       free(ch_name);
+
+       end:
+       return retval;
 }
 
-void *listener_main(void *p)
+static int do_cmd_get_n_subbufs(const char *recvbuf, struct ustcomm_source *src)
 {
+       int retval = 0;
+       struct ust_trace *trace;
+       char trace_name[] = "auto";
+       int i;
+       char *channel_and_cpu;
+       int found = 0;
        int result;
+       char *ch_name;
+       int ch_cpu;
 
-       ust_register_thread();
+       DBG("get_n_subbufs");
 
-       DBG("LISTENER");
+       channel_and_cpu = nth_token(recvbuf, 1);
+       if(channel_and_cpu == NULL) {
+               ERR("cannot parse channel");
+               goto end;
+       }
 
-       for(;;) {
-               char trace_name[] = "auto";
-               char trace_type[] = "ustrelay";
-               char *recvbuf;
-               int len;
-               struct ustcomm_source src;
+       seperate_channel_cpu(channel_and_cpu, &ch_name, &ch_cpu);
+       if(ch_cpu == -1) {
+               ERR("problem parsing channel name");
+               goto free_short_chan_name;
+       }
 
-               process_blocked_consumers();
+       ltt_lock_traces();
+       trace = _ltt_trace_find(trace_name);
+       ltt_unlock_traces();
 
-               result = ustcomm_app_recv_message(&ustcomm_app, &recvbuf, &src, 5);
-               if(result < 0) {
-                       WARN("error in ustcomm_app_recv_message");
-                       continue;
-               }
-               else if(result == 0) {
-                       /* no message */
-                       continue;
-               }
+       if(trace == NULL) {
+               ERR("cannot find trace!");
+               retval = -1;
+               goto free_short_chan_name;
+       }
 
-               DBG("received a message! it's: %s", recvbuf);
-               len = strlen(recvbuf);
+       for(i=0; i<trace->nr_channels; i++) {
+               struct ust_channel *channel = &trace->channels[i];
 
-               if(!strcmp(recvbuf, "print_markers")) {
-                       print_markers(stderr);
-               }
-               else if(!strcmp(recvbuf, "list_markers")) {
-                       char *ptr;
-                       size_t size;
-                       FILE *fp;
+               if(!strcmp(trace->channels[i].channel_name, ch_name)) {
+                       char *reply;
 
-                       fp = open_memstream(&ptr, &size);
-                       print_markers(fp);
-                       fclose(fp);
+                       DBG("the n_subbufs for the requested channel is %d", channel->subbuf_cnt);
+                       asprintf(&reply, "%d", channel->subbuf_cnt);
 
-                       result = ustcomm_send_reply(&ustcomm_app.server, ptr, &src);
+                       result = ustcomm_send_reply(&ustcomm_app.server, reply, src);
+                       if(result) {
+                               ERR("ustcomm_send_reply failed");
+                               free(reply);
+                               retval = -1;
+                               goto free_short_chan_name;
+                       }
 
-                       free(ptr);
+                       free(reply);
+                       found = 1;
+                       break;
                }
-               else if(!strcmp(recvbuf, "start")) {
-                       /* start is an operation that setups the trace, allocates it and starts it */
-                       result = ltt_trace_setup(trace_name);
-                       if(result < 0) {
-                               ERR("ltt_trace_setup failed");
-                               return (void *)1;
-                       }
+       }
+       if(found == 0) {
+               ERR("unable to find channel");
+       }
 
-                       result = ltt_trace_set_type(trace_name, trace_type);
-                       if(result < 0) {
-                               ERR("ltt_trace_set_type failed");
-                               return (void *)1;
-                       }
+       free_short_chan_name:
+       free(ch_name);
 
-                       result = ltt_trace_alloc(trace_name);
-                       if(result < 0) {
-                               ERR("ltt_trace_alloc failed");
-                               return (void *)1;
-                       }
+       end:
+       return retval;
+}
 
-                       inform_consumer_daemon(trace_name);
+static int do_cmd_get_subbuf_size(const char *recvbuf, struct ustcomm_source *src)
+{
+       int retval = 0;
+       struct ust_trace *trace;
+       char trace_name[] = "auto";
+       int i;
+       char *channel_and_cpu;
+       int found = 0;
+       int result;
+       char *ch_name;
+       int ch_cpu;
 
-                       result = ltt_trace_start(trace_name);
-                       if(result < 0) {
-                               ERR("ltt_trace_start failed");
-                               continue;
-                       }
-               }
-               else if(!strcmp(recvbuf, "trace_setup")) {
-                       DBG("trace setup");
+       DBG("get_subbuf_size");
 
-                       result = ltt_trace_setup(trace_name);
-                       if(result < 0) {
-                               ERR("ltt_trace_setup failed");
-                               return (void *)1;
-                       }
+       channel_and_cpu = nth_token(recvbuf, 1);
+       if(channel_and_cpu == NULL) {
+               ERR("cannot parse channel");
+               goto end;
+       }
 
-                       result = ltt_trace_set_type(trace_name, trace_type);
-                       if(result < 0) {
-                               ERR("ltt_trace_set_type failed");
-                               return (void *)1;
+       seperate_channel_cpu(channel_and_cpu, &ch_name, &ch_cpu);
+       if(ch_cpu == -1) {
+               ERR("problem parsing channel name");
+               goto free_short_chan_name;
+       }
+
+       ltt_lock_traces();
+       trace = _ltt_trace_find(trace_name);
+       ltt_unlock_traces();
+
+       if(trace == NULL) {
+               ERR("cannot find trace!");
+               retval = -1;
+               goto free_short_chan_name;
+       }
+
+       for(i=0; i<trace->nr_channels; i++) {
+               struct ust_channel *channel = &trace->channels[i];
+
+               if(!strcmp(trace->channels[i].channel_name, ch_name)) {
+                       char *reply;
+
+                       DBG("the subbuf_size for the requested channel is %zd", channel->subbuf_size);
+                       asprintf(&reply, "%zd", channel->subbuf_size);
+
+                       result = ustcomm_send_reply(&ustcomm_app.server, reply, src);
+                       if(result) {
+                               ERR("ustcomm_send_reply failed");
+                               free(reply);
+                               retval = -1;
+                               goto free_short_chan_name;
                        }
+
+                       free(reply);
+                       found = 1;
+                       break;
                }
-               else if(!strcmp(recvbuf, "trace_alloc")) {
-                       DBG("trace alloc");
+       }
+       if(found == 0) {
+               ERR("unable to find channel");
+       }
 
-                       result = ltt_trace_alloc(trace_name);
-                       if(result < 0) {
-                               ERR("ltt_trace_alloc failed");
-                               return (void *)1;
-                       }
+       free_short_chan_name:
+       free(ch_name);
+
+       end:
+       return retval;
+}
+
+/* Return the power of two which is equal or higher to v */
+
+static unsigned int pow2_higher_or_eq(unsigned int v)
+{
+       int hb = fls(v);
+       int hbm1 = hb-1;
+       int retval = 1<<(hb-1);
+
+       if(v-retval == 0)
+               return retval;
+       else
+               return retval<<1;
+}
+
+static int do_cmd_set_subbuf_size(const char *recvbuf, struct ustcomm_source *src)
+{
+       char *channel_slash_size;
+       char ch_name[256]="";
+       unsigned int size, power;
+       int retval = 0;
+       struct ust_trace *trace;
+       char trace_name[] = "auto";
+       int i;
+       int found = 0;
+
+       DBG("set_subbuf_size");
+
+       channel_slash_size = nth_token(recvbuf, 1);
+       sscanf(channel_slash_size, "%255[^/]/%u", ch_name, &size);
+
+       if(ch_name == NULL) {
+               ERR("cannot parse channel");
+               goto end;
+       }
+
+       power = pow2_higher_or_eq(size);
+       power = max_t(unsigned int, 2u, power);
+       if (power != size)
+               WARN("using the next power of two for buffer size = %u\n", power);
+
+       ltt_lock_traces();
+       trace = _ltt_trace_find_setup(trace_name);
+       if(trace == NULL) {
+               ERR("cannot find trace!");
+               retval = -1;
+               goto end;
+       }
+
+       for(i = 0; i < trace->nr_channels; i++) {
+               struct ust_channel *channel = &trace->channels[i];
+
+               if(!strcmp(trace->channels[i].channel_name, ch_name)) {
+
+                       channel->subbuf_size = power;
+                       DBG("the set_subbuf_size for the requested channel is %zd", channel->subbuf_size);
+
+                       found = 1;
+                       break;
                }
-               else if(!strcmp(recvbuf, "trace_start")) {
-                       DBG("trace start");
+       }
+       if(found == 0) {
+               ERR("unable to find channel");
+       }
 
-                       result = ltt_trace_start(trace_name);
-                       if(result < 0) {
-                               ERR("ltt_trace_start failed");
-                               continue;
-                       }
+       end:
+       ltt_unlock_traces();
+       return retval;
+}
+
+static int do_cmd_set_subbuf_num(const char *recvbuf, struct ustcomm_source *src)
+{
+       char *channel_slash_num;
+       char ch_name[256]="";
+       unsigned int num;
+       int retval = 0;
+       struct ust_trace *trace;
+       char trace_name[] = "auto";
+       int i;
+       int found = 0;
+
+       DBG("set_subbuf_num");
+
+       channel_slash_num = nth_token(recvbuf, 1);
+       sscanf(channel_slash_num, "%255[^/]/%u", ch_name, &num);
+
+       if(ch_name == NULL) {
+               ERR("cannot parse channel");
+               goto end;
+       }
+       if (num < 2) {
+               ERR("subbuffer count should be greater than 2");
+               goto end;
+       }
+
+       ltt_lock_traces();
+       trace = _ltt_trace_find_setup(trace_name);
+       if(trace == NULL) {
+               ERR("cannot find trace!");
+               retval = -1;
+               goto end;
+       }
+
+       for(i = 0; i < trace->nr_channels; i++) {
+               struct ust_channel *channel = &trace->channels[i];
+
+               if(!strcmp(trace->channels[i].channel_name, ch_name)) {
+
+                       channel->subbuf_cnt = num;
+                       DBG("the set_subbuf_cnt for the requested channel is %zd", channel->subbuf_cnt);
+
+                       found = 1;
+                       break;
                }
-               else if(!strcmp(recvbuf, "trace_stop")) {
-                       DBG("trace stop");
+       }
+       if(found == 0) {
+               ERR("unable to find channel");
+       }
 
-                       result = ltt_trace_stop(trace_name);
-                       if(result < 0) {
-                               ERR("ltt_trace_stop failed");
-                               return (void *)1;
-                       }
+       end:
+       ltt_unlock_traces();
+       return retval;
+}
+
+static int do_cmd_get_subbuffer(const char *recvbuf, struct ustcomm_source *src)
+{
+       int retval = 0;
+       struct ust_trace *trace;
+       char trace_name[] = "auto";
+       int i;
+       char *channel_and_cpu;
+       int found = 0;
+       char *ch_name;
+       int ch_cpu;
+
+       DBG("get_subbuf");
+
+       channel_and_cpu = nth_token(recvbuf, 1);
+       if(channel_and_cpu == NULL) {
+               ERR("cannot parse channel");
+               goto end;
+       }
+
+       seperate_channel_cpu(channel_and_cpu, &ch_name, &ch_cpu);
+       if(ch_cpu == -1) {
+               ERR("problem parsing channel name");
+               goto free_short_chan_name;
+       }
+
+       ltt_lock_traces();
+       trace = _ltt_trace_find(trace_name);
+
+       if(trace == NULL) {
+               int result;
+
+               DBG("Cannot find trace. It was likely destroyed by the user.");
+               result = ustcomm_send_reply(&ustcomm_app.server, "NOTFOUND", src);
+               if(result) {
+                       ERR("ustcomm_send_reply failed");
+                       retval = -1;
+                       goto unlock_traces;
                }
-               else if(!strcmp(recvbuf, "trace_destroy")) {
 
-                       DBG("trace destroy");
+               goto unlock_traces;
+       }
 
-                       result = ltt_trace_destroy(trace_name);
-                       if(result < 0) {
-                               ERR("ltt_trace_destroy failed");
-                               return (void *)1;
+       for(i=0; i<trace->nr_channels; i++) {
+               struct ust_channel *channel = &trace->channels[i];
+
+               if(!strcmp(trace->channels[i].channel_name, ch_name)) {
+                       struct ust_buffer *buf = channel->buf[ch_cpu];
+                       struct blocked_consumer *bc;
+
+                       found = 1;
+
+                       bc = (struct blocked_consumer *) malloc(sizeof(struct blocked_consumer));
+                       if(bc == NULL) {
+                               ERR("malloc returned NULL");
+                               goto unlock_traces;
                        }
-               }
-               else if(nth_token_is(recvbuf, "get_shmid", 0) == 1) {
-                       struct ltt_trace_struct *trace;
-                       char trace_name[] = "auto";
-                       int i;
-                       char *channel_name;
-
-                       DBG("get_shmid");
-
-                       channel_name = nth_token(recvbuf, 1);
-                       if(channel_name == NULL) {
-                               ERR("get_shmid: cannot parse channel");
-                               goto next_cmd;
+                       bc->fd_consumer = src->fd;
+                       bc->fd_producer = buf->data_ready_fd_read;
+                       bc->buf = buf;
+                       bc->src = *src;
+                       bc->server = ustcomm_app.server;
+
+                       list_add(&bc->list, &blocked_consumers);
+
+                       /* Being here is the proof the daemon has mapped the buffer in its
+                        * memory. We may now decrement buffers_to_export.
+                        */
+                       if(uatomic_read(&buf->consumed) == 0) {
+                               DBG("decrementing buffers_to_export");
+                               STORE_SHARED(buffers_to_export, LOAD_SHARED(buffers_to_export)-1);
                        }
 
-                       ltt_lock_traces();
-                       trace = _ltt_trace_find(trace_name);
-                       ltt_unlock_traces();
+                       break;
+               }
+       }
+       if(found == 0) {
+               ERR("unable to find channel");
+       }
 
-                       if(trace == NULL) {
-                               ERR("cannot find trace!");
-                               return (void *)1;
-                       }
+       unlock_traces:
+       ltt_unlock_traces();
+
+       free_short_chan_name:
+       free(ch_name);
+
+       end:
+       return retval;
+}
+
+static int do_cmd_put_subbuffer(const char *recvbuf, struct ustcomm_source *src)
+{
+       int retval = 0;
+       struct ust_trace *trace;
+       char trace_name[] = "auto";
+       int i;
+       char *channel_and_cpu;
+       int found = 0;
+       int result;
+       char *ch_name;
+       int ch_cpu;
+       long consumed_old;
+       char *consumed_old_str;
+       char *endptr;
+       char *reply = NULL;
+
+       DBG("put_subbuf");
+
+       channel_and_cpu = strdup(nth_token(recvbuf, 1));
+       if(channel_and_cpu == NULL) {
+               ERR("cannot parse channel");
+               retval = -1;
+               goto end;
+       }
+
+       consumed_old_str = strdup(nth_token(recvbuf, 2));
+       if(consumed_old_str == NULL) {
+               ERR("cannot parse consumed_old");
+               retval = -1;
+               goto free_channel_and_cpu;
+       }
+       consumed_old = strtol(consumed_old_str, &endptr, 10);
+       if(*endptr != '\0') {
+               ERR("invalid value for consumed_old");
+               retval = -1;
+               goto free_consumed_old_str;
+       }
 
-                       for(i=0; i<trace->nr_channels; i++) {
-                               struct rchan *rchan = trace->channels[i].trans_channel_data;
-                               struct rchan_buf *rbuf = rchan->buf;
-                               struct ltt_channel_struct *ltt_channel = (struct ltt_channel_struct *)rchan->private_data;
+       seperate_channel_cpu(channel_and_cpu, &ch_name, &ch_cpu);
+       if(ch_cpu == -1) {
+               ERR("problem parsing channel name");
+               retval = -1;
+               goto free_short_chan_name;
+       }
 
-                               if(!strcmp(trace->channels[i].channel_name, channel_name)) {
-                                       char *reply;
+       ltt_lock_traces();
+       trace = _ltt_trace_find(trace_name);
 
-                                       DBG("the shmid for the requested channel is %d", rbuf->shmid);
-                                       DBG("the shmid for its buffer structure is %d", ltt_channel->buf_shmid);
-                                       asprintf(&reply, "%d %d", rbuf->shmid, ltt_channel->buf_shmid);
+       if(trace == NULL) {
+               DBG("Cannot find trace. It was likely destroyed by the user.");
+               result = ustcomm_send_reply(&ustcomm_app.server, "NOTFOUND", src);
+               if(result) {
+                       ERR("ustcomm_send_reply failed");
+                       retval = -1;
+                       goto unlock_traces;
+               }
+
+               goto unlock_traces;
+       }
 
-                                       result = ustcomm_send_reply(&ustcomm_app.server, reply, &src);
-                                       if(result) {
-                                               ERR("listener: get_shmid: ustcomm_send_reply failed");
-                                               goto next_cmd;
-                                       }
+       for(i=0; i<trace->nr_channels; i++) {
+               struct ust_channel *channel = &trace->channels[i];
 
-                                       free(reply);
+               if(!strcmp(trace->channels[i].channel_name, ch_name)) {
+                       struct ust_buffer *buf = channel->buf[ch_cpu];
 
-                                       break;
-                               }
+                       found = 1;
+
+                       result = ust_buffers_put_subbuf(buf, consumed_old);
+                       if(result < 0) {
+                               WARN("ust_buffers_put_subbuf: error (subbuf=%s)", channel_and_cpu);
+                               asprintf(&reply, "%s", "ERROR");
                        }
-               }
-               else if(nth_token_is(recvbuf, "get_n_subbufs", 0) == 1) {
-                       struct ltt_trace_struct *trace;
-                       char trace_name[] = "auto";
-                       int i;
-                       char *channel_name;
-
-                       DBG("get_n_subbufs");
-
-                       channel_name = nth_token(recvbuf, 1);
-                       if(channel_name == NULL) {
-                               ERR("get_n_subbufs: cannot parse channel");
-                               goto next_cmd;
+                       else {
+                               DBG("ust_buffers_put_subbuf: success (subbuf=%s)", channel_and_cpu);
+                               asprintf(&reply, "%s", "OK");
                        }
 
-                       ltt_lock_traces();
-                       trace = _ltt_trace_find(trace_name);
-                       ltt_unlock_traces();
+                       result = ustcomm_send_reply(&ustcomm_app.server, reply, src);
+                       if(result) {
+                               ERR("ustcomm_send_reply failed");
+                               free(reply);
+                               retval = -1;
+                               goto unlock_traces;
+                       }
+
+                       free(reply);
+                       break;
+               }
+       }
+       if(found == 0) {
+               ERR("unable to find channel");
+       }
+
+       unlock_traces:
+       ltt_unlock_traces();
+       free_short_chan_name:
+       free(ch_name);
+       free_consumed_old_str:
+       free(consumed_old_str);
+       free_channel_and_cpu:
+       free(channel_and_cpu);
+
+       end:
+       return retval;
+}
+
+static void listener_cleanup(void *ptr)
+{
+       ustcomm_fini_app(&ustcomm_app, 0);
+}
 
-                       if(trace == NULL) {
-                               ERR("cannot find trace!");
-                               return (void *)1;
-                       }
+int process_client_cmd(char *recvbuf, struct ustcomm_source *src)
+{
+       int result;
+       char trace_name[] = "auto";
+       char trace_type[] = "ustrelay";
+       int len;
 
-                       for(i=0; i<trace->nr_channels; i++) {
-                               struct rchan *rchan = trace->channels[i].trans_channel_data;
+       DBG("received a message! it's: %s", recvbuf);
+       len = strlen(recvbuf);
 
-                               if(!strcmp(trace->channels[i].channel_name, channel_name)) {
-                                       char *reply;
+       if(!strcmp(recvbuf, "print_markers")) {
+               print_markers(stderr);
+       }
+       else if(!strcmp(recvbuf, "list_markers")) {
+               char *ptr;
+               size_t size;
+               FILE *fp;
 
-                                       DBG("the n_subbufs for the requested channel is %zd", rchan->n_subbufs);
-                                       asprintf(&reply, "%zd", rchan->n_subbufs);
+               fp = open_memstream(&ptr, &size);
+               print_markers(fp);
+               fclose(fp);
 
-                                       result = ustcomm_send_reply(&ustcomm_app.server, reply, &src);
-                                       if(result) {
-                                               ERR("listener: get_n_subbufs: ustcomm_send_reply failed");
-                                               goto next_cmd;
-                                       }
+               result = ustcomm_send_reply(&ustcomm_app.server, ptr, src);
 
-                                       free(reply);
+               free(ptr);
+       }
+       else if(!strcmp(recvbuf, "start")) {
+               /* start is an operation that setups the trace, allocates it and starts it */
+               result = ltt_trace_setup(trace_name);
+               if(result < 0) {
+                       ERR("ltt_trace_setup failed");
+                       return -1;
+               }
 
-                                       break;
-                               }
-                       }
+               result = ltt_trace_set_type(trace_name, trace_type);
+               if(result < 0) {
+                       ERR("ltt_trace_set_type failed");
+                       return -1;
                }
-               else if(nth_token_is(recvbuf, "get_subbuf_size", 0) == 1) {
-                       struct ltt_trace_struct *trace;
-                       char trace_name[] = "auto";
-                       int i;
-                       char *channel_name;
-
-                       DBG("get_subbuf_size");
-
-                       channel_name = nth_token(recvbuf, 1);
-                       if(channel_name == NULL) {
-                               ERR("get_subbuf_size: cannot parse channel");
-                               goto next_cmd;
-                       }
 
-                       ltt_lock_traces();
-                       trace = _ltt_trace_find(trace_name);
-                       ltt_unlock_traces();
+               result = ltt_trace_alloc(trace_name);
+               if(result < 0) {
+                       ERR("ltt_trace_alloc failed");
+                       return -1;
+               }
 
-                       if(trace == NULL) {
-                               ERR("cannot find trace!");
-                               return (void *)1;
-                       }
+               inform_consumer_daemon(trace_name);
 
-                       for(i=0; i<trace->nr_channels; i++) {
-                               struct rchan *rchan = trace->channels[i].trans_channel_data;
+               result = ltt_trace_start(trace_name);
+               if(result < 0) {
+                       ERR("ltt_trace_start failed");
+                       return -1;
+               }
+       }
+       else if(!strcmp(recvbuf, "trace_setup")) {
+               DBG("trace setup");
 
-                               if(!strcmp(trace->channels[i].channel_name, channel_name)) {
-                                       char *reply;
+               result = ltt_trace_setup(trace_name);
+               if(result < 0) {
+                       ERR("ltt_trace_setup failed");
+                       return -1;
+               }
 
-                                       DBG("the subbuf_size for the requested channel is %zd", rchan->subbuf_size);
-                                       asprintf(&reply, "%zd", rchan->subbuf_size);
+               result = ltt_trace_set_type(trace_name, trace_type);
+               if(result < 0) {
+                       ERR("ltt_trace_set_type failed");
+                       return -1;
+               }
+       }
+       else if(!strcmp(recvbuf, "trace_alloc")) {
+               DBG("trace alloc");
 
-                                       result = ustcomm_send_reply(&ustcomm_app.server, reply, &src);
-                                       if(result) {
-                                               ERR("listener: get_subbuf_size: ustcomm_send_reply failed");
-                                               goto next_cmd;
-                                       }
+               result = ltt_trace_alloc(trace_name);
+               if(result < 0) {
+                       ERR("ltt_trace_alloc failed");
+                       return -1;
+               }
+               inform_consumer_daemon(trace_name);
+       }
+       else if(!strcmp(recvbuf, "trace_create")) {
+               DBG("trace create");
 
-                                       free(reply);
+               result = ltt_trace_setup(trace_name);
+               if(result < 0) {
+                       ERR("ltt_trace_setup failed");
+                       return -1;
+               }
 
-                                       break;
-                               }
-                       }
+               result = ltt_trace_set_type(trace_name, trace_type);
+               if(result < 0) {
+                       ERR("ltt_trace_set_type failed");
+                       return -1;
                }
-               else if(nth_token_is(recvbuf, "load_probe_lib", 0) == 1) {
-                       char *libfile;
+       }
+       else if(!strcmp(recvbuf, "trace_start")) {
+               DBG("trace start");
 
-                       libfile = nth_token(recvbuf, 1);
+               result = ltt_trace_alloc(trace_name);
+               if(result < 0) {
+                       ERR("ltt_trace_alloc failed");
+                       return -1;
+               }
+               if(!result) {
+                       inform_consumer_daemon(trace_name);
+               }
 
-                       DBG("load_probe_lib loading %s", libfile);
+               result = ltt_trace_start(trace_name);
+               if(result < 0) {
+                       ERR("ltt_trace_start failed");
+                       return -1;
                }
-               else if(nth_token_is(recvbuf, "get_subbuffer", 0) == 1) {
-                       struct ltt_trace_struct *trace;
-                       char trace_name[] = "auto";
-                       int i;
-                       char *channel_name;
-
-                       DBG("get_subbuf");
-
-                       channel_name = nth_token(recvbuf, 1);
-                       if(channel_name == NULL) {
-                               ERR("get_subbuf: cannot parse channel");
-                               goto next_cmd;
-                       }
+       }
+       else if(!strcmp(recvbuf, "trace_stop")) {
+               DBG("trace stop");
 
-                       ltt_lock_traces();
-                       trace = _ltt_trace_find(trace_name);
-                       ltt_unlock_traces();
+               result = ltt_trace_stop(trace_name);
+               if(result < 0) {
+                       ERR("ltt_trace_stop failed");
+                       return -1;
+               }
+       }
+       else if(!strcmp(recvbuf, "trace_destroy")) {
 
-                       if(trace == NULL) {
-                               ERR("cannot find trace!");
-                               return (void *)1;
-                       }
+               DBG("trace destroy");
 
-                       for(i=0; i<trace->nr_channels; i++) {
-                               struct rchan *rchan = trace->channels[i].trans_channel_data;
-
-                               if(!strcmp(trace->channels[i].channel_name, channel_name)) {
-                                       struct rchan_buf *rbuf = rchan->buf;
-                                       struct ltt_channel_buf_struct *lttbuf = trace->channels[i].buf;
-                                       struct blocked_consumer *bc;
-
-                                       bc = (struct blocked_consumer *) malloc(sizeof(struct blocked_consumer));
-                                       if(bc == NULL) {
-                                               ERR("malloc returned NULL");
-                                               goto next_cmd;
-                                       }
-                                       bc->fd_consumer = src.fd;
-                                       bc->fd_producer = lttbuf->data_ready_fd_read;
-                                       bc->rbuf = rbuf;
-                                       bc->lttbuf = lttbuf;
-                                       bc->src = src;
-                                       bc->server = ustcomm_app.server;
-
-                                       list_add(&bc->list, &blocked_consumers);
-
-                                       break;
-                               }
-                       }
+               result = ltt_trace_destroy(trace_name, 0);
+               if(result < 0) {
+                       ERR("ltt_trace_destroy failed");
+                       return -1;
                }
-               else if(nth_token_is(recvbuf, "put_subbuffer", 0) == 1) {
-                       struct ltt_trace_struct *trace;
-                       char trace_name[] = "auto";
-                       int i;
-                       char *channel_name;
-                       long consumed_old;
-                       char *consumed_old_str;
-                       char *endptr;
-
-                       DBG("put_subbuf");
-
-                       channel_name = strdup_malloc(nth_token(recvbuf, 1));
-                       if(channel_name == NULL) {
-                               ERR("put_subbuf_size: cannot parse channel");
-                               goto next_cmd;
-                       }
+       }
+       else if(nth_token_is(recvbuf, "get_shmid", 0) == 1) {
+               do_cmd_get_shmid(recvbuf, src);
+       }
+       else if(nth_token_is(recvbuf, "get_n_subbufs", 0) == 1) {
+               do_cmd_get_n_subbufs(recvbuf, src);
+       }
+       else if(nth_token_is(recvbuf, "get_subbuf_size", 0) == 1) {
+               do_cmd_get_subbuf_size(recvbuf, src);
+       }
+       else if(nth_token_is(recvbuf, "load_probe_lib", 0) == 1) {
+               char *libfile;
 
-                       consumed_old_str = strdup_malloc(nth_token(recvbuf, 2));
-                       if(consumed_old_str == NULL) {
-                               ERR("put_subbuf: cannot parse consumed_old");
-                               goto next_cmd;
-                       }
-                       consumed_old = strtol(consumed_old_str, &endptr, 10);
-                       if(*endptr != '\0') {
-                               ERR("put_subbuf: invalid value for consumed_old");
-                               goto next_cmd;
-                       }
+               libfile = nth_token(recvbuf, 1);
 
-                       ltt_lock_traces();
-                       trace = _ltt_trace_find(trace_name);
-                       ltt_unlock_traces();
+               DBG("load_probe_lib loading %s", libfile);
 
-                       if(trace == NULL) {
-                               ERR("cannot find trace!");
-                               return (void *)1;
-                       }
+               free(libfile);
+       }
+       else if(nth_token_is(recvbuf, "get_subbuffer", 0) == 1) {
+               do_cmd_get_subbuffer(recvbuf, src);
+       }
+       else if(nth_token_is(recvbuf, "put_subbuffer", 0) == 1) {
+               do_cmd_put_subbuffer(recvbuf, src);
+       }
+       else if(nth_token_is(recvbuf, "set_subbuf_size", 0) == 1) {
+               do_cmd_set_subbuf_size(recvbuf, src);
+       }
+       else if(nth_token_is(recvbuf, "set_subbuf_num", 0) == 1) {
+               do_cmd_set_subbuf_num(recvbuf, src);
+       }
+       else if(nth_token_is(recvbuf, "enable_marker", 0) == 1) {
+               char *channel_slash_name = nth_token(recvbuf, 1);
+               char channel_name[256]="";
+               char marker_name[256]="";
 
-                       for(i=0; i<trace->nr_channels; i++) {
-                               struct rchan *rchan = trace->channels[i].trans_channel_data;
-
-                               if(!strcmp(trace->channels[i].channel_name, channel_name)) {
-                                       struct rchan_buf *rbuf = rchan->buf;
-                                       struct ltt_channel_buf_struct *lttbuf = trace->channels[i].buf;
-                                       char *reply;
-                                       long consumed_old=0;
-
-                                       result = ltt_do_put_subbuf(rbuf, lttbuf, consumed_old);
-                                       if(result < 0) {
-                                               WARN("ltt_do_put_subbuf: error (subbuf=%s)", channel_name);
-                                               asprintf(&reply, "%s", "ERROR");
-                                       }
-                                       else {
-                                               DBG("ltt_do_put_subbuf: success (subbuf=%s)", channel_name);
-                                               asprintf(&reply, "%s", "OK");
-                                       }
-
-                                       result = ustcomm_send_reply(&ustcomm_app.server, reply, &src);
-                                       if(result) {
-                                               ERR("listener: put_subbuf: ustcomm_send_reply failed");
-                                               goto next_cmd;
-                                       }
-
-                                       free(reply);
-
-                                       break;
-                               }
-                       }
+               result = sscanf(channel_slash_name, "%255[^/]/%255s", channel_name, marker_name);
 
-                       free(channel_name);
-                       free(consumed_old_str);
+               if(channel_name == NULL || marker_name == NULL) {
+                       WARN("invalid marker name");
+                       goto next_cmd;
                }
-               else if(nth_token_is(recvbuf, "enable_marker", 0) == 1) {
-                       char *channel_slash_name = nth_token(recvbuf, 1);
-                       char channel_name[256]="";
-                       char marker_name[256]="";
 
-                       result = sscanf(channel_slash_name, "%255[^/]/%255s", channel_name, marker_name);
+               result = ltt_marker_connect(channel_name, marker_name, "default");
+               if(result < 0) {
+                       WARN("could not enable marker; channel=%s, name=%s", channel_name, marker_name);
+               }
+       }
+       else if(nth_token_is(recvbuf, "disable_marker", 0) == 1) {
+               char *channel_slash_name = nth_token(recvbuf, 1);
+               char *marker_name;
+               char *channel_name;
 
-                       if(channel_name == NULL || marker_name == NULL) {
-                               WARN("invalid marker name");
-                               goto next_cmd;
-                       }
-                       printf("%s %s\n", channel_name, marker_name);
+               result = sscanf(channel_slash_name, "%a[^/]/%as", &channel_name, &marker_name);
 
-                       result = ltt_marker_connect(channel_name, marker_name, "default");
-                       if(result < 0) {
-                               WARN("could not enable marker; channel=%s, name=%s", channel_name, marker_name);
-                       }
+               if(marker_name == NULL) {
                }
-               else if(nth_token_is(recvbuf, "disable_marker", 0) == 1) {
-                       char *channel_slash_name = nth_token(recvbuf, 1);
-                       char *marker_name;
-                       char *channel_name;
 
-                       result = sscanf(channel_slash_name, "%a[^/]/%as", &channel_name, &marker_name);
+               result = ltt_marker_disconnect(channel_name, marker_name, "default");
+               if(result < 0) {
+                       WARN("could not disable marker; channel=%s, name=%s", channel_name, marker_name);
+               }
+       }
+       else if(nth_token_is(recvbuf, "get_pidunique", 0) == 1) {
+               char *reply;
 
-                       if(marker_name == NULL) {
-                       }
-                       printf("%s %s\n", channel_name, marker_name);
+               asprintf(&reply, "%lld", pidunique);
 
-                       result = ltt_marker_disconnect(channel_name, marker_name, "default");
-                       if(result < 0) {
-                               WARN("could not disable marker; channel=%s, name=%s", channel_name, marker_name);
-                       }
-               }
-//             else if(nth_token_is(recvbuf, "get_notifications", 0) == 1) {
-//                     struct ltt_trace_struct *trace;
-//                     char trace_name[] = "auto";
-//                     int i;
-//                     char *channel_name;
-//
-//                     DBG("get_notifications");
-//
-//                     channel_name = strdup_malloc(nth_token(recvbuf, 1));
-//                     if(channel_name == NULL) {
-//                             ERR("put_subbuf_size: cannot parse channel");
-//                             goto next_cmd;
-//                     }
-//
-//                     ltt_lock_traces();
-//                     trace = _ltt_trace_find(trace_name);
-//                     ltt_unlock_traces();
-//
-//                     if(trace == NULL) {
-//                             ERR("cannot find trace!");
-//                             return (void *)1;
-//                     }
-//
-//                     for(i=0; i<trace->nr_channels; i++) {
-//                             struct rchan *rchan = trace->channels[i].trans_channel_data;
-//                             int fd;
-//
-//                             if(!strcmp(trace->channels[i].channel_name, channel_name)) {
-//                                     struct rchan_buf *rbuf = rchan->buf;
-//                                     struct ltt_channel_buf_struct *lttbuf = trace->channels[i].buf;
-//
-//                                     result = fd = ustcomm_app_detach_client(&ustcomm_app, &src);
-//                                     if(result == -1) {
-//                                             ERR("ustcomm_app_detach_client failed");
-//                                             goto next_cmd;
-//                                     }
-//
-//                                     lttbuf->wake_consumer_arg = (void *) fd;
-//
-//                                     smp_wmb();
-//
-//                                     lttbuf->call_wake_consumer = 1;
-//
-//                                     break;
-//                             }
-//                     }
-//
-//                     free(channel_name);
-//             }
-               else {
-                       ERR("unable to parse message: %s", recvbuf);
+               result = ustcomm_send_reply(&ustcomm_app.server, reply, src);
+               if(result) {
+                       ERR("listener: get_pidunique: ustcomm_send_reply failed");
+                       goto next_cmd;
                }
 
-       next_cmd:
-               free(recvbuf);
+               free(reply);
+       }
+       else {
+               ERR("unable to parse message: %s", recvbuf);
        }
-}
 
-int have_listener = 0;
+next_cmd:
 
-void create_listener(void)
+       return 0;
+}
+
+void *listener_main(void *p)
 {
-#ifdef USE_CLONE
-       static char listener_stack[16384];
-#else
-       pthread_t thread;
-#endif
+       int result;
 
-       if(have_listener)
-               return;
+       DBG("LISTENER");
 
-#ifdef USE_CLONE
-       result = clone(listener_main, listener_stack+sizeof(listener_stack)-1, CLONE_FS | CLONE_FILES | CLONE_VM | CLONE_SIGHAND | CLONE_THREAD, NULL);
-       if(result == -1) {
-               perror("clone");
-               return;
-       }
-#else
+       pthread_cleanup_push(listener_cleanup, NULL);
 
-       pthread_create(&thread, NULL, listener_main, NULL);
-#endif
+       for(;;) {
+               struct mpentries mpent;
 
-       have_listener = 1;
-}
+               multipoll_init(&mpent);
 
-/* The signal handler itself. Signals must be setup so there cannot be
-   nested signals. */
+               blocked_consumers_add_to_mp(&mpent);
+               ustcomm_mp_add_app_clients(&mpent, &ustcomm_app, process_client_cmd);
 
-void sighandler(int sig)
-{
-       static char have_listener = 0;
-       DBG("sighandler");
+               result = multipoll_poll(&mpent, -1);
+               if(result == -1) {
+                       ERR("error in multipoll_poll");
+               }
 
-       if(!have_listener) {
-               create_listener();
-               have_listener = 1;
+               multipoll_destroy(&mpent);
        }
-}
-
-/* Called by the app signal handler to chain it to us. */
 
-void chain_signal(void)
-{
-       sighandler(USTSIGNAL);
-}
-
-static int init_socket(void)
-{
-       return ustcomm_init_app(getpid(), &ustcomm_app);
+       pthread_cleanup_pop(1);
 }
 
-/* FIXME: reenable this to delete socket file. */
+/* These should only be accessed in the parent thread,
+ * not the listener.
+ */
+static volatile sig_atomic_t have_listener = 0;
+static pthread_t listener_thread;
 
-#if 0
-static void destroy_socket(void)
+void create_listener(void)
 {
        int result;
+       sigset_t sig_all_blocked;
+       sigset_t orig_parent_mask;
 
-       if(mysocketfile[0] == '\0')
+       if(have_listener) {
+               WARN("not creating listener because we already had one");
                return;
-
-       result = unlink(mysocketfile);
-       if(result == -1) {
-               PERROR("unlink");
        }
-}
-#endif
 
-static int init_signal_handler(void)
-{
-       /* Attempt to handler SIGIO. If the main program wants to
-        * handle it, fine, it'll override us. They it'll have to
-        * use the chaining function.
+       /* A new thread created by pthread_create inherits the signal mask
+        * from the parent. To avoid any signal being received by the
+        * listener thread, we block all signals temporarily in the parent,
+        * while we create the listener thread.
         */
 
-       int result;
-       struct sigaction act;
+       sigfillset(&sig_all_blocked);
 
-       result = sigemptyset(&act.sa_mask);
-       if(result == -1) {
-               PERROR("sigemptyset");
-               return -1;
+       result = pthread_sigmask(SIG_SETMASK, &sig_all_blocked, &orig_parent_mask);
+       if(result) {
+               PERROR("pthread_sigmask: %s", strerror(result));
        }
 
-       act.sa_handler = sighandler;
-       act.sa_flags = SA_RESTART;
-
-       /* Only defer ourselves. Also, try to restart interrupted
-        * syscalls to disturb the traced program as little as possible.
-        */
-       result = sigaction(SIGIO, &act, NULL);
+       result = pthread_create(&listener_thread, NULL, listener_main, NULL);
        if(result == -1) {
-               PERROR("sigaction");
-               return -1;
+               PERROR("pthread_create");
        }
 
-       return 0;
+       /* Restore original signal mask in parent */
+       result = pthread_sigmask(SIG_SETMASK, &orig_parent_mask, NULL);
+       if(result) {
+               PERROR("pthread_sigmask: %s", strerror(result));
+       }
+
+       have_listener = 1;
+}
+
+static int init_socket(void)
+{
+       return ustcomm_init_app(getpid(), &ustcomm_app);
 }
 
 #define AUTOPROBE_DISABLED      0
@@ -869,35 +1106,29 @@ static void auto_probe_connect(struct marker *m)
        if(result && result != -EEXIST)
                ERR("ltt_marker_connect (marker = %s/%s, errno = %d)", m->channel, m->name, -result);
 
-       DBG("auto connected marker %s %s to probe default", m->channel, m->name);
+       DBG("auto connected marker %s (addr: %p) %s to probe default", m->channel, m, m->name);
 
 }
 
-static void __attribute__((constructor(1000))) init()
+static void __attribute__((constructor)) init()
 {
        int result;
        char* autoprobe_val = NULL;
 
-       /* Initialize RCU in case the constructor order is not good. */
-       urcu_init();
-
-       /* It is important to do this before events start to be generated. */
-       ust_register_thread();
+       /* Assign the pidunique, to be able to differentiate the processes with same
+        * pid, (before and after an exec).
+        */
+       pidunique = make_pidunique();
 
        DBG("Tracectl constructor");
 
-       /* Must create socket before signal handler to prevent races.
-         */
        result = init_socket();
        if(result == -1) {
                ERR("init_socket error");
                return;
        }
-       result = init_signal_handler();
-       if(result == -1) {
-               ERR("init_signal_handler error");
-               return;
-       }
+
+       create_listener();
 
        autoprobe_val = getenv("UST_AUTOPROBE");
        if(autoprobe_val) {
@@ -946,6 +1177,26 @@ static void __attribute__((constructor(1000))) init()
                }
        }
 
+       if(getenv("UST_OVERWRITE")) {
+               int val = atoi(getenv("UST_OVERWRITE"));
+               if(val == 0 || val == 1) {
+                       STORE_SHARED(ust_channels_overwrite_by_default, val);
+               }
+               else {
+                       WARN("invalid value for UST_OVERWRITE");
+               }
+       }
+
+       if(getenv("UST_AUTOCOLLECT")) {
+               int val = atoi(getenv("UST_AUTOCOLLECT"));
+               if(val == 0 || val == 1) {
+                       STORE_SHARED(ust_channels_request_collection_by_default, val);
+               }
+               else {
+                       WARN("invalid value for UST_AUTOCOLLECT");
+               }
+       }
+
        if(getenv("UST_TRACE")) {
                char trace_name[] = "auto";
                char trace_type[] = "ustrelay";
@@ -955,13 +1206,32 @@ static void __attribute__((constructor(1000))) init()
                /* Ensure marker control is initialized */
                init_marker_control();
 
-               /* Ensure relay is initialized */
-               init_ustrelay_transport();
-
                /* Ensure markers are initialized */
                init_markers();
 
-               /* In case. */
+               /* Ensure buffers are initialized, for the transport to be available.
+                * We are about to set a trace type and it will fail without this.
+                */
+               init_ustrelay_transport();
+
+               /* FIXME: When starting early tracing (here), depending on the
+                * order of constructors, it is very well possible some marker
+                * sections are not yet registered. Because of this, some
+                * channels may not be registered. Yet, we are about to ask the
+                * daemon to collect the channels. Channels which are not yet
+                * registered will not be collected.
+                *
+                * Currently, in LTTng, there is no way to add a channel after
+                * trace start. The reason for this is that it induces complex
+                * concurrency issues on the trace structures, which can only
+                * be resolved using RCU. This has not been done yet. As a
+                * workaround, we are forcing the registration of the "ust"
+                * channel here. This is the only channel (apart from metadata)
+                * that can be reliably used in early tracing.
+                *
+                * Non-early tracing does not have this problem and can use
+                * arbitrary channel names.
+                */
                ltt_channels_register("ust");
 
                result = ltt_trace_setup(trace_name);
@@ -982,13 +1252,15 @@ static void __attribute__((constructor(1000))) init()
                        return;
                }
 
-               inform_consumer_daemon(trace_name);
-
                result = ltt_trace_start(trace_name);
                if(result < 0) {
                        ERR("ltt_trace_start failed");
                        return;
                }
+
+               /* Do this after the trace is started in order to avoid creating confusion
+                * if the trace fails to start. */
+               inform_consumer_daemon(trace_name);
        }
 
 
@@ -999,11 +1271,17 @@ static void __attribute__((constructor(1000))) init()
 }
 
 /* This is only called if we terminate normally, not with an unhandled signal,
- * so we cannot rely on it. */
+ * so we cannot rely on it. However, for now, LTTV requires that the header of
+ * the last sub-buffer contain a valid end time for the trace. This is done
+ * automatically only when the trace is properly stopped.
+ *
+ * If the traced program crashed, it is always possible to manually add the
+ * right value in the header, or to open the trace in text mode.
+ *
+ * FIXME: Fix LTTV so it doesn't need this.
+ */
 
-/* This destructor probably isn't needed, because ustd can do crash recovery. */
-#if 0
-static void __attribute__((destructor)) fini()
+static void destroy_traces(void)
 {
        int result;
 
@@ -1016,20 +1294,16 @@ static void __attribute__((destructor)) fini()
                ERR("ltt_trace_stop error");
        }
 
-       result = ltt_trace_destroy("auto");
+       result = ltt_trace_destroy("auto", 0);
        if(result == -1) {
                ERR("ltt_trace_destroy error");
        }
-
-       destroy_socket();
 }
-#endif
 
-#if 0
 static int trace_recording(void)
 {
        int retval = 0;
-       struct ltt_trace_struct *trace;
+       struct ust_trace *trace;
 
        ltt_lock_traces();
 
@@ -1045,22 +1319,20 @@ static int trace_recording(void)
        return retval;
 }
 
+#if 0
 static int have_consumer(void)
 {
        return !list_empty(&blocked_consumers);
 }
+#endif
 
-/* This destructor keeps the process alive for a few seconds in order
- * to leave time to ustd to consume its buffers.
- */
-
-int restarting_sleep(int secs)
+int restarting_usleep(useconds_t usecs)
 {
         struct timespec tv; 
         int result; 
  
-        tv.tv_sec = secs
-        tv.tv_nsec = 0; 
+        tv.tv_sec = 0
+        tv.tv_nsec = usecs * 1000; 
  
         do { 
                 result = nanosleep(&tv, &tv); 
@@ -1069,67 +1341,168 @@ int restarting_sleep(int secs)
        return result;
 }
 
+static void stop_listener()
+{
+       int result;
+
+       result = pthread_cancel(listener_thread);
+       if(result != 0) {
+               ERR("pthread_cancel: %s", strerror(result));
+       }
+       result = pthread_join(listener_thread, NULL);
+       if(result != 0) {
+               ERR("pthread_join: %s", strerror(result));
+       }
+}
+
+/* This destructor keeps the process alive for a few seconds in order
+ * to leave time to ustd to connect to its buffers. This is necessary
+ * for programs whose execution is very short. It is also useful in all
+ * programs when tracing is started close to the end of the program
+ * execution.
+ *
+ * FIXME: For now, this only works for the first trace created in a
+ * process.
+ */
+
 static void __attribute__((destructor)) keepalive()
 {
-//     struct ustcomm_ustd ustd;
-//     int result;
-//     sigset_t sigset;
-//
-//     result = sigemptyset(&sigset);
-//     if(result == -1) {
-//             perror("sigemptyset");
-//             return;
-//     }
-//     result = sigaddset(&sigset, SIGIO);
-//     if(result == -1) {
-//             perror("sigaddset");
-//             return;
-//     }
-//     result = sigprocmask(SIG_BLOCK, &sigset, NULL);
-//     if(result == -1) {
-//             perror("sigprocmask");
-//             return;
-//     }
-//
-//     if(trace_recording()) {
-//             if(!have_consumer()) {
-//                     /* Request listener creation. We've blocked SIGIO's in
-//                      * order to not interrupt sleep(), so we will miss the
-//                      * one sent by the daemon and therefore won't create
-//                      * the listener automatically.
-//                      */
-//                     create_listener();
-//
-                       printf("Keeping process alive for consumer daemon...\n");
-                       restarting_sleep(3);
-                       printf("Finally dying...\n");
-//             }
-//     }
-//
-//     result = sigprocmask(SIG_UNBLOCK, &sigset, NULL);
-//     if(result == -1) {
-//             perror("sigprocmask");
-//             return;
-//     }
+       if(trace_recording() && LOAD_SHARED(buffers_to_export)) {
+               int total = 0;
+               DBG("Keeping process alive for consumer daemon...");
+               while(LOAD_SHARED(buffers_to_export)) {
+                       const int interv = 200000;
+                       restarting_usleep(interv);
+                       total += interv;
+
+                       if(total >= 3000000) {
+                               WARN("non-consumed buffers remaining after wait limit; not waiting anymore");
+                               break;
+                       }
+               }
+               DBG("Finally dying...");
+       }
+
+       destroy_traces();
+
+       /* Ask the listener to stop and clean up. */
+       stop_listener();
+}
+
+void ust_potential_exec(void)
+{
+       trace_mark(ust, potential_exec, MARK_NOARGS);
+
+       DBG("test");
+
+       keepalive();
 }
-#endif
 
 /* Notify ust that there was a fork. This needs to be called inside
  * the new process, anytime a process whose memory is not shared with
  * the parent is created. If this function is not called, the events
  * of the new process will not be collected.
+ *
+ * Signals should be disabled before the fork and reenabled only after
+ * this call in order to guarantee tracing is not started before ust_fork()
+ * sanitizes the new process.
  */
 
-void ust_fork(void)
+static void ust_fork(void)
 {
+       struct blocked_consumer *bc;
+       struct blocked_consumer *deletable_bc = NULL;
+       int result;
+
+       /* FIXME: technically, the locks could have been taken before the fork */
        DBG("ust: forking");
+
+       /* break lock if necessary */
+       ltt_unlock_traces();
+
        ltt_trace_stop("auto");
-       ltt_trace_destroy("auto");
-       ltt_trace_alloc("auto");
-       ltt_trace_start("auto");
-       init_socket();
+       ltt_trace_destroy("auto", 1);
+       /* Delete all active connections */
+       ustcomm_close_all_connections(&ustcomm_app.server);
+
+       /* Delete all blocked consumers */
+       list_for_each_entry(bc, &blocked_consumers, list) {
+               result = close(bc->fd_producer);
+               if(result == -1) {
+                       PERROR("close");
+               }
+               free(deletable_bc);
+               deletable_bc = bc;
+               list_del(&bc->list);
+       }
+
+       /* free app, keeping socket file */
+       ustcomm_fini_app(&ustcomm_app, 1);
+
+       STORE_SHARED(buffers_to_export, 0);
        have_listener = 0;
+       init_socket();
        create_listener();
+       ltt_trace_setup("auto");
+       result = ltt_trace_set_type("auto", "ustrelay");
+       if(result < 0) {
+               ERR("ltt_trace_set_type failed");
+               return;
+       }
+
+       ltt_trace_alloc("auto");
+       ltt_trace_start("auto");
        inform_consumer_daemon("auto");
 }
 
+void ust_before_fork(ust_fork_info_t *fork_info)
+{
+        /* Disable signals. This is to avoid that the child
+         * intervenes before it is properly setup for tracing. It is
+         * safer to disable all signals, because then we know we are not
+         * breaking anything by restoring the original mask.
+         */
+       sigset_t all_sigs;
+       int result;
+
+        /* FIXME:
+                - only do this if tracing is active
+        */
+
+        /* Disable signals */
+        sigfillset(&all_sigs);
+        result = sigprocmask(SIG_BLOCK, &all_sigs, &fork_info->orig_sigs);
+        if(result == -1) {
+                PERROR("sigprocmask");
+                return;
+        }
+}
+
+/* Don't call this function directly in a traced program */
+static void ust_after_fork_common(ust_fork_info_t *fork_info)
+{
+       int result;
+
+        /* Restore signals */
+        result = sigprocmask(SIG_SETMASK, &fork_info->orig_sigs, NULL);
+        if(result == -1) {
+                PERROR("sigprocmask");
+                return;
+        }
+}
+
+void ust_after_fork_parent(ust_fork_info_t *fork_info)
+{
+       /* Reenable signals */
+       ust_after_fork_common(fork_info);
+}
+
+void ust_after_fork_child(ust_fork_info_t *fork_info)
+{
+       /* First sanitize the child */
+       ust_fork();
+
+       /* Then reenable interrupts */
+       ust_after_fork_common(fork_info);
+}
+
This page took 0.043612 seconds and 4 git commands to generate.