[ust.git] / ustd / ustd.c

/*
 * Copyright (C) 2009  Pierre-Marc Fournier
 *
 * This program is free software: you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * (at your option) any later version.
 *
 * This program 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 General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program.  If not, see <http://www.gnu.org/licenses/>.
 */

#define _GNU_SOURCE

#include <sys/types.h>
#include <sys/shm.h>
#include <fcntl.h>
#include <unistd.h>
#include <pthread.h>

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

#include "ustd.h"
#include "localerr.h"
#include "ustcomm.h"

struct list_head buffers = LIST_HEAD_INIT(buffers);

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

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(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 = 1;
	}
	else {
		ERR("invalid response to put_subbuffer");
	}

	free(rep_code);
	return retval;
}

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("error putting subbuffer");
			break;
		}
	}

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

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

	return NULL;
}

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;

	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) {
		ERR("unable to connect to process");
		return -1;
	}

	/* get shmid */
	asprintf(&send_msg, "get_shmid %s", buf->name);
	ustcomm_send_request(&buf->conn, send_msg, &received_msg);
	free(send_msg);
	DBG("got buffer name %s", buf->name);

	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);
	ustcomm_send_request(&buf->conn, send_msg, &received_msg);
	free(send_msg);

	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");

	/* open file for output */
	asprintf(&tmp, "/tmp/trace/%s_0", buf->name);
	result = fd = open(tmp, O_WRONLY | O_CREAT | O_TRUNC, 00600);
	if(result == -1) {
		PERROR("open");
		return -1;
	}
	buf->file_fd = fd;
	free(tmp);

	//list_add(&buf->list, &buffers);

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

	return 0;
}

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

	result = ustcomm_init_ustd(&ustd);
	if(result == -1) {
		ERR("failed to initialize socket");
		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
1f8b0dff PMF	1	/*
	2	* Copyright (C) 2009 Pierre-Marc Fournier
	3	*
	4	* This program is free software: you can redistribute it and/or modify
	5	* it under the terms of the GNU General Public License as published by
	6	* the Free Software Foundation, either version 3 of the License, or
	7	* (at your option) any later version.
	8	*
	9	* This program 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
	12	* GNU General Public License for more details.
	13	*
	14	* You should have received a copy of the GNU General Public License
	15	* along with this program. If not, see <http://www.gnu.org/licenses/>.
	16	*/
	17
3796af9b PMF	18	#define _GNU_SOURCE
	19
	20	#include <sys/types.h>
	21	#include <sys/shm.h>
688760ef PMF	22	#include <fcntl.h>
688760ef PMF	23	#include <unistd.h>
3a7b90de	24	#include <pthread.h>
3796af9b PMF	25
	26	#include <stdlib.h>
	27	#include <stdio.h>
	28	#include <string.h>
	29
0b0cd937	30	#include "ustd.h"
3796af9b PMF	31	#include "localerr.h"
	32	#include "ustcomm.h"
	33
811e4b93 PMF	34	struct list_head buffers = LIST_HEAD_INIT(buffers);
811e4b93 PMF	35
3a7b90de PMF	36	/* return value: 0 = subbuffer is finished, it won't produce data anymore
	37	* 1 = got subbuffer successfully
	38	* <0 = error
	39	*/
3796af9b	40
8cefc145 PMF	41	#define GET_SUBBUF_OK 1
	42	#define GET_SUBBUF_DONE 0
	43	#define GET_SUBBUF_DIED 2
	44
688760ef PMF	45	int get_subbuffer(struct buffer_info *buf)
	46	{
	47	char *send_msg;
	48	char *received_msg;
	49	char *rep_code;
	50	int retval;
	51	int result;
	52
	53	asprintf(&send_msg, "get_subbuffer %s", buf->name);
3bb56863	54	result = ustcomm_send_request(&buf->conn, send_msg, &received_msg);
8cefc145	55	free(send_msg);
688760ef	56	if(result < 0) {
3bb56863	57	ERR("get_subbuffer: ustcomm_send_request failed");
688760ef PMF	58	return -1;
688760ef PMF	59	}
8cefc145 PMF	60	else if(result == 0) {
	61	DBG("app died while being traced");
	62	return GET_SUBBUF_DIED;
	63	}
688760ef PMF	64
688760ef PMF	65	result = sscanf(received_msg, "%as %ld", &rep_code, &buf->consumed_old);
3a7b90de	66	if(result != 2 && result != 1) {
688760ef PMF	67	ERR("unable to parse response to get_subbuffer");
	68	return -1;
	69	}
3a7b90de PMF	70
3a7b90de PMF	71	DBG("received msg is %s", received_msg);
688760ef PMF	72
	73	if(!strcmp(rep_code, "OK")) {
	74	DBG("got subbuffer %s", buf->name);
8cefc145	75	retval = GET_SUBBUF_OK;
688760ef	76	}
3a7b90de	77	else if(nth_token_is(received_msg, "END", 0) == 1) {
8cefc145	78	return GET_SUBBUF_DONE;
3a7b90de	79	}
688760ef	80	else {
3a7b90de PMF	81	DBG("error getting subbuffer %s", buf->name);
3a7b90de PMF	82	retval = -1;
688760ef PMF	83	}
688760ef PMF	84
3a7b90de PMF	85	/* FIMXE: free correctly the stuff */
3a7b90de PMF	86	free(received_msg);
688760ef PMF	87	free(rep_code);
	88	return retval;
	89	}
	90
	91	int put_subbuffer(struct buffer_info *buf)
	92	{
	93	char *send_msg;
	94	char *received_msg;
	95	char *rep_code;
	96	int retval;
	97	int result;
	98
	99	asprintf(&send_msg, "put_subbuffer %s %ld", buf->name, buf->consumed_old);
3bb56863	100	result = ustcomm_send_request(&buf->conn, send_msg, &received_msg);
688760ef PMF	101	if(result < 0) {
	102	ERR("put_subbuffer: send_message failed");
	103	return -1;
	104	}
	105	free(send_msg);
	106
	107	result = sscanf(received_msg, "%as", &rep_code);
	108	if(result != 1) {
	109	ERR("unable to parse response to put_subbuffer");
	110	return -1;
	111	}
	112	free(received_msg);
	113
	114	if(!strcmp(rep_code, "OK")) {
	115	DBG("subbuffer put %s", buf->name);
	116	retval = 1;
	117	}
	118	else {
	119	ERR("invalid response to put_subbuffer");
	120	}
	121
	122	free(rep_code);
	123	return retval;
	124	}
	125
	126	ssize_t patient_write(int fd, const void *buf, size_t count)
	127	{
	128	const char bufc = (const char ) buf;
	129	int result;
	130
	131	for(;;) {
	132	result = write(fd, bufc, count);
	133	if(result <= 0) {
	134	return result;
	135	}
	136	count -= result;
	137	bufc += result;
	138
	139	if(count == 0) {
	140	break;
	141	}
	142	}
	143
	144	return bufc-(const char *)buf;
	145	}
	146
3a7b90de PMF	147	void consumer_thread(void arg)
	148	{
	149	struct buffer_info buf = (struct buffer_info ) arg;
	150	int result;
	151
	152	for(;;) {
8cefc145	153	/* get the subbuffer */
0b0cd937 PMF	154	result = get_subbuffer(buf);
	155	if(result == -1) {
	156	ERR("error getting subbuffer");
	157	continue;
3a7b90de	158	}
0b0cd937 PMF	159	else if(result == GET_SUBBUF_DONE) {
	160	/* this is done */
	161	break;
	162	}
	163	else if(result == GET_SUBBUF_DIED) {
	164	finish_consuming_dead_subbuffer(buf);
	165	break;
3a7b90de PMF	166	}
	167
	168	/* write data to file */
	169	result = patient_write(buf->file_fd, buf->mem + (buf->consumed_old & (buf->n_subbufs * buf->subbuf_size-1)), buf->subbuf_size);
	170	if(result == -1) {
	171	PERROR("write");
	172	/* FIXME: maybe drop this trace */
	173	}
	174
8cefc145	175	/* put the subbuffer */
0b0cd937 PMF	176	result = put_subbuffer(buf);
	177	if(result == -1) {
	178	ERR("error putting subbuffer");
	179	break;
3a7b90de PMF	180	}
	181	}
	182
	183	DBG("thread for buffer %s is stopping", buf->name);
	184
8cefc145 PMF	185	/* FIXME: destroy, unalloc... */
8cefc145 PMF	186
3a7b90de PMF	187	return NULL;
	188	}
	189
	190	int add_buffer(pid_t pid, char *bufname)
	191	{
	192	struct buffer_info *buf;
	193	char *send_msg;
	194	char *received_msg;
	195	int result;
	196	char *tmp;
	197	int fd;
	198	pthread_t thr;
	199
	200	buf = (struct buffer_info *) malloc(sizeof(struct buffer_info));
	201	if(buf == NULL) {
	202	ERR("add_buffer: insufficient memory");
	203	return -1;
	204	}
	205
	206	buf->name = bufname;
	207	buf->pid = pid;
	208
4e2a8808 PMF	209	/* connect to app */
	210	result = ustcomm_connect_app(buf->pid, &buf->conn);
	211	if(result) {
	212	ERR("unable to connect to process");
	213	return -1;
	214	}
	215
3a7b90de PMF	216	/* get shmid */
3a7b90de PMF	217	asprintf(&send_msg, "get_shmid %s", buf->name);
4e2a8808	218	ustcomm_send_request(&buf->conn, send_msg, &received_msg);
3a7b90de PMF	219	free(send_msg);
	220	DBG("got buffer name %s", buf->name);
	221
8cefc145 PMF	222	result = sscanf(received_msg, "%d %d", &buf->shmid, &buf->bufstruct_shmid);
8cefc145 PMF	223	if(result != 2) {
3a7b90de PMF	224	ERR("unable to parse response to get_shmid");
	225	return -1;
	226	}
	227	free(received_msg);
8cefc145	228	DBG("got shmids %d %d", buf->shmid, buf->bufstruct_shmid);
3a7b90de PMF	229
	230	/* get n_subbufs */
	231	asprintf(&send_msg, "get_n_subbufs %s", buf->name);
4e2a8808	232	ustcomm_send_request(&buf->conn, send_msg, &received_msg);
3a7b90de PMF	233	free(send_msg);
	234
	235	result = sscanf(received_msg, "%d", &buf->n_subbufs);
	236	if(result != 1) {
	237	ERR("unable to parse response to get_n_subbufs");
	238	return -1;
	239	}
	240	free(received_msg);
	241	DBG("got n_subbufs %d", buf->n_subbufs);
	242
	243	/* get subbuf size */
	244	asprintf(&send_msg, "get_subbuf_size %s", buf->name);
4e2a8808	245	ustcomm_send_request(&buf->conn, send_msg, &received_msg);
3a7b90de PMF	246	free(send_msg);
	247
	248	result = sscanf(received_msg, "%d", &buf->subbuf_size);
	249	if(result != 1) {
	250	ERR("unable to parse response to get_subbuf_size");
	251	return -1;
	252	}
	253	free(received_msg);
	254	DBG("got subbuf_size %d", buf->subbuf_size);
	255
	256	/* attach memory */
	257	buf->mem = shmat(buf->shmid, NULL, 0);
	258	if(buf->mem == (void *) 0) {
	259	perror("shmat");
	260	return -1;
	261	}
8cefc145 PMF	262	DBG("successfully attached buffer memory");
	263
	264	buf->bufstruct_mem = shmat(buf->bufstruct_shmid, NULL, 0);
	265	if(buf->bufstruct_mem == (void *) 0) {
	266	perror("shmat");
	267	return -1;
	268	}
	269	DBG("successfully attached buffer bufstruct memory");
3a7b90de PMF	270
	271	/* open file for output */
	272	asprintf(&tmp, "/tmp/trace/%s_0", buf->name);
	273	result = fd = open(tmp, O_WRONLY \| O_CREAT \| O_TRUNC, 00600);
	274	if(result == -1) {
	275	PERROR("open");
	276	return -1;
	277	}
	278	buf->file_fd = fd;
	279	free(tmp);
	280
	281	//list_add(&buf->list, &buffers);
	282
	283	pthread_create(&thr, NULL, consumer_thread, buf);
	284
	285	return 0;
	286	}
	287
3796af9b PMF	288	int main(int argc, char **argv)
	289	{
	290	struct ustcomm_ustd ustd;
	291	int result;
	292
	293	result = ustcomm_init_ustd(&ustd);
	294	if(result == -1) {
	295	ERR("failed to initialize socket");
	296	return 1;
	297	}
	298
688760ef	299	/* app loop */
3796af9b PMF	300	for(;;) {
	301	char *recvbuf;
	302
3a7b90de	303	/* check for requests on our public socket */
688760ef PMF	304	result = ustcomm_ustd_recv_message(&ustd, &recvbuf, NULL, 100);
	305	if(result == -1) {
	306	ERR("error in ustcomm_ustd_recv_message");
	307	continue;
	308	}
	309	if(result > 0) {
	310	if(!strncmp(recvbuf, "collect", 7)) {
	311	pid_t pid;
	312	char *bufname;
	313	int result;
3796af9b	314
688760ef PMF	315	result = sscanf(recvbuf, "%*s %d %50as", &pid, &bufname);
	316	if(result != 2) {
	317	fprintf(stderr, "parsing error: %s\n", recvbuf);
	318	}
3796af9b	319
688760ef PMF	320	result = add_buffer(pid, bufname);
	321	if(result < 0) {
	322	ERR("error in add_buffer");
	323	continue;
	324	}
3796af9b PMF	325	}
3796af9b PMF	326
688760ef	327	free(recvbuf);
3796af9b	328	}
3796af9b PMF	329	}
	330
	331	return 0;
	332	}