[ust.git] / libustd / libustd.c

/* Copyright (C) 2009  Pierre-Marc Fournier
 *               2010  Alexis Halle
 *
 * This library is free software; you can redistribute it and/or
 * modify it under the terms of the GNU Lesser General Public
 * License as published by the Free Software Foundation; either
 * version 2.1 of the License, or (at your option) any later version.
 *
 * This library is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
 * Lesser General Public License for more details.
 *
 * You should have received a copy of the GNU Lesser General Public
 * License along with this library; if not, write to the Free Software
 * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301 USA
 */

#define _GNU_SOURCE

#include <sys/shm.h>
#include <unistd.h>
#include <pthread.h>
#include <signal.h>

#include <stdlib.h>
#include <stdio.h>
#include <string.h>
#include <errno.h>
#include <assert.h>

#include "libustd.h"
#include "usterr.h"
#include "ustcomm.h"

/* return value: 0 = subbuffer is finished, it won't produce data anymore
 *               1 = got subbuffer successfully
 *               <0 = error
 */

#define GET_SUBBUF_OK 1
#define GET_SUBBUF_DONE 0
#define GET_SUBBUF_DIED 2

#define PUT_SUBBUF_OK 1
#define PUT_SUBBUF_DIED 0
#define PUT_SUBBUF_PUSHED 2
#define PUT_SUBBUF_DONE 3

#define UNIX_PATH_MAX 108

int get_subbuffer(struct buffer_info *buf)
{
	char *send_msg=NULL;
	char *received_msg=NULL;
	char *rep_code=NULL;
	int retval;
	int result;

	asprintf(&send_msg, "get_subbuffer %s", buf->name);
	result = ustcomm_send_request(buf->conn, send_msg, &received_msg);
	if((result == -1 && (errno == ECONNRESET || errno == EPIPE)) || result == 0) {
		DBG("app died while being traced");
		retval = GET_SUBBUF_DIED;
		goto end;
	}
	else if(result < 0) {
		ERR("get_subbuffer: ustcomm_send_request failed");
		retval = -1;
		goto end;
	}

	result = sscanf(received_msg, "%as %ld", &rep_code, &buf->consumed_old);
	if(result != 2 && result != 1) {
		ERR("unable to parse response to get_subbuffer");
		retval = -1;
		free(received_msg);
		goto end_rep;
	}

	if(!strcmp(rep_code, "OK")) {
		DBG("got subbuffer %s", buf->name);
		retval = GET_SUBBUF_OK;
	}
	else if(nth_token_is(received_msg, "END", 0) == 1) {
		retval = GET_SUBBUF_DONE;
		goto end_rep;
	}
	else if(!strcmp(received_msg, "NOTFOUND")) {
		DBG("For buffer %s, the trace was not found. This likely means it was destroyed by the user.", buf->name);
		retval = GET_SUBBUF_DIED;
		goto end_rep;
	}
	else {
		DBG("error getting subbuffer %s", buf->name);
		retval = -1;
	}

	/* FIXME: free correctly the stuff */
end_rep:
	if(rep_code)
		free(rep_code);
end:
	if(send_msg)
		free(send_msg);
	if(received_msg)
		free(received_msg);

	return retval;
}

int put_subbuffer(struct buffer_info *buf)
{
	char *send_msg=NULL;
	char *received_msg=NULL;
	char *rep_code=NULL;
	int retval;
	int result;

	asprintf(&send_msg, "put_subbuffer %s %ld", buf->name, buf->consumed_old);
	result = ustcomm_send_request(buf->conn, send_msg, &received_msg);
	if(result < 0 && (errno == ECONNRESET || errno == EPIPE)) {
		retval = PUT_SUBBUF_DIED;
		goto end;
	}
	else if(result < 0) {
		ERR("put_subbuffer: send_message failed");
		retval = -1;
		goto end;
	}
	else if(result == 0) {
		/* Program seems finished. However this might not be
		 * the last subbuffer that has to be collected.
		 */
		retval = PUT_SUBBUF_DIED;
		goto end;
	}

	result = sscanf(received_msg, "%as", &rep_code);
	if(result != 1) {
		ERR("unable to parse response to put_subbuffer");
		retval = -1;
		goto end_rep;
	}

	if(!strcmp(rep_code, "OK")) {
		DBG("subbuffer put %s", buf->name);
		retval = PUT_SUBBUF_OK;
	}
	else if(!strcmp(received_msg, "NOTFOUND")) {
		DBG("For buffer %s, the trace was not found. This likely means it was destroyed by the user.", buf->name);
		/* However, maybe this was not the last subbuffer. So
		 * we return the program died.
		 */
		retval = PUT_SUBBUF_DIED;
		goto end_rep;
	}
	else {
		DBG("put_subbuffer: received error, we were pushed");
		retval = PUT_SUBBUF_PUSHED;
		goto end_rep;
	}

end_rep:
	if(rep_code)
		free(rep_code);

end:
	if(send_msg)
		free(send_msg);
	if(received_msg)
		free(received_msg);

	return retval;
}

void decrement_active_buffers(void *arg)
{
	struct libustd_instance *instance = arg;
	pthread_mutex_lock(&instance->mutex);
	instance->active_buffers--;
	pthread_mutex_unlock(&instance->mutex);
}

struct buffer_info *connect_buffer(struct libustd_instance *instance, pid_t pid, const char *bufname)
{
	struct buffer_info *buf;
	char *send_msg;
	char *received_msg;
	int result;
	struct shmid_ds shmds;

	buf = (struct buffer_info *) malloc(sizeof(struct buffer_info));
	if(buf == NULL) {
		ERR("add_buffer: insufficient memory");
		return NULL;
	}

	buf->conn = malloc(sizeof(struct ustcomm_connection));
	if(buf->conn == NULL) {
		ERR("add_buffer: insufficient memory");
		free(buf);
		return NULL;
	}

	buf->name = bufname;
	buf->pid = pid;

	/* connect to app */
	result = ustcomm_connect_app(buf->pid, buf->conn);
	if(result) {
		WARN("unable to connect to process, it probably died before we were able to connect");
		return NULL;
	}

	/* get pidunique */
	asprintf(&send_msg, "get_pidunique");
	result = ustcomm_send_request(buf->conn, send_msg, &received_msg);
	free(send_msg);
	if(result == -1) {
		ERR("problem in ustcomm_send_request(get_pidunique)");
		return NULL;
	}
	if(result == 0) {
		goto error;
	}

	result = sscanf(received_msg, "%lld", &buf->pidunique);
	if(result != 1) {
		ERR("unable to parse response to get_pidunique");
		return NULL;
	}
	free(received_msg);
	DBG("got pidunique %lld", buf->pidunique);

	/* get shmid */
	asprintf(&send_msg, "get_shmid %s", buf->name);
	result = ustcomm_send_request(buf->conn, send_msg, &received_msg);
	free(send_msg);
	if(result == -1) {
		ERR("problem in ustcomm_send_request(get_shmid)");
		return NULL;
	}
	if(result == 0) {
		goto error;
	}

	result = sscanf(received_msg, "%d %d", &buf->shmid, &buf->bufstruct_shmid);
	if(result != 2) {
		ERR("unable to parse response to get_shmid (\"%s\")", received_msg);
		return NULL;
	}
	free(received_msg);
	DBG("got shmids %d %d", buf->shmid, buf->bufstruct_shmid);

	/* get n_subbufs */
	asprintf(&send_msg, "get_n_subbufs %s", buf->name);
	result = ustcomm_send_request(buf->conn, send_msg, &received_msg);
	free(send_msg);
	if(result == -1) {
		ERR("problem in ustcomm_send_request(g_n_subbufs)");
		return NULL;
	}
	if(result == 0) {
		goto error;
	}

	result = sscanf(received_msg, "%d", &buf->n_subbufs);
	if(result != 1) {
		ERR("unable to parse response to get_n_subbufs");
		return NULL;
	}
	free(received_msg);
	DBG("got n_subbufs %d", buf->n_subbufs);

	/* get subbuf size */
	asprintf(&send_msg, "get_subbuf_size %s", buf->name);
	result = ustcomm_send_request(buf->conn, send_msg, &received_msg);
	free(send_msg);
	if(result == -1) {
		ERR("problem in ustcomm_send_request(get_subbuf_size)");
		return NULL;
	}
	if(result == 0) {
		goto error;
	}

	result = sscanf(received_msg, "%d", &buf->subbuf_size);
	if(result != 1) {
		ERR("unable to parse response to get_subbuf_size");
		return NULL;
	}
	free(received_msg);
	DBG("got subbuf_size %d", buf->subbuf_size);

	/* attach memory */
	buf->mem = shmat(buf->shmid, NULL, 0);
	if(buf->mem == (void *) 0) {
		PERROR("shmat");
		return NULL;
	}
	DBG("successfully attached buffer memory");

	buf->bufstruct_mem = shmat(buf->bufstruct_shmid, NULL, 0);
	if(buf->bufstruct_mem == (void *) 0) {
		PERROR("shmat");
		return NULL;
	}
	DBG("successfully attached buffer bufstruct memory");

	/* obtain info on the memory segment */
	result = shmctl(buf->shmid, IPC_STAT, &shmds);
	if(result == -1) {
		PERROR("shmctl");
		return NULL;
	}
	buf->memlen = shmds.shm_segsz;

	if(instance->callbacks->on_open_buffer)
		instance->callbacks->on_open_buffer(instance->callbacks, buf);

	pthread_mutex_lock(&instance->mutex);
	instance->active_buffers++;
	pthread_mutex_unlock(&instance->mutex);

	return buf;

error:
	free(buf);
	return NULL;
}

static void destroy_buffer(struct libustd_callbacks *callbacks,
			struct buffer_info *buf)
{
	int result;

	result = ustcomm_close_app(buf->conn);
	if(result == -1) {
		WARN("problem calling ustcomm_close_app");
	}

	result = shmdt(buf->mem);
	if(result == -1) {
		PERROR("shmdt");
	}

	result = shmdt(buf->bufstruct_mem);
	if(result == -1) {
		PERROR("shmdt");
	}

	if(callbacks->on_close_buffer)
		callbacks->on_close_buffer(callbacks, buf);

	free(buf->conn);
	free(buf);
}

int consumer_loop(struct libustd_instance *instance, struct buffer_info *buf)
{
	int result;

	pthread_cleanup_push(decrement_active_buffers, instance);

	for(;;) {
		/* get the subbuffer */
		result = get_subbuffer(buf);
		if(result == -1) {
			ERR("error getting subbuffer");
			continue;
		}
		else if(result == GET_SUBBUF_DONE) {
			/* this is done */
			break;
		}
		else if(result == GET_SUBBUF_DIED) {
			finish_consuming_dead_subbuffer(instance->callbacks, buf);
			break;
		}

		if(instance->callbacks->on_read_subbuffer)
			instance->callbacks->on_read_subbuffer(instance->callbacks, buf);

		/* put the subbuffer */
		result = put_subbuffer(buf);
		if(result == -1) {
			ERR("unknown error putting subbuffer (channel=%s)", buf->name);
			break;
		}
		else if(result == PUT_SUBBUF_PUSHED) {
			ERR("Buffer overflow (channel=%s), reader pushed. This channel will not be usable passed this point.", buf->name);
			break;
		}
		else if(result == PUT_SUBBUF_DIED) {
			DBG("application died while putting subbuffer");
			/* Skip the first subbuffer. We are not sure it is trustable
			 * because the put_subbuffer() did not complete.
			 */
			if(instance->callbacks->on_put_error)
				instance->callbacks->on_put_error(instance->callbacks, buf);

			finish_consuming_dead_subbuffer(instance->callbacks, buf);
			break;
		}
		else if(result == PUT_SUBBUF_DONE) {
			/* Done with this subbuffer */
			/* FIXME: add a case where this branch is used? Upon
			 * normal trace termination, at put_subbuf time, a
			 * special last-subbuffer code could be returned by
			 * the listener.
			 */
			break;
		}
		else if(result == PUT_SUBBUF_OK) {
		}
	}

	DBG("thread for buffer %s is stopping", buf->name);

	/* FIXME: destroy, unalloc... */

	pthread_cleanup_pop(1);

	return 0;
}

struct consumer_thread_args {
	pid_t pid;
	const char *bufname;
	struct libustd_instance *instance;
};

void *consumer_thread(void *arg)
{
	struct buffer_info *buf;
	struct consumer_thread_args *args = (struct consumer_thread_args *) arg;
	int result;
	sigset_t sigset;

	DBG("GOT ARGS: pid %d bufname %s", args->pid, args->bufname);

	if(args->instance->callbacks->on_new_thread)
		args->instance->callbacks->on_new_thread(args->instance->callbacks);

	/* Block signals that should be handled by the main thread. */
	result = sigemptyset(&sigset);
	if(result == -1) {
		PERROR("sigemptyset");
		goto end;
	}
	result = sigaddset(&sigset, SIGTERM);
	if(result == -1) {
		PERROR("sigaddset");
		goto end;
	}
	result = sigaddset(&sigset, SIGINT);
	if(result == -1) {
		PERROR("sigaddset");
		goto end;
	}
	result = sigprocmask(SIG_BLOCK, &sigset, NULL);
	if(result == -1) {
		PERROR("sigprocmask");
		goto end;
	}

	buf = connect_buffer(args->instance, args->pid, args->bufname);
	if(buf == NULL) {
		ERR("failed to connect to buffer");
		goto end;
	}

	consumer_loop(args->instance, buf);

	destroy_buffer(args->instance->callbacks, buf);

	end:

	if(args->instance->callbacks->on_close_thread)
		args->instance->callbacks->on_close_thread(args->instance->callbacks);

	free((void *)args->bufname);
	free(args);
	return NULL;
}

int start_consuming_buffer(
	struct libustd_instance *instance, pid_t pid, const char *bufname)
{
	pthread_t thr;
	struct consumer_thread_args *args;
	int result;

	DBG("beginning of start_consuming_buffer: args: pid %d bufname %s", pid, bufname);

	args = (struct consumer_thread_args *) malloc(sizeof(struct consumer_thread_args));

	args->pid = pid;
	args->bufname = strdup(bufname);
	args->instance = instance;
	DBG("beginning2 of start_consuming_buffer: args: pid %d bufname %s", args->pid, args->bufname);

	result = pthread_create(&thr, NULL, consumer_thread, args);
	if(result == -1) {
		ERR("pthread_create failed");
		return -1;
	}
	result = pthread_detach(thr);
	if(result == -1) {
		ERR("pthread_detach failed");
		return -1;
	}
	DBG("end of start_consuming_buffer: args: pid %d bufname %s", args->pid, args->bufname);

	return 0;
}

int libustd_start_instance(struct libustd_instance *instance)
{
	int result;
	int timeout = -1;

	if(!instance->is_init) {
		ERR("libustd instance not initialized");
		return 1;
	}

	/* app loop */
	for(;;) {
		char *recvbuf;

		/* check for requests on our public socket */
		result = ustcomm_ustd_recv_message(instance->comm, &recvbuf, NULL, timeout);
		if(result == -1 && errno == EINTR) {
			/* Caught signal */
		}
		else if(result == -1) {
			ERR("error in ustcomm_ustd_recv_message");
			goto loop_end;
		}
		else if(result > 0) {
			if(!strncmp(recvbuf, "collect", 7)) {
				pid_t pid;
				char *bufname;
				int result;

				result = sscanf(recvbuf, "%*s %d %50as", &pid, &bufname);
				if(result != 2) {
					ERR("parsing error: %s", recvbuf);
					goto free_bufname;
				}

				result = start_consuming_buffer(instance, pid, bufname);
				if(result < 0) {
					ERR("error in add_buffer");
					goto free_bufname;
				}

				free_bufname:
				free(bufname);
			}
			else if(!strncmp(recvbuf, "exit", 4)) {
				/* Only there to force poll to return */
			}
			else {
				WARN("unknown command: %s", recvbuf);
			}

			free(recvbuf);
		}

		loop_end:

		if(instance->quit_program) {
			pthread_mutex_lock(&instance->mutex);
			if(instance->active_buffers == 0) {
				pthread_mutex_unlock(&instance->mutex);
				break;
			}
			pthread_mutex_unlock(&instance->mutex);
			timeout = 100;
		}
	}

	if(instance->callbacks->on_trace_end)
		instance->callbacks->on_trace_end(instance);

	libustd_delete_instance(instance);

	return 0;
}

void libustd_delete_instance(struct libustd_instance *instance)
{
	if(instance->is_init)
		ustcomm_fini_ustd(instance->comm);

	pthread_mutex_destroy(&instance->mutex);
	free(instance->sock_path);
	free(instance->comm);
	free(instance);
}

int libustd_stop_instance(struct libustd_instance *instance, int send_msg)
{
	int result;
	int fd;
	int bytes = 0;

	char msg[] = "exit";

	instance->quit_program = 1;

	if(!send_msg)
		return 0;

	/* Send a message through the socket to force poll to return */

	struct sockaddr_un addr;

	result = fd = socket(PF_UNIX, SOCK_STREAM, 0);
	if(result == -1) {
		PERROR("socket");
		return 1;
	}

	addr.sun_family = AF_UNIX;

	strncpy(addr.sun_path, instance->sock_path, UNIX_PATH_MAX);
	addr.sun_path[UNIX_PATH_MAX-1] = '\0';

	result = connect(fd, (struct sockaddr *)&addr, sizeof(addr));
	if(result == -1) {
		PERROR("connect");
	}

	while(bytes != sizeof(msg))
		bytes += send(fd, msg, sizeof(msg), 0);

	close(fd);

	return 0;
}

struct libustd_instance *libustd_new_instance(
	struct libustd_callbacks *callbacks, char *sock_path)
{
	struct libustd_instance *instance =
		malloc(sizeof(struct libustd_instance));
	if(!instance)
		return NULL;

	instance->comm = malloc(sizeof(struct ustcomm_ustd));
	if(!instance->comm) {
		free(instance);
		return NULL;
	}

	instance->callbacks = callbacks;
	instance->quit_program = 0;
	instance->is_init = 0;
	instance->active_buffers = 0;
	pthread_mutex_init(&instance->mutex, NULL);

	if(sock_path)
		instance->sock_path = strdup(sock_path);
	else
		instance->sock_path = NULL;

	return instance;
}

int libustd_init_instance(struct libustd_instance *instance)
{
	int result;
	result = ustcomm_init_ustd(instance->comm, instance->sock_path);
	if(result == -1) {
		ERR("failed to initialize socket");
		return 1;
	}
	instance->is_init = 1;
	return 0;
}
Commit	Line	Data
d159ac37 AH	1	/* Copyright (C) 2009 Pierre-Marc Fournier
	2	* 2010 Alexis Halle
	3	*
	4	* This library is free software; you can redistribute it and/or
	5	* modify it under the terms of the GNU Lesser General Public
	6	* License as published by the Free Software Foundation; either
	7	* version 2.1 of the License, or (at your option) any later version.
	8	*
	9	* This library is distributed in the hope that it will be useful,
	10	* but WITHOUT ANY WARRANTY; without even the implied warranty of
	11	* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
	12	* Lesser General Public License for more details.
	13	*
	14	* You should have received a copy of the GNU Lesser General Public
	15	* License along with this library; if not, write to the Free Software
	16	* Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
	17	*/
	18
	19	#define _GNU_SOURCE
	20
	21	#include <sys/shm.h>
	22	#include <unistd.h>
	23	#include <pthread.h>
	24	#include <signal.h>
	25
	26	#include <stdlib.h>
	27	#include <stdio.h>
	28	#include <string.h>
	29	#include <errno.h>
	30	#include <assert.h>
	31
	32	#include "libustd.h"
	33	#include "usterr.h"
	34	#include "ustcomm.h"
	35
	36	/* return value: 0 = subbuffer is finished, it won't produce data anymore
	37	* 1 = got subbuffer successfully
	38	* <0 = error
	39	*/
	40
	41	#define GET_SUBBUF_OK 1
	42	#define GET_SUBBUF_DONE 0
	43	#define GET_SUBBUF_DIED 2
	44
	45	#define PUT_SUBBUF_OK 1
	46	#define PUT_SUBBUF_DIED 0
	47	#define PUT_SUBBUF_PUSHED 2
	48	#define PUT_SUBBUF_DONE 3
	49
	50	#define UNIX_PATH_MAX 108
	51
	52	int get_subbuffer(struct buffer_info *buf)
	53	{
	54	char *send_msg=NULL;
	55	char *received_msg=NULL;
	56	char *rep_code=NULL;
	57	int retval;
	58	int result;
	59
	60	asprintf(&send_msg, "get_subbuffer %s", buf->name);
f3f8cc91	61	result = ustcomm_send_request(buf->conn, send_msg, &received_msg);
d159ac37 AH	62	if((result == -1 && (errno == ECONNRESET \|\| errno == EPIPE)) \|\| result == 0) {
	63	DBG("app died while being traced");
	64	retval = GET_SUBBUF_DIED;
	65	goto end;
	66	}
	67	else if(result < 0) {
	68	ERR("get_subbuffer: ustcomm_send_request failed");
	69	retval = -1;
	70	goto end;
	71	}
	72
	73	result = sscanf(received_msg, "%as %ld", &rep_code, &buf->consumed_old);
	74	if(result != 2 && result != 1) {
	75	ERR("unable to parse response to get_subbuffer");
	76	retval = -1;
	77	free(received_msg);
	78	goto end_rep;
	79	}
	80
	81	if(!strcmp(rep_code, "OK")) {
	82	DBG("got subbuffer %s", buf->name);
	83	retval = GET_SUBBUF_OK;
	84	}
	85	else if(nth_token_is(received_msg, "END", 0) == 1) {
	86	retval = GET_SUBBUF_DONE;
	87	goto end_rep;
	88	}
	89	else if(!strcmp(received_msg, "NOTFOUND")) {
	90	DBG("For buffer %s, the trace was not found. This likely means it was destroyed by the user.", buf->name);
	91	retval = GET_SUBBUF_DIED;
	92	goto end_rep;
	93	}
	94	else {
	95	DBG("error getting subbuffer %s", buf->name);
	96	retval = -1;
	97	}
	98
	99	/* FIXME: free correctly the stuff */
	100	end_rep:
	101	if(rep_code)
	102	free(rep_code);
	103	end:
	104	if(send_msg)
	105	free(send_msg);
	106	if(received_msg)
	107	free(received_msg);
	108
	109	return retval;
	110	}
	111
	112	int put_subbuffer(struct buffer_info *buf)
	113	{
	114	char *send_msg=NULL;
	115	char *received_msg=NULL;
	116	char *rep_code=NULL;
	117	int retval;
	118	int result;
	119
	120	asprintf(&send_msg, "put_subbuffer %s %ld", buf->name, buf->consumed_old);
f3f8cc91	121	result = ustcomm_send_request(buf->conn, send_msg, &received_msg);
d159ac37 AH	122	if(result < 0 && (errno == ECONNRESET \|\| errno == EPIPE)) {
	123	retval = PUT_SUBBUF_DIED;
	124	goto end;
	125	}
	126	else if(result < 0) {
	127	ERR("put_subbuffer: send_message failed");
	128	retval = -1;
	129	goto end;
	130	}
	131	else if(result == 0) {
	132	/* Program seems finished. However this might not be
	133	* the last subbuffer that has to be collected.
	134	*/
	135	retval = PUT_SUBBUF_DIED;
	136	goto end;
	137	}
	138
	139	result = sscanf(received_msg, "%as", &rep_code);
	140	if(result != 1) {
	141	ERR("unable to parse response to put_subbuffer");
	142	retval = -1;
	143	goto end_rep;
	144	}
	145
	146	if(!strcmp(rep_code, "OK")) {
	147	DBG("subbuffer put %s", buf->name);
	148	retval = PUT_SUBBUF_OK;
	149	}
	150	else if(!strcmp(received_msg, "NOTFOUND")) {
	151	DBG("For buffer %s, the trace was not found. This likely means it was destroyed by the user.", buf->name);
	152	/* However, maybe this was not the last subbuffer. So
	153	* we return the program died.
	154	*/
	155	retval = PUT_SUBBUF_DIED;
	156	goto end_rep;
	157	}
	158	else {
	159	DBG("put_subbuffer: received error, we were pushed");
	160	retval = PUT_SUBBUF_PUSHED;
	161	goto end_rep;
	162	}
	163
	164	end_rep:
	165	if(rep_code)
	166	free(rep_code);
	167
	168	end:
	169	if(send_msg)
	170	free(send_msg);
	171	if(received_msg)
	172	free(received_msg);
	173
	174	return retval;
	175	}
	176
	177	void decrement_active_buffers(void *arg)
	178	{
	179	struct libustd_instance *instance = arg;
	180	pthread_mutex_lock(&instance->mutex);
	181	instance->active_buffers--;
	182	pthread_mutex_unlock(&instance->mutex);
	183	}
	184
	185	struct buffer_info connect_buffer(struct libustd_instance instance, pid_t pid, const char *bufname)
186	{
187	struct buffer_info *buf;
188	char *send_msg;
189	char *received_msg;
190	int result;
d159ac37 AH	191	struct shmid_ds shmds;
	192
	193	buf = (struct buffer_info *) malloc(sizeof(struct buffer_info));
	194	if(buf == NULL) {
	195	ERR("add_buffer: insufficient memory");
	196	return NULL;
	197	}
	198
f3f8cc91 AH	199	buf->conn = malloc(sizeof(struct ustcomm_connection));
	200	if(buf->conn == NULL) {
	201	ERR("add_buffer: insufficient memory");
	202	free(buf);
	203	return NULL;
	204	}
	205
d159ac37 AH	206	buf->name = bufname;
	207	buf->pid = pid;
	208
	209	/* connect to app */
f3f8cc91	210	result = ustcomm_connect_app(buf->pid, buf->conn);
d159ac37 AH	211	if(result) {
	212	WARN("unable to connect to process, it probably died before we were able to connect");
	213	return NULL;
	214	}
	215
	216	/* get pidunique */
	217	asprintf(&send_msg, "get_pidunique");
f3f8cc91	218	result = ustcomm_send_request(buf->conn, send_msg, &received_msg);
d159ac37 AH	219	free(send_msg);
	220	if(result == -1) {
	221	ERR("problem in ustcomm_send_request(get_pidunique)");
	222	return NULL;
	223	}
	224	if(result == 0) {
	225	goto error;
	226	}
	227
	228	result = sscanf(received_msg, "%lld", &buf->pidunique);
	229	if(result != 1) {
	230	ERR("unable to parse response to get_pidunique");
	231	return NULL;
	232	}
	233	free(received_msg);
	234	DBG("got pidunique %lld", buf->pidunique);
	235
	236	/* get shmid */
	237	asprintf(&send_msg, "get_shmid %s", buf->name);
f3f8cc91	238	result = ustcomm_send_request(buf->conn, send_msg, &received_msg);
d159ac37 AH	239	free(send_msg);
	240	if(result == -1) {
	241	ERR("problem in ustcomm_send_request(get_shmid)");
	242	return NULL;
	243	}
	244	if(result == 0) {
	245	goto error;
	246	}
	247
	248	result = sscanf(received_msg, "%d %d", &buf->shmid, &buf->bufstruct_shmid);
	249	if(result != 2) {
	250	ERR("unable to parse response to get_shmid (\"%s\")", received_msg);
	251	return NULL;
	252	}
	253	free(received_msg);
	254	DBG("got shmids %d %d", buf->shmid, buf->bufstruct_shmid);
	255
	256	/* get n_subbufs */
	257	asprintf(&send_msg, "get_n_subbufs %s", buf->name);
f3f8cc91	258	result = ustcomm_send_request(buf->conn, send_msg, &received_msg);
d159ac37 AH	259	free(send_msg);
	260	if(result == -1) {
	261	ERR("problem in ustcomm_send_request(g_n_subbufs)");
	262	return NULL;
	263	}
	264	if(result == 0) {
	265	goto error;
	266	}
	267
	268	result = sscanf(received_msg, "%d", &buf->n_subbufs);
	269	if(result != 1) {
	270	ERR("unable to parse response to get_n_subbufs");
	271	return NULL;
	272	}
	273	free(received_msg);
	274	DBG("got n_subbufs %d", buf->n_subbufs);
	275
	276	/* get subbuf size */
	277	asprintf(&send_msg, "get_subbuf_size %s", buf->name);
f3f8cc91	278	result = ustcomm_send_request(buf->conn, send_msg, &received_msg);
d159ac37 AH	279	free(send_msg);
	280	if(result == -1) {
	281	ERR("problem in ustcomm_send_request(get_subbuf_size)");
	282	return NULL;
	283	}
	284	if(result == 0) {
	285	goto error;
	286	}
	287
	288	result = sscanf(received_msg, "%d", &buf->subbuf_size);
	289	if(result != 1) {
	290	ERR("unable to parse response to get_subbuf_size");
	291	return NULL;
	292	}
	293	free(received_msg);
	294	DBG("got subbuf_size %d", buf->subbuf_size);
	295
	296	/* attach memory */
	297	buf->mem = shmat(buf->shmid, NULL, 0);
	298	if(buf->mem == (void *) 0) {
	299	PERROR("shmat");
	300	return NULL;
	301	}
	302	DBG("successfully attached buffer memory");
	303
	304	buf->bufstruct_mem = shmat(buf->bufstruct_shmid, NULL, 0);
	305	if(buf->bufstruct_mem == (void *) 0) {
	306	PERROR("shmat");
	307	return NULL;
	308	}
	309	DBG("successfully attached buffer bufstruct memory");
	310
	311	/* obtain info on the memory segment */
	312	result = shmctl(buf->shmid, IPC_STAT, &shmds);
	313	if(result == -1) {
	314	PERROR("shmctl");
	315	return NULL;
	316	}
	317	buf->memlen = shmds.shm_segsz;
	318
	319	if(instance->callbacks->on_open_buffer)
	320	instance->callbacks->on_open_buffer(instance->callbacks, buf);
	321
	322	pthread_mutex_lock(&instance->mutex);
	323	instance->active_buffers++;
	324	pthread_mutex_unlock(&instance->mutex);
	325
	326	return buf;
	327
	328	error:
	329	free(buf);
	330	return NULL;
	331	}
	332
	333	static void destroy_buffer(struct libustd_callbacks *callbacks,
	334	struct buffer_info *buf)
	335	{
	336	int result;
	337
f3f8cc91	338	result = ustcomm_close_app(buf->conn);
d159ac37 AH	339	if(result == -1) {
	340	WARN("problem calling ustcomm_close_app");
	341	}
	342
	343	result = shmdt(buf->mem);
	344	if(result == -1) {
	345	PERROR("shmdt");
	346	}
	347
	348	result = shmdt(buf->bufstruct_mem);
	349	if(result == -1) {
	350	PERROR("shmdt");
	351	}
	352
	353	if(callbacks->on_close_buffer)
	354	callbacks->on_close_buffer(callbacks, buf);
	355
f3f8cc91	356	free(buf->conn);
d159ac37 AH	357	free(buf);
	358	}
	359
	360	int consumer_loop(struct libustd_instance instance, struct buffer_info buf)
	361	{
	362	int result;
	363
	364	pthread_cleanup_push(decrement_active_buffers, instance);
	365
	366	for(;;) {
	367	/* get the subbuffer */
	368	result = get_subbuffer(buf);
	369	if(result == -1) {
	370	ERR("error getting subbuffer");
	371	continue;
	372	}
	373	else if(result == GET_SUBBUF_DONE) {
	374	/* this is done */
	375	break;
	376	}
	377	else if(result == GET_SUBBUF_DIED) {
	378	finish_consuming_dead_subbuffer(instance->callbacks, buf);
	379	break;
	380	}
	381
	382	if(instance->callbacks->on_read_subbuffer)
	383	instance->callbacks->on_read_subbuffer(instance->callbacks, buf);
	384
	385	/* put the subbuffer */
	386	result = put_subbuffer(buf);
	387	if(result == -1) {
	388	ERR("unknown error putting subbuffer (channel=%s)", buf->name);
	389	break;
	390	}
	391	else if(result == PUT_SUBBUF_PUSHED) {
	392	ERR("Buffer overflow (channel=%s), reader pushed. This channel will not be usable passed this point.", buf->name);
	393	break;
	394	}
	395	else if(result == PUT_SUBBUF_DIED) {
	396	DBG("application died while putting subbuffer");
	397	/* Skip the first subbuffer. We are not sure it is trustable
	398	* because the put_subbuffer() did not complete.
	399	*/
	400	if(instance->callbacks->on_put_error)
	401	instance->callbacks->on_put_error(instance->callbacks, buf);
	402
	403	finish_consuming_dead_subbuffer(instance->callbacks, buf);
	404	break;
	405	}
	406	else if(result == PUT_SUBBUF_DONE) {
	407	/* Done with this subbuffer */
	408	/* FIXME: add a case where this branch is used? Upon
	409	* normal trace termination, at put_subbuf time, a
	410	* special last-subbuffer code could be returned by
	411	* the listener.
	412	*/
	413	break;
	414	}
	415	else if(result == PUT_SUBBUF_OK) {
	416	}
	417	}
	418
	419	DBG("thread for buffer %s is stopping", buf->name);
	420
421	/* FIXME: destroy, unalloc... */
422
423	pthread_cleanup_pop(1);
424
425	return 0;
426	}
427
428	struct consumer_thread_args {
429	pid_t pid;
430	const char *bufname;
431	struct libustd_instance *instance;
432	};
433
434	void consumer_thread(void arg)
435	{
436	struct buffer_info *buf;
437	struct consumer_thread_args args = (struct consumer_thread_args ) arg;
438	int result;
439	sigset_t sigset;
440
441	DBG("GOT ARGS: pid %d bufname %s", args->pid, args->bufname);
442
443	if(args->instance->callbacks->on_new_thread)
444	args->instance->callbacks->on_new_thread(args->instance->callbacks);
445
446	/* Block signals that should be handled by the main thread. */
447	result = sigemptyset(&sigset);
448	if(result == -1) {
449	PERROR("sigemptyset");
450	goto end;
451	}
452	result = sigaddset(&sigset, SIGTERM);
453	if(result == -1) {
454	PERROR("sigaddset");
455	goto end;
456	}
457	result = sigaddset(&sigset, SIGINT);
458	if(result == -1) {
459	PERROR("sigaddset");
460	goto end;
461	}
462	result = sigprocmask(SIG_BLOCK, &sigset, NULL);
463	if(result == -1) {
464	PERROR("sigprocmask");
465	goto end;
466	}
467
468	buf = connect_buffer(args->instance, args->pid, args->bufname);
469	if(buf == NULL) {
470	ERR("failed to connect to buffer");
471	goto end;
472	}
473
474	consumer_loop(args->instance, buf);
475
476	destroy_buffer(args->instance->callbacks, buf);
477
478	end:
479
480	if(args->instance->callbacks->on_close_thread)
481	args->instance->callbacks->on_close_thread(args->instance->callbacks);
482
483	free((void *)args->bufname);
484	free(args);
485	return NULL;
486	}
487
488	int start_consuming_buffer(
489	struct libustd_instance instance, pid_t pid, const char bufname)
490	{
491	pthread_t thr;
492	struct consumer_thread_args *args;
493	int result;
494
495	DBG("beginning of start_consuming_buffer: args: pid %d bufname %s", pid, bufname);
496
497	args = (struct consumer_thread_args *) malloc(sizeof(struct consumer_thread_args));
498
499	args->pid = pid;
500	args->bufname = strdup(bufname);
501	args->instance = instance;
502	DBG("beginning2 of start_consuming_buffer: args: pid %d bufname %s", args->pid, args->bufname);
503
504	result = pthread_create(&thr, NULL, consumer_thread, args);
505	if(result == -1) {
506	ERR("pthread_create failed");
507	return -1;
508	}
509	result = pthread_detach(thr);
510	if(result == -1) {
511	ERR("pthread_detach failed");
512	return -1;
513	}
514	DBG("end of start_consuming_buffer: args: pid %d bufname %s", args->pid, args->bufname);
515
516	return 0;
517	}
518
519	int libustd_start_instance(struct libustd_instance *instance)
520	{
521	int result;
522	int timeout = -1;
523
524	if(!instance->is_init) {
525	ERR("libustd instance not initialized");
526	return 1;
527	}
528
529	/* app loop */
530	for(;;) {
531	char *recvbuf;
532
533	/* check for requests on our public socket */
f3f8cc91	534	result = ustcomm_ustd_recv_message(instance->comm, &recvbuf, NULL, timeout);
d159ac37 AH	535	if(result == -1 && errno == EINTR) {
	536	/* Caught signal */
	537	}
	538	else if(result == -1) {
	539	ERR("error in ustcomm_ustd_recv_message");
	540	goto loop_end;
	541	}
	542	else if(result > 0) {
	543	if(!strncmp(recvbuf, "collect", 7)) {
	544	pid_t pid;
	545	char *bufname;
	546	int result;
	547
	548	result = sscanf(recvbuf, "%*s %d %50as", &pid, &bufname);
	549	if(result != 2) {
	550	ERR("parsing error: %s", recvbuf);
	551	goto free_bufname;
	552	}
	553
	554	result = start_consuming_buffer(instance, pid, bufname);
	555	if(result < 0) {
	556	ERR("error in add_buffer");
	557	goto free_bufname;
	558	}
	559
	560	free_bufname:
	561	free(bufname);
	562	}
	563	else if(!strncmp(recvbuf, "exit", 4)) {
	564	/* Only there to force poll to return */
	565	}
	566	else {
	567	WARN("unknown command: %s", recvbuf);
	568	}
	569
	570	free(recvbuf);
	571	}
	572
	573	loop_end:
	574
	575	if(instance->quit_program) {
	576	pthread_mutex_lock(&instance->mutex);
	577	if(instance->active_buffers == 0) {
	578	pthread_mutex_unlock(&instance->mutex);
	579	break;
	580	}
	581	pthread_mutex_unlock(&instance->mutex);
	582	timeout = 100;
	583	}
	584	}
	585
	586	if(instance->callbacks->on_trace_end)
	587	instance->callbacks->on_trace_end(instance);
	588
	589	libustd_delete_instance(instance);
	590
	591	return 0;
	592	}
	593
	594	void libustd_delete_instance(struct libustd_instance *instance)
	595	{
	596	if(instance->is_init)
f3f8cc91	597	ustcomm_fini_ustd(instance->comm);
d159ac37 AH	598
	599	pthread_mutex_destroy(&instance->mutex);
	600	free(instance->sock_path);
f3f8cc91	601	free(instance->comm);
d159ac37 AH	602	free(instance);
	603	}
	604
	605	int libustd_stop_instance(struct libustd_instance *instance, int send_msg)
	606	{
	607	int result;
	608	int fd;
	609	int bytes = 0;
	610
	611	char msg[] = "exit";
	612
	613	instance->quit_program = 1;
	614
	615	if(!send_msg)
	616	return 0;
	617
	618	/* Send a message through the socket to force poll to return */
	619
	620	struct sockaddr_un addr;
	621
	622	result = fd = socket(PF_UNIX, SOCK_STREAM, 0);
	623	if(result == -1) {
	624	PERROR("socket");
	625	return 1;
	626	}
	627
	628	addr.sun_family = AF_UNIX;
	629
	630	strncpy(addr.sun_path, instance->sock_path, UNIX_PATH_MAX);
	631	addr.sun_path[UNIX_PATH_MAX-1] = '\0';
	632
	633	result = connect(fd, (struct sockaddr *)&addr, sizeof(addr));
	634	if(result == -1) {
	635	PERROR("connect");
	636	}
	637
	638	while(bytes != sizeof(msg))
	639	bytes += send(fd, msg, sizeof(msg), 0);
	640
	641	close(fd);
	642
	643	return 0;
	644	}
	645
	646	struct libustd_instance *libustd_new_instance(
	647	struct libustd_callbacks callbacks, char sock_path)
	648	{
	649	struct libustd_instance *instance =
	650	malloc(sizeof(struct libustd_instance));
	651	if(!instance)
	652	return NULL;
	653
f3f8cc91 AH	654	instance->comm = malloc(sizeof(struct ustcomm_ustd));
	655	if(!instance->comm) {
	656	free(instance);
	657	return NULL;
	658	}
	659
d159ac37 AH	660	instance->callbacks = callbacks;
	661	instance->quit_program = 0;
	662	instance->is_init = 0;
	663	instance->active_buffers = 0;
	664	pthread_mutex_init(&instance->mutex, NULL);
	665
	666	if(sock_path)
	667	instance->sock_path = strdup(sock_path);
	668	else
	669	instance->sock_path = NULL;
	670
	671	return instance;
	672	}
	673
	674	int libustd_init_instance(struct libustd_instance *instance)
	675	{
	676	int result;
f3f8cc91	677	result = ustcomm_init_ustd(instance->comm, instance->sock_path);
d159ac37 AH	678	if(result == -1) {
	679	ERR("failed to initialize socket");
	680	return 1;
	681	}
	682	instance->is_init = 1;
	683	return 0;
	684	}
	685