[lttng-tools.git] / tests / regression / tools / live / live_test.c

/*
 * Copyright (c) - 2013 Julien Desfossez <jdesfossez@efficios.com>
 *
 * 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 as
 * published by the Free Software Foundation; only version 2 of the License.
 *
 * 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, write to the Free Software Foundation, Inc., 51
 * Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
 */

#define _GNU_SOURCE
#include <assert.h>
#include <errno.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <time.h>
#include <sys/types.h>
#include <inttypes.h>
#include <stdlib.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <netdb.h>
#include <fcntl.h>
#include <sys/mman.h>
#include <sys/stat.h>

#include <tap/tap.h>
#include <lttng/lttng.h>

#include <urcu/list.h>
#include <bin/lttng-sessiond/session.h>
#include <common/common.h>

#include <bin/lttng-relayd/lttng-viewer-abi.h>
#include <common/index/ctf-index.h>

#define SESSION1 "test1"
#define RELAYD_URL "net://localhost"
#define LIVE_TIMER 2000000

/* Number of TAP tests in this file */
#define NUM_TESTS 8
#define mmap_size 524288

int ust_consumerd32_fd;
int ust_consumerd64_fd;

static int control_sock;
struct live_session *session;

static int first_packet_offset;
static int first_packet_len;
static int first_packet_stream_id;

struct viewer_stream {
	uint64_t id;
	uint64_t ctf_trace_id;
	void *mmap_base;
	int fd;
	int metadata_flag;
	int first_read;
	char path[PATH_MAX];
};

struct live_session {
	struct viewer_stream *streams;
	uint64_t live_timer_interval;
	uint64_t stream_count;
};

static
int connect_viewer(char *hostname)
{
	struct hostent *host;
	struct sockaddr_in server_addr;
	int ret;

	host = gethostbyname(hostname);
	if (!host) {
		ret = -1;
		goto end;
	}

	if ((control_sock = socket(AF_INET, SOCK_STREAM, 0)) == -1) {
		perror("Socket");
		ret = -1;
		goto end;
	}

	server_addr.sin_family = AF_INET;
	server_addr.sin_port = htons(5344);
	server_addr.sin_addr = *((struct in_addr *) host->h_addr);
	bzero(&(server_addr.sin_zero), 8);

	if (connect(control_sock, (struct sockaddr *) &server_addr,
				sizeof(struct sockaddr)) == -1) {
		perror("Connect");
		ret = -1;
		goto end;
	}

	server_addr.sin_family = AF_INET;
	server_addr.sin_port = htons(5345);
	server_addr.sin_addr = *((struct in_addr *) host->h_addr);
	bzero(&(server_addr.sin_zero), 8);

	ret = 0;

end:
	return ret;
}

int establish_connection(void)
{
	struct lttng_viewer_cmd cmd;
	struct lttng_viewer_connect connect;
	int ret;

	cmd.cmd = htobe32(LTTNG_VIEWER_CONNECT);
	cmd.data_size = sizeof(connect);
	cmd.cmd_version = 0;

	memset(&connect, 0, sizeof(connect));
	connect.major = htobe32(VERSION_MAJOR);
	connect.minor = htobe32(VERSION_MINOR);
	connect.type = htobe32(LTTNG_VIEWER_CLIENT_COMMAND);

	do {
		ret = send(control_sock, &cmd, sizeof(cmd), 0);
	} while (ret < 0 && errno == EINTR);
	if (ret < 0) {
		fprintf(stderr, "Error sending cmd\n");
		goto error;
	}
	do {
		ret = send(control_sock, &connect, sizeof(connect), 0);
	} while (ret < 0 && errno == EINTR);
	if (ret < 0) {
		fprintf(stderr, "Error sending version\n");
		goto error;
	}

	do {
		ret = recv(control_sock, &connect, sizeof(connect), 0);
	} while (ret < 0 && errno == EINTR);
	if (ret < 0) {
		fprintf(stderr, "Error receiving version\n");
		goto error;
	}
	ret = 0;

error:
	return ret;
}

/*
 * Returns the number of sessions, should be 1 during the unit test.
 */
int list_sessions(int *session_id)
{
	struct lttng_viewer_cmd cmd;
	struct lttng_viewer_list_sessions list;
	struct lttng_viewer_session lsession;
	int i, ret;
	int first_session = 0;

	cmd.cmd = htobe32(LTTNG_VIEWER_LIST_SESSIONS);
	cmd.data_size = 0;
	cmd.cmd_version = 0;

	do {
		ret = send(control_sock, &cmd, sizeof(cmd), 0);
	} while (ret < 0 && errno == EINTR);
	if (ret < 0) {
		fprintf(stderr, "Error sending cmd\n");
		goto error;
	}

	do {
		ret = recv(control_sock, &list, sizeof(list), 0);
	} while (ret < 0 && errno == EINTR);
	if (ret < 0) {
		fprintf(stderr, "Error receiving session list\n");
		goto error;
	}

	for (i = 0; i < be32toh(list.sessions_count); i++) {
		do {
			ret = recv(control_sock, &lsession, sizeof(lsession), 0);
		} while (ret < 0 && errno == EINTR);
		if (ret < 0) {
			fprintf(stderr, "Error receiving session\n");
			goto error;
		}
		if (lsession.streams > 0 && first_session <= 0) {
			first_session = be64toh(lsession.id);
			*session_id = first_session;
		}
	}

	ret = be32toh(list.sessions_count);

error:
	return ret;
}

int create_viewer_session()
{
	struct lttng_viewer_cmd cmd;
	struct lttng_viewer_create_session_response resp;
	int ret;
	ssize_t ret_len;

	cmd.cmd = htobe32(LTTNG_VIEWER_CREATE_SESSION);
	cmd.data_size = 0;
	cmd.cmd_version = 0;

	do {
		ret_len = send(control_sock, &cmd, sizeof(cmd), 0);
	} while (ret_len < 0 && errno == EINTR);
	if (ret_len < 0) {
		fprintf(stderr, "[error] Error sending cmd\n");
		ret = ret_len;
		goto error;
	}
	assert(ret_len == sizeof(cmd));

	do {
		ret_len = recv(control_sock, &resp, sizeof(resp), 0);
	} while (ret_len < 0 && errno == EINTR);
	if (ret_len < 0) {
		fprintf(stderr, "[error] Error receiving create session reply\n");
		ret = ret_len;
		goto error;
	}
	assert(ret_len == sizeof(resp));

	if (be32toh(resp.status) != LTTNG_VIEWER_CREATE_SESSION_OK) {
		fprintf(stderr, "[error] Error creating viewer session\n");
		ret = -1;
		goto error;
	}
	ret = 0;

error:
	return ret;
}

int attach_session(int id)
{
	struct lttng_viewer_cmd cmd;
	struct lttng_viewer_attach_session_request rq;
	struct lttng_viewer_attach_session_response rp;
	struct lttng_viewer_stream stream;
	int ret, i;

	session = zmalloc(sizeof(struct live_session));
	if (!session) {
		ret = -1;
		goto error;
	}

	cmd.cmd = htobe32(LTTNG_VIEWER_ATTACH_SESSION);
	cmd.data_size = sizeof(rq);
	cmd.cmd_version = 0;

	memset(&rq, 0, sizeof(rq));
	rq.session_id = htobe64(id);
	rq.seek = htobe32(LTTNG_VIEWER_SEEK_BEGINNING);

	do {
		ret = send(control_sock, &cmd, sizeof(cmd), 0);
	} while (ret < 0 && errno == EINTR);
	if (ret < 0) {
		fprintf(stderr, "Error sending cmd\n");
		goto error;
	}
	do {
		ret = send(control_sock, &rq, sizeof(rq), 0);
	} while (ret < 0 && errno == EINTR);
	if (ret < 0) {
		fprintf(stderr, "Error sending attach request\n");
		goto error;
	}

	do {
		ret = recv(control_sock, &rp, sizeof(rp), 0);
	} while (ret < 0 && errno == EINTR);
	if (ret < 0) {
		fprintf(stderr, "Error receiving attach response\n");
		goto error;
	}
	if (be32toh(rp.status) != LTTNG_VIEWER_ATTACH_OK) {
		ret = -1;
		goto end;
	}

	session->stream_count = be32toh(rp.streams_count);
	session->streams = zmalloc(session->stream_count *
			sizeof(struct viewer_stream));
	if (!session->streams) {
		ret = -1;
		goto error;
	}

	for (i = 0; i < be32toh(rp.streams_count); i++) {
		do {
			ret = recv(control_sock, &stream, sizeof(stream), 0);
		} while (ret < 0 && errno == EINTR);
		if (ret < 0) {
			fprintf(stderr, "Error receiving stream\n");
			goto error;
		}
		session->streams[i].id = be64toh(stream.id);

		session->streams[i].ctf_trace_id = be64toh(stream.ctf_trace_id);
		session->streams[i].first_read = 1;
		session->streams[i].mmap_base = mmap(NULL, mmap_size, PROT_READ | PROT_WRITE,
				MAP_PRIVATE | MAP_ANONYMOUS, -1, 0);
		if (session->streams[i].mmap_base == MAP_FAILED) {
			fprintf(stderr, "mmap error\n");
			ret = -1;
			goto error;
		}

		if (be32toh(stream.metadata_flag)) {
			session->streams[i].metadata_flag = 1;
		}
	}
	ret = session->stream_count;

end:
error:
	return ret;
}

int get_metadata(void)
{
	struct lttng_viewer_cmd cmd;
	struct lttng_viewer_get_metadata rq;
	struct lttng_viewer_metadata_packet rp;
	int ret;
	uint64_t i;
	char *data = NULL;
	uint64_t len = 0;
	int metadata_stream_id = -1;

	cmd.cmd = htobe32(LTTNG_VIEWER_GET_METADATA);
	cmd.data_size = sizeof(rq);
	cmd.cmd_version = 0;

	for (i = 0; i < session->stream_count; i++) {
		if (session->streams[i].metadata_flag) {
			metadata_stream_id = i;
			break;
		}
	}

	if (metadata_stream_id < 0) {
		fprintf(stderr, "No metadata stream found\n");
		ret = -1;
		goto error;
	}

	rq.stream_id = htobe64(session->streams[metadata_stream_id].id);

	do {
		ret = send(control_sock, &cmd, sizeof(cmd), 0);
	} while (ret < 0 && errno == EINTR);
	if (ret < 0) {
		fprintf(stderr, "Error sending cmd\n");
		goto error;
	}
	do {
		ret = send(control_sock, &rq, sizeof(rq), 0);
	} while (ret < 0 && errno == EINTR);
	if (ret < 0) {
		fprintf(stderr, "Error sending get_metadata request\n");
		goto error;
	}
	do {
		ret = recv(control_sock, &rp, sizeof(rp), 0);
	} while (ret < 0 && errno == EINTR);
	if (ret < 0) {
		fprintf(stderr, "Error receiving metadata response\n");
		goto error;
	}
	switch (be32toh(rp.status)) {
		case LTTNG_VIEWER_METADATA_OK:
			break;
		case LTTNG_VIEWER_NO_NEW_METADATA:
			fprintf(stderr, "NO NEW\n");
			ret = -1;
			goto end;
		case LTTNG_VIEWER_METADATA_ERR:
			fprintf(stderr, "ERR\n");
			ret = -1;
			goto end;
		default:
			fprintf(stderr, "UNKNOWN\n");
			ret = -1;
			goto end;
	}

	len = be64toh(rp.len);
	if (len <= 0) {
		goto end;
	}

	data = zmalloc(len);
	if (!data) {
		perror("relay data zmalloc");
		goto error;
	}
	do {
		ret = recv(control_sock, data, len, MSG_WAITALL);
	} while (ret < 0 && errno == EINTR);
	if (ret < 0) {
		fprintf(stderr, "Error receiving trace packet\n");
		free(data);
		goto error;
	}
	free(data);

	ret = (int) len;
end:
error:
	return ret;
}

int get_next_index(void)
{
	struct lttng_viewer_cmd cmd;
	struct lttng_viewer_get_next_index rq;
	struct lttng_viewer_index rp;
	int ret;
	int id;

	cmd.cmd = htobe32(LTTNG_VIEWER_GET_NEXT_INDEX);
	cmd.data_size = sizeof(rq);
	cmd.cmd_version = 0;

	for (id = 0; id < session->stream_count; id++) {
		if (session->streams[id].metadata_flag) {
			continue;
		}
		rq.stream_id = htobe64(session->streams[id].id);

retry:
		do {
			ret = send(control_sock, &cmd, sizeof(cmd), 0);
		} while (ret < 0 && errno == EINTR);
		if (ret < 0) {
			fprintf(stderr, "Error sending cmd\n");
			goto error;
		}
		do {
			ret = send(control_sock, &rq, sizeof(rq), 0);
		} while (ret < 0 && errno == EINTR);
		if (ret < 0) {
			fprintf(stderr, "Error sending get_next_index request\n");
			goto error;
		}
		do {
			ret = recv(control_sock, &rp, sizeof(rp), 0);
		} while (ret < 0 && errno == EINTR);
		if (ret < 0) {
			fprintf(stderr, "Error receiving index response\n");
			goto error;
		}

		rp.flags = be32toh(rp.flags);

		switch (be32toh(rp.status)) {
			case LTTNG_VIEWER_INDEX_INACTIVE:
				fprintf(stderr, "(INACTIVE)\n");
				break;
			case LTTNG_VIEWER_INDEX_OK:
				break;
			case LTTNG_VIEWER_INDEX_RETRY:
				sleep(1);
				goto retry;
			case LTTNG_VIEWER_INDEX_HUP:
				fprintf(stderr, "(HUP)\n");
				session->streams[id].id = -1ULL;
				session->streams[id].fd = -1;
				break;
			case LTTNG_VIEWER_INDEX_ERR:
				fprintf(stderr, "(ERR)\n");
				ret = -1;
				goto error;
			default:
				fprintf(stderr, "SHOULD NOT HAPPEN\n");
				ret = -1;
				goto error;
		}
		if (!first_packet_stream_id) {
			first_packet_offset = be64toh(rp.offset);
			first_packet_len = be64toh(rp.packet_size) / CHAR_BIT;
			first_packet_stream_id = id;
		}
	}
	ret = 0;

error:
	return ret;
}

static
int get_data_packet(int id, uint64_t offset,
		uint64_t len)
{
	struct lttng_viewer_cmd cmd;
	struct lttng_viewer_get_packet rq;
	struct lttng_viewer_trace_packet rp;
	int ret;

	cmd.cmd = htobe32(LTTNG_VIEWER_GET_PACKET);
	cmd.data_size = sizeof(rq);
	cmd.cmd_version = 0;

	rq.stream_id = htobe64(session->streams[id].id);
	/* Already in big endian. */
	rq.offset = offset;
	rq.len = htobe32(len);

	do {
		ret = send(control_sock, &cmd, sizeof(cmd), 0);
	} while (ret < 0 && errno == EINTR);
	if (ret < 0) {
		fprintf(stderr, "Error sending cmd\n");
		goto error;
	}
	do {
		ret = send(control_sock, &rq, sizeof(rq), 0);
	} while (ret < 0 && errno == EINTR);
	if (ret < 0) {
		fprintf(stderr, "Error sending get_data_packet request\n");
		goto error;
	}
	do {
		ret = recv(control_sock, &rp, sizeof(rp), 0);
	} while (ret < 0 && errno == EINTR);
	if (ret < 0) {
		fprintf(stderr, "Error receiving data response\n");
		goto error;
	}
	rp.flags = be32toh(rp.flags);

	switch (be32toh(rp.status)) {
	case LTTNG_VIEWER_GET_PACKET_OK:
		break;
	case LTTNG_VIEWER_GET_PACKET_RETRY:
		fprintf(stderr, "RETRY\n");
		ret = -1;
		goto end;
	case LTTNG_VIEWER_GET_PACKET_ERR:
		if (rp.flags & LTTNG_VIEWER_FLAG_NEW_METADATA) {
			fprintf(stderr, "NEW_METADATA\n");
			ret = 0;
			goto end;
		}
		fprintf(stderr, "ERR\n");
		ret = -1;
		goto end;
	default:
		fprintf(stderr, "UNKNOWN\n");
		ret = -1;
		goto end;
	}

	len = be32toh(rp.len);
	if (len <= 0) {
		goto end;
	}

	if (len > mmap_size) {
		fprintf(stderr, "mmap_size not big enough\n");
		ret = -1;
		goto error;
	}

	do {
		ret = recv(control_sock, session->streams[id].mmap_base, len, MSG_WAITALL);
	} while (ret < 0 && errno == EINTR);
	if (ret < 0) {
		fprintf(stderr, "Error receiving trace packet\n");
		goto error;
	}
	ret = len;

end:
error:
	return ret;
}

int main(int argc, char **argv)
{
	int ret;
	int session_id;

	plan_tests(NUM_TESTS);

	diag("Live unit tests");

	ret = connect_viewer("localhost");
	ok(ret == 0, "Connect viewer to relayd");

	ret = establish_connection();
	ok(ret == 0, "Established connection and version check with %d.%d",
			VERSION_MAJOR, VERSION_MINOR);

	ret = list_sessions(&session_id);
	ok(ret > 0, "List sessions : %d session(s)", ret);

	ret = create_viewer_session();
	ok(ret == 0, "Create viewer session");

	ret = attach_session(session_id);
	ok(ret > 0, "Attach to session, %d streams received", ret);

	ret = get_metadata();
	ok(ret > 0, "Get metadata, received %d bytes", ret);

	ret = get_next_index();
	ok(ret == 0, "Get one index per stream");

	ret = get_data_packet(first_packet_stream_id, first_packet_offset,
			first_packet_len);
	ok(ret == first_packet_len,
			"Get one data packet for stream %d, offset %d, len %d",
			first_packet_stream_id, first_packet_offset,
			first_packet_len);

	return exit_status();
}
Commit	Line	Data
1b368955 JD	1	/*
	2	* Copyright (c) - 2013 Julien Desfossez <jdesfossez@efficios.com>
	3	*
	4	* This program is free software; you can redistribute it and/or modify it
	5	* under the terms of the GNU General Public License as published by as
	6	* published by the Free Software Foundation; only version 2 of the License.
	7	*
	8	* This program is distributed in the hope that it will be useful, but WITHOUT
	9	* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
	10	* FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
	11	* more details.
	12	*
	13	* You should have received a copy of the GNU General Public License along with
	14	* this program; if not, write to the Free Software Foundation, Inc., 51
	15	* Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
	16	*/
	17
	18	#define _GNU_SOURCE
	19	#include <assert.h>
	20	#include <errno.h>
	21	#include <stdio.h>
	22	#include <stdlib.h>
	23	#include <string.h>
	24	#include <unistd.h>
	25	#include <time.h>
	26	#include <sys/types.h>
	27	#include <inttypes.h>
	28	#include <stdlib.h>
	29	#include <sys/socket.h>
	30	#include <netinet/in.h>
	31	#include <netdb.h>
	32	#include <fcntl.h>
	33	#include <sys/mman.h>
	34	#include <sys/stat.h>
	35
	36	#include <tap/tap.h>
	37	#include <lttng/lttng.h>
	38
	39	#include <urcu/list.h>
	40	#include <bin/lttng-sessiond/session.h>
	41	#include <common/common.h>
	42
2f8f53af	43	#include <bin/lttng-relayd/lttng-viewer-abi.h>
50adc264	44	#include <common/index/ctf-index.h>
1b368955 JD	45
	46	#define SESSION1 "test1"
	47	#define RELAYD_URL "net://localhost"
	48	#define LIVE_TIMER 2000000
	49
	50	/* Number of TAP tests in this file */
25b50f8f	51	#define NUM_TESTS 8
1b368955 JD	52	#define mmap_size 524288
	53
	54	int ust_consumerd32_fd;
	55	int ust_consumerd64_fd;
	56
	57	static int control_sock;
	58	struct live_session *session;
	59
	60	static int first_packet_offset;
	61	static int first_packet_len;
	62	static int first_packet_stream_id;
	63
	64	struct viewer_stream {
	65	uint64_t id;
	66	uint64_t ctf_trace_id;
	67	void *mmap_base;
	68	int fd;
	69	int metadata_flag;
	70	int first_read;
	71	char path[PATH_MAX];
	72	};
	73
	74	struct live_session {
	75	struct viewer_stream *streams;
	76	uint64_t live_timer_interval;
	77	uint64_t stream_count;
	78	};
	79
	80	static
	81	int connect_viewer(char *hostname)
	82	{
	83	struct hostent *host;
	84	struct sockaddr_in server_addr;
	85	int ret;
	86
	87	host = gethostbyname(hostname);
	88	if (!host) {
	89	ret = -1;
	90	goto end;
	91	}
	92
	93	if ((control_sock = socket(AF_INET, SOCK_STREAM, 0)) == -1) {
	94	perror("Socket");
	95	ret = -1;
	96	goto end;
	97	}
	98
	99	server_addr.sin_family = AF_INET;
	100	server_addr.sin_port = htons(5344);
	101	server_addr.sin_addr = ((struct in_addr ) host->h_addr);
	102	bzero(&(server_addr.sin_zero), 8);
	103
	104	if (connect(control_sock, (struct sockaddr *) &server_addr,
	105	sizeof(struct sockaddr)) == -1) {
	106	perror("Connect");
	107	ret = -1;
	108	goto end;
	109	}
	110
	111	server_addr.sin_family = AF_INET;
	112	server_addr.sin_port = htons(5345);
	113	server_addr.sin_addr = ((struct in_addr ) host->h_addr);
	114	bzero(&(server_addr.sin_zero), 8);
	115
116	ret = 0;
117
118	end:
119	return ret;
120	}
121
122	int establish_connection(void)
123	{
124	struct lttng_viewer_cmd cmd;
125	struct lttng_viewer_connect connect;
126	int ret;
127
c4e361a4	128	cmd.cmd = htobe32(LTTNG_VIEWER_CONNECT);
1b368955 JD	129	cmd.data_size = sizeof(connect);
	130	cmd.cmd_version = 0;
	131
dcf5c25e	132	memset(&connect, 0, sizeof(connect));
1b368955 JD	133	connect.major = htobe32(VERSION_MAJOR);
1b368955 JD	134	connect.minor = htobe32(VERSION_MINOR);
c4e361a4	135	connect.type = htobe32(LTTNG_VIEWER_CLIENT_COMMAND);
1b368955 JD	136
	137	do {
	138	ret = send(control_sock, &cmd, sizeof(cmd), 0);
	139	} while (ret < 0 && errno == EINTR);
	140	if (ret < 0) {
	141	fprintf(stderr, "Error sending cmd\n");
	142	goto error;
	143	}
	144	do {
	145	ret = send(control_sock, &connect, sizeof(connect), 0);
	146	} while (ret < 0 && errno == EINTR);
	147	if (ret < 0) {
	148	fprintf(stderr, "Error sending version\n");
	149	goto error;
	150	}
	151
	152	do {
	153	ret = recv(control_sock, &connect, sizeof(connect), 0);
	154	} while (ret < 0 && errno == EINTR);
	155	if (ret < 0) {
	156	fprintf(stderr, "Error receiving version\n");
	157	goto error;
	158	}
	159	ret = 0;
	160
	161	error:
	162	return ret;
	163	}
	164
	165	/*
	166	* Returns the number of sessions, should be 1 during the unit test.
	167	*/
	168	int list_sessions(int *session_id)
	169	{
	170	struct lttng_viewer_cmd cmd;
	171	struct lttng_viewer_list_sessions list;
	172	struct lttng_viewer_session lsession;
	173	int i, ret;
	174	int first_session = 0;
	175
c4e361a4	176	cmd.cmd = htobe32(LTTNG_VIEWER_LIST_SESSIONS);
1b368955 JD	177	cmd.data_size = 0;
	178	cmd.cmd_version = 0;
	179
	180	do {
	181	ret = send(control_sock, &cmd, sizeof(cmd), 0);
	182	} while (ret < 0 && errno == EINTR);
	183	if (ret < 0) {
	184	fprintf(stderr, "Error sending cmd\n");
	185	goto error;
	186	}
	187
	188	do {
	189	ret = recv(control_sock, &list, sizeof(list), 0);
	190	} while (ret < 0 && errno == EINTR);
	191	if (ret < 0) {
	192	fprintf(stderr, "Error receiving session list\n");
	193	goto error;
	194	}
	195
	196	for (i = 0; i < be32toh(list.sessions_count); i++) {
	197	do {
	198	ret = recv(control_sock, &lsession, sizeof(lsession), 0);
	199	} while (ret < 0 && errno == EINTR);
	200	if (ret < 0) {
	201	fprintf(stderr, "Error receiving session\n");
	202	goto error;
	203	}
	204	if (lsession.streams > 0 && first_session <= 0) {
	205	first_session = be64toh(lsession.id);
	206	*session_id = first_session;
	207	}
	208	}
	209
	210	ret = be32toh(list.sessions_count);
	211
	212	error:
	213	return ret;
	214	}
	215
25b50f8f JD	216	int create_viewer_session()
	217	{
	218	struct lttng_viewer_cmd cmd;
	219	struct lttng_viewer_create_session_response resp;
	220	int ret;
	221	ssize_t ret_len;
	222
	223	cmd.cmd = htobe32(LTTNG_VIEWER_CREATE_SESSION);
	224	cmd.data_size = 0;
	225	cmd.cmd_version = 0;
	226
	227	do {
	228	ret_len = send(control_sock, &cmd, sizeof(cmd), 0);
	229	} while (ret_len < 0 && errno == EINTR);
	230	if (ret_len < 0) {
	231	fprintf(stderr, "[error] Error sending cmd\n");
	232	ret = ret_len;
	233	goto error;
	234	}
	235	assert(ret_len == sizeof(cmd));
	236
	237	do {
	238	ret_len = recv(control_sock, &resp, sizeof(resp), 0);
	239	} while (ret_len < 0 && errno == EINTR);
	240	if (ret_len < 0) {
	241	fprintf(stderr, "[error] Error receiving create session reply\n");
	242	ret = ret_len;
	243	goto error;
	244	}
	245	assert(ret_len == sizeof(resp));
	246
	247	if (be32toh(resp.status) != LTTNG_VIEWER_CREATE_SESSION_OK) {
	248	fprintf(stderr, "[error] Error creating viewer session\n");
	249	ret = -1;
	250	goto error;
	251	}
	252	ret = 0;
	253
	254	error:
	255	return ret;
	256	}
	257
1b368955 JD	258	int attach_session(int id)
	259	{
	260	struct lttng_viewer_cmd cmd;
	261	struct lttng_viewer_attach_session_request rq;
	262	struct lttng_viewer_attach_session_response rp;
	263	struct lttng_viewer_stream stream;
	264	int ret, i;
	265
	266	session = zmalloc(sizeof(struct live_session));
	267	if (!session) {
	268	ret = -1;
	269	goto error;
	270	}
	271
c4e361a4	272	cmd.cmd = htobe32(LTTNG_VIEWER_ATTACH_SESSION);
1b368955 JD	273	cmd.data_size = sizeof(rq);
	274	cmd.cmd_version = 0;
	275
dcf5c25e	276	memset(&rq, 0, sizeof(rq));
1b368955	277	rq.session_id = htobe64(id);
c4e361a4	278	rq.seek = htobe32(LTTNG_VIEWER_SEEK_BEGINNING);
1b368955 JD	279
	280	do {
	281	ret = send(control_sock, &cmd, sizeof(cmd), 0);
	282	} while (ret < 0 && errno == EINTR);
	283	if (ret < 0) {
	284	fprintf(stderr, "Error sending cmd\n");
	285	goto error;
	286	}
	287	do {
	288	ret = send(control_sock, &rq, sizeof(rq), 0);
	289	} while (ret < 0 && errno == EINTR);
	290	if (ret < 0) {
	291	fprintf(stderr, "Error sending attach request\n");
	292	goto error;
	293	}
	294
	295	do {
	296	ret = recv(control_sock, &rp, sizeof(rp), 0);
	297	} while (ret < 0 && errno == EINTR);
	298	if (ret < 0) {
	299	fprintf(stderr, "Error receiving attach response\n");
	300	goto error;
	301	}
c4e361a4	302	if (be32toh(rp.status) != LTTNG_VIEWER_ATTACH_OK) {
1b368955 JD	303	ret = -1;
	304	goto end;
	305	}
	306
	307	session->stream_count = be32toh(rp.streams_count);
	308	session->streams = zmalloc(session->stream_count *
	309	sizeof(struct viewer_stream));
	310	if (!session->streams) {
	311	ret = -1;
	312	goto error;
	313	}
	314
	315	for (i = 0; i < be32toh(rp.streams_count); i++) {
	316	do {
	317	ret = recv(control_sock, &stream, sizeof(stream), 0);
	318	} while (ret < 0 && errno == EINTR);
	319	if (ret < 0) {
	320	fprintf(stderr, "Error receiving stream\n");
	321	goto error;
	322	}
	323	session->streams[i].id = be64toh(stream.id);
	324
	325	session->streams[i].ctf_trace_id = be64toh(stream.ctf_trace_id);
	326	session->streams[i].first_read = 1;
	327	session->streams[i].mmap_base = mmap(NULL, mmap_size, PROT_READ \| PROT_WRITE,
	328	MAP_PRIVATE \| MAP_ANONYMOUS, -1, 0);
	329	if (session->streams[i].mmap_base == MAP_FAILED) {
	330	fprintf(stderr, "mmap error\n");
	331	ret = -1;
	332	goto error;
	333	}
	334
	335	if (be32toh(stream.metadata_flag)) {
	336	session->streams[i].metadata_flag = 1;
	337	}
	338	}
	339	ret = session->stream_count;
	340
	341	end:
	342	error:
	343	return ret;
	344	}
	345
	346	int get_metadata(void)
	347	{
	348	struct lttng_viewer_cmd cmd;
	349	struct lttng_viewer_get_metadata rq;
	350	struct lttng_viewer_metadata_packet rp;
	351	int ret;
	352	uint64_t i;
	353	char *data = NULL;
	354	uint64_t len = 0;
	355	int metadata_stream_id = -1;
	356
c4e361a4	357	cmd.cmd = htobe32(LTTNG_VIEWER_GET_METADATA);
1b368955 JD	358	cmd.data_size = sizeof(rq);
	359	cmd.cmd_version = 0;
	360
	361	for (i = 0; i < session->stream_count; i++) {
	362	if (session->streams[i].metadata_flag) {
	363	metadata_stream_id = i;
	364	break;
	365	}
	366	}
	367
	368	if (metadata_stream_id < 0) {
	369	fprintf(stderr, "No metadata stream found\n");
	370	ret = -1;
	371	goto error;
	372	}
	373
	374	rq.stream_id = htobe64(session->streams[metadata_stream_id].id);
	375
	376	do {
	377	ret = send(control_sock, &cmd, sizeof(cmd), 0);
	378	} while (ret < 0 && errno == EINTR);
	379	if (ret < 0) {
	380	fprintf(stderr, "Error sending cmd\n");
	381	goto error;
	382	}
	383	do {
	384	ret = send(control_sock, &rq, sizeof(rq), 0);
	385	} while (ret < 0 && errno == EINTR);
	386	if (ret < 0) {
	387	fprintf(stderr, "Error sending get_metadata request\n");
	388	goto error;
	389	}
	390	do {
	391	ret = recv(control_sock, &rp, sizeof(rp), 0);
	392	} while (ret < 0 && errno == EINTR);
	393	if (ret < 0) {
	394	fprintf(stderr, "Error receiving metadata response\n");
	395	goto error;
	396	}
	397	switch (be32toh(rp.status)) {
c4e361a4	398	case LTTNG_VIEWER_METADATA_OK:
1b368955	399	break;
c4e361a4	400	case LTTNG_VIEWER_NO_NEW_METADATA:
1b368955 JD	401	fprintf(stderr, "NO NEW\n");
	402	ret = -1;
	403	goto end;
c4e361a4	404	case LTTNG_VIEWER_METADATA_ERR:
1b368955 JD	405	fprintf(stderr, "ERR\n");
	406	ret = -1;
	407	goto end;
	408	default:
	409	fprintf(stderr, "UNKNOWN\n");
	410	ret = -1;
	411	goto end;
	412	}
	413
	414	len = be64toh(rp.len);
	415	if (len <= 0) {
	416	goto end;
	417	}
	418
	419	data = zmalloc(len);
	420	if (!data) {
	421	perror("relay data zmalloc");
	422	goto error;
	423	}
	424	do {
	425	ret = recv(control_sock, data, len, MSG_WAITALL);
	426	} while (ret < 0 && errno == EINTR);
	427	if (ret < 0) {
	428	fprintf(stderr, "Error receiving trace packet\n");
	429	free(data);
	430	goto error;
	431	}
	432	free(data);
	433
	434	ret = (int) len;
	435	end:
	436	error:
	437	return ret;
	438	}
	439
	440	int get_next_index(void)
	441	{
	442	struct lttng_viewer_cmd cmd;
	443	struct lttng_viewer_get_next_index rq;
	444	struct lttng_viewer_index rp;
	445	int ret;
	446	int id;
	447
c4e361a4	448	cmd.cmd = htobe32(LTTNG_VIEWER_GET_NEXT_INDEX);
1b368955 JD	449	cmd.data_size = sizeof(rq);
	450	cmd.cmd_version = 0;
	451
	452	for (id = 0; id < session->stream_count; id++) {
	453	if (session->streams[id].metadata_flag) {
	454	continue;
	455	}
	456	rq.stream_id = htobe64(session->streams[id].id);
	457
	458	retry:
	459	do {
	460	ret = send(control_sock, &cmd, sizeof(cmd), 0);
	461	} while (ret < 0 && errno == EINTR);
	462	if (ret < 0) {
	463	fprintf(stderr, "Error sending cmd\n");
	464	goto error;
	465	}
	466	do {
	467	ret = send(control_sock, &rq, sizeof(rq), 0);
	468	} while (ret < 0 && errno == EINTR);
	469	if (ret < 0) {
	470	fprintf(stderr, "Error sending get_next_index request\n");
	471	goto error;
	472	}
	473	do {
	474	ret = recv(control_sock, &rp, sizeof(rp), 0);
	475	} while (ret < 0 && errno == EINTR);
	476	if (ret < 0) {
	477	fprintf(stderr, "Error receiving index response\n");
	478	goto error;
	479	}
	480
	481	rp.flags = be32toh(rp.flags);
	482
	483	switch (be32toh(rp.status)) {
c4e361a4	484	case LTTNG_VIEWER_INDEX_INACTIVE:
1b368955 JD	485	fprintf(stderr, "(INACTIVE)\n");
1b368955 JD	486	break;
c4e361a4	487	case LTTNG_VIEWER_INDEX_OK:
1b368955	488	break;
c4e361a4	489	case LTTNG_VIEWER_INDEX_RETRY:
1b368955 JD	490	sleep(1);
1b368955 JD	491	goto retry;
c4e361a4	492	case LTTNG_VIEWER_INDEX_HUP:
1b368955 JD	493	fprintf(stderr, "(HUP)\n");
	494	session->streams[id].id = -1ULL;
	495	session->streams[id].fd = -1;
	496	break;
c4e361a4	497	case LTTNG_VIEWER_INDEX_ERR:
1b368955 JD	498	fprintf(stderr, "(ERR)\n");
	499	ret = -1;
	500	goto error;
	501	default:
	502	fprintf(stderr, "SHOULD NOT HAPPEN\n");
	503	ret = -1;
	504	goto error;
	505	}
	506	if (!first_packet_stream_id) {
	507	first_packet_offset = be64toh(rp.offset);
	508	first_packet_len = be64toh(rp.packet_size) / CHAR_BIT;
	509	first_packet_stream_id = id;
	510	}
	511	}
	512	ret = 0;
	513
	514	error:
	515	return ret;
	516	}
	517
	518	static
	519	int get_data_packet(int id, uint64_t offset,
	520	uint64_t len)
	521	{
	522	struct lttng_viewer_cmd cmd;
	523	struct lttng_viewer_get_packet rq;
	524	struct lttng_viewer_trace_packet rp;
	525	int ret;
	526
c4e361a4	527	cmd.cmd = htobe32(LTTNG_VIEWER_GET_PACKET);
1b368955 JD	528	cmd.data_size = sizeof(rq);
	529	cmd.cmd_version = 0;
	530
	531	rq.stream_id = htobe64(session->streams[id].id);
	532	/* Already in big endian. */
	533	rq.offset = offset;
	534	rq.len = htobe32(len);
	535
	536	do {
	537	ret = send(control_sock, &cmd, sizeof(cmd), 0);
	538	} while (ret < 0 && errno == EINTR);
	539	if (ret < 0) {
	540	fprintf(stderr, "Error sending cmd\n");
	541	goto error;
	542	}
	543	do {
	544	ret = send(control_sock, &rq, sizeof(rq), 0);
	545	} while (ret < 0 && errno == EINTR);
	546	if (ret < 0) {
	547	fprintf(stderr, "Error sending get_data_packet request\n");
	548	goto error;
	549	}
	550	do {
	551	ret = recv(control_sock, &rp, sizeof(rp), 0);
	552	} while (ret < 0 && errno == EINTR);
	553	if (ret < 0) {
	554	fprintf(stderr, "Error receiving data response\n");
	555	goto error;
	556	}
	557	rp.flags = be32toh(rp.flags);
	558
	559	switch (be32toh(rp.status)) {
c4e361a4	560	case LTTNG_VIEWER_GET_PACKET_OK:
1b368955	561	break;
c4e361a4	562	case LTTNG_VIEWER_GET_PACKET_RETRY:
1b368955 JD	563	fprintf(stderr, "RETRY\n");
	564	ret = -1;
	565	goto end;
c4e361a4	566	case LTTNG_VIEWER_GET_PACKET_ERR:
1b368955 JD	567	if (rp.flags & LTTNG_VIEWER_FLAG_NEW_METADATA) {
	568	fprintf(stderr, "NEW_METADATA\n");
	569	ret = 0;
	570	goto end;
	571	}
	572	fprintf(stderr, "ERR\n");
	573	ret = -1;
	574	goto end;
	575	default:
	576	fprintf(stderr, "UNKNOWN\n");
	577	ret = -1;
	578	goto end;
	579	}
	580
	581	len = be32toh(rp.len);
	582	if (len <= 0) {
	583	goto end;
	584	}
	585
	586	if (len > mmap_size) {
	587	fprintf(stderr, "mmap_size not big enough\n");
	588	ret = -1;
	589	goto error;
	590	}
	591
	592	do {
	593	ret = recv(control_sock, session->streams[id].mmap_base, len, MSG_WAITALL);
	594	} while (ret < 0 && errno == EINTR);
	595	if (ret < 0) {
	596	fprintf(stderr, "Error receiving trace packet\n");
	597	goto error;
	598	}
	599	ret = len;
	600
	601	end:
	602	error:
	603	return ret;
	604	}
	605
	606	int main(int argc, char **argv)
	607	{
	608	int ret;
	609	int session_id;
	610
	611	plan_tests(NUM_TESTS);
	612
	613	diag("Live unit tests");
	614
	615	ret = connect_viewer("localhost");
	616	ok(ret == 0, "Connect viewer to relayd");
	617
	618	ret = establish_connection();
	619	ok(ret == 0, "Established connection and version check with %d.%d",
	620	VERSION_MAJOR, VERSION_MINOR);
	621
	622	ret = list_sessions(&session_id);
	623	ok(ret > 0, "List sessions : %d session(s)", ret);
	624
25b50f8f JD	625	ret = create_viewer_session();
	626	ok(ret == 0, "Create viewer session");
	627
1b368955 JD	628	ret = attach_session(session_id);
	629	ok(ret > 0, "Attach to session, %d streams received", ret);
	630
	631	ret = get_metadata();
	632	ok(ret > 0, "Get metadata, received %d bytes", ret);
	633
	634	ret = get_next_index();
	635	ok(ret == 0, "Get one index per stream");
	636
	637	ret = get_data_packet(first_packet_stream_id, first_packet_offset,
	638	first_packet_len);
	639	ok(ret == first_packet_len,
	640	"Get one data packet for stream %d, offset %d, len %d",
	641	first_packet_stream_id, first_packet_offset,
	642	first_packet_len);
	643
	644	return exit_status();
	645	}