[ust.git] / libtracectl / tracectl.c

#include <stdio.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 "marker.h"
#include "tracer.h"
#include "localerr.h"
#include "ustcomm.h"

//#define USE_CLONE

#define UNIX_PATH_MAX 108

#define SOCKETDIR "/tmp/socks"
#define SOCKETDIRLEN sizeof(SOCKETDIR)
#define USTSIGNAL SIGIO

#define MAX_MSG_SIZE (100)
#define MSG_NOTIF 1
#define MSG_REGISTER_NOTIF 2

char consumer_stack[10000];

static struct ustcomm_app ustcomm_app;

struct tracecmd { /* no padding */
	uint32_t size;
	uint16_t command;
};

//struct listener_arg {
//	int pipe_fd;
//};

struct trctl_msg {
	/* size: the size of all the fields except size itself */
	uint32_t size;
	uint16_t type;
	/* Only the necessary part of the payload is transferred. It
         * may even be none of it.
         */
	char payload[94];
};

char mysocketfile[UNIX_PATH_MAX] = "";
//int pfd = -1;

struct consumer_channel {
	int fd;
	struct ltt_channel_struct *chan;
};

int consumer(void *arg)
{
	int result;
	int fd;
	char str[] = "Hello, this is the consumer.\n";
	struct ltt_trace_struct *trace;
	struct consumer_channel *consumer_channels;
	int i;
	char trace_name[] = "auto";

	ltt_lock_traces();
	trace = _ltt_trace_find(trace_name);
	ltt_unlock_traces();

	if(trace == NULL) {
		CPRINTF("cannot find trace!");
		return 1;
	}

	consumer_channels = (struct consumer_channel *) malloc(trace->nr_channels * sizeof(struct consumer_channel));
	if(consumer_channels == NULL) {
		ERR("malloc returned NULL");
		return 1;
	}

	CPRINTF("opening trace files");
	for(i=0; i<trace->nr_channels; i++) {
		char tmp[100];
		struct ltt_channel_struct *chan = &trace->channels[i];

		consumer_channels[i].chan = chan;

		snprintf(tmp, sizeof(tmp), "trace/%s_0", chan->channel_name);
		result = consumer_channels[i].fd = open(tmp, O_WRONLY | O_CREAT | O_TRUNC, 00600);
		if(result == -1) {
			perror("open");
			return -1;
		}
		CPRINTF("\topened trace file %s", tmp);
		
	}
	CPRINTF("done opening trace files");

	for(;;) {
		/*wait*/

		for(i=0; i<trace->nr_channels; i++) {
			struct rchan *rchan = consumer_channels[i].chan->trans_channel_data;
			struct rchan_buf *rbuf = rchan->buf;
			struct ltt_channel_buf_struct *lttbuf = consumer_channels[i].chan->buf;
			long consumed_old;

			result = ltt_do_get_subbuf(rbuf, lttbuf, &consumed_old);
			if(result < 0) {
				DBG("ltt_do_get_subbuf: error: %s", strerror(-result));
			}
			else {
				DBG("success!");

				result = write(consumer_channels[i].fd, rbuf->buf_data + (consumed_old & (2 * 4096-1)), 4096);
				ltt_do_put_subbuf(rbuf, lttbuf, consumed_old);
			}
		}

		sleep(1);
	}
}

void start_consumer(void)
{
#ifdef USE_CLONE
	int result;

	result = clone(consumer, consumer_stack+sizeof(consumer_stack)-1, CLONE_FS | CLONE_FILES | CLONE_VM | CLONE_SIGHAND | CLONE_THREAD, NULL);
	if(result == -1) {
		perror("clone");
	}
#else
	pthread_t thread;

	pthread_create(&thread, NULL, consumer, NULL);
#endif
}

static void print_markers(void)
{
	struct marker_iter iter;

	lock_markers();
	marker_iter_reset(&iter);
	marker_iter_start(&iter);

	while(iter.marker) {
		fprintf(stderr, "marker: %s_%s \"%s\"\n", iter.marker->channel, iter.marker->name, iter.marker->format);
		marker_iter_next(&iter);
	}
	unlock_markers();
}

void do_command(struct tracecmd *cmd)
{
}

void receive_commands()
{
}

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;
	}
}

#define CONSUMER_DAEMON_SOCK SOCKETDIR "/ustd"

static int inform_consumer_daemon(void)
{
	ustcomm_request_consumer(getpid(), "metadata");
	ustcomm_request_consumer(getpid(), "ust");
}

int listener_main(void *p)
{
	int result;

	DBG("LISTENER");

	for(;;) {
		uint32_t size;
		struct sockaddr_un addr;
		socklen_t addrlen = sizeof(addr);
		char trace_name[] = "auto";
		char trace_type[] = "ustrelay";
		char *recvbuf;
		int len;
		struct ustcomm_source src;

		result = ustcomm_app_recv_message(&ustcomm_app, &recvbuf, &src, -1);
		if(result <= 0) {
			WARN("error in ustcomm_app_recv_message");
			continue;
		}

		DBG("received a message! it's: %s\n", recvbuf);
		len = strlen(recvbuf);

		if(!strcmp(recvbuf, "print_markers")) {
			print_markers();
		}
		else if(!strcmp(recvbuf, "trace_setup")) {
			DBG("trace setup");

			result = ltt_trace_setup(trace_name);
			if(result < 0) {
				ERR("ltt_trace_setup failed");
				return;
			}

			result = ltt_trace_set_type(trace_name, trace_type);
			if(result < 0) {
				ERR("ltt_trace_set_type failed");
				return;
			}
		}
		else if(!strcmp(recvbuf, "trace_alloc")) {
			DBG("trace alloc");

			result = ltt_trace_alloc(trace_name);
			if(result < 0) {
				ERR("ltt_trace_alloc failed");
				return;
			}
		}
		else if(!strcmp(recvbuf, "trace_start")) {
			DBG("trace start");

			result = ltt_trace_start(trace_name);
			if(result < 0) {
				ERR("ltt_trace_start failed");
				continue;
			}
		}
		else if(!strcmp(recvbuf, "trace_stop")) {
			DBG("trace stop");

			result = ltt_trace_stop(trace_name);
			if(result < 0) {
				ERR("ltt_trace_stop failed");
				return;
			}
		}
		else if(!strcmp(recvbuf, "trace_destroy")) {

			DBG("trace destroy");

			result = ltt_trace_destroy(trace_name);
			if(result < 0) {
				ERR("ltt_trace_destroy failed");
				return;
			}
		}
		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;
			}

			ltt_lock_traces();
			trace = _ltt_trace_find(trace_name);
			ltt_unlock_traces();

			if(trace == NULL) {
				CPRINTF("cannot find trace!");
				return 1;
			}

			for(i=0; i<trace->nr_channels; i++) {
				struct rchan *rchan = trace->channels[i].trans_channel_data;
				struct rchan_buf *rbuf = rchan->buf;

				if(!strcmp(trace->channels[i].channel_name, channel_name)) {
					char *reply;

					DBG("the shmid for the requested channel is %d", rbuf->shmid);
					asprintf(&reply, "%d", rbuf->shmid);

					result = ustcomm_send_reply(&ustcomm_app.server, reply, &src);
					if(result) {
						ERR("listener: get_shmid: ustcomm_send_reply failed");
						goto next_cmd;
					}

					free(reply);

					break;
				}
			}
		}
		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;
			}

			ltt_lock_traces();
			trace = _ltt_trace_find(trace_name);
			ltt_unlock_traces();

			if(trace == NULL) {
				CPRINTF("cannot find trace!");
				return 1;
			}

			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)) {
					char *reply;

					DBG("the n_subbufs for the requested channel is %d", rchan->n_subbufs);
					asprintf(&reply, "%d", rchan->n_subbufs);

					result = ustcomm_send_reply(&ustcomm_app.server, reply, &src);
					if(result) {
						ERR("listener: get_n_subbufs: ustcomm_send_reply failed");
						goto next_cmd;
					}

					free(reply);

					break;
				}
			}
		}
		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();

			if(trace == NULL) {
				CPRINTF("cannot find trace!");
				return 1;
			}

			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)) {
					char *reply;

					DBG("the subbuf_size for the requested channel is %d", rchan->subbuf_size);
					asprintf(&reply, "%d", rchan->subbuf_size);

					result = ustcomm_send_reply(&ustcomm_app.server, reply, &src);
					if(result) {
						ERR("listener: get_subbuf_size: ustcomm_send_reply failed");
						goto next_cmd;
					}

					free(reply);

					break;
				}
			}
		}
		else if(nth_token_is(recvbuf, "load_probe_lib", 0) == 1) {
			char *libfile;

			libfile = nth_token(recvbuf, 1);

			DBG("load_probe_lib loading %s", libfile);
		}
		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;
			}

			ltt_lock_traces();
			trace = _ltt_trace_find(trace_name);
			ltt_unlock_traces();

			if(trace == NULL) {
				CPRINTF("cannot find trace!");
				return 1;
			}

			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_get_subbuf(rbuf, lttbuf, &consumed_old);
					if(result < 0) {
						DBG("ltt_do_get_subbuf: error: %s", strerror(-result));
						asprintf(&reply, "%s %ld", "UNAVAIL", 0);
					}
					else {
						DBG("ltt_do_get_subbuf: success");
						asprintf(&reply, "%s %ld", "OK", consumed_old);
					}

					result = ustcomm_send_reply(&ustcomm_app.server, reply, &src);
					if(result) {
						ERR("listener: get_subbuf: ustcomm_send_reply failed");
						goto next_cmd;
					}

					free(reply);

					break;
				}
			}
		}
		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;
			}

			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;
			}

			ltt_lock_traces();
			trace = _ltt_trace_find(trace_name);
			ltt_unlock_traces();

			if(trace == NULL) {
				CPRINTF("cannot find trace!");
				return 1;
			}

			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");
					}
					else {
						DBG("ltt_do_put_subbuf: success");
					}
					asprintf(&reply, "%s", "OK", consumed_old);

					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;
				}
			}

			free(channel_name);
			free(consumed_old_str);
		}
		else {
			ERR("unable to parse message: %s", recvbuf);
		}

	next_cmd:
		free(recvbuf);
	}
}

static char listener_stack[16384];

void create_listener(void)
{
	int result;
	static char listener_stack[16384];
	//char *listener_stack = malloc(16384);

#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");
	}
#else
	pthread_t thread;

	pthread_create(&thread, NULL, listener_main, NULL);
#endif
}

/* The signal handler itself. Signals must be setup so there cannot be
   nested signals. */

void sighandler(int sig)
{
	static char have_listener = 0;
	DBG("sighandler");

	if(!have_listener) {
		create_listener();
		have_listener = 1;
	}
}

/* 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);
}

static void destroy_socket(void)
{
	int result;

	if(mysocketfile[0] == '\0')
		return;

	result = unlink(mysocketfile);
	if(result == -1) {
		PERROR("unlink");
	}
}

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.
	 */

	int result;
	struct sigaction act;

	result = sigemptyset(&act.sa_mask);
	if(result == -1) {
		PERROR("sigemptyset");
		return -1;
	}

	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);
	if(result == -1) {
		PERROR("sigaction");
		return -1;
	}

	return 0;
}

static void auto_probe_connect(struct marker *m)
{
	int result;

	result = ltt_marker_connect(m->channel, m->name, "default");
	if(result)
		ERR("ltt_marker_connect");

	DBG("just auto connected marker %s %s to probe default", m->channel, m->name);
}

static void __attribute__((constructor(101))) init0()
{
	DBG("UST_AUTOPROBE constructor");
	if(getenv("UST_AUTOPROBE")) {
		marker_set_new_marker_cb(auto_probe_connect);
	}
}

static void fini(void);

static void __attribute__((constructor(1000))) init()
{
	int result;

	DBG("UST_TRACE 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;
	}

	if(getenv("UST_TRACE")) {
		char trace_name[] = "auto";
		char trace_type[] = "ustrelay";

		DBG("starting early tracing");

		/* Ensure marker control is initialized */
		init_marker_control();

		/* Ensure relay is initialized */
		init_ustrelay_transport();

		/* Ensure markers are initialized */
		init_markers();

		/* In case. */
		ltt_channels_register("ust");

		result = ltt_trace_setup(trace_name);
		if(result < 0) {
			ERR("ltt_trace_setup failed");
			return;
		}

		result = ltt_trace_set_type(trace_name, trace_type);
		if(result < 0) {
			ERR("ltt_trace_set_type failed");
			return;
		}

		result = ltt_trace_alloc(trace_name);
		if(result < 0) {
			ERR("ltt_trace_alloc failed");
			return;
		}

		result = ltt_trace_start(trace_name);
		if(result < 0) {
			ERR("ltt_trace_start failed");
			return;
		}
		//start_consumer();
		inform_consumer_daemon();
	}


	return;

	/* should decrementally destroy stuff if error */

}

/* This is only called if we terminate normally, not with an unhandled signal,
 * so we cannot rely on it. */

static void __attribute__((destructor)) fini()
{
	int result;

	/* if trace running, finish it */

	DBG("destructor stopping traces");

	result = ltt_trace_stop("auto");
	if(result == -1) {
		ERR("ltt_trace_stop error");
	}

	result = ltt_trace_destroy("auto");
	if(result == -1) {
		ERR("ltt_trace_destroy error");
	}

	/* FIXME: wait for the consumer to be done */
	DBG("waiting 5 sec for consume");
	sleep(5);

	destroy_socket();
}
Commit	Line	Data
68c1021b PMF	1	#include <stdio.h>
	2	#include <stdint.h>
	3	#include <signal.h>
	4	#include <sys/types.h>
	5	#include <sys/socket.h>
	6	#include <sys/un.h>
98963de4	7	#include <sched.h>
a584bc4e	8	#include <fcntl.h>
fbd8191b PMF	9
fbd8191b PMF	10	#include "marker.h"
a584bc4e	11	#include "tracer.h"
d0b5f2b9 PMF	12	#include "localerr.h"
d0b5f2b9 PMF	13	#include "ustcomm.h"
fbd8191b	14
b02e31e5	15	//#define USE_CLONE
3847c3ba	16
68c1021b PMF	17	#define UNIX_PATH_MAX 108
68c1021b PMF	18
68c1021b PMF	19	#define SOCKETDIR "/tmp/socks"
	20	#define SOCKETDIRLEN sizeof(SOCKETDIR)
	21	#define USTSIGNAL SIGIO
	22
98963de4 PMF	23	#define MAX_MSG_SIZE (100)
	24	#define MSG_NOTIF 1
	25	#define MSG_REGISTER_NOTIF 2
	26
a584bc4e PMF	27	char consumer_stack[10000];
a584bc4e PMF	28
d0b5f2b9 PMF	29	static struct ustcomm_app ustcomm_app;
d0b5f2b9 PMF	30
68c1021b PMF	31	struct tracecmd { /* no padding */
	32	uint32_t size;
	33	uint16_t command;
	34	};
	35
98963de4 PMF	36	//struct listener_arg {
	37	// int pipe_fd;
	38	//};
	39
	40	struct trctl_msg {
	41	/* size: the size of all the fields except size itself */
	42	uint32_t size;
	43	uint16_t type;
	44	/* Only the necessary part of the payload is transferred. It
	45	* may even be none of it.
	46	*/
	47	char payload[94];
	48	};
68c1021b	49
68c1021b	50	char mysocketfile[UNIX_PATH_MAX] = "";
d0b5f2b9	51	//int pfd = -1;
68c1021b	52
a584bc4e PMF	53	struct consumer_channel {
	54	int fd;
	55	struct ltt_channel_struct *chan;
	56	};
	57
	58	int consumer(void *arg)
	59	{
	60	int result;
	61	int fd;
	62	char str[] = "Hello, this is the consumer.\n";
	63	struct ltt_trace_struct *trace;
	64	struct consumer_channel *consumer_channels;
	65	int i;
	66	char trace_name[] = "auto";
	67
	68	ltt_lock_traces();
	69	trace = _ltt_trace_find(trace_name);
	70	ltt_unlock_traces();
	71
	72	if(trace == NULL) {
	73	CPRINTF("cannot find trace!");
	74	return 1;
	75	}
	76
	77	consumer_channels = (struct consumer_channel ) malloc(trace->nr_channels sizeof(struct consumer_channel));
	78	if(consumer_channels == NULL) {
	79	ERR("malloc returned NULL");
	80	return 1;
	81	}
	82
	83	CPRINTF("opening trace files");
	84	for(i=0; i<trace->nr_channels; i++) {
	85	char tmp[100];
	86	struct ltt_channel_struct *chan = &trace->channels[i];
	87
	88	consumer_channels[i].chan = chan;
	89
1c184644	90	snprintf(tmp, sizeof(tmp), "trace/%s_0", chan->channel_name);
688760ef	91	result = consumer_channels[i].fd = open(tmp, O_WRONLY \| O_CREAT \| O_TRUNC, 00600);
a584bc4e PMF	92	if(result == -1) {
	93	perror("open");
	94	return -1;
	95	}
	96	CPRINTF("\topened trace file %s", tmp);
	97
	98	}
	99	CPRINTF("done opening trace files");
	100
	101	for(;;) {
	102	/wait/
	103
	104	for(i=0; i<trace->nr_channels; i++) {
	105	struct rchan *rchan = consumer_channels[i].chan->trans_channel_data;
	106	struct rchan_buf *rbuf = rchan->buf;
	107	struct ltt_channel_buf_struct *lttbuf = consumer_channels[i].chan->buf;
	108	long consumed_old;
	109
	110	result = ltt_do_get_subbuf(rbuf, lttbuf, &consumed_old);
	111	if(result < 0) {
1c184644	112	DBG("ltt_do_get_subbuf: error: %s", strerror(-result));
a584bc4e PMF	113	}
a584bc4e PMF	114	else {
1c184644	115	DBG("success!");
a584bc4e PMF	116
	117	result = write(consumer_channels[i].fd, rbuf->buf_data + (consumed_old & (2 * 4096-1)), 4096);
	118	ltt_do_put_subbuf(rbuf, lttbuf, consumed_old);
	119	}
	120	}
	121
	122	sleep(1);
	123	}
a584bc4e PMF	124	}
	125
	126	void start_consumer(void)
	127	{
3847c3ba	128	#ifdef USE_CLONE
a584bc4e PMF	129	int result;
	130
	131	result = clone(consumer, consumer_stack+sizeof(consumer_stack)-1, CLONE_FS \| CLONE_FILES \| CLONE_VM \| CLONE_SIGHAND \| CLONE_THREAD, NULL);
	132	if(result == -1) {
	133	perror("clone");
	134	}
3847c3ba PMF	135	#else
	136	pthread_t thread;
	137
	138	pthread_create(&thread, NULL, consumer, NULL);
	139	#endif
a584bc4e	140	}
fbd8191b PMF	141
	142	static void print_markers(void)
	143	{
	144	struct marker_iter iter;
	145
d0b5f2b9	146	lock_markers();
fbd8191b PMF	147	marker_iter_reset(&iter);
	148	marker_iter_start(&iter);
	149
	150	while(iter.marker) {
	151	fprintf(stderr, "marker: %s_%s \"%s\"\n", iter.marker->channel, iter.marker->name, iter.marker->format);
	152	marker_iter_next(&iter);
	153	}
d0b5f2b9	154	unlock_markers();
fbd8191b PMF	155	}
fbd8191b PMF	156
68c1021b PMF	157	void do_command(struct tracecmd *cmd)
	158	{
	159	}
	160
	161	void receive_commands()
	162	{
	163	}
	164
98963de4 PMF	165	int fd_notif = -1;
	166	void notif_cb(void)
	167	{
	168	int result;
	169	struct trctl_msg msg;
	170
	171	/* FIXME: fd_notif should probably be protected by a spinlock */
	172
	173	if(fd_notif == -1)
	174	return;
	175
	176	msg.type = MSG_NOTIF;
	177	msg.size = sizeof(msg.type);
	178
	179	/* FIXME: don't block here */
	180	result = write(fd_notif, &msg, msg.size+sizeof(msg.size));
	181	if(result == -1) {
	182	PERROR("write");
	183	return;
	184	}
	185	}
	186
d0b5f2b9 PMF	187	#define CONSUMER_DAEMON_SOCK SOCKETDIR "/ustd"
	188
	189	static int inform_consumer_daemon(void)
	190	{
3847c3ba PMF	191	ustcomm_request_consumer(getpid(), "metadata");
3847c3ba PMF	192	ustcomm_request_consumer(getpid(), "ust");
d0b5f2b9	193	}
fbd8191b	194
98963de4 PMF	195	int listener_main(void *p)
	196	{
	197	int result;
	198
b0540e11 PMF	199	DBG("LISTENER");
b0540e11 PMF	200
98963de4	201	for(;;) {
98963de4	202	uint32_t size;
98963de4 PMF	203	struct sockaddr_un addr;
98963de4 PMF	204	socklen_t addrlen = sizeof(addr);
aafb1650 PMF	205	char trace_name[] = "auto";
aafb1650 PMF	206	char trace_type[] = "ustrelay";
d0b5f2b9 PMF	207	char *recvbuf;
d0b5f2b9 PMF	208	int len;
b02e31e5	209	struct ustcomm_source src;
98963de4	210
688760ef PMF	211	result = ustcomm_app_recv_message(&ustcomm_app, &recvbuf, &src, -1);
688760ef PMF	212	if(result <= 0) {
d0b5f2b9 PMF	213	WARN("error in ustcomm_app_recv_message");
	214	continue;
	215	}
98963de4	216
d0b5f2b9 PMF	217	DBG("received a message! it's: %s\n", recvbuf);
d0b5f2b9 PMF	218	len = strlen(recvbuf);
98963de4	219
d0b5f2b9 PMF	220	if(!strcmp(recvbuf, "print_markers")) {
	221	print_markers();
	222	}
	223	else if(!strcmp(recvbuf, "trace_setup")) {
	224	DBG("trace setup");
fbd8191b	225
d0b5f2b9 PMF	226	result = ltt_trace_setup(trace_name);
	227	if(result < 0) {
	228	ERR("ltt_trace_setup failed");
	229	return;
fbd8191b	230	}
d0b5f2b9 PMF	231
	232	result = ltt_trace_set_type(trace_name, trace_type);
	233	if(result < 0) {
	234	ERR("ltt_trace_set_type failed");
	235	return;
fbd8191b	236	}
d0b5f2b9 PMF	237	}
	238	else if(!strcmp(recvbuf, "trace_alloc")) {
	239	DBG("trace alloc");
	240
	241	result = ltt_trace_alloc(trace_name);
	242	if(result < 0) {
	243	ERR("ltt_trace_alloc failed");
	244	return;
fbd8191b	245	}
d0b5f2b9 PMF	246	}
	247	else if(!strcmp(recvbuf, "trace_start")) {
	248	DBG("trace start");
	249
	250	result = ltt_trace_start(trace_name);
	251	if(result < 0) {
	252	ERR("ltt_trace_start failed");
	253	continue;
fbd8191b	254	}
d0b5f2b9 PMF	255	}
	256	else if(!strcmp(recvbuf, "trace_stop")) {
	257	DBG("trace stop");
	258
	259	result = ltt_trace_stop(trace_name);
	260	if(result < 0) {
	261	ERR("ltt_trace_stop failed");
	262	return;
aafb1650	263	}
d0b5f2b9 PMF	264	}
d0b5f2b9 PMF	265	else if(!strcmp(recvbuf, "trace_destroy")) {
aafb1650	266
d0b5f2b9	267	DBG("trace destroy");
aafb1650	268
d0b5f2b9 PMF	269	result = ltt_trace_destroy(trace_name);
	270	if(result < 0) {
	271	ERR("ltt_trace_destroy failed");
	272	return;
fbd8191b	273	}
98963de4	274	}
b02e31e5	275	else if(nth_token_is(recvbuf, "get_shmid", 0) == 1) {
3847c3ba PMF	276	struct ltt_trace_struct *trace;
	277	char trace_name[] = "auto";
	278	int i;
811e4b93	279	char *channel_name;
3847c3ba PMF	280
	281	DBG("get_shmid");
	282
811e4b93 PMF	283	channel_name = nth_token(recvbuf, 1);
	284	if(channel_name == NULL) {
	285	ERR("get_shmid: cannot parse channel");
	286	goto next_cmd;
	287	}
	288
3847c3ba PMF	289	ltt_lock_traces();
	290	trace = _ltt_trace_find(trace_name);
	291	ltt_unlock_traces();
	292
	293	if(trace == NULL) {
	294	CPRINTF("cannot find trace!");
	295	return 1;
	296	}
	297
	298	for(i=0; i<trace->nr_channels; i++) {
	299	struct rchan *rchan = trace->channels[i].trans_channel_data;
	300	struct rchan_buf *rbuf = rchan->buf;
	301
811e4b93 PMF	302	if(!strcmp(trace->channels[i].channel_name, channel_name)) {
	303	char *reply;
	304
	305	DBG("the shmid for the requested channel is %d", rbuf->shmid);
	306	asprintf(&reply, "%d", rbuf->shmid);
	307
	308	result = ustcomm_send_reply(&ustcomm_app.server, reply, &src);
	309	if(result) {
	310	ERR("listener: get_shmid: ustcomm_send_reply failed");
	311	goto next_cmd;
	312	}
	313
	314	free(reply);
	315
	316	break;
	317	}
	318	}
	319	}
	320	else if(nth_token_is(recvbuf, "get_n_subbufs", 0) == 1) {
	321	struct ltt_trace_struct *trace;
	322	char trace_name[] = "auto";
	323	int i;
	324	char *channel_name;
	325
	326	DBG("get_n_subbufs");
	327
	328	channel_name = nth_token(recvbuf, 1);
	329	if(channel_name == NULL) {
	330	ERR("get_n_subbufs: cannot parse channel");
	331	goto next_cmd;
	332	}
	333
	334	ltt_lock_traces();
	335	trace = _ltt_trace_find(trace_name);
	336	ltt_unlock_traces();
	337
	338	if(trace == NULL) {
	339	CPRINTF("cannot find trace!");
	340	return 1;
	341	}
	342
	343	for(i=0; i<trace->nr_channels; i++) {
	344	struct rchan *rchan = trace->channels[i].trans_channel_data;
	345
	346	if(!strcmp(trace->channels[i].channel_name, channel_name)) {
	347	char *reply;
	348
	349	DBG("the n_subbufs for the requested channel is %d", rchan->n_subbufs);
	350	asprintf(&reply, "%d", rchan->n_subbufs);
	351
	352	result = ustcomm_send_reply(&ustcomm_app.server, reply, &src);
	353	if(result) {
	354	ERR("listener: get_n_subbufs: ustcomm_send_reply failed");
	355	goto next_cmd;
	356	}
	357
	358	free(reply);
	359
	360	break;
	361	}
	362	}
	363	}
	364	else if(nth_token_is(recvbuf, "get_subbuf_size", 0) == 1) {
	365	struct ltt_trace_struct *trace;
366	char trace_name[] = "auto";
367	int i;
368	char *channel_name;
369
370	DBG("get_subbuf_size");
371
372	channel_name = nth_token(recvbuf, 1);
373	if(channel_name == NULL) {
374	ERR("get_subbuf_size: cannot parse channel");
375	goto next_cmd;
376	}
377
378	ltt_lock_traces();
379	trace = _ltt_trace_find(trace_name);
380	ltt_unlock_traces();
381
382	if(trace == NULL) {
383	CPRINTF("cannot find trace!");
384	return 1;
385	}
386
387	for(i=0; i<trace->nr_channels; i++) {
388	struct rchan *rchan = trace->channels[i].trans_channel_data;
389
390	if(!strcmp(trace->channels[i].channel_name, channel_name)) {
391	char *reply;
392
393	DBG("the subbuf_size for the requested channel is %d", rchan->subbuf_size);
394	asprintf(&reply, "%d", rchan->subbuf_size);
395
396	result = ustcomm_send_reply(&ustcomm_app.server, reply, &src);
397	if(result) {
398	ERR("listener: get_subbuf_size: ustcomm_send_reply failed");
399	goto next_cmd;
400	}
401
402	free(reply);
3847c3ba	403
811e4b93 PMF	404	break;
811e4b93 PMF	405	}
3847c3ba PMF	406	}
3847c3ba PMF	407	}
b02e31e5 PMF	408	else if(nth_token_is(recvbuf, "load_probe_lib", 0) == 1) {
	409	char *libfile;
	410
	411	libfile = nth_token(recvbuf, 1);
	412
	413	DBG("load_probe_lib loading %s", libfile);
	414	}
688760ef PMF	415	else if(nth_token_is(recvbuf, "get_subbuffer", 0) == 1) {
	416	struct ltt_trace_struct *trace;
	417	char trace_name[] = "auto";
	418	int i;
	419	char *channel_name;
	420
	421	DBG("get_subbuf");
	422
	423	channel_name = nth_token(recvbuf, 1);
	424	if(channel_name == NULL) {
	425	ERR("get_subbuf: cannot parse channel");
	426	goto next_cmd;
	427	}
	428
	429	ltt_lock_traces();
	430	trace = _ltt_trace_find(trace_name);
	431	ltt_unlock_traces();
	432
	433	if(trace == NULL) {
	434	CPRINTF("cannot find trace!");
	435	return 1;
	436	}
	437
	438	for(i=0; i<trace->nr_channels; i++) {
	439	struct rchan *rchan = trace->channels[i].trans_channel_data;
	440
	441	if(!strcmp(trace->channels[i].channel_name, channel_name)) {
	442	struct rchan_buf *rbuf = rchan->buf;
	443	struct ltt_channel_buf_struct *lttbuf = trace->channels[i].buf;
	444	char *reply;
	445	long consumed_old=0;
	446
	447	result = ltt_do_get_subbuf(rbuf, lttbuf, &consumed_old);
	448	if(result < 0) {
	449	DBG("ltt_do_get_subbuf: error: %s", strerror(-result));
	450	asprintf(&reply, "%s %ld", "UNAVAIL", 0);
	451	}
	452	else {
	453	DBG("ltt_do_get_subbuf: success");
	454	asprintf(&reply, "%s %ld", "OK", consumed_old);
	455	}
	456
	457	result = ustcomm_send_reply(&ustcomm_app.server, reply, &src);
	458	if(result) {
	459	ERR("listener: get_subbuf: ustcomm_send_reply failed");
	460	goto next_cmd;
	461	}
	462
	463	free(reply);
	464
	465	break;
	466	}
	467	}
	468	}
	469	else if(nth_token_is(recvbuf, "put_subbuffer", 0) == 1) {
	470	struct ltt_trace_struct *trace;
	471	char trace_name[] = "auto";
	472	int i;
	473	char *channel_name;
	474	long consumed_old;
	475	char *consumed_old_str;
	476	char *endptr;
	477
	478	DBG("put_subbuf");
479
480	channel_name = strdup_malloc(nth_token(recvbuf, 1));
481	if(channel_name == NULL) {
482	ERR("put_subbuf_size: cannot parse channel");
483	goto next_cmd;
484	}
485
486	consumed_old_str = strdup_malloc(nth_token(recvbuf, 2));
487	if(consumed_old_str == NULL) {
488	ERR("put_subbuf: cannot parse consumed_old");
489	goto next_cmd;
490	}
491	consumed_old = strtol(consumed_old_str, &endptr, 10);
492	if(*endptr != '\0') {
493	ERR("put_subbuf: invalid value for consumed_old");
494	goto next_cmd;
495	}
496
497	ltt_lock_traces();
498	trace = _ltt_trace_find(trace_name);
499	ltt_unlock_traces();
500
501	if(trace == NULL) {
502	CPRINTF("cannot find trace!");
503	return 1;
504	}
505
506	for(i=0; i<trace->nr_channels; i++) {
507	struct rchan *rchan = trace->channels[i].trans_channel_data;
508
509	if(!strcmp(trace->channels[i].channel_name, channel_name)) {
510	struct rchan_buf *rbuf = rchan->buf;
511	struct ltt_channel_buf_struct *lttbuf = trace->channels[i].buf;
512	char *reply;
513	long consumed_old=0;
514
515	result = ltt_do_put_subbuf(rbuf, lttbuf, consumed_old);
516	if(result < 0) {
517	WARN("ltt_do_put_subbuf: error");
518	}
519	else {
520	DBG("ltt_do_put_subbuf: success");
521	}
522	asprintf(&reply, "%s", "OK", consumed_old);
523
524	result = ustcomm_send_reply(&ustcomm_app.server, reply, &src);
525	if(result) {
526	ERR("listener: put_subbuf: ustcomm_send_reply failed");
527	goto next_cmd;
528	}
529
530	free(reply);
531
532	break;
533	}
534	}
535
536	free(channel_name);
537	free(consumed_old_str);
538	}
539	else {
540	ERR("unable to parse message: %s", recvbuf);
541	}
d0b5f2b9	542
811e4b93	543	next_cmd:
d0b5f2b9	544	free(recvbuf);
98963de4 PMF	545	}
	546	}
	547
b0540e11 PMF	548	static char listener_stack[16384];
b0540e11 PMF	549
98963de4 PMF	550	void create_listener(void)
	551	{
	552	int result;
	553	static char listener_stack[16384];
b0540e11	554	//char *listener_stack = malloc(16384);
98963de4	555
3847c3ba	556	#ifdef USE_CLONE
fbd8191b	557	result = clone(listener_main, listener_stack+sizeof(listener_stack)-1, CLONE_FS \| CLONE_FILES \| CLONE_VM \| CLONE_SIGHAND \| CLONE_THREAD, NULL);
98963de4 PMF	558	if(result == -1) {
	559	perror("clone");
	560	}
3847c3ba PMF	561	#else
3847c3ba PMF	562	pthread_t thread;
b0540e11	563
3847c3ba PMF	564	pthread_create(&thread, NULL, listener_main, NULL);
3847c3ba PMF	565	#endif
98963de4 PMF	566	}
98963de4 PMF	567
d0b5f2b9 PMF	568	/* The signal handler itself. Signals must be setup so there cannot be
d0b5f2b9 PMF	569	nested signals. */
68c1021b PMF	570
	571	void sighandler(int sig)
	572	{
d0b5f2b9	573	static char have_listener = 0;
68c1021b	574	DBG("sighandler");
d0b5f2b9 PMF	575
	576	if(!have_listener) {
	577	create_listener();
	578	have_listener = 1;
	579	}
68c1021b PMF	580	}
	581
	582	/* Called by the app signal handler to chain it to us. */
	583
98963de4	584	void chain_signal(void)
68c1021b PMF	585	{
	586	sighandler(USTSIGNAL);
	587	}
	588
98963de4	589	static int init_socket(void)
68c1021b	590	{
d0b5f2b9	591	return ustcomm_init_app(getpid(), &ustcomm_app);
68c1021b PMF	592	}
68c1021b PMF	593
98963de4	594	static void destroy_socket(void)
68c1021b PMF	595	{
	596	int result;
	597
	598	if(mysocketfile[0] == '\0')
	599	return;
	600
	601	result = unlink(mysocketfile);
	602	if(result == -1) {
	603	PERROR("unlink");
	604	}
	605	}
	606
98963de4	607	static int init_signal_handler(void)
68c1021b PMF	608	{
	609	/* Attempt to handler SIGIO. If the main program wants to
	610	* handle it, fine, it'll override us. They it'll have to
	611	* use the chaining function.
	612	*/
	613
	614	int result;
	615	struct sigaction act;
	616
	617	result = sigemptyset(&act.sa_mask);
	618	if(result == -1) {
	619	PERROR("sigemptyset");
	620	return -1;
	621	}
	622
	623	act.sa_handler = sighandler;
	624	act.sa_flags = SA_RESTART;
	625
	626	/* Only defer ourselves. Also, try to restart interrupted
	627	* syscalls to disturb the traced program as little as possible.
	628	*/
	629	result = sigaction(SIGIO, &act, NULL);
	630	if(result == -1) {
	631	PERROR("sigaction");
	632	return -1;
	633	}
	634
	635	return 0;
	636	}
	637
20b37a31	638	static void auto_probe_connect(struct marker *m)
68c1021b PMF	639	{
	640	int result;
	641
20b37a31 PMF	642	result = ltt_marker_connect(m->channel, m->name, "default");
	643	if(result)
	644	ERR("ltt_marker_connect");
	645
	646	DBG("just auto connected marker %s %s to probe default", m->channel, m->name);
	647	}
	648
	649	static void __attribute__((constructor(101))) init0()
	650	{
	651	DBG("UST_AUTOPROBE constructor");
	652	if(getenv("UST_AUTOPROBE")) {
	653	marker_set_new_marker_cb(auto_probe_connect);
	654	}
	655	}
	656
a584bc4e PMF	657	static void fini(void);
a584bc4e PMF	658
20b37a31 PMF	659	static void __attribute__((constructor(1000))) init()
	660	{
	661	int result;
	662
	663	DBG("UST_TRACE constructor");
	664
3847c3ba PMF	665	/* Must create socket before signal handler to prevent races.
	666	*/
	667	result = init_socket();
	668	if(result == -1) {
	669	ERR("init_socket error");
	670	return;
	671	}
	672	result = init_signal_handler();
	673	if(result == -1) {
	674	ERR("init_signal_handler error");
	675	return;
	676	}
68c1021b	677
4db647c5 PMF	678	if(getenv("UST_TRACE")) {
	679	char trace_name[] = "auto";
	680	char trace_type[] = "ustrelay";
	681
	682	DBG("starting early tracing");
	683
	684	/* Ensure marker control is initialized */
	685	init_marker_control();
	686
	687	/* Ensure relay is initialized */
	688	init_ustrelay_transport();
	689
	690	/* Ensure markers are initialized */
	691	init_markers();
	692
20b37a31 PMF	693	/* In case. */
20b37a31 PMF	694	ltt_channels_register("ust");
4db647c5 PMF	695
	696	result = ltt_trace_setup(trace_name);
	697	if(result < 0) {
	698	ERR("ltt_trace_setup failed");
	699	return;
	700	}
	701
	702	result = ltt_trace_set_type(trace_name, trace_type);
	703	if(result < 0) {
	704	ERR("ltt_trace_set_type failed");
	705	return;
	706	}
	707
	708	result = ltt_trace_alloc(trace_name);
	709	if(result < 0) {
	710	ERR("ltt_trace_alloc failed");
	711	return;
	712	}
	713
	714	result = ltt_trace_start(trace_name);
	715	if(result < 0) {
	716	ERR("ltt_trace_start failed");
	717	return;
	718	}
3847c3ba PMF	719	//start_consumer();
3847c3ba PMF	720	inform_consumer_daemon();
4db647c5 PMF	721	}
4db647c5 PMF	722
68c1021b PMF	723
	724	return;
	725
	726	/* should decrementally destroy stuff if error */
	727
	728	}
	729
	730	/* This is only called if we terminate normally, not with an unhandled signal,
	731	* so we cannot rely on it. */
	732
98963de4	733	static void __attribute__((destructor)) fini()
68c1021b	734	{
a584bc4e PMF	735	int result;
	736
	737	/* if trace running, finish it */
	738
	739	DBG("destructor stopping traces");
	740
	741	result = ltt_trace_stop("auto");
	742	if(result == -1) {
	743	ERR("ltt_trace_stop error");
	744	}
	745
	746	result = ltt_trace_destroy("auto");
	747	if(result == -1) {
	748	ERR("ltt_trace_destroy error");
	749	}
	750
	751	/* FIXME: wait for the consumer to be done */
688760ef PMF	752	DBG("waiting 5 sec for consume");
688760ef PMF	753	sleep(5);
a584bc4e	754
68c1021b PMF	755	destroy_socket();
68c1021b PMF	756	}