[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 <ust/ustd.h>
#include "lowlevel.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;

	if (asprintf(&send_msg, "get_subbuffer %s", buf->name) < 0) {
		ERR("get_subbuffer : asprintf failed (%s)",
		    buf->name);
		retval = -1;
		goto end;
	}
	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;

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