[ust.git] / ustd / ustd.c

/* Copyright (C) 2009  Pierre-Marc Fournier
 *
 * 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/types.h>
#include <sys/stat.h>
#include <sys/shm.h>
#include <fcntl.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 <getopt.h>

#include "ustd.h"
#include "localerr.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

char *sock_path=NULL;
char *trace_path=NULL;

int test_sigpipe(void)
{
	sigset_t sigset;
	int result;

	result = sigemptyset(&sigset);
	if(result == -1) {
		perror("sigemptyset");
		return -1;
	}
	result = sigaddset(&sigset, SIGPIPE);
	if(result == -1) {
		perror("sigaddset");
		return -1;
	}

	result = sigtimedwait(&sigset, NULL, &(struct timespec){0,0});
	if(result == -1 && errno == EAGAIN) {
		/* no signal received */
		return 0;
	}
	else if(result == -1) {
		perror("sigtimedwait");
		return -1;
	}
	else if(result == SIGPIPE) {
		/* received sigpipe */
		return 1;
	}
	else {
		assert(0);
	}
}

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

	asprintf(&send_msg, "get_subbuffer %s", buf->name);
	result = ustcomm_send_request(&buf->conn, send_msg, &received_msg);
	free(send_msg);
	if(test_sigpipe()) {
		WARN("process %d destroyed before we could connect to it", buf->pid);
		return GET_SUBBUF_DONE;
	}
	else if(result < 0) {
		ERR("get_subbuffer: ustcomm_send_request failed");
		return -1;
	}
	else if(result == 0) {
		DBG("app died while being traced");
		return GET_SUBBUF_DIED;
	}

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

	DBG("received msg is %s", received_msg);

	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) {
		return GET_SUBBUF_DONE;
	}
	else {
		DBG("error getting subbuffer %s", buf->name);
		retval = -1;
	}

	/* FIMXE: free correctly the stuff */
	free(received_msg);
	free(rep_code);
	return retval;
}

int put_subbuffer(struct buffer_info *buf)
{
	char *send_msg;
	char *received_msg;
	char *rep_code;
	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) {
		ERR("put_subbuffer: send_message failed");
		return -1;
	}
	free(send_msg);

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

	if(!strcmp(rep_code, "OK")) {
		DBG("subbuffer put %s", buf->name);
		retval = PUT_SUBBUF_OK;
	}
	else {
		DBG("put_subbuffer: received error, we were pushed");
		return PUT_SUBBUF_PUSHED;
	}

	free(rep_code);
	return retval;
}

/* This write is patient because it restarts if it was incomplete.
 */

ssize_t patient_write(int fd, const void *buf, size_t count)
{
	const char *bufc = (const char *) buf;
	int result;

	for(;;) {
		result = write(fd, bufc, count);
		if(result <= 0) {
			return result;
		}
		count -= result;
		bufc += result;

		if(count == 0) {
			break;
		}
	}

	return bufc-(const char *)buf;
}

void *consumer_thread(void *arg)
{
	struct buffer_info *buf = (struct buffer_info *) arg;
	int result;

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

		/* write data to file */
		result = patient_write(buf->file_fd, buf->mem + (buf->consumed_old & (buf->n_subbufs * buf->subbuf_size-1)), buf->subbuf_size);
		if(result == -1) {
			PERROR("write");
			/* FIXME: maybe drop this trace */
		}

		/* 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) {
			WARN("application died while putting subbuffer");
			/* FIXME: probably need to skip the first subbuffer in finish_consuming_dead_subbuffer */
			finish_consuming_dead_subbuffer(buf);
		}
		else if(result == PUT_SUBBUF_OK) {
		}
	}

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

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

	return NULL;
}

int create_dir_if_needed(char *dir)
{
	int result;
	result = mkdir(dir, 0777);
	if(result == -1) {
		if(errno != EEXIST) {
			perror("mkdir");
			return -1;
		}
	}

	return 0;
}

int is_directory(const char *dir)
{
	int result;
	struct stat st;

	result = stat(dir, &st);
	if(result == -1) {
		PERROR("stat");
		return 0;
	}

	if(!S_ISDIR(st.st_mode)) {
		return 0;
	}

	return 1;
}

int add_buffer(pid_t pid, char *bufname)
{
	struct buffer_info *buf;
	char *send_msg;
	char *received_msg;
	int result;
	char *tmp;
	int fd;
	pthread_t thr;
	struct shmid_ds shmds;

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

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

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

	result = sscanf(received_msg, "%d %d", &buf->shmid, &buf->bufstruct_shmid);
	if(result != 2) {
		ERR("unable to parse response to get_shmid");
		return -1;
	}
	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 -1;
	}

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

	/* get subbuf size */
	asprintf(&send_msg, "get_subbuf_size %s", buf->name);
	ustcomm_send_request(&buf->conn, send_msg, &received_msg);
	free(send_msg);

	result = sscanf(received_msg, "%d", &buf->subbuf_size);
	if(result != 1) {
		ERR("unable to parse response to get_subbuf_size");
		return -1;
	}
	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 -1;
	}
	DBG("successfully attached buffer memory");

	buf->bufstruct_mem = shmat(buf->bufstruct_shmid, NULL, 0);
	if(buf->bufstruct_mem == (void *) 0) {
		perror("shmat");
		return -1;
	}
	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 -1;
	}
	buf->memlen = shmds.shm_segsz;

	/* open file for output */
	if(!trace_path) {
		/* Only create the directory if using the default path, because
		 * of the risk of typo when using trace path override. We don't
		 * want to risk creating plenty of useless directories in that case.
		 */
		result = create_dir_if_needed(USTD_DEFAULT_TRACE_PATH);
		if(result == -1) {
			ERR("could not create directory %s", USTD_DEFAULT_TRACE_PATH);
			return -1;
		}

		trace_path = USTD_DEFAULT_TRACE_PATH;
	}

	asprintf(&tmp, "%s/%u", trace_path, buf->pid);
	result = create_dir_if_needed(tmp);
	if(result == -1) {
		ERR("could not create directory %s", tmp);
		free(tmp);
		return -1;
	}
	free(tmp);

	asprintf(&tmp, "%s/%u/%s_0", trace_path, buf->pid, buf->name);
	result = fd = open(tmp, O_WRONLY | O_CREAT | O_TRUNC, 00600);
	if(result == -1) {
		PERROR("open");
		ERR("failed opening trace file %s", tmp);
		return -1;
	}
	buf->file_fd = fd;
	free(tmp);

	pthread_create(&thr, NULL, consumer_thread, buf);

	return 0;
}

void usage(void)
{
	fprintf(stderr, "Usage:\nustd OPTIONS\n\nOptions:\n"
			"\t-h\t\tDisplay this usage.\n"
			"\t-o DIR\t\tSpecify the directory where to output the traces.\n"
			"\t-s PATH\t\tSpecify the path to use for the daemon socket.\n");
}

int parse_args(int argc, char **argv)
{
	int c;

	while (1) {
		int option_index = 0;
		static struct option long_options[] = {
			{"help", 0, 0, 'h'},
			{"version", 0, 0, 'V'},
			{0, 0, 0, 0}
		};

		c = getopt_long(argc, argv, "hs:o:", long_options, &option_index);
		if (c == -1)
			break;

		switch (c) {
		case 0:
			printf("option %s", long_options[option_index].name);
			if (optarg)
				printf(" with arg %s", optarg);
			printf("\n");
			break;
		case 's':
			sock_path = optarg;
			break;
		case 'o':
			trace_path = optarg;
			if(!is_directory(trace_path)) {
				ERR("Not a valid directory. (%s)", trace_path);
				return -1;
			}
			break;
		case 'h':
			usage();
			exit(0);
		case 'V':
			printf("Version 0.0\n");
			break;

		default:
			/* unknown option or other error; error is
			printed by getopt, just return */
			return -1;
		}
	}

	return 0;
}

int main(int argc, char **argv)
{
	struct ustcomm_ustd ustd;
	int result;
	sigset_t sigset;

	result = parse_args(argc, argv);
	if(result == -1) {
		exit(1);
	}

	result = ustcomm_init_ustd(&ustd, sock_path);
	if(result == -1) {
		ERR("failed to initialize socket");
		return 1;
	}

	result = sigemptyset(&sigset);
	if(result == -1) {
		perror("sigemptyset");
		return 1;
	}
	result = sigaddset(&sigset, SIGPIPE);
	if(result == -1) {
		perror("sigaddset");
		return 1;
	}
	result = sigprocmask(SIG_BLOCK, &sigset, NULL);
	if(result == -1) {
		perror("sigprocmask");
		return 1;
	}

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

		/* check for requests on our public socket */
		result = ustcomm_ustd_recv_message(&ustd, &recvbuf, NULL, 100);
		if(result == -1) {
			ERR("error in ustcomm_ustd_recv_message");
			continue;
		}
		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) {
					fprintf(stderr, "parsing error: %s\n", recvbuf);
				}

				result = add_buffer(pid, bufname);
				if(result < 0) {
					ERR("error in add_buffer");
					continue;
				}
			}

			free(recvbuf);
		}
	}

	return 0;
}
Commit	Line	Data
c39c72ee	1	/* Copyright (C) 2009 Pierre-Marc Fournier
1f8b0dff	2	*
c39c72ee PMF	3	* This library is free software; you can redistribute it and/or
	4	* modify it under the terms of the GNU Lesser General Public
	5	* License as published by the Free Software Foundation; either
	6	* version 2.1 of the License, or (at your option) any later version.
1f8b0dff	7	*
c39c72ee	8	* This library is distributed in the hope that it will be useful,
1f8b0dff	9	* but WITHOUT ANY WARRANTY; without even the implied warranty of
c39c72ee PMF	10	* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
c39c72ee PMF	11	* Lesser General Public License for more details.
1f8b0dff	12	*
c39c72ee PMF	13	* You should have received a copy of the GNU Lesser General Public
	14	* License along with this library; if not, write to the Free Software
	15	* Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
1f8b0dff PMF	16	*/
1f8b0dff PMF	17
3796af9b PMF	18	#define _GNU_SOURCE
	19
	20	#include <sys/types.h>
cd226f25	21	#include <sys/stat.h>
3796af9b	22	#include <sys/shm.h>
688760ef PMF	23	#include <fcntl.h>
688760ef PMF	24	#include <unistd.h>
3a7b90de	25	#include <pthread.h>
a3cdd4a7	26	#include <signal.h>
3796af9b PMF	27
	28	#include <stdlib.h>
	29	#include <stdio.h>
	30	#include <string.h>
a3cdd4a7 PMF	31	#include <errno.h>
a3cdd4a7 PMF	32	#include <assert.h>
cd226f25	33	#include <getopt.h>
3796af9b	34
0b0cd937	35	#include "ustd.h"
3796af9b PMF	36	#include "localerr.h"
	37	#include "ustcomm.h"
	38
3a7b90de PMF	39	/* return value: 0 = subbuffer is finished, it won't produce data anymore
	40	* 1 = got subbuffer successfully
	41	* <0 = error
	42	*/
3796af9b	43
8cefc145 PMF	44	#define GET_SUBBUF_OK 1
	45	#define GET_SUBBUF_DONE 0
	46	#define GET_SUBBUF_DIED 2
	47
a3cdd4a7 PMF	48	#define PUT_SUBBUF_OK 1
	49	#define PUT_SUBBUF_DIED 0
	50	#define PUT_SUBBUF_PUSHED 2
	51
c97d4437	52	char *sock_path=NULL;
cd226f25 PMF	53	char *trace_path=NULL;
cd226f25 PMF	54
a3cdd4a7 PMF	55	int test_sigpipe(void)
	56	{
	57	sigset_t sigset;
	58	int result;
	59
	60	result = sigemptyset(&sigset);
	61	if(result == -1) {
	62	perror("sigemptyset");
	63	return -1;
	64	}
	65	result = sigaddset(&sigset, SIGPIPE);
	66	if(result == -1) {
	67	perror("sigaddset");
	68	return -1;
	69	}
	70
	71	result = sigtimedwait(&sigset, NULL, &(struct timespec){0,0});
	72	if(result == -1 && errno == EAGAIN) {
	73	/* no signal received */
	74	return 0;
	75	}
	76	else if(result == -1) {
	77	perror("sigtimedwait");
	78	return -1;
	79	}
	80	else if(result == SIGPIPE) {
	81	/* received sigpipe */
	82	return 1;
	83	}
	84	else {
	85	assert(0);
	86	}
	87	}
	88
688760ef PMF	89	int get_subbuffer(struct buffer_info *buf)
	90	{
	91	char *send_msg;
	92	char *received_msg;
	93	char *rep_code;
	94	int retval;
	95	int result;
	96
	97	asprintf(&send_msg, "get_subbuffer %s", buf->name);
3bb56863	98	result = ustcomm_send_request(&buf->conn, send_msg, &received_msg);
8cefc145	99	free(send_msg);
a3cdd4a7 PMF	100	if(test_sigpipe()) {
	101	WARN("process %d destroyed before we could connect to it", buf->pid);
	102	return GET_SUBBUF_DONE;
	103	}
	104	else if(result < 0) {
3bb56863	105	ERR("get_subbuffer: ustcomm_send_request failed");
688760ef PMF	106	return -1;
688760ef PMF	107	}
8cefc145 PMF	108	else if(result == 0) {
	109	DBG("app died while being traced");
	110	return GET_SUBBUF_DIED;
	111	}
688760ef PMF	112
688760ef PMF	113	result = sscanf(received_msg, "%as %ld", &rep_code, &buf->consumed_old);
3a7b90de	114	if(result != 2 && result != 1) {
688760ef PMF	115	ERR("unable to parse response to get_subbuffer");
	116	return -1;
	117	}
3a7b90de PMF	118
3a7b90de PMF	119	DBG("received msg is %s", received_msg);
688760ef PMF	120
	121	if(!strcmp(rep_code, "OK")) {
	122	DBG("got subbuffer %s", buf->name);
8cefc145	123	retval = GET_SUBBUF_OK;
688760ef	124	}
3a7b90de	125	else if(nth_token_is(received_msg, "END", 0) == 1) {
8cefc145	126	return GET_SUBBUF_DONE;
3a7b90de	127	}
688760ef	128	else {
3a7b90de PMF	129	DBG("error getting subbuffer %s", buf->name);
3a7b90de PMF	130	retval = -1;
688760ef PMF	131	}
688760ef PMF	132
3a7b90de PMF	133	/* FIMXE: free correctly the stuff */
3a7b90de PMF	134	free(received_msg);
688760ef PMF	135	free(rep_code);
	136	return retval;
	137	}
	138
	139	int put_subbuffer(struct buffer_info *buf)
	140	{
	141	char *send_msg;
	142	char *received_msg;
	143	char *rep_code;
	144	int retval;
	145	int result;
	146
	147	asprintf(&send_msg, "put_subbuffer %s %ld", buf->name, buf->consumed_old);
3bb56863	148	result = ustcomm_send_request(&buf->conn, send_msg, &received_msg);
688760ef PMF	149	if(result < 0) {
	150	ERR("put_subbuffer: send_message failed");
	151	return -1;
	152	}
	153	free(send_msg);
	154
	155	result = sscanf(received_msg, "%as", &rep_code);
	156	if(result != 1) {
	157	ERR("unable to parse response to put_subbuffer");
	158	return -1;
	159	}
	160	free(received_msg);
	161
	162	if(!strcmp(rep_code, "OK")) {
	163	DBG("subbuffer put %s", buf->name);
a3cdd4a7	164	retval = PUT_SUBBUF_OK;
688760ef PMF	165	}
688760ef PMF	166	else {
a3cdd4a7 PMF	167	DBG("put_subbuffer: received error, we were pushed");
a3cdd4a7 PMF	168	return PUT_SUBBUF_PUSHED;
688760ef PMF	169	}
	170
	171	free(rep_code);
	172	return retval;
	173	}
	174
a3cdd4a7 PMF	175	/* This write is patient because it restarts if it was incomplete.
	176	*/
	177
688760ef PMF	178	ssize_t patient_write(int fd, const void *buf, size_t count)
	179	{
	180	const char bufc = (const char ) buf;
	181	int result;
	182
	183	for(;;) {
	184	result = write(fd, bufc, count);
	185	if(result <= 0) {
	186	return result;
	187	}
	188	count -= result;
	189	bufc += result;
	190
	191	if(count == 0) {
	192	break;
	193	}
	194	}
	195
	196	return bufc-(const char *)buf;
	197	}
	198
3a7b90de PMF	199	void consumer_thread(void arg)
	200	{
	201	struct buffer_info buf = (struct buffer_info ) arg;
	202	int result;
	203
	204	for(;;) {
8cefc145	205	/* get the subbuffer */
0b0cd937 PMF	206	result = get_subbuffer(buf);
	207	if(result == -1) {
	208	ERR("error getting subbuffer");
	209	continue;
3a7b90de	210	}
0b0cd937 PMF	211	else if(result == GET_SUBBUF_DONE) {
	212	/* this is done */
	213	break;
	214	}
	215	else if(result == GET_SUBBUF_DIED) {
	216	finish_consuming_dead_subbuffer(buf);
	217	break;
3a7b90de PMF	218	}
	219
	220	/* write data to file */
	221	result = patient_write(buf->file_fd, buf->mem + (buf->consumed_old & (buf->n_subbufs * buf->subbuf_size-1)), buf->subbuf_size);
	222	if(result == -1) {
	223	PERROR("write");
	224	/* FIXME: maybe drop this trace */
	225	}
	226
8cefc145	227	/* put the subbuffer */
0b0cd937 PMF	228	result = put_subbuffer(buf);
0b0cd937 PMF	229	if(result == -1) {
a3cdd4a7 PMF	230	ERR("unknown error putting subbuffer (channel=%s)", buf->name);
	231	break;
	232	}
	233	else if(result == PUT_SUBBUF_PUSHED) {
	234	ERR("Buffer overflow (channel=%s), reader pushed. This channel will not be usable passed this point.", buf->name);
0b0cd937	235	break;
3a7b90de	236	}
a3cdd4a7 PMF	237	else if(result == PUT_SUBBUF_DIED) {
	238	WARN("application died while putting subbuffer");
	239	/* FIXME: probably need to skip the first subbuffer in finish_consuming_dead_subbuffer */
	240	finish_consuming_dead_subbuffer(buf);
	241	}
	242	else if(result == PUT_SUBBUF_OK) {
	243	}
3a7b90de PMF	244	}
	245
	246	DBG("thread for buffer %s is stopping", buf->name);
	247
8cefc145 PMF	248	/* FIXME: destroy, unalloc... */
8cefc145 PMF	249
3a7b90de PMF	250	return NULL;
	251	}
	252
72ebd39a PMF	253	int create_dir_if_needed(char *dir)
	254	{
	255	int result;
	256	result = mkdir(dir, 0777);
	257	if(result == -1) {
	258	if(errno != EEXIST) {
	259	perror("mkdir");
	260	return -1;
	261	}
	262	}
	263
	264	return 0;
	265	}
	266
cd226f25 PMF	267	int is_directory(const char *dir)
	268	{
	269	int result;
	270	struct stat st;
	271
	272	result = stat(dir, &st);
	273	if(result == -1) {
	274	PERROR("stat");
	275	return 0;
	276	}
	277
	278	if(!S_ISDIR(st.st_mode)) {
	279	return 0;
	280	}
	281
	282	return 1;
	283	}
	284
3a7b90de PMF	285	int add_buffer(pid_t pid, char *bufname)
	286	{
	287	struct buffer_info *buf;
	288	char *send_msg;
	289	char *received_msg;
	290	int result;
	291	char *tmp;
	292	int fd;
	293	pthread_t thr;
a3cdd4a7	294	struct shmid_ds shmds;
3a7b90de PMF	295
	296	buf = (struct buffer_info *) malloc(sizeof(struct buffer_info));
	297	if(buf == NULL) {
	298	ERR("add_buffer: insufficient memory");
	299	return -1;
	300	}
	301
	302	buf->name = bufname;
	303	buf->pid = pid;
	304
4e2a8808 PMF	305	/* connect to app */
	306	result = ustcomm_connect_app(buf->pid, &buf->conn);
	307	if(result) {
a3cdd4a7	308	WARN("unable to connect to process, it probably died before we were able to connect");
4e2a8808 PMF	309	return -1;
	310	}
	311
3a7b90de PMF	312	/* get shmid */
3a7b90de PMF	313	asprintf(&send_msg, "get_shmid %s", buf->name);
a3cdd4a7	314	result = ustcomm_send_request(&buf->conn, send_msg, &received_msg);
3a7b90de	315	free(send_msg);
a3cdd4a7 PMF	316	if(result == -1) {
	317	ERR("problem in ustcomm_send_request(get_shmid)");
	318	return -1;
	319	}
3a7b90de	320
8cefc145 PMF	321	result = sscanf(received_msg, "%d %d", &buf->shmid, &buf->bufstruct_shmid);
8cefc145 PMF	322	if(result != 2) {
3a7b90de PMF	323	ERR("unable to parse response to get_shmid");
	324	return -1;
	325	}
	326	free(received_msg);
8cefc145	327	DBG("got shmids %d %d", buf->shmid, buf->bufstruct_shmid);
3a7b90de PMF	328
	329	/* get n_subbufs */
	330	asprintf(&send_msg, "get_n_subbufs %s", buf->name);
a3cdd4a7	331	result = ustcomm_send_request(&buf->conn, send_msg, &received_msg);
3a7b90de	332	free(send_msg);
a3cdd4a7 PMF	333	if(result == -1) {
	334	ERR("problem in ustcomm_send_request(g_n_subbufs)");
	335	return -1;
	336	}
3a7b90de PMF	337
	338	result = sscanf(received_msg, "%d", &buf->n_subbufs);
	339	if(result != 1) {
	340	ERR("unable to parse response to get_n_subbufs");
	341	return -1;
	342	}
	343	free(received_msg);
	344	DBG("got n_subbufs %d", buf->n_subbufs);
	345
	346	/* get subbuf size */
	347	asprintf(&send_msg, "get_subbuf_size %s", buf->name);
4e2a8808	348	ustcomm_send_request(&buf->conn, send_msg, &received_msg);
3a7b90de PMF	349	free(send_msg);
	350
	351	result = sscanf(received_msg, "%d", &buf->subbuf_size);
	352	if(result != 1) {
	353	ERR("unable to parse response to get_subbuf_size");
	354	return -1;
	355	}
	356	free(received_msg);
	357	DBG("got subbuf_size %d", buf->subbuf_size);
	358
	359	/* attach memory */
	360	buf->mem = shmat(buf->shmid, NULL, 0);
	361	if(buf->mem == (void *) 0) {
	362	perror("shmat");
	363	return -1;
	364	}
8cefc145 PMF	365	DBG("successfully attached buffer memory");
	366
	367	buf->bufstruct_mem = shmat(buf->bufstruct_shmid, NULL, 0);
	368	if(buf->bufstruct_mem == (void *) 0) {
	369	perror("shmat");
	370	return -1;
	371	}
	372	DBG("successfully attached buffer bufstruct memory");
3a7b90de	373
a3cdd4a7 PMF	374	/* obtain info on the memory segment */
	375	result = shmctl(buf->shmid, IPC_STAT, &shmds);
	376	if(result == -1) {
	377	perror("shmctl");
	378	return -1;
	379	}
	380	buf->memlen = shmds.shm_segsz;
	381
3a7b90de	382	/* open file for output */
cd226f25 PMF	383	if(!trace_path) {
	384	/* Only create the directory if using the default path, because
	385	* of the risk of typo when using trace path override. We don't
	386	* want to risk creating plenty of useless directories in that case.
	387	*/
	388	result = create_dir_if_needed(USTD_DEFAULT_TRACE_PATH);
	389	if(result == -1) {
	390	ERR("could not create directory %s", USTD_DEFAULT_TRACE_PATH);
	391	return -1;
	392	}
	393
	394	trace_path = USTD_DEFAULT_TRACE_PATH;
72ebd39a PMF	395	}
72ebd39a PMF	396
cd226f25	397	asprintf(&tmp, "%s/%u", trace_path, buf->pid);
72ebd39a PMF	398	result = create_dir_if_needed(tmp);
	399	if(result == -1) {
	400	ERR("could not create directory %s", tmp);
	401	free(tmp);
	402	return -1;
	403	}
	404	free(tmp);
	405
cd226f25	406	asprintf(&tmp, "%s/%u/%s_0", trace_path, buf->pid, buf->name);
3a7b90de PMF	407	result = fd = open(tmp, O_WRONLY \| O_CREAT \| O_TRUNC, 00600);
	408	if(result == -1) {
	409	PERROR("open");
6cb88bc0	410	ERR("failed opening trace file %s", tmp);
3a7b90de PMF	411	return -1;
	412	}
	413	buf->file_fd = fd;
	414	free(tmp);
	415
3a7b90de PMF	416	pthread_create(&thr, NULL, consumer_thread, buf);
	417
	418	return 0;
	419	}
	420
cd226f25 PMF	421	void usage(void)
	422	{
	423	fprintf(stderr, "Usage:\nustd OPTIONS\n\nOptions:\n"
	424	"\t-h\t\tDisplay this usage.\n"
	425	"\t-o DIR\t\tSpecify the directory where to output the traces.\n"
	426	"\t-s PATH\t\tSpecify the path to use for the daemon socket.\n");
	427	}
	428
	429	int parse_args(int argc, char **argv)
	430	{
	431	int c;
	432
	433	while (1) {
	434	int option_index = 0;
	435	static struct option long_options[] = {
	436	{"help", 0, 0, 'h'},
	437	{"version", 0, 0, 'V'},
	438	{0, 0, 0, 0}
	439	};
	440
	441	c = getopt_long(argc, argv, "hs:o:", long_options, &option_index);
	442	if (c == -1)
	443	break;
	444
	445	switch (c) {
	446	case 0:
	447	printf("option %s", long_options[option_index].name);
	448	if (optarg)
	449	printf(" with arg %s", optarg);
	450	printf("\n");
	451	break;
	452	case 's':
	453	sock_path = optarg;
	454	break;
	455	case 'o':
	456	trace_path = optarg;
	457	if(!is_directory(trace_path)) {
	458	ERR("Not a valid directory. (%s)", trace_path);
	459	return -1;
	460	}
	461	break;
	462	case 'h':
	463	usage();
	464	exit(0);
	465	case 'V':
	466	printf("Version 0.0\n");
	467	break;
	468
	469	default:
	470	/* unknown option or other error; error is
	471	printed by getopt, just return */
	472	return -1;
	473	}
	474	}
	475
	476	return 0;
	477	}
	478
3796af9b PMF	479	int main(int argc, char **argv)
	480	{
	481	struct ustcomm_ustd ustd;
	482	int result;
a3cdd4a7	483	sigset_t sigset;
3796af9b	484
cd226f25 PMF	485	result = parse_args(argc, argv);
	486	if(result == -1) {
	487	exit(1);
	488	}
	489
	490	result = ustcomm_init_ustd(&ustd, sock_path);
3796af9b PMF	491	if(result == -1) {
	492	ERR("failed to initialize socket");
	493	return 1;
	494	}
	495
a3cdd4a7 PMF	496	result = sigemptyset(&sigset);
	497	if(result == -1) {
	498	perror("sigemptyset");
	499	return 1;
	500	}
	501	result = sigaddset(&sigset, SIGPIPE);
	502	if(result == -1) {
	503	perror("sigaddset");
	504	return 1;
	505	}
	506	result = sigprocmask(SIG_BLOCK, &sigset, NULL);
	507	if(result == -1) {
	508	perror("sigprocmask");
	509	return 1;
	510	}
	511
688760ef	512	/* app loop */
3796af9b PMF	513	for(;;) {
	514	char *recvbuf;
	515
3a7b90de	516	/* check for requests on our public socket */
688760ef PMF	517	result = ustcomm_ustd_recv_message(&ustd, &recvbuf, NULL, 100);
	518	if(result == -1) {
	519	ERR("error in ustcomm_ustd_recv_message");
	520	continue;
	521	}
	522	if(result > 0) {
	523	if(!strncmp(recvbuf, "collect", 7)) {
	524	pid_t pid;
	525	char *bufname;
	526	int result;
3796af9b	527
688760ef PMF	528	result = sscanf(recvbuf, "%*s %d %50as", &pid, &bufname);
	529	if(result != 2) {
	530	fprintf(stderr, "parsing error: %s\n", recvbuf);
	531	}
3796af9b	532
688760ef PMF	533	result = add_buffer(pid, bufname);
	534	if(result < 0) {
	535	ERR("error in add_buffer");
	536	continue;
	537	}
3796af9b PMF	538	}
3796af9b PMF	539
688760ef	540	free(recvbuf);
3796af9b	541	}
3796af9b PMF	542	}
	543
	544	return 0;
	545	}