X-Git-Url: https://git.lttng.org/?a=blobdiff_plain;f=src%2Fcommon%2Fconsumer.c;h=7acb8560ca4e07bfdccc89518477dad6a8d79051;hb=e5d1a9b33aa0920bbd9f6948bd0676156da67c61;hp=6e99a7f52d0d24ac1473db3f1c2bf12880a8df78;hpb=75d83e506938a3e673ad19bb11838ee88e977e72;p=lttng-tools.git

diff --git a/src/common/consumer.c b/src/common/consumer.c
index 6e99a7f52..7acb8560c 100644
--- a/src/common/consumer.c
+++ b/src/common/consumer.c
@@ -457,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,
@@ -485,6 +498,7 @@ struct lttng_consumer_stream *consumer_allocate_stream(uint64_t channel_key,
 	stream->key = stream_key;
 	stream->out_fd = -1;
 	stream->out_fd_offset = 0;
+	stream->output_written = 0;
 	stream->state = state;
 	stream->uid = uid;
 	stream->gid = gid;
@@ -539,9 +553,9 @@ 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;
 
 	assert(stream);
@@ -551,6 +565,7 @@ static int add_stream(struct lttng_consumer_stream *stream,
 
 	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();
 
@@ -588,12 +603,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.
@@ -838,6 +859,7 @@ struct lttng_consumer_channel *consumer_allocate_channel(uint64_t key,
 	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
@@ -884,6 +906,7 @@ int consumer_add_channel(struct lttng_consumer_channel *channel,
 
 	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);
@@ -900,6 +923,7 @@ 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);
 
@@ -1311,6 +1335,7 @@ ssize_t lttng_consumer_on_read_subbuffer_mmap(
 	if (stream->net_seq_idx != (uint64_t) -1ULL) {
 		relayd = consumer_find_relayd(stream->net_seq_idx);
 		if (relayd == NULL) {
+			ret = -EPIPE;
 			goto end;
 		}
 	}
@@ -1320,28 +1345,31 @@ ssize_t lttng_consumer_on_read_subbuffer_mmap(
 	case LTTNG_CONSUMER_KERNEL:
 		mmap_base = stream->mmap_base;
 		ret = kernctl_get_mmap_read_offset(stream->wait_fd, &mmap_offset);
+		if (ret != 0) {
+			PERROR("tracer ctl get_mmap_read_offset");
+			written = -errno;
+			goto end;
+		}
 		break;
 	case LTTNG_CONSUMER32_UST:
 	case LTTNG_CONSUMER64_UST:
 		mmap_base = lttng_ustctl_get_mmap_base(stream);
 		if (!mmap_base) {
 			ERR("read mmap get mmap base for stream %s", stream->name);
-			written = -1;
+			written = -EPERM;
 			goto end;
 		}
 		ret = lttng_ustctl_get_mmap_read_offset(stream, &mmap_offset);
-
+		if (ret != 0) {
+			PERROR("tracer ctl get_mmap_read_offset");
+			written = ret;
+			goto end;
+		}
 		break;
 	default:
 		ERR("Unknown consumer_data type");
 		assert(0);
 	}
-	if (ret != 0) {
-		errno = -ret;
-		PERROR("tracer ctl get_mmap_read_offset");
-		written = ret;
-		goto end;
-	}
 
 	/* Handle stream on the relayd if the output is on the network */
 	if (relayd) {
@@ -1404,6 +1432,8 @@ ssize_t lttng_consumer_on_read_subbuffer_mmap(
 			outfd = stream->out_fd = ret;
 			/* Reset current size because we just perform a rotation. */
 			stream->tracefile_size_current = 0;
+			stream->out_fd_offset = 0;
+			orig_offset = 0;
 		}
 		stream->tracefile_size_current += len;
 	}
@@ -1422,7 +1452,7 @@ ssize_t lttng_consumer_on_read_subbuffer_mmap(
 			 */
 			DBG("Error in file write mmap");
 			if (written == 0) {
-				written = ret;
+				written = -errno;
 			}
 			/* Socket operation failed. We consider the relayd dead */
 			if (errno == EPIPE || errno == EINVAL) {
@@ -1446,6 +1476,7 @@ ssize_t lttng_consumer_on_read_subbuffer_mmap(
 					SYNC_FILE_RANGE_WRITE);
 			stream->out_fd_offset += ret;
 		}
+		stream->output_written += ret;
 		written += ret;
 	}
 	lttng_consumer_sync_trace_file(stream, orig_offset);
@@ -1510,6 +1541,7 @@ ssize_t lttng_consumer_on_read_subbuffer_splice(
 	if (stream->net_seq_idx != (uint64_t) -1ULL) {
 		relayd = consumer_find_relayd(stream->net_seq_idx);
 		if (relayd == NULL) {
+			ret = -EPIPE;
 			goto end;
 		}
 	}
@@ -1586,6 +1618,8 @@ ssize_t lttng_consumer_on_read_subbuffer_splice(
 			outfd = stream->out_fd = ret;
 			/* Reset current size because we just perform a rotation. */
 			stream->tracefile_size_current = 0;
+			stream->out_fd_offset = 0;
+			orig_offset = 0;
 		}
 		stream->tracefile_size_current += len;
 	}
@@ -1656,6 +1690,7 @@ ssize_t lttng_consumer_on_read_subbuffer_splice(
 					SYNC_FILE_RANGE_WRITE);
 			stream->out_fd_offset += ret_splice;
 		}
+		stream->output_written += ret_splice;
 		written += ret_splice;
 	}
 	lttng_consumer_sync_trace_file(stream, orig_offset);
@@ -1956,8 +1991,8 @@ 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;
 
@@ -1977,9 +2012,9 @@ 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 lttng_ht_iter iter;
 	struct lttng_ht_node_u64 *node;
@@ -1991,6 +2026,7 @@ static int add_metadata_stream(struct lttng_consumer_stream *stream,
 
 	pthread_mutex_lock(&consumer_data.lock);
 	pthread_mutex_lock(&stream->chan->lock);
+	pthread_mutex_lock(&stream->chan->timer_lock);
 	pthread_mutex_lock(&stream->lock);
 
 	/*
@@ -2037,6 +2073,7 @@ static int add_metadata_stream(struct lttng_consumer_stream *stream,
 
 	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;
 }
@@ -2182,7 +2219,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.
 						 */
@@ -2199,14 +2236,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);
@@ -2405,7 +2434,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;
 			}
@@ -2420,17 +2449,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;
 		}
@@ -3385,6 +3403,15 @@ int consumer_data_pending(uint64_t id)
 		 */
 		ret = cds_lfht_is_node_deleted(&stream->node.node);
 		if (!ret) {
+			/*
+			 * An empty output file is not valid. We need at least one packet
+			 * generated per stream, even if it contains no event, so it
+			 * contains at least one packet header.
+			 */
+			if (stream->output_written == 0) {
+				pthread_mutex_unlock(&stream->lock);
+				goto data_pending;
+			}
 			/* Check the stream if there is data in the buffers. */
 			ret = data_pending(stream);
 			if (ret == 1) {