X-Git-Url: https://git.lttng.org/?p=lttng-tools.git;a=blobdiff_plain;f=src%2Fcommon%2Fust-consumer%2Fust-consumer.c;h=e5c5256c7b767675c7bbb748d1a5e25b84131580;hp=8ab2b819dcee52a237518b1ce6f8cba5f3e1aa1f;hb=c617c0c651432f9d5ae7adf4c5c1a5fd92ad828e;hpb=91dfef6e2f50460b9597fc8a8186949c48557c14

diff --git a/src/common/ust-consumer/ust-consumer.c b/src/common/ust-consumer/ust-consumer.c
index 8ab2b819d..e5c5256c7 100644
--- a/src/common/ust-consumer/ust-consumer.c
+++ b/src/common/ust-consumer/ust-consumer.c
@@ -101,6 +101,7 @@ int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
 		int sock, struct pollfd *consumer_sockpoll)
 {
 	ssize_t ret;
+	enum lttng_error_code ret_code = LTTNG_OK;
 	struct lttcomm_consumer_msg msg;
 
 	ret = lttcomm_recv_unix_sock(sock, &msg, sizeof(msg));
@@ -111,6 +112,14 @@ int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
 		return ret;
 	}
 	if (msg.cmd_type == LTTNG_CONSUMER_STOP) {
+		/*
+		 * Notify the session daemon that the command is completed.
+		 *
+		 * On transport layer error, the function call will print an error
+		 * message so handling the returned code is a bit useless since we
+		 * return an error code anyway.
+		 */
+		(void) consumer_send_status_msg(sock, ret_code);
 		return -ENOENT;
 	}
 
@@ -120,9 +129,10 @@ int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
 	switch (msg.cmd_type) {
 	case LTTNG_CONSUMER_ADD_RELAYD_SOCKET:
 	{
+		/* Session daemon status message are handled in the following call. */
 		ret = consumer_add_relayd_socket(msg.u.relayd_sock.net_index,
 				msg.u.relayd_sock.type, ctx, sock, consumer_sockpoll,
-				&msg.u.relayd_sock.sock);
+				&msg.u.relayd_sock.sock, msg.u.relayd_sock.session_id);
 		goto end_nosignal;
 	}
 	case LTTNG_CONSUMER_ADD_CHANNEL:
@@ -133,6 +143,13 @@ int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
 
 		DBG("UST Consumer adding channel");
 
+		/* First send a status message before receiving the fds. */
+		ret = consumer_send_status_msg(sock, ret_code);
+		if (ret < 0) {
+			/* Somehow, the session daemon is not responding anymore. */
+			goto end_nosignal;
+		}
+
 		/* block */
 		if (lttng_consumer_poll_socket(consumer_sockpoll) < 0) {
 			rcu_read_unlock();
@@ -145,12 +162,24 @@ int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
 			return ret;
 		}
 
+		/*
+		 * Send status code to session daemon only if the recv works. If the
+		 * above recv() failed, the session daemon is notified through the
+		 * error socket and the teardown is eventually done.
+		 */
+		ret = consumer_send_status_msg(sock, ret_code);
+		if (ret < 0) {
+			/* Somehow, the session daemon is not responding anymore. */
+			goto end_nosignal;
+		}
+
 		DBG("consumer_add_channel %d", msg.u.channel.channel_key);
 
 		new_channel = consumer_allocate_channel(msg.u.channel.channel_key,
 				fds[0], -1,
 				msg.u.channel.mmap_len,
-				msg.u.channel.max_sb_size);
+				msg.u.channel.max_sb_size,
+				msg.u.channel.nb_init_streams);
 		if (new_channel == NULL) {
 			lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_OUTFD_ERROR);
 			goto end_nosignal;
@@ -170,12 +199,20 @@ int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
 	case LTTNG_CONSUMER_ADD_STREAM:
 	{
 		struct lttng_consumer_stream *new_stream;
-		int fds[2];
+		int fds[2], stream_pipe;
 		size_t nb_fd = 2;
 		struct consumer_relayd_sock_pair *relayd = NULL;
+		int alloc_ret = 0;
 
 		DBG("UST Consumer adding stream");
 
+		/* First send a status message before receiving the fds. */
+		ret = consumer_send_status_msg(sock, ret_code);
+		if (ret < 0) {
+			/* Somehow, the session daemon is not responding anymore. */
+			goto end_nosignal;
+		}
+
 		/* block */
 		if (lttng_consumer_poll_socket(consumer_sockpoll) < 0) {
 			rcu_read_unlock();
@@ -188,9 +225,19 @@ int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
 			return ret;
 		}
 
-		DBG("consumer_add_stream chan %d stream %d",
-				msg.u.stream.channel_key,
-				msg.u.stream.stream_key);
+		/*
+		 * Send status code to session daemon only if the recv works. If the
+		 * above recv() failed, the session daemon is notified through the
+		 * error socket and the teardown is eventually done.
+		 */
+		ret = consumer_send_status_msg(sock, ret_code);
+		if (ret < 0) {
+			/* Somehow, the session daemon is not responding anymore. */
+			goto end_nosignal;
+		}
+
+		DBG("Consumer command ADD_STREAM chan %d stream %d",
+				msg.u.stream.channel_key, msg.u.stream.stream_key);
 
 		assert(msg.u.stream.output == LTTNG_EVENT_MMAP);
 		new_stream = consumer_allocate_stream(msg.u.stream.channel_key,
@@ -203,9 +250,24 @@ int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
 				msg.u.stream.uid,
 				msg.u.stream.gid,
 				msg.u.stream.net_index,
-				msg.u.stream.metadata_flag);
+				msg.u.stream.metadata_flag,
+				msg.u.stream.session_id,
+				&alloc_ret);
 		if (new_stream == NULL) {
-			lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_OUTFD_ERROR);
+			switch (alloc_ret) {
+			case -ENOMEM:
+			case -EINVAL:
+			default:
+				lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_OUTFD_ERROR);
+				break;
+			case -ENOENT:
+				/*
+				 * We could not find the channel. Can happen if cpu hotplug
+				 * happens while tearing down.
+				 */
+				DBG3("Could not find channel");
+				break;
+			}
 			goto end_nosignal;
 		}
 
@@ -219,42 +281,44 @@ int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
 					&new_stream->relayd_stream_id);
 			pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
 			if (ret < 0) {
+				consumer_del_stream(new_stream, NULL);
 				goto end_nosignal;
 			}
 		} else if (msg.u.stream.net_index != -1) {
 			ERR("Network sequence index %d unknown. Not adding stream.",
 					msg.u.stream.net_index);
-			free(new_stream);
+			consumer_del_stream(new_stream, NULL);
 			goto end_nosignal;
 		}
 
-		/* Send stream to the metadata thread */
-		if (new_stream->metadata_flag) {
-			if (ctx->on_recv_stream) {
-				ret = ctx->on_recv_stream(new_stream);
-				if (ret < 0) {
-					goto end_nosignal;
-				}
-			}
-
-			do {
-				ret = write(ctx->consumer_metadata_pipe[1], new_stream,
-						sizeof(struct lttng_consumer_stream));
-			} while (ret < 0 && errno == EINTR);
+		/* Do actions once stream has been received. */
+		if (ctx->on_recv_stream) {
+			ret = ctx->on_recv_stream(new_stream);
 			if (ret < 0) {
-				PERROR("write metadata pipe");
+				consumer_del_stream(new_stream, NULL);
+				goto end_nosignal;
 			}
+		}
+
+		/* Get the right pipe where the stream will be sent. */
+		if (new_stream->metadata_flag) {
+			stream_pipe = ctx->consumer_metadata_pipe[1];
 		} else {
-			if (ctx->on_recv_stream) {
-				ret = ctx->on_recv_stream(new_stream);
-				if (ret < 0) {
-					goto end_nosignal;
-				}
-			}
-			consumer_add_stream(new_stream);
+			stream_pipe = ctx->consumer_data_pipe[1];
 		}
 
-		DBG("UST consumer_add_stream %s (%d,%d) with relayd id %" PRIu64,
+		do {
+			ret = write(stream_pipe, &new_stream, sizeof(new_stream));
+		} while (ret < 0 && errno == EINTR);
+		if (ret < 0) {
+			PERROR("Consumer write %s stream to pipe %d",
+					new_stream->metadata_flag ? "metadata" : "data",
+					stream_pipe);
+			consumer_del_stream(new_stream, NULL);
+			goto end_nosignal;
+		}
+
+		DBG("UST consumer ADD_STREAM %s (%d,%d) with relayd id %" PRIu64,
 				msg.u.stream.path_name, fds[0], fds[1],
 				new_stream->relayd_stream_id);
 		break;
@@ -270,7 +334,7 @@ int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
 		relayd = consumer_find_relayd(index);
 		if (relayd == NULL) {
 			ERR("Unable to find relayd %" PRIu64, index);
-			goto end_nosignal;
+			ret_code = LTTNG_ERR_NO_CONSUMER;
 		}
 
 		/*
@@ -283,7 +347,15 @@ int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
 		 *
 		 * The destroy can happen either here or when a stream fd hangs up.
 		 */
-		consumer_flag_relayd_for_destroy(relayd);
+		if (relayd) {
+			consumer_flag_relayd_for_destroy(relayd);
+		}
+
+		ret = consumer_send_status_msg(sock, ret_code);
+		if (ret < 0) {
+			/* Somehow, the session daemon is not responding anymore. */
+			goto end_nosignal;
+		}
 
 		goto end_nosignal;
 	}
@@ -291,39 +363,32 @@ int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
 	{
 		rcu_read_unlock();
 		return -ENOSYS;
-#if 0
-		if (ctx->on_update_stream != NULL) {
-			ret = ctx->on_update_stream(msg.u.stream.stream_key, msg.u.stream.state);
-			if (ret == 0) {
-				consumer_change_stream_state(msg.u.stream.stream_key, msg.u.stream.state);
-			} else if (ret < 0) {
-				goto end;
-			}
-		} else {
-			consumer_change_stream_state(msg.u.stream.stream_key,
-				msg.u.stream.state);
+	}
+	case LTTNG_CONSUMER_DATA_PENDING:
+	{
+		int32_t ret;
+		uint64_t id = msg.u.data_pending.session_id;
+
+		DBG("UST consumer data pending command for id %" PRIu64, id);
+
+		ret = consumer_data_pending(id);
+
+		/* Send back returned value to session daemon */
+		ret = lttcomm_send_unix_sock(sock, &ret, sizeof(ret));
+		if (ret < 0) {
+			PERROR("send data pending ret code");
 		}
+
+		/*
+		 * No need to send back a status message since the data pending
+		 * returned value is the response.
+		 */
 		break;
-#endif
 	}
 	default:
 		break;
 	}
 
-	/*
-	 * Wake-up the other end by writing a null byte in the pipe (non-blocking).
-	 * Important note: Because writing into the pipe is non-blocking (and
-	 * therefore we allow dropping wakeup data, as long as there is wakeup data
-	 * present in the pipe buffer to wake up the other end), the other end
-	 * should perform the following sequence for waiting:
-	 *
-	 * 1) empty the pipe (reads).
-	 * 2) perform update operation.
-	 * 3) wait on the pipe (poll).
-	 */
-	do {
-		ret = write(ctx->consumer_poll_pipe[1], "", 1);
-	} while (ret < 0 && errno == EINTR);
 end_nosignal:
 	rcu_read_unlock();
 
@@ -363,7 +428,7 @@ void lttng_ustconsumer_del_channel(struct lttng_consumer_channel *chan)
 	ustctl_unmap_channel(chan->handle);
 }
 
-int lttng_ustconsumer_allocate_stream(struct lttng_consumer_stream *stream)
+int lttng_ustconsumer_add_stream(struct lttng_consumer_stream *stream)
 {
 	struct lttng_ust_object_data obj;
 	int ret;
@@ -373,21 +438,35 @@ int lttng_ustconsumer_allocate_stream(struct lttng_consumer_stream *stream)
 	obj.wait_fd = stream->wait_fd;
 	obj.memory_map_size = stream->mmap_len;
 	ret = ustctl_add_stream(stream->chan->handle, &obj);
-	if (ret)
-		return ret;
+	if (ret) {
+		ERR("UST ctl add_stream failed with ret %d", ret);
+		goto error;
+	}
+
 	stream->buf = ustctl_open_stream_read(stream->chan->handle, stream->cpu);
-	if (!stream->buf)
-		return -EBUSY;
+	if (!stream->buf) {
+		ERR("UST ctl open_stream_read failed");
+		ret = -EBUSY;
+		goto error;
+	}
+
 	/* ustctl_open_stream_read has closed the shm fd. */
 	stream->wait_fd_is_copy = 1;
 	stream->shm_fd = -1;
 
 	stream->mmap_base = ustctl_get_mmap_base(stream->chan->handle, stream->buf);
 	if (!stream->mmap_base) {
-		return -EINVAL;
+		ERR("UST ctl get_mmap_base failed");
+		ret = -EINVAL;
+		goto mmap_error;
 	}
 
 	return 0;
+
+mmap_error:
+	ustctl_close_stream_read(stream->chan->handle, stream->buf);
+error:
+	return ret;
 }
 
 void lttng_ustconsumer_del_stream(struct lttng_consumer_stream *stream)
@@ -405,13 +484,14 @@ int lttng_ustconsumer_read_subbuffer(struct lttng_consumer_stream *stream,
 	struct lttng_ust_shm_handle *handle;
 	struct lttng_ust_lib_ring_buffer *buf;
 	char dummy;
-	ssize_t readlen;
 
 	DBG("In read_subbuffer (wait_fd: %d, stream key: %d)",
 		stream->wait_fd, stream->key);
 
 	/* We can consume the 1 byte written into the wait_fd by UST */
 	if (!stream->hangup_flush_done) {
+		ssize_t readlen;
+
 		do {
 			readlen = read(stream->wait_fd, &dummy, 1);
 		} while (readlen == -1 && errno == EINTR);
@@ -465,7 +545,6 @@ int lttng_ustconsumer_read_subbuffer(struct lttng_consumer_stream *stream,
 		ERR("Error writing to tracefile "
 				"(ret: %zd != len: %lu != subbuf_size: %lu)",
 				ret, len, subbuf_size);
-
 	}
 	err = ustctl_put_next_subbuf(handle, buf);
 	assert(err == 0);
@@ -491,9 +570,48 @@ int lttng_ustconsumer_on_recv_stream(struct lttng_consumer_stream *stream)
 		stream->out_fd = ret;
 	}
 
+	ret = lttng_ustconsumer_add_stream(stream);
+	if (ret) {
+		consumer_del_stream(stream, NULL);
+		ret = -1;
+		goto error;
+	}
+
 	/* we return 0 to let the library handle the FD internally */
 	return 0;
 
 error:
 	return ret;
 }
+
+/*
+ * Check if data is still being extracted from the buffers for a specific
+ * stream. Consumer data lock MUST be acquired before calling this function
+ * and the stream lock.
+ *
+ * Return 1 if the traced data are still getting read else 0 meaning that the
+ * data is available for trace viewer reading.
+ */
+int lttng_ustconsumer_data_pending(struct lttng_consumer_stream *stream)
+{
+	int ret;
+
+	assert(stream);
+
+	DBG("UST consumer checking data pending");
+
+	ret = ustctl_get_next_subbuf(stream->chan->handle, stream->buf);
+	if (ret == 0) {
+		/* There is still data so let's put back this subbuffer. */
+		ret = ustctl_put_subbuf(stream->chan->handle, stream->buf);
+		assert(ret == 0);
+		ret = 1;  /* Data is pending */
+		goto end;
+	}
+
+	/* Data is NOT pending so ready to be read. */
+	ret = 0;
+
+end:
+	return ret;
+}