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

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