X-Git-Url: http://git.efficios.com/?p=babeltrace.git;a=blobdiff_plain;f=formats%2Flttng-live%2Flttng-live-comm.c;h=9feb7185c76d0d1cdea4c1e5dfd79dc300849dbc;hp=945df4317bdec9d0ca1aa324405f110d229d8143;hb=e29395a8120118fa59d3e464442135b0b1940fc7;hpb=f312b3aa35382b62e9f8bd2c4cbdd9180490845e

diff --git a/formats/lttng-live/lttng-live-comm.c b/formats/lttng-live/lttng-live-comm.c
index 945df431..9feb7185 100644
--- a/formats/lttng-live/lttng-live-comm.c
+++ b/formats/lttng-live/lttng-live-comm.c
@@ -113,16 +113,21 @@ int lttng_live_connect_viewer(struct lttng_live_ctx *ctx)
 	struct sockaddr_in server_addr;
 	int ret;
 
-	host = gethostbyname(ctx->relay_hostname);
-	if (!host) {
+	if (lttng_live_should_quit()) {
 		ret = -1;
 		goto end;
 	}
 
+	host = gethostbyname(ctx->relay_hostname);
+	if (!host) {
+		fprintf(stderr, "[error] Cannot lookup hostname %s\n",
+			ctx->relay_hostname);
+		goto error;
+	}
+
 	if ((ctx->control_sock = socket(AF_INET, SOCK_STREAM, 0)) == -1) {
 		perror("Socket");
-		ret = -1;
-		goto end;
+		goto error;
 	}
 
 	server_addr.sin_family = AF_INET;
@@ -133,14 +138,17 @@ int lttng_live_connect_viewer(struct lttng_live_ctx *ctx)
 	if (connect(ctx->control_sock, (struct sockaddr *) &server_addr,
 				sizeof(struct sockaddr)) == -1) {
 		perror("Connect");
-		ret = -1;
-		goto end;
+		goto error;
 	}
 
 	ret = 0;
 
 end:
 	return ret;
+
+error:
+	fprintf(stderr, "[error] Connection failed\n");
+	return -1;
 }
 
 int lttng_live_establish_connection(struct lttng_live_ctx *ctx)
@@ -150,6 +158,11 @@ int lttng_live_establish_connection(struct lttng_live_ctx *ctx)
 	int ret;
 	ssize_t ret_len;
 
+	if (lttng_live_should_quit()) {
+		ret = -1;
+		goto end;
+	}
+
 	cmd.cmd = htobe32(LTTNG_VIEWER_CONNECT);
 	cmd.data_size = sizeof(connect);
 	cmd.cmd_version = 0;
@@ -162,7 +175,6 @@ int lttng_live_establish_connection(struct lttng_live_ctx *ctx)
 	ret_len = lttng_live_send(ctx->control_sock, &cmd, sizeof(cmd));
 	if (ret_len < 0) {
 		perror("[error] Error sending cmd");
-		ret = ret_len;
 		goto error;
 	}
 	assert(ret_len == sizeof(cmd));
@@ -170,7 +182,6 @@ int lttng_live_establish_connection(struct lttng_live_ctx *ctx)
 	ret_len = lttng_live_send(ctx->control_sock, &connect, sizeof(connect));
 	if (ret_len < 0) {
 		perror("[error] Error sending version");
-		ret = ret_len;
 		goto error;
 	}
 	assert(ret_len == sizeof(connect));
@@ -178,12 +189,10 @@ int lttng_live_establish_connection(struct lttng_live_ctx *ctx)
 	ret_len = lttng_live_recv(ctx->control_sock, &connect, sizeof(connect));
 	if (ret_len == 0) {
 		fprintf(stderr, "[error] Remote side has closed connection\n");
-		ret = -1;
 		goto error;
 	}
 	if (ret_len < 0) {
 		perror("[error] Error receiving version");
-		ret = ret_len;
 		goto error;
 	}
 	assert(ret_len == sizeof(connect));
@@ -192,11 +201,13 @@ int lttng_live_establish_connection(struct lttng_live_ctx *ctx)
 			be64toh(connect.viewer_session_id));
 	printf_verbose("Relayd version : %u.%u\n", be32toh(connect.major),
 			be32toh(connect.minor));
-
 	ret = 0;
+end:
+	return ret;
 
 error:
-	return ret;
+	fprintf(stderr, "[error] Unable to establish connection\n");
+	return -1;
 }
 
 static
@@ -271,6 +282,11 @@ int lttng_live_list_sessions(struct lttng_live_ctx *ctx, const char *path)
 	uint64_t session_id;
 	GPtrArray *session_list = NULL;
 
+	if (lttng_live_should_quit()) {
+		ret = -1;
+		goto end;
+	}
+
 	if (strlen(ctx->session_name) == 0) {
 		print_list = 1;
 		session_list = g_ptr_array_new();
@@ -283,7 +299,6 @@ int lttng_live_list_sessions(struct lttng_live_ctx *ctx, const char *path)
 	ret_len = lttng_live_send(ctx->control_sock, &cmd, sizeof(cmd));
 	if (ret_len < 0) {
 		perror("[error] Error sending cmd");
-		ret = ret_len;
 		goto error;
 	}
 	assert(ret_len == sizeof(cmd));
@@ -291,12 +306,10 @@ int lttng_live_list_sessions(struct lttng_live_ctx *ctx, const char *path)
 	ret_len = lttng_live_recv(ctx->control_sock, &list, sizeof(list));
 	if (ret_len == 0) {
 		fprintf(stderr, "[error] Remote side has closed connection\n");
-		ret = -1;
 		goto error;
 	}
 	if (ret_len < 0) {
 		perror("[error] Error receiving session list");
-		ret = ret_len;
 		goto error;
 	}
 	assert(ret_len == sizeof(list));
@@ -306,12 +319,10 @@ int lttng_live_list_sessions(struct lttng_live_ctx *ctx, const char *path)
 		ret_len = lttng_live_recv(ctx->control_sock, &lsession, sizeof(lsession));
 		if (ret_len == 0) {
 			fprintf(stderr, "[error] Remote side has closed connection\n");
-			ret = -1;
 			goto error;
 		}
 		if (ret_len < 0) {
 			perror("[error] Error receiving session");
-			ret = ret_len;
 			goto error;
 		}
 		assert(ret_len == sizeof(lsession));
@@ -342,11 +353,13 @@ int lttng_live_list_sessions(struct lttng_live_ctx *ctx, const char *path)
 		print_session_list(session_list, path);
 		free_session_list(session_list);
 	}
-
 	ret = 0;
+end:
+	return ret;
 
 error:
-	return ret;
+	fprintf(stderr, "[error] Unable to list sessions\n");
+	return -1;
 }
 
 int lttng_live_ctf_trace_assign(struct lttng_live_viewer_stream *stream,
@@ -399,6 +412,11 @@ int lttng_live_attach_session(struct lttng_live_ctx *ctx, uint64_t id)
 	int ret, i;
 	ssize_t ret_len;
 
+	if (lttng_live_should_quit()) {
+		ret = -1;
+		goto end;
+	}
+
 	cmd.cmd = htobe32(LTTNG_VIEWER_ATTACH_SESSION);
 	cmd.data_size = sizeof(rq);
 	cmd.cmd_version = 0;
@@ -412,7 +430,6 @@ int lttng_live_attach_session(struct lttng_live_ctx *ctx, uint64_t id)
 	ret_len = lttng_live_send(ctx->control_sock, &cmd, sizeof(cmd));
 	if (ret_len < 0) {
 		perror("[error] Error sending cmd");
-		ret = ret_len;
 		goto error;
 	}
 	assert(ret_len == sizeof(cmd));
@@ -420,7 +437,6 @@ int lttng_live_attach_session(struct lttng_live_ctx *ctx, uint64_t id)
 	ret_len = lttng_live_send(ctx->control_sock, &rq, sizeof(rq));
 	if (ret_len < 0) {
 		perror("[error] Error sending attach request");
-		ret = ret_len;
 		goto error;
 	}
 	assert(ret_len == sizeof(rq));
@@ -428,12 +444,10 @@ int lttng_live_attach_session(struct lttng_live_ctx *ctx, uint64_t id)
 	ret_len = lttng_live_recv(ctx->control_sock, &rp, sizeof(rp));
 	if (ret_len == 0) {
 		fprintf(stderr, "[error] Remote side has closed connection\n");
-		ret = -1;
 		goto error;
 	}
 	if (ret_len < 0) {
 		perror("[error] Error receiving attach response");
-		ret = ret_len;
 		goto error;
 	}
 	assert(ret_len == sizeof(rp));
@@ -446,25 +460,20 @@ int lttng_live_attach_session(struct lttng_live_ctx *ctx, uint64_t id)
 		goto end;
 	case LTTNG_VIEWER_ATTACH_ALREADY:
 		fprintf(stderr, "[error] There is already a viewer attached to this session\n");
-		ret = -1;
-		goto end;
+		goto error;
 	case LTTNG_VIEWER_ATTACH_NOT_LIVE:
 		fprintf(stderr, "[error] Not a live session\n");
-		ret = -1;
-		goto end;
+		goto error;
 	case LTTNG_VIEWER_ATTACH_SEEK_ERR:
 		fprintf(stderr, "[error] Wrong seek parameter\n");
-		ret = -1;
-		goto end;
+		goto error;
 	default:
 		fprintf(stderr, "[error] Unknown attach return code %u\n",
 				be32toh(rp.status));
-		ret = -1;
-		goto end;
+		goto error;
 	}
 	if (be32toh(rp.status) != LTTNG_VIEWER_ATTACH_OK) {
-		ret = -1;
-		goto end;
+		goto error;
 	}
 
 	ctx->session->stream_count += be32toh(rp.streams_count);
@@ -485,12 +494,10 @@ int lttng_live_attach_session(struct lttng_live_ctx *ctx, uint64_t id)
 		ret_len = lttng_live_recv(ctx->control_sock, &stream, sizeof(stream));
 		if (ret_len == 0) {
 			fprintf(stderr, "[error] Remote side has closed connection\n");
-			ret = -1;
 			goto error;
 		}
 		if (ret_len < 0) {
 			perror("[error] Error receiving stream");
-			ret = ret_len;
 			goto error;
 		}
 		assert(ret_len == sizeof(stream));
@@ -517,10 +524,11 @@ int lttng_live_attach_session(struct lttng_live_ctx *ctx, uint64_t id)
 
 	}
 	ret = 0;
-
 end:
-error:
 	return ret;
+
+error:
+	return -1;
 }
 
 static
@@ -536,6 +544,9 @@ restart:
 		printf_verbose("Asking for new streams returns %d\n",
 				ret);
 		if (ret < 0) {
+			if (lttng_live_should_quit()) {
+				goto end;
+			}
 			if (ret == -LTTNG_VIEWER_NEW_STREAMS_HUP) {
 				printf_verbose("Session %" PRIu64 " closed\n",
 						id);
@@ -613,20 +624,22 @@ int get_data_packet(struct lttng_live_ctx *ctx,
 	int ret;
 
 retry:
+	if (lttng_live_should_quit()) {
+		ret = -1;
+		goto end;
+	}
 	cmd.cmd = htobe32(LTTNG_VIEWER_GET_PACKET);
 	cmd.data_size = sizeof(rq);
 	cmd.cmd_version = 0;
 
 	memset(&rq, 0, sizeof(rq));
 	rq.stream_id = htobe64(stream->id);
-	/* Already in big endian. */
-	rq.offset = offset;
+	rq.offset = htobe64(offset);
 	rq.len = htobe32(len);
 
 	ret_len = lttng_live_send(ctx->control_sock, &cmd, sizeof(cmd));
 	if (ret_len < 0) {
 		perror("[error] Error sending cmd");
-		ret = ret_len;
 		goto error;
 	}
 	assert(ret_len == sizeof(cmd));
@@ -634,7 +647,6 @@ retry:
 	ret_len = lttng_live_send(ctx->control_sock, &rq, sizeof(rq));
 	if (ret_len < 0) {
 		perror("[error] Error sending get_data_packet request");
-		ret = ret_len;
 		goto error;
 	}
 	assert(ret_len == sizeof(rq));
@@ -642,19 +654,16 @@ retry:
 	ret_len = lttng_live_recv(ctx->control_sock, &rp, sizeof(rp));
 	if (ret_len == 0) {
 		fprintf(stderr, "[error] Remote side has closed connection\n");
-		ret = -1;
 		goto error;
 	}
 	if (ret_len < 0) {
 		perror("[error] Error receiving data response");
-		ret = ret_len;
 		goto error;
 	}
 	if (ret_len != sizeof(rp)) {
 		fprintf(stderr, "[error] get_data_packet: expected %" PRId64
-				", received %" PRId64 "\n", ret_len,
-				sizeof(rp));
-		ret = -1;
+				", received %" PRId64 "\n", sizeof(rp),
+				ret_len);
 		goto error;
 	}
 
@@ -667,9 +676,9 @@ retry:
 				"\n", len);
 		break;
 	case LTTNG_VIEWER_GET_PACKET_RETRY:
+		/* Unimplemented by relay daemon */
 		printf_verbose("get_data_packet: retry\n");
-		ret = -1;
-		goto end;
+		goto error;
 	case LTTNG_VIEWER_GET_PACKET_ERR:
 		if (rp.flags & LTTNG_VIEWER_FLAG_NEW_METADATA) {
 			printf_verbose("get_data_packet: new metadata needed\n");
@@ -682,29 +691,25 @@ retry:
 			ret = ask_new_streams(ctx);
 			if (ret < 0)
 				goto error;
-			g_hash_table_foreach(ctx->session->ctf_traces, add_traces,
-					ctx->bt_ctx);
+			g_hash_table_foreach(ctx->session->ctf_traces,
+					add_traces, ctx->bt_ctx);
 		}
 		if (rp.flags & (LTTNG_VIEWER_FLAG_NEW_METADATA
 				| LTTNG_VIEWER_FLAG_NEW_STREAM)) {
 			goto retry;
 		}
 		fprintf(stderr, "[error] get_data_packet: error\n");
-		ret = -1;
-		goto end;
+		goto error;
 	case LTTNG_VIEWER_GET_PACKET_EOF:
 		ret = -2;
-		goto error;
+		goto end;
 	default:
 		printf_verbose("get_data_packet: unknown\n");
-		assert(0);
-		ret = -1;
-		goto end;
+		goto error;
 	}
 
-	if (len <= 0) {
-		ret = -1;
-		goto end;
+	if (len == 0) {
+		goto error;
 	}
 
 	if (len > stream->mmap_size) {
@@ -716,7 +721,6 @@ retry:
 			ret = munmap_align(pos->base_mma);
 			if (ret) {
 				perror("[error] Unable to unmap old base");
-				ret = -1;
 				goto error;
 			}
 			pos->base_mma = NULL;
@@ -727,7 +731,6 @@ retry:
 		if (pos->base_mma == MAP_FAILED) {
 			perror("[error] mmap error");
 			pos->base_mma = NULL;
-			ret = -1;
 			goto error;
 		}
 
@@ -740,19 +743,19 @@ retry:
 			mmap_align_addr(pos->base_mma), len);
 	if (ret_len == 0) {
 		fprintf(stderr, "[error] Remote side has closed connection\n");
-		ret = -1;
 		goto error;
 	}
 	if (ret_len < 0) {
 		perror("[error] Error receiving trace packet");
-		ret = ret_len;
 		goto error;
 	}
 	assert(ret_len == len);
 	ret = 0;
 end:
-error:
 	return ret;
+
+error:
+	return -1;
 }
 
 static
@@ -767,6 +770,11 @@ int get_one_metadata_packet(struct lttng_live_ctx *ctx,
 	char *data = NULL;
 	ssize_t ret_len;
 
+	if (lttng_live_should_quit()) {
+		ret = -1;
+		goto end;
+	}
+
 	rq.stream_id = htobe64(metadata_stream->id);
 	cmd.cmd = htobe32(LTTNG_VIEWER_GET_METADATA);
 	cmd.data_size = sizeof(rq);
@@ -775,7 +783,6 @@ int get_one_metadata_packet(struct lttng_live_ctx *ctx,
 	ret_len = lttng_live_send(ctx->control_sock, &cmd, sizeof(cmd));
 	if (ret_len < 0) {
 		perror("[error] Error sending cmd");
-		ret = ret_len;
 		goto error;
 	}
 	assert(ret_len == sizeof(cmd));
@@ -783,7 +790,6 @@ int get_one_metadata_packet(struct lttng_live_ctx *ctx,
 	ret_len = lttng_live_send(ctx->control_sock, &rq, sizeof(rq));
 	if (ret_len < 0) {
 		perror("[error] Error sending get_metadata request");
-		ret = ret_len;
 		goto error;
 	}
 	assert(ret_len == sizeof(rq));
@@ -791,12 +797,10 @@ int get_one_metadata_packet(struct lttng_live_ctx *ctx,
 	ret_len = lttng_live_recv(ctx->control_sock, &rp, sizeof(rp));
 	if (ret_len == 0) {
 		fprintf(stderr, "[error] Remote side has closed connection\n");
-		ret = -1;
 		goto error;
 	}
 	if (ret_len < 0) {
 		perror("[error] Error receiving metadata response");
-		ret = ret_len;
 		goto error;
 	}
 	assert(ret_len == sizeof(rp));
@@ -811,39 +815,31 @@ int get_one_metadata_packet(struct lttng_live_ctx *ctx,
 			goto end;
 		case LTTNG_VIEWER_METADATA_ERR:
 			printf_verbose("get_metadata : ERR\n");
-			ret = -1;
-			goto end;
+			goto error;
 		default:
 			printf_verbose("get_metadata : UNKNOWN\n");
-			ret = -1;
-			goto end;
+			goto error;
 	}
 
 	len = be64toh(rp.len);
 	printf_verbose("Writing %" PRIu64" bytes to metadata\n", len);
 	if (len <= 0) {
-		ret = -1;
-		goto end;
+		goto error;
 	}
 
 	data = zmalloc(len);
 	if (!data) {
 		perror("relay data zmalloc");
-		ret = -1;
 		goto error;
 	}
 	ret_len = lttng_live_recv(ctx->control_sock, data, len);
 	if (ret_len == 0) {
 		fprintf(stderr, "[error] Remote side has closed connection\n");
-		ret = -1;
-		free(data);
-		goto error;
+		goto error_free_data;
 	}
 	if (ret_len < 0) {
 		perror("[error] Error receiving trace packet");
-		ret = ret_len;
-		free(data);
-		goto error;
+		goto error_free_data;
 	}
 	assert(ret_len == len);
 
@@ -853,19 +849,19 @@ int get_one_metadata_packet(struct lttng_live_ctx *ctx,
 	} while (ret_len < 0 && errno == EINTR);
 	if (ret_len < 0) {
 		fprintf(stderr, "[error] Writing in the metadata fp\n");
-		free(data);
-		ret = ret_len;
-		goto error;
+		goto error_free_data;
 	}
 	assert(ret_len == len);
 	metadata_stream->metadata_len += len;
+	free(data);
 	ret = len;
+end:
+	return ret;
 
+error_free_data:
 	free(data);
-
-end:
 error:
-	return ret;
+	return -1;
 }
 
 /*
@@ -939,10 +935,13 @@ int get_next_index(struct lttng_live_ctx *ctx,
 	rq.stream_id = htobe64(viewer_stream->id);
 
 retry:
+	if (lttng_live_should_quit()) {
+		ret = -1;
+		goto end;
+	}
 	ret_len = lttng_live_send(ctx->control_sock, &cmd, sizeof(cmd));
 	if (ret_len < 0) {
 		perror("[error] Error sending cmd");
-		ret = ret_len;
 		goto error;
 	}
 	assert(ret_len == sizeof(cmd));
@@ -950,7 +949,6 @@ retry:
 	ret_len = lttng_live_send(ctx->control_sock, &rq, sizeof(rq));
 	if (ret_len < 0) {
 		perror("[error] Error sending get_next_index request");
-		ret = ret_len;
 		goto error;
 	}
 	assert(ret_len == sizeof(rq));
@@ -958,12 +956,10 @@ retry:
 	ret_len = lttng_live_recv(ctx->control_sock, &rp, sizeof(rp));
 	if (ret_len == 0) {
 		fprintf(stderr, "[error] Remote side has closed connection\n");
-		ret = -1;
 		goto error;
 	}
 	if (ret_len < 0) {
 		perror("[error] Error receiving index response");
-		ret = ret_len;
 		goto error;
 	}
 	assert(ret_len == sizeof(rp));
@@ -995,8 +991,8 @@ retry:
 			ret = ask_new_streams(ctx);
 			if (ret < 0)
 				goto error;
-			g_hash_table_foreach(ctx->session->ctf_traces, add_traces,
-					ctx->bt_ctx);
+			g_hash_table_foreach(ctx->session->ctf_traces,
+					add_traces, ctx->bt_ctx);
 		}
 		break;
 	case LTTNG_VIEWER_INDEX_RETRY:
@@ -1011,18 +1007,17 @@ retry:
 		break;
 	case LTTNG_VIEWER_INDEX_ERR:
 		fprintf(stderr, "[error] get_next_index: error\n");
-		ret = -1;
 		goto error;
 	default:
 		fprintf(stderr, "[error] get_next_index: unkwown value\n");
-		ret = -1;
 		goto error;
 	}
-
 	ret = 0;
+end:
+	return ret;
 
 error:
-	return ret;
+	return -1;
 }
 
 static
@@ -1073,7 +1068,9 @@ retry:
 	ret = get_next_index(session->ctx, viewer_stream, cur_index);
 	if (ret < 0) {
 		pos->offset = EOF;
-		fprintf(stderr, "[error] get_next_index failed\n");
+		if (!lttng_live_should_quit()) {
+			fprintf(stderr, "[error] get_next_index failed\n");
+		}
 		return;
 	}
 
@@ -1117,13 +1114,15 @@ retry:
 	printf_verbose("get_data_packet for stream %" PRIu64 "\n",
 			viewer_stream->id);
 	ret = get_data_packet(session->ctx, pos, viewer_stream,
-			be64toh(cur_index->offset),
+			cur_index->offset,
 			cur_index->packet_size / CHAR_BIT);
 	if (ret == -2) {
 		goto retry;
 	} else if (ret < 0) {
 		pos->offset = EOF;
-		fprintf(stderr, "[error] get_data_packet failed\n");
+		if (!lttng_live_should_quit()) {
+			fprintf(stderr, "[error] get_data_packet failed\n");
+		}
 		return;
 	}
 
@@ -1166,6 +1165,11 @@ int lttng_live_create_viewer_session(struct lttng_live_ctx *ctx)
 	int ret;
 	ssize_t ret_len;
 
+	if (lttng_live_should_quit()) {
+		ret = -1;
+		goto end;
+	}
+
 	cmd.cmd = htobe32(LTTNG_VIEWER_CREATE_SESSION);
 	cmd.data_size = 0;
 	cmd.cmd_version = 0;
@@ -1173,7 +1177,6 @@ int lttng_live_create_viewer_session(struct lttng_live_ctx *ctx)
 	ret_len = lttng_live_send(ctx->control_sock, &cmd, sizeof(cmd));
 	if (ret_len < 0) {
 		perror("[error] Error sending cmd");
-		ret = ret_len;
 		goto error;
 	}
 	assert(ret_len == sizeof(cmd));
@@ -1181,25 +1184,24 @@ int lttng_live_create_viewer_session(struct lttng_live_ctx *ctx)
 	ret_len = lttng_live_recv(ctx->control_sock, &resp, sizeof(resp));
 	if (ret_len == 0) {
 		fprintf(stderr, "[error] Remote side has closed connection\n");
-		ret = -1;
 		goto error;
 	}
 	if (ret_len < 0) {
 		perror("[error] Error receiving create session reply");
-		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;
+end:
+	return ret;
 
 error:
-	return ret;
+	return -1;
 }
 
 static
@@ -1323,6 +1325,11 @@ int lttng_live_get_new_streams(struct lttng_live_ctx *ctx, uint64_t id)
 	ssize_t ret_len;
 	uint32_t stream_count;
 
+	if (lttng_live_should_quit()) {
+		ret = -1;
+		goto end;
+	}
+
 	cmd.cmd = htobe32(LTTNG_VIEWER_GET_NEW_STREAMS);
 	cmd.data_size = sizeof(rq);
 	cmd.cmd_version = 0;
@@ -1333,7 +1340,6 @@ int lttng_live_get_new_streams(struct lttng_live_ctx *ctx, uint64_t id)
 	ret_len = lttng_live_send(ctx->control_sock, &cmd, sizeof(cmd));
 	if (ret_len < 0) {
 		perror("[error] Error sending cmd");
-		ret = ret_len;
 		goto error;
 	}
 	assert(ret_len == sizeof(cmd));
@@ -1341,7 +1347,6 @@ int lttng_live_get_new_streams(struct lttng_live_ctx *ctx, uint64_t id)
 	ret_len = lttng_live_send(ctx->control_sock, &rq, sizeof(rq));
 	if (ret_len < 0) {
 		perror("[error] Error sending get_new_streams request");
-		ret = ret_len;
 		goto error;
 	}
 	assert(ret_len == sizeof(rq));
@@ -1349,12 +1354,10 @@ int lttng_live_get_new_streams(struct lttng_live_ctx *ctx, uint64_t id)
 	ret_len = lttng_live_recv(ctx->control_sock, &rp, sizeof(rp));
 	if (ret_len == 0) {
 		fprintf(stderr, "[error] Remote side has closed connection\n");
-		ret = -1;
 		goto error;
 	}
 	if (ret_len < 0) {
 		perror("[error] Error receiving get_new_streams response");
-		ret = ret_len;
 		goto error;
 	}
 	assert(ret_len == sizeof(rp));
@@ -1370,13 +1373,11 @@ int lttng_live_get_new_streams(struct lttng_live_ctx *ctx, uint64_t id)
 		goto end;
 	case LTTNG_VIEWER_NEW_STREAMS_ERR:
 		fprintf(stderr, "[error] get_new_streams error\n");
-		ret = -1;
-		goto end;
+		goto error;
 	default:
 		fprintf(stderr, "[error] Unknown return code %u\n",
 				be32toh(rp.status));
-		ret = -1;
-		goto end;
+		goto error;
 	}
 
 	stream_count = be32toh(rp.streams_count);
@@ -1398,12 +1399,10 @@ int lttng_live_get_new_streams(struct lttng_live_ctx *ctx, uint64_t id)
 		ret_len = lttng_live_recv(ctx->control_sock, &stream, sizeof(stream));
 		if (ret_len == 0) {
 			fprintf(stderr, "[error] Remote side has closed connection\n");
-			ret = -1;
 			goto error;
 		}
 		if (ret_len < 0) {
 			perror("[error] Error receiving stream");
-			ret = ret_len;
 			goto error;
 		}
 		assert(ret_len == sizeof(stream));
@@ -1430,15 +1429,17 @@ int lttng_live_get_new_streams(struct lttng_live_ctx *ctx, uint64_t id)
 
 	}
 	ret = 0;
-
 end:
-error:
 	return ret;
+
+error:
+	return -1;
 }
 
-void lttng_live_read(struct lttng_live_ctx *ctx)
+int lttng_live_read(struct lttng_live_ctx *ctx)
 {
-	int ret, i;
+	int ret = -1;
+	int i;
 	struct bt_ctf_iter *iter;
 	const struct bt_ctf_event *event;
 	struct bt_iter_pos begin_pos;
@@ -1467,8 +1468,9 @@ void lttng_live_read(struct lttng_live_ctx *ctx)
 
 	sout = container_of(td_write, struct ctf_text_stream_pos,
 			trace_descriptor);
-	if (!sout->parent.event_cb)
+	if (!sout->parent.event_cb) {
 		goto end_free;
+	}
 
 	ret = lttng_live_create_viewer_session(ctx);
 	if (ret < 0) {
@@ -1494,14 +1496,24 @@ void lttng_live_read(struct lttng_live_ctx *ctx)
 	for (;;) {
 		int flags;
 
+		if (lttng_live_should_quit()) {
+			ret = 0;
+			goto end_free;
+		}
+
 		while (!ctx->session->stream_count) {
-			if (ctx->session_ids->len == 0)
+			if (lttng_live_should_quit()
+					|| ctx->session_ids->len == 0) {
+				ret = 0;
 				goto end_free;
+			}
 			ret = ask_new_streams(ctx);
-			if (ret < 0)
+			if (ret < 0) {
 				goto end_free;
-			if (!ctx->session->stream_count)
+			}
+			if (!ctx->session->stream_count) {
 				(void) poll(NULL, 0, ACTIVE_POLL_DELAY);
+			}
 		}
 
 		g_hash_table_foreach(ctx->session->ctf_traces, add_traces,
@@ -1510,10 +1522,18 @@ void lttng_live_read(struct lttng_live_ctx *ctx)
 		begin_pos.type = BT_SEEK_BEGIN;
 		iter = bt_ctf_iter_create(ctx->bt_ctx, &begin_pos, NULL);
 		if (!iter) {
+			if (lttng_live_should_quit()) {
+				ret = 0;
+				goto end;
+			}
 			fprintf(stderr, "[error] Iterator creation error\n");
 			goto end;
 		}
 		for (;;) {
+			if (lttng_live_should_quit()) {
+				ret = 0;
+				goto end_free;
+			}
 			event = bt_ctf_iter_read_event_flags(iter, &flags);
 			if (!(flags & BT_ITER_FLAG_RETRY)) {
 				if (!event) {
@@ -1542,5 +1562,8 @@ void lttng_live_read(struct lttng_live_ctx *ctx)
 end_free:
 	bt_context_put(ctx->bt_ctx);
 end:
-	return;
+	if (lttng_live_should_quit()) {
+		ret = 0;
+	}
+	return ret;
 }