X-Git-Url: https://git.lttng.org/?a=blobdiff_plain;f=libust%2Ftracectl.c;h=2fccb1caab9d73deef930469a29cd842ccb50a50;hb=0e4b45ace8b034dc4352a30d89b74f5e59e2e27d;hp=e84d28904d0db3286a68c8e92db88f4312eb767e;hpb=27e84572611e54fa1158f72bdd18482e0385ae47;p=ust.git

diff --git a/libust/tracectl.c b/libust/tracectl.c
index e84d289..2fccb1c 100644
--- a/libust/tracectl.c
+++ b/libust/tracectl.c
@@ -17,17 +17,16 @@
 
 #define _GNU_SOURCE
 #include <stdio.h>
+#include <stdlib.h>
 #include <stdint.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-bp.h>
+#include <urcu/uatomic_arch.h>
 
 #include <ust/marker.h>
 #include <ust/tracectl.h>
@@ -36,8 +35,7 @@
 #include "ustcomm.h"
 #include "buffers.h"
 #include "marker-control.h"
-
-//#define USE_CLONE
+#include "multipoll.h"
 
 #define USTSIGNAL SIGIO
 
@@ -125,15 +123,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)
 {
@@ -197,84 +186,67 @@ 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++;
-	}
-
-	fds = (struct pollfd *) malloc(n_fds * sizeof(struct pollfd));
-	if(fds == NULL) {
-		ERR("malloc returned NULL");
-		return;
-	}
-
-	list_for_each_entry(bc, &blocked_consumers, list) {
-		fds[idx].fd = bc->fd_producer;
-		fds[idx].events = POLLIN;
-		bc->tmp_poll_idx = idx;
-		idx++;
-	}
-
-	while((result = poll(fds, n_fds, 0)) == -1 && errno == EINTR)
-		/* nothing */;
+	result = read(bc->fd_producer, &inbuf, 1);
 	if(result == -1) {
-		PERROR("poll");
-		return;
+		PERROR("read");
+		return -1;
 	}
+	if(result == 0) {
+		int res;
+		DBG("listener: got messsage that a buffer ended");
 
-	list_for_each_entry(bc, &blocked_consumers, list) {
-		if(fds[bc->tmp_poll_idx].revents) {
-			long consumed_old = 0;
-			char *reply;
+		res = close(bc->fd_producer);
+		if(res == -1) {
+			PERROR("close");
+		}
 
-			result = read(bc->fd_producer, &inbuf, 1);
-			if(result == -1) {
-				PERROR("read");
-				continue;
-			}
-			if(result == 0) {
-				DBG("PRODUCER END");
+		list_del(&bc->list);
 
-				close(bc->fd_producer);
+		result = ustcomm_send_reply(&bc->server, "END", &bc->src);
+		if(result < 0) {
+			ERR("ustcomm_send_reply failed");
+			return -1;
+		}
 
-				list_del(&bc->list);
+		return 0;
+	}
 
-				result = ustcomm_send_reply(&bc->server, "END", &bc->src);
-				if(result < 0) {
-					ERR("ustcomm_send_reply failed");
-					continue;
-				}
+	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);
 
-				continue;
-			}
+	list_del(&bc->list);
 
-			result = ust_buffers_get_subbuf(bc->buf, &consumed_old);
-			if(result == -EAGAIN) {
-				WARN("missed buffer?");
-				continue;
-			}
-			else if(result < 0) {
-				DBG("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);
-				continue;
-			}
-			free(reply);
+	return 0;
+}
 
-			list_del(&bc->list);
-		}
+void blocked_consumers_add_to_mp(struct mpentries *ent)
+{
+	struct blocked_consumer *bc;
+
+	list_for_each_entry(bc, &blocked_consumers, list) {
+		multipoll_add(ent, bc->fd_producer, POLLIN, process_blkd_consumer_act, bc, NULL);
 	}
 
 }
@@ -357,10 +329,7 @@ static int do_cmd_get_shmid(const char *recvbuf, struct ustcomm_source *src)
 		}
 	}
 
-	if(found) {
-		buffers_to_export--;
-	}
-	else {
+	if(!found) {
 		ERR("channel not found (%s)", channel_and_cpu);
 	}
 
@@ -509,6 +478,128 @@ static int do_cmd_get_subbuf_size(const char *recvbuf, struct ustcomm_source *sr
 	return retval;
 }
 
+static unsigned int poweroftwo(unsigned int x)
+{
+    unsigned int power2 = 1;
+    unsigned int hardcoded = 2147483648; /* FIX max 2^31 */
+
+    if (x < 2)
+        return 2;
+
+    while (power2 < x && power2 < hardcoded)
+        power2 *= 2;
+
+    return power2;
+}
+
+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 = poweroftwo(size);
+	if (power != size)
+		WARN("using the next 2^n = %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;
+		}
+	}
+	if(found == 0) {
+		ERR("unable to find channel");
+	}
+
+	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;
+		}
+	}
+	if(found == 0) {
+		ERR("unable to find channel");
+	}
+
+	end:
+	ltt_unlock_traces();
+	return retval;
+}
+
 static int do_cmd_get_subbuffer(const char *recvbuf, struct ustcomm_source *src)
 {
 	int retval = 0;
@@ -536,24 +627,19 @@ static int do_cmd_get_subbuffer(const char *recvbuf, struct ustcomm_source *src)
 
 	ltt_lock_traces();
 	trace = _ltt_trace_find(trace_name);
-	ltt_unlock_traces();
 
 	if(trace == NULL) {
-		char *reply;
 		int result;
 
-		WARN("Cannot find trace. It was likely destroyed by the user.");
-		asprintf(&reply, "%s", "NOTFOUND");
-		result = ustcomm_send_reply(&ustcomm_app.server, reply, src);
+		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");
-			free(reply);
 			retval = -1;
-			goto free_short_chan_name;
+			goto unlock_traces;
 		}
 
-		free(reply);
-		goto free_short_chan_name;
+		goto unlock_traces;
 	}
 
 	for(i=0; i<trace->nr_channels; i++) {
@@ -568,7 +654,7 @@ static int do_cmd_get_subbuffer(const char *recvbuf, struct ustcomm_source *src)
 			bc = (struct blocked_consumer *) malloc(sizeof(struct blocked_consumer));
 			if(bc == NULL) {
 				ERR("malloc returned NULL");
-				goto free_short_chan_name;
+				goto unlock_traces;
 			}
 			bc->fd_consumer = src->fd;
 			bc->fd_producer = buf->data_ready_fd_read;
@@ -578,6 +664,14 @@ static int do_cmd_get_subbuffer(const char *recvbuf, struct ustcomm_source *src)
 
 			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");
+				buffers_to_export--;
+			}
+
 			break;
 		}
 	}
@@ -585,6 +679,9 @@ static int do_cmd_get_subbuffer(const char *recvbuf, struct ustcomm_source *src)
 		ERR("unable to find channel");
 	}
 
+	unlock_traces:
+	ltt_unlock_traces();
+
 	free_short_chan_name:
 	free(ch_name);
 
@@ -639,21 +736,17 @@ static int do_cmd_put_subbuffer(const char *recvbuf, struct ustcomm_source *src)
 
 	ltt_lock_traces();
 	trace = _ltt_trace_find(trace_name);
-	ltt_unlock_traces();
 
 	if(trace == NULL) {
-		WARN("Cannot find trace. It was likely destroyed by the user.");
-		asprintf(&reply, "%s", "NOTFOUND");
-		result = ustcomm_send_reply(&ustcomm_app.server, reply, src);
+		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");
-			free(reply);
 			retval = -1;
-			goto free_short_chan_name;
+			goto unlock_traces;
 		}
 
-		free(reply);
-		goto free_short_chan_name;
+		goto unlock_traces;
 	}
 
 	for(i=0; i<trace->nr_channels; i++) {
@@ -679,7 +772,7 @@ static int do_cmd_put_subbuffer(const char *recvbuf, struct ustcomm_source *src)
 				ERR("ustcomm_send_reply failed");
 				free(reply);
 				retval = -1;
-				goto free_channel_and_cpu;
+				goto unlock_traces;
 			}
 
 			free(reply);
@@ -690,6 +783,8 @@ static int do_cmd_put_subbuffer(const char *recvbuf, struct ustcomm_source *src)
 		ERR("unable to find channel");
 	}
 
+	unlock_traces:
+	ltt_unlock_traces();
 	free_short_chan_name:
 	free(ch_name);
 	free_consumed_old_str:
@@ -701,224 +796,217 @@ static int do_cmd_put_subbuffer(const char *recvbuf, struct ustcomm_source *src)
 	return retval;
 }
 
-void *listener_main(void *p)
+static void listener_cleanup(void *ptr)
+{
+	ustcomm_fini_app(&ustcomm_app, 0);
+}
+
+int process_client_cmd(char *recvbuf, struct ustcomm_source *src)
 {
 	int result;
+	char trace_name[] = "auto";
+	char trace_type[] = "ustrelay";
+	int len;
 
-	ust_register_thread();
+	DBG("received a message! it's: %s", recvbuf);
+	len = strlen(recvbuf);
 
-	DBG("LISTENER");
+	if(!strcmp(recvbuf, "print_markers")) {
+		print_markers(stderr);
+	}
+	else if(!strcmp(recvbuf, "list_markers")) {
+		char *ptr;
+		size_t size;
+		FILE *fp;
 
-	for(;;) {
-		char trace_name[] = "auto";
-		char trace_type[] = "ustrelay";
-		char *recvbuf;
-		int len;
-		struct ustcomm_source src;
+		fp = open_memstream(&ptr, &size);
+		print_markers(fp);
+		fclose(fp);
 
-		process_blocked_consumers();
+		result = ustcomm_send_reply(&ustcomm_app.server, ptr, src);
 
-		result = ustcomm_app_recv_message(&ustcomm_app, &recvbuf, &src, 5);
+		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) {
-			WARN("error in ustcomm_app_recv_message");
-			continue;
-		}
-		else if(result == 0) {
-			/* no message */
-			continue;
+			ERR("ltt_trace_setup failed");
+			return -1;
 		}
 
-		DBG("received a message! it's: %s", recvbuf);
-		len = strlen(recvbuf);
-
-		if(!strcmp(recvbuf, "print_markers")) {
-			print_markers(stderr);
+		result = ltt_trace_set_type(trace_name, trace_type);
+		if(result < 0) {
+			ERR("ltt_trace_set_type failed");
+			return -1;
 		}
-		else if(!strcmp(recvbuf, "list_markers")) {
-			char *ptr;
-			size_t size;
-			FILE *fp;
-
-			fp = open_memstream(&ptr, &size);
-			print_markers(fp);
-			fclose(fp);
-
-			result = ustcomm_send_reply(&ustcomm_app.server, ptr, &src);
 
-			free(ptr);
+		result = ltt_trace_alloc(trace_name);
+		if(result < 0) {
+			ERR("ltt_trace_alloc failed");
+			return -1;
 		}
-		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;
-			}
-
-			result = ltt_trace_set_type(trace_name, trace_type);
-			if(result < 0) {
-				ERR("ltt_trace_set_type failed");
-				return (void *)1;
-			}
 
-			result = ltt_trace_alloc(trace_name);
-			if(result < 0) {
-				ERR("ltt_trace_alloc failed");
-				return (void *)1;
-			}
-
-			inform_consumer_daemon(trace_name);
+		inform_consumer_daemon(trace_name);
 
-			result = ltt_trace_start(trace_name);
-			if(result < 0) {
-				ERR("ltt_trace_start failed");
-				continue;
-			}
+		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");
-
-			result = ltt_trace_setup(trace_name);
-			if(result < 0) {
-				ERR("ltt_trace_setup failed");
-				return (void *)1;
-			}
+	}
+	else if(!strcmp(recvbuf, "trace_setup")) {
+		DBG("trace setup");
 
-			result = ltt_trace_set_type(trace_name, trace_type);
-			if(result < 0) {
-				ERR("ltt_trace_set_type failed");
-				return (void *)1;
-			}
+		result = ltt_trace_setup(trace_name);
+		if(result < 0) {
+			ERR("ltt_trace_setup failed");
+			return -1;
 		}
-		else if(!strcmp(recvbuf, "trace_alloc")) {
-			DBG("trace alloc");
 
-			result = ltt_trace_alloc(trace_name);
-			if(result < 0) {
-				ERR("ltt_trace_alloc failed");
-				return (void *)1;
-			}
+		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_create")) {
-			DBG("trace create");
+	}
+	else if(!strcmp(recvbuf, "trace_alloc")) {
+		DBG("trace alloc");
 
-			result = ltt_trace_setup(trace_name);
-			if(result < 0) {
-				ERR("ltt_trace_setup failed");
-				return (void *)1;
-			}
+		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");
 
-			result = ltt_trace_set_type(trace_name, trace_type);
-			if(result < 0) {
-				ERR("ltt_trace_set_type failed");
-				return (void *)1;
-			}
+		result = ltt_trace_setup(trace_name);
+		if(result < 0) {
+			ERR("ltt_trace_setup failed");
+			return -1;
+		}
 
-			result = ltt_trace_alloc(trace_name);
-			if(result < 0) {
-				ERR("ltt_trace_alloc failed");
-				return (void *)1;
-			}
+		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_start")) {
+		DBG("trace start");
 
+		result = ltt_trace_alloc(trace_name);
+		if(result < 0) {
+			ERR("ltt_trace_alloc failed");
+			return -1;
+		}
+		if(!result) {
 			inform_consumer_daemon(trace_name);
 		}
-		else if(!strcmp(recvbuf, "trace_start")) {
-			DBG("trace start");
 
-			result = ltt_trace_start(trace_name);
-			if(result < 0) {
-				ERR("ltt_trace_start failed");
-				continue;
-			}
+		result = ltt_trace_start(trace_name);
+		if(result < 0) {
+			ERR("ltt_trace_start failed");
+			return -1;
 		}
-		else if(!strcmp(recvbuf, "trace_stop")) {
-			DBG("trace stop");
+	}
+	else if(!strcmp(recvbuf, "trace_stop")) {
+		DBG("trace stop");
 
-			result = ltt_trace_stop(trace_name);
-			if(result < 0) {
-				ERR("ltt_trace_stop failed");
-				return (void *)1;
-			}
+		result = ltt_trace_stop(trace_name);
+		if(result < 0) {
+			ERR("ltt_trace_stop failed");
+			return -1;
 		}
-		else if(!strcmp(recvbuf, "trace_destroy")) {
+	}
+	else if(!strcmp(recvbuf, "trace_destroy")) {
 
-			DBG("trace destroy");
+		DBG("trace destroy");
 
-			result = ltt_trace_destroy(trace_name);
-			if(result < 0) {
-				ERR("ltt_trace_destroy failed");
-				return (void *)1;
-			}
-		}
-		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);
+		result = ltt_trace_destroy(trace_name, 0);
+		if(result < 0) {
+			ERR("ltt_trace_destroy failed");
+			return -1;
 		}
-		else if(nth_token_is(recvbuf, "load_probe_lib", 0) == 1) {
-			char *libfile;
+	}
+	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;
 
-			libfile = nth_token(recvbuf, 1);
+		libfile = nth_token(recvbuf, 1);
 
-			DBG("load_probe_lib loading %s", libfile);
+		DBG("load_probe_lib loading %s", libfile);
 
-			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, "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);
+		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]="";
 
-			if(channel_name == NULL || marker_name == NULL) {
-				WARN("invalid marker name");
-				goto next_cmd;
-			}
+		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);
-			}
+		if(channel_name == NULL || marker_name == NULL) {
+			WARN("invalid marker name");
+			goto next_cmd;
 		}
-		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_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(marker_name == NULL) {
-			}
+		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);
-			}
+		if(marker_name == NULL) {
 		}
-		else if(nth_token_is(recvbuf, "get_pidunique", 0) == 1) {
-			char *reply;
 
-			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_pidunique", 0) == 1) {
+		char *reply;
 
-			result = ustcomm_send_reply(&ustcomm_app.server, reply, &src);
-			if(result) {
-				ERR("listener: get_pidunique: ustcomm_send_reply failed");
-				goto next_cmd;
-			}
+		asprintf(&reply, "%lld", pidunique);
 
-			free(reply);
+		result = ustcomm_send_reply(&ustcomm_app.server, reply, src);
+		if(result) {
+			ERR("listener: get_pidunique: ustcomm_send_reply failed");
+			goto next_cmd;
 		}
+
+		free(reply);
+	}
 //		else if(nth_token_is(recvbuf, "get_notifications", 0) == 1) {
 //			struct ust_trace *trace;
 //			char trace_name[] = "auto";
@@ -950,7 +1038,7 @@ void *listener_main(void *p)
 //					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);
+//					result = fd = ustcomm_app_detach_client(&ustcomm_app, src);
 //					if(result == -1) {
 //						ERR("ustcomm_app_detach_client failed");
 //						goto next_cmd;
@@ -968,41 +1056,61 @@ void *listener_main(void *p)
 //
 //			free(channel_name);
 //		}
-		else {
-			ERR("unable to parse message: %s", recvbuf);
+	else {
+		ERR("unable to parse message: %s", recvbuf);
+	}
+
+next_cmd:
+
+	return 0;
+}
+
+void *listener_main(void *p)
+{
+	int result;
+
+	DBG("LISTENER");
+
+	pthread_cleanup_push(listener_cleanup, NULL);
+
+	for(;;) {
+		struct mpentries mpent;
+
+		multipoll_init(&mpent);
+
+		blocked_consumers_add_to_mp(&mpent);
+		ustcomm_mp_add_app_clients(&mpent, &ustcomm_app, process_client_cmd);
+
+		result = multipoll_poll(&mpent, -1);
+		if(result == -1) {
+			ERR("error in multipoll_poll");
 		}
 
-	next_cmd:
-		free(recvbuf);
+		multipoll_destroy(&mpent);
 	}
+
+	pthread_cleanup_pop(1);
 }
 
-volatile sig_atomic_t have_listener = 0;
+/* 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;
 
 void create_listener(void)
 {
-#ifdef USE_CLONE
-	static char listener_stack[16384];
 	int result;
-#else
-	pthread_t thread;
-#endif
 
 	if(have_listener) {
 		WARN("not creating listener because we already had one");
 		return;
 	}
 
-#ifdef USE_CLONE
-	result = clone((int (*)(void *)) listener_main, listener_stack+sizeof(listener_stack)-1, CLONE_FS | CLONE_FILES | CLONE_VM | CLONE_SIGHAND | CLONE_THREAD, NULL);
+	result = pthread_create(&listener_thread, NULL, listener_main, NULL);
 	if(result == -1) {
-		perror("clone");
-		return;
+		PERROR("pthread_create");
 	}
-#else
-
-	pthread_create(&thread, NULL, listener_main, NULL);
-#endif
 
 	have_listener = 1;
 }
@@ -1060,16 +1168,8 @@ static void __attribute__((constructor)) init()
 	 */
 	pidunique = make_pidunique();
 
-	/* Initialize RCU in case the constructor order is not good. */
-	rcu_init();
-
-	/* It is important to do this before events start to be generated. */
-	ust_register_thread();
-
 	DBG("Tracectl constructor");
 
-	/* Must create socket before signal handler to prevent races.
-         */
 	result = init_socket();
 	if(result == -1) {
 		ERR("init_socket error");
@@ -1222,7 +1322,7 @@ static void destroy_traces(void)
 		ERR("ltt_trace_stop error");
 	}
 
-	result = ltt_trace_destroy("auto");
+	result = ltt_trace_destroy("auto", 0);
 	if(result == -1) {
 		ERR("ltt_trace_destroy error");
 	}
@@ -1269,6 +1369,20 @@ int restarting_usleep(useconds_t usecs)
 	return result;
 }
 
+static void stop_listener()
+{
+	int result;
+
+	result = pthread_cancel(listener_thread);
+	if(result == -1) {
+		PERROR("pthread_cancel");
+	}
+	result = pthread_join(listener_thread, NULL);
+	if(result == -1) {
+		PERROR("pthread_join");
+	}
+}
+
 /* 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
@@ -1299,7 +1413,8 @@ static void __attribute__((destructor)) keepalive()
 
 	destroy_traces();
 
-	ustcomm_fini_app(&ustcomm_app);
+	/* Ask the listener to stop and clean up. */
+	stop_listener();
 }
 
 void ust_potential_exec(void)
@@ -1327,24 +1442,35 @@ static void ust_fork(void)
 	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_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) {
-		close(bc->fd_producer);
-		close(bc->fd_consumer);
+		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);
+
+	buffers_to_export = 0;
 	have_listener = 0;
-	create_listener();
 	init_socket();
+	create_listener();
 	ltt_trace_setup("auto");
 	result = ltt_trace_set_type("auto", "ustrelay");
 	if(result < 0) {