X-Git-Url: https://git.lttng.org/?p=lttng-tools.git;a=blobdiff_plain;f=src%2Fcommon%2Fconsumer.c;h=4b657f33249f7a4f777b7b3ff784b379b15f5878;hp=3aafb519380b334099d584b83ca17d9b201ed6e1;hb=6a00837f8cb0431a2ad90974d67fae138ea97dd5;hpb=4891ece8d4eeb2645efdf1467680b037c63ab425

diff --git a/src/common/consumer.c b/src/common/consumer.c
index 3aafb5193..4b657f332 100644
--- a/src/common/consumer.c
+++ b/src/common/consumer.c
@@ -291,6 +291,7 @@ void consumer_del_channel(struct lttng_consumer_channel *channel)
 	DBG("Consumer delete channel key %" PRIu64, channel->key);
 
 	pthread_mutex_lock(&consumer_data.lock);
+	pthread_mutex_lock(&channel->lock);
 
 	/* Delete streams that might have been left in the stream list. */
 	cds_list_for_each_entry_safe(stream, stmp, &channel->streams.head,
@@ -324,6 +325,7 @@ void consumer_del_channel(struct lttng_consumer_channel *channel)
 
 	call_rcu(&channel->node.head, free_channel_rcu);
 end:
+	pthread_mutex_unlock(&channel->lock);
 	pthread_mutex_unlock(&consumer_data.lock);
 }
 
@@ -455,6 +457,19 @@ void consumer_del_stream(struct lttng_consumer_stream *stream,
 	consumer_stream_destroy(stream, ht);
 }
 
+/*
+ * XXX naming of del vs destroy is all mixed up.
+ */
+void consumer_del_stream_for_data(struct lttng_consumer_stream *stream)
+{
+	consumer_stream_destroy(stream, data_ht);
+}
+
+void consumer_del_stream_for_metadata(struct lttng_consumer_stream *stream)
+{
+	consumer_stream_destroy(stream, metadata_ht);
+}
+
 struct lttng_consumer_stream *consumer_allocate_stream(uint64_t channel_key,
 		uint64_t stream_key,
 		enum lttng_consumer_stream_state state,
@@ -489,6 +504,7 @@ struct lttng_consumer_stream *consumer_allocate_stream(uint64_t channel_key,
 	stream->net_seq_idx = relayd_id;
 	stream->session_id = session_id;
 	stream->monitor = monitor;
+	stream->endpoint_status = CONSUMER_ENDPOINT_ACTIVE;
 	pthread_mutex_init(&stream->lock, NULL);
 
 	/* If channel is the metadata, flag this stream as metadata. */
@@ -536,11 +552,10 @@ end:
 /*
  * Add a stream to the global list protected by a mutex.
  */
-static int add_stream(struct lttng_consumer_stream *stream,
-		struct lttng_ht *ht)
+int consumer_add_data_stream(struct lttng_consumer_stream *stream)
 {
+	struct lttng_ht *ht = data_ht;
 	int ret = 0;
-	struct consumer_relayd_sock_pair *relayd;
 
 	assert(stream);
 	assert(ht);
@@ -548,6 +563,8 @@ static int add_stream(struct lttng_consumer_stream *stream,
 	DBG3("Adding consumer stream %" PRIu64, stream->key);
 
 	pthread_mutex_lock(&consumer_data.lock);
+	pthread_mutex_lock(&stream->chan->lock);
+	pthread_mutex_lock(&stream->chan->timer_lock);
 	pthread_mutex_lock(&stream->lock);
 	rcu_read_lock();
 
@@ -566,12 +583,6 @@ static int add_stream(struct lttng_consumer_stream *stream,
 	 */
 	lttng_ht_add_u64(consumer_data.stream_list_ht, &stream->node_session_id);
 
-	/* Check and cleanup relayd */
-	relayd = consumer_find_relayd(stream->net_seq_idx);
-	if (relayd != NULL) {
-		uatomic_inc(&relayd->refcount);
-	}
-
 	/*
 	 * When nb_init_stream_left reaches 0, we don't need to trigger any action
 	 * in terms of destroying the associated channel, because the action that
@@ -591,11 +602,18 @@ static int add_stream(struct lttng_consumer_stream *stream,
 
 	rcu_read_unlock();
 	pthread_mutex_unlock(&stream->lock);
+	pthread_mutex_unlock(&stream->chan->timer_lock);
+	pthread_mutex_unlock(&stream->chan->lock);
 	pthread_mutex_unlock(&consumer_data.lock);
 
 	return ret;
 }
 
+void consumer_del_data_stream(struct lttng_consumer_stream *stream)
+{
+	consumer_del_stream(stream, data_ht);
+}
+
 /*
  * Add relayd socket to global consumer data hashtable. RCU read side lock MUST
  * be acquired before calling this.
@@ -709,6 +727,7 @@ int consumer_send_relayd_stream(struct lttng_consumer_stream *stream,
 			goto end;
 		}
 		uatomic_inc(&relayd->refcount);
+		stream->sent_to_relayd = 1;
 	} else {
 		ERR("Stream %" PRIu64 " relayd ID %" PRIu64 " unknown. Can't send it.",
 				stream->key, stream->net_seq_idx);
@@ -838,6 +857,8 @@ struct lttng_consumer_channel *consumer_allocate_channel(uint64_t key,
 	channel->tracefile_size = tracefile_size;
 	channel->tracefile_count = tracefile_count;
 	channel->monitor = monitor;
+	pthread_mutex_init(&channel->lock, NULL);
+	pthread_mutex_init(&channel->timer_lock, NULL);
 
 	/*
 	 * In monitor mode, the streams associated with the channel will be put in
@@ -883,6 +904,8 @@ int consumer_add_channel(struct lttng_consumer_channel *channel,
 	struct lttng_ht_iter iter;
 
 	pthread_mutex_lock(&consumer_data.lock);
+	pthread_mutex_lock(&channel->lock);
+	pthread_mutex_lock(&channel->timer_lock);
 	rcu_read_lock();
 
 	lttng_ht_lookup(consumer_data.channel_ht, &channel->key, &iter);
@@ -899,6 +922,8 @@ int consumer_add_channel(struct lttng_consumer_channel *channel,
 
 end:
 	rcu_read_unlock();
+	pthread_mutex_unlock(&channel->timer_lock);
+	pthread_mutex_unlock(&channel->lock);
 	pthread_mutex_unlock(&consumer_data.lock);
 
 	if (!ret && channel->wait_fd != -1 &&
@@ -1134,7 +1159,7 @@ struct lttng_consumer_local_data *lttng_consumer_create(
 			struct lttng_consumer_local_data *ctx),
 		int (*recv_channel)(struct lttng_consumer_channel *channel),
 		int (*recv_stream)(struct lttng_consumer_stream *stream),
-		int (*update_stream)(int stream_key, uint32_t state))
+		int (*update_stream)(uint64_t stream_key, uint32_t state))
 {
 	int ret;
 	struct lttng_consumer_local_data *ctx;
@@ -1151,6 +1176,7 @@ struct lttng_consumer_local_data *lttng_consumer_create(
 
 	ctx->consumer_error_socket = -1;
 	ctx->consumer_metadata_socket = -1;
+	pthread_mutex_init(&ctx->metadata_socket_lock, NULL);
 	/* assign the callbacks */
 	ctx->on_buffer_ready = buffer_ready;
 	ctx->on_recv_channel = recv_channel;
@@ -1858,6 +1884,7 @@ void consumer_del_metadata_stream(struct lttng_consumer_stream *stream,
 	}
 
 	pthread_mutex_lock(&consumer_data.lock);
+	pthread_mutex_lock(&stream->chan->lock);
 	pthread_mutex_lock(&stream->lock);
 
 	switch (consumer_data.type) {
@@ -1877,6 +1904,13 @@ void consumer_del_metadata_stream(struct lttng_consumer_stream *stream,
 		break;
 	case LTTNG_CONSUMER32_UST:
 	case LTTNG_CONSUMER64_UST:
+		if (stream->monitor) {
+			/* close the write-side in close_metadata */
+			ret = close(stream->ust_metadata_poll_pipe[0]);
+			if (ret < 0) {
+				PERROR("Close UST metadata read-side poll pipe");
+			}
+		}
 		lttng_ustconsumer_del_stream(stream);
 		break;
 	default:
@@ -1945,12 +1979,13 @@ void consumer_del_metadata_stream(struct lttng_consumer_stream *stream,
 end:
 	/*
 	 * Nullify the stream reference so it is not used after deletion. The
-	 * consumer data lock MUST be acquired before being able to check for a
-	 * NULL pointer value.
+	 * channel lock MUST be acquired before being able to check for
+	 * a NULL pointer value.
 	 */
 	stream->chan->metadata_stream = NULL;
 
 	pthread_mutex_unlock(&stream->lock);
+	pthread_mutex_unlock(&stream->chan->lock);
 	pthread_mutex_unlock(&consumer_data.lock);
 
 	if (free_chan) {
@@ -1965,11 +2000,10 @@ free_stream_rcu:
  * Action done with the metadata stream when adding it to the consumer internal
  * data structures to handle it.
  */
-static int add_metadata_stream(struct lttng_consumer_stream *stream,
-		struct lttng_ht *ht)
+int consumer_add_metadata_stream(struct lttng_consumer_stream *stream)
 {
+	struct lttng_ht *ht = metadata_ht;
 	int ret = 0;
-	struct consumer_relayd_sock_pair *relayd;
 	struct lttng_ht_iter iter;
 	struct lttng_ht_node_u64 *node;
 
@@ -1979,6 +2013,8 @@ static int add_metadata_stream(struct lttng_consumer_stream *stream,
 	DBG3("Adding metadata stream %" PRIu64 " to hash table", stream->key);
 
 	pthread_mutex_lock(&consumer_data.lock);
+	pthread_mutex_lock(&stream->chan->lock);
+	pthread_mutex_lock(&stream->chan->timer_lock);
 	pthread_mutex_lock(&stream->lock);
 
 	/*
@@ -1996,12 +2032,6 @@ static int add_metadata_stream(struct lttng_consumer_stream *stream,
 	node = lttng_ht_iter_get_node_u64(&iter);
 	assert(!node);
 
-	/* Find relayd and, if one is found, increment refcount. */
-	relayd = consumer_find_relayd(stream->net_seq_idx);
-	if (relayd != NULL) {
-		uatomic_inc(&relayd->refcount);
-	}
-
 	/*
 	 * When nb_init_stream_left reaches 0, we don't need to trigger any action
 	 * in terms of destroying the associated channel, because the action that
@@ -2030,6 +2060,8 @@ static int add_metadata_stream(struct lttng_consumer_stream *stream,
 	rcu_read_unlock();
 
 	pthread_mutex_unlock(&stream->lock);
+	pthread_mutex_unlock(&stream->chan->lock);
+	pthread_mutex_unlock(&stream->chan->timer_lock);
 	pthread_mutex_unlock(&consumer_data.lock);
 	return ret;
 }
@@ -2175,7 +2207,7 @@ restart:
 					pipe_len = lttng_pipe_read(ctx->consumer_metadata_pipe,
 							&stream, sizeof(stream));
 					if (pipe_len < 0) {
-						ERR("read metadata stream, ret: %ld", pipe_len);
+						ERR("read metadata stream, ret: %zd", pipe_len);
 						/*
 						 * Continue here to handle the rest of the streams.
 						 */
@@ -2192,14 +2224,6 @@ restart:
 					DBG("Adding metadata stream %d to poll set",
 							stream->wait_fd);
 
-					ret = add_metadata_stream(stream, metadata_ht);
-					if (ret) {
-						ERR("Unable to add metadata stream");
-						/* Stream was not setup properly. Continuing. */
-						consumer_del_metadata_stream(stream, NULL);
-						continue;
-					}
-
 					/* Add metadata stream to the global poll events list */
 					lttng_poll_add(&events, stream->wait_fd,
 							LPOLLIN | LPOLLPRI);
@@ -2253,14 +2277,21 @@ restart:
 				DBG("Metadata available on fd %d", pollfd);
 				assert(stream->wait_fd == pollfd);
 
-				len = ctx->on_buffer_ready(stream, ctx);
+				do {
+					len = ctx->on_buffer_ready(stream, ctx);
+					/*
+					 * We don't check the return value here since if we get
+					 * a negative len, it means an error occured thus we
+					 * simply remove it from the poll set and free the
+					 * stream.
+					 */
+				} while (len > 0);
+
 				/* It's ok to have an unavailable sub-buffer */
 				if (len < 0 && len != -EAGAIN && len != -ENODATA) {
 					/* Clean up stream from consumer and free it. */
 					lttng_poll_del(&events, stream->wait_fd);
 					consumer_del_metadata_stream(stream, metadata_ht);
-				} else if (len > 0) {
-					stream->data_read = 1;
 				}
 			}
 
@@ -2304,7 +2335,11 @@ void *consumer_thread_data_poll(void *data)
 		goto end;
 	}
 
-	local_stream = zmalloc(sizeof(struct lttng_consumer_stream));
+	local_stream = zmalloc(sizeof(struct lttng_consumer_stream *));
+	if (local_stream == NULL) {
+		PERROR("local_stream malloc");
+		goto end;
+	}
 
 	while (1) {
 		high_prio = 0;
@@ -2387,7 +2422,7 @@ void *consumer_thread_data_poll(void *data)
 			pipe_readlen = lttng_pipe_read(ctx->consumer_data_pipe,
 					&new_stream, sizeof(new_stream));
 			if (pipe_readlen < 0) {
-				ERR("Consumer data pipe ret %ld", pipe_readlen);
+				ERR("Consumer data pipe ret %zd", pipe_readlen);
 				/* Continue so we can at least handle the current stream(s). */
 				continue;
 			}
@@ -2402,17 +2437,6 @@ void *consumer_thread_data_poll(void *data)
 				continue;
 			}
 
-			ret = add_stream(new_stream, data_ht);
-			if (ret) {
-				ERR("Consumer add stream %" PRIu64 " failed. Continuing",
-						new_stream->key);
-				/*
-				 * At this point, if the add_stream fails, it is not in the
-				 * hash table thus passing the NULL value here.
-				 */
-				consumer_del_stream(new_stream, NULL);
-			}
-
 			/* Continue to update the local streams and handle prio ones */
 			continue;
 		}
@@ -3044,7 +3068,7 @@ void lttng_consumer_init(void)
 int consumer_add_relayd_socket(uint64_t net_seq_idx, int sock_type,
 		struct lttng_consumer_local_data *ctx, int sock,
 		struct pollfd *consumer_sockpoll,
-		struct lttcomm_relayd_sock *relayd_sock, unsigned int sessiond_id)
+		struct lttcomm_relayd_sock *relayd_sock, uint64_t sessiond_id)
 {
 	int fd = -1, ret = -1, relayd_created = 0;
 	enum lttng_error_code ret_code = LTTNG_OK;
@@ -3066,7 +3090,7 @@ int consumer_add_relayd_socket(uint64_t net_seq_idx, int sock_type,
 			ret_code = LTTCOMM_CONSUMERD_ENOMEM;
 			goto error;
 		} else {
-			relayd->sessiond_session_id = (uint64_t) sessiond_id;
+			relayd->sessiond_session_id = sessiond_id;
 			relayd_created = 1;
 		}
 
@@ -3466,3 +3490,23 @@ int consumer_send_status_channel(int sock,
 
 	return lttcomm_send_unix_sock(sock, &msg, sizeof(msg));
 }
+
+/*
+ * Using a maximum stream size with the produced and consumed position of a
+ * stream, computes the new consumed position to be as close as possible to the
+ * maximum possible stream size.
+ *
+ * If maximum stream size is lower than the possible buffer size (produced -
+ * consumed), the consumed_pos given is returned untouched else the new value
+ * is returned.
+ */
+unsigned long consumer_get_consumed_maxsize(unsigned long consumed_pos,
+		unsigned long produced_pos, uint64_t max_stream_size)
+{
+	if (max_stream_size && max_stream_size < (produced_pos - consumed_pos)) {
+		/* Offset from the produced position to get the latest buffers. */
+		return produced_pos - max_stream_size;
+	}
+
+	return consumed_pos;
+}