X-Git-Url: http://git.efficios.com/?a=blobdiff_plain;f=src%2Fcommon%2Fconsumer.c;h=2dd463c1431f72e9fd1bf35d8551fddad5b28f4a;hb=e56251fcd578c85ad24a36b3768fa6b095fc59e1;hp=b6e440a486fc5995f9a7caca0eac39631bf3340c;hpb=1624d5b7ba86633d36f3a5c86ea1df5d308c4360;p=lttng-tools.git

diff --git a/src/common/consumer.c b/src/common/consumer.c
index b6e440a48..2dd463c14 100644
--- a/src/common/consumer.c
+++ b/src/common/consumer.c
@@ -50,12 +50,14 @@ struct lttng_consumer_global_data consumer_data = {
 
 enum consumer_channel_action {
 	CONSUMER_CHANNEL_ADD,
+	CONSUMER_CHANNEL_DEL,
 	CONSUMER_CHANNEL_QUIT,
 };
 
 struct consumer_channel_msg {
 	enum consumer_channel_action action;
-	struct lttng_consumer_channel *chan;
+	struct lttng_consumer_channel *chan;	/* add */
+	uint64_t key;				/* del */
 };
 
 /*
@@ -91,11 +93,14 @@ static void notify_thread_pipe(int wpipe)
 
 static void notify_channel_pipe(struct lttng_consumer_local_data *ctx,
 		struct lttng_consumer_channel *chan,
+		uint64_t key,
 		enum consumer_channel_action action)
 {
 	struct consumer_channel_msg msg;
 	int ret;
 
+	memset(&msg, 0, sizeof(msg));
+
 	msg.action = action;
 	msg.chan = chan;
 	do {
@@ -103,8 +108,15 @@ static void notify_channel_pipe(struct lttng_consumer_local_data *ctx,
 	} while (ret < 0 && errno == EINTR);
 }
 
+void notify_thread_del_channel(struct lttng_consumer_local_data *ctx,
+		uint64_t key)
+{
+	notify_channel_pipe(ctx, NULL, key, CONSUMER_CHANNEL_DEL);
+}
+
 static int read_channel_pipe(struct lttng_consumer_local_data *ctx,
 		struct lttng_consumer_channel **chan,
+		uint64_t *key,
 		enum consumer_channel_action *action)
 {
 	struct consumer_channel_msg msg;
@@ -116,6 +128,7 @@ static int read_channel_pipe(struct lttng_consumer_local_data *ctx,
 	if (ret > 0) {
 		*action = msg.action;
 		*chan = msg.chan;
+		*key = msg.key;
 	}
 	return ret;
 }
@@ -319,9 +332,9 @@ static void cleanup_relayd_ht(void)
 		destroy_relayd(relayd);
 	}
 
-	lttng_ht_destroy(consumer_data.relayd_ht);
-
 	rcu_read_unlock();
+
+	lttng_ht_destroy(consumer_data.relayd_ht);
 }
 
 /*
@@ -516,8 +529,7 @@ void consumer_del_stream(struct lttng_consumer_stream *stream,
 	}
 	rcu_read_unlock();
 
-	uatomic_dec(&stream->chan->refcount);
-	if (!uatomic_read(&stream->chan->refcount)
+	if (!uatomic_sub_return(&stream->chan->refcount, 1)
 			&& !uatomic_read(&stream->chan->nb_init_stream_left)) {
 		free_chan = stream->chan;
 	}
@@ -659,6 +671,8 @@ static int add_stream(struct lttng_consumer_stream *stream,
 	 * stream.
 	 */
 	if (uatomic_read(&stream->chan->nb_init_stream_left) > 0) {
+		/* Increment refcount before decrementing nb_init_stream_left */
+		cmm_smp_wmb();
 		uatomic_dec(&stream->chan->nb_init_stream_left);
 	}
 
@@ -719,6 +733,8 @@ struct consumer_relayd_sock_pair *consumer_allocate_relayd_sock_pair(
 	obj->net_seq_idx = net_seq_idx;
 	obj->refcount = 0;
 	obj->destroy_flag = 0;
+	obj->control_sock.sock.fd = -1;
+	obj->data_sock.sock.fd = -1;
 	lttng_ht_node_init_u64(&obj->node, obj->net_seq_idx);
 	pthread_mutex_init(&obj->ctrl_sock_mutex, NULL);
 
@@ -783,7 +799,7 @@ static int write_relayd_stream_header(struct lttng_consumer_stream *stream,
 		}
 
 		/* Metadata are always sent on the control socket. */
-		outfd = relayd->control_sock.fd;
+		outfd = relayd->control_sock.sock.fd;
 	} else {
 		/* Set header with stream information */
 		data_hdr.stream_id = htobe64(stream->relayd_stream_id);
@@ -808,7 +824,7 @@ static int write_relayd_stream_header(struct lttng_consumer_stream *stream,
 		++stream->next_net_seq_num;
 
 		/* Set to go on data socket */
-		outfd = relayd->data_sock.fd;
+		outfd = relayd->data_sock.sock.fd;
 	}
 
 error:
@@ -899,7 +915,7 @@ end:
 
 	if (!ret && channel->wait_fd != -1 &&
 			channel->metadata_stream == NULL) {
-		notify_channel_pipe(ctx, channel, CONSUMER_CHANNEL_ADD);
+		notify_channel_pipe(ctx, channel, -1, CONSUMER_CHANNEL_ADD);
 	}
 	return ret;
 }
@@ -1287,93 +1303,6 @@ end:
 	return ret;
 }
 
-/*
- * Create the tracefile on disk.
- *
- * Return 0 on success or else a negative value.
- */
-int lttng_create_output_file(struct lttng_consumer_stream *stream)
-{
-	int ret;
-	char full_path[PATH_MAX];
-	char *path_name_id = NULL;
-	char *path;
-
-	assert(stream);
-	assert(stream->net_seq_idx == (uint64_t) -1ULL);
-
-	ret = snprintf(full_path, sizeof(full_path), "%s/%s",
-			stream->chan->pathname, stream->name);
-	if (ret < 0) {
-		PERROR("snprintf create output file");
-		goto error;
-	}
-
-	/*
-	 * If we split the trace in multiple files, we have to add the tracefile
-	 * current count at the end of the tracefile name
-	 */
-	if (stream->chan->tracefile_size > 0) {
-		ret = asprintf(&path_name_id, "%s_%" PRIu64, full_path,
-				stream->tracefile_count_current);
-		if (ret < 0) {
-			PERROR("Allocating path name ID");
-			goto error;
-		}
-		path = path_name_id;
-	} else {
-		path = full_path;
-	}
-
-	ret = run_as_open(path, O_WRONLY | O_CREAT | O_TRUNC,
-			S_IRWXU | S_IRWXG | S_IRWXO, stream->uid, stream->gid);
-	if (ret < 0) {
-		PERROR("open stream path %s", path);
-		goto error_open;
-	}
-	stream->out_fd = ret;
-	stream->tracefile_size_current = 0;
-
-error_open:
-	free(path_name_id);
-error:
-	return ret;
-}
-
-/*
- * Change the output tracefile according to the tracefile_size and
- * tracefile_count parameters. The stream lock MUST be held before calling this
- * function because we are modifying the stream status.
- *
- * Return 0 on success or else a negative value.
- */
-static int rotate_output_file(struct lttng_consumer_stream *stream)
-{
-	int ret;
-
-	assert(stream);
-	assert(stream->tracefile_size_current);
-
-	ret = close(stream->out_fd);
-	if (ret < 0) {
-		PERROR("Closing tracefile");
-		goto end;
-	}
-
-	if (stream->chan->tracefile_count > 0) {
-		stream->tracefile_count_current =
-			(stream->tracefile_count_current + 1) %
-			stream->chan->tracefile_count;
-	} else {
-		stream->tracefile_count_current++;
-	}
-
-	return lttng_create_output_file(stream);
-
-end:
-	return ret;
-}
-
 /*
  * Mmap the ring buffer, read it and write the data to the tracefile. This is a
  * core function for writing trace buffers to either the local filesystem or
@@ -1488,12 +1417,17 @@ ssize_t lttng_consumer_on_read_subbuffer_mmap(
 		if (stream->chan->tracefile_size > 0 &&
 				(stream->tracefile_size_current + len) >
 				stream->chan->tracefile_size) {
-			ret = rotate_output_file(stream);
+			ret = utils_rotate_stream_file(stream->chan->pathname,
+					stream->name, stream->chan->tracefile_size,
+					stream->chan->tracefile_count, stream->uid, stream->gid,
+					stream->out_fd, &(stream->tracefile_count_current));
 			if (ret < 0) {
 				ERR("Rotating output file");
 				goto end;
 			}
-			outfd = stream->out_fd;
+			outfd = stream->out_fd = ret;
+			/* Reset current size because we just perform a rotation. */
+			stream->tracefile_size_current = 0;
 		}
 		stream->tracefile_size_current += len;
 	}
@@ -1665,12 +1599,17 @@ ssize_t lttng_consumer_on_read_subbuffer_splice(
 		if (stream->chan->tracefile_size > 0 &&
 				(stream->tracefile_size_current + len) >
 				stream->chan->tracefile_size) {
-			ret = rotate_output_file(stream);
+			ret = utils_rotate_stream_file(stream->chan->pathname,
+					stream->name, stream->chan->tracefile_size,
+					stream->chan->tracefile_count, stream->uid, stream->gid,
+					stream->out_fd, &(stream->tracefile_count_current));
 			if (ret < 0) {
 				ERR("Rotating output file");
 				goto end;
 			}
-			outfd = stream->out_fd;
+			outfd = stream->out_fd = ret;
+			/* Reset current size because we just perform a rotation. */
+			stream->tracefile_size_current = 0;
 		}
 		stream->tracefile_size_current += len;
 	}
@@ -2018,8 +1957,7 @@ void consumer_del_metadata_stream(struct lttng_consumer_stream *stream,
 	rcu_read_unlock();
 
 	/* Atomically decrement channel refcount since other threads can use it. */
-	uatomic_dec(&stream->chan->refcount);
-	if (!uatomic_read(&stream->chan->refcount)
+	if (!uatomic_sub_return(&stream->chan->refcount, 1)
 			&& !uatomic_read(&stream->chan->nb_init_stream_left)) {
 		/* Go for channel deletion! */
 		free_chan = stream->chan;
@@ -2089,6 +2027,8 @@ static int add_metadata_stream(struct lttng_consumer_stream *stream,
 	 * stream.
 	 */
 	if (uatomic_read(&stream->chan->nb_init_stream_left) > 0) {
+		/* Increment refcount before decrementing nb_init_stream_left */
+		cmm_smp_wmb();
 		uatomic_dec(&stream->chan->nb_init_stream_left);
 	}
 
@@ -2638,6 +2578,13 @@ void consumer_close_channel_streams(struct lttng_consumer_channel *channel)
 			ht->hash_fct(&channel->key, lttng_ht_seed),
 			ht->match_fct, &channel->key,
 			&iter.iter, stream, node_channel_id.node) {
+		/*
+		 * Protect against teardown with mutex.
+		 */
+		pthread_mutex_lock(&stream->lock);
+		if (cds_lfht_is_node_deleted(&stream->node.node)) {
+			goto next;
+		}
 		switch (consumer_data.type) {
 		case LTTNG_CONSUMER_KERNEL:
 			break;
@@ -2654,6 +2601,8 @@ void consumer_close_channel_streams(struct lttng_consumer_channel *channel)
 			ERR("Unknown consumer_data type");
 			assert(0);
 		}
+	next:
+		pthread_mutex_unlock(&stream->lock);
 	}
 	rcu_read_unlock();
 }
@@ -2761,8 +2710,9 @@ restart:
 					continue;
 				} else if (revents & LPOLLIN) {
 					enum consumer_channel_action action;
+					uint64_t key;
 
-					ret = read_channel_pipe(ctx, &chan, &action);
+					ret = read_channel_pipe(ctx, &chan, &key, &action);
 					if (ret <= 0) {
 						ERR("Error reading channel pipe");
 						continue;
@@ -2781,6 +2731,27 @@ restart:
 						lttng_poll_add(&events, chan->wait_fd,
 								LPOLLIN | LPOLLPRI);
 						break;
+					case CONSUMER_CHANNEL_DEL:
+					{
+						chan = consumer_find_channel(key);
+						if (!chan) {
+							ERR("UST consumer get channel key %" PRIu64 " not found for del channel", key);
+							break;
+						}
+						lttng_poll_del(&events, chan->wait_fd);
+						ret = lttng_ht_del(channel_ht, &iter);
+						assert(ret == 0);
+						consumer_close_channel_streams(chan);
+
+						/*
+						 * Release our own refcount. Force channel deletion even if
+						 * streams were not initialized.
+						 */
+						if (!uatomic_sub_return(&chan->refcount, 1)) {
+							consumer_del_channel(chan);
+						}
+						goto restart;
+					}
 					case CONSUMER_CHANNEL_QUIT:
 						/*
 						 * Remove the pipe from the poll set and continue the loop
@@ -2818,6 +2789,12 @@ restart:
 				ret = lttng_ht_del(channel_ht, &iter);
 				assert(ret == 0);
 				consumer_close_channel_streams(chan);
+
+				/* Release our own refcount */
+				if (!uatomic_sub_return(&chan->refcount, 1)
+						&& !uatomic_read(&chan->nb_init_stream_left)) {
+					consumer_del_channel(chan);
+				}
 			}
 
 			/* Release RCU lock for the channel looked up */
@@ -2995,7 +2972,7 @@ end:
 	 */
 	notify_thread_pipe(ctx->consumer_data_pipe[1]);
 
-	notify_channel_pipe(ctx, NULL, CONSUMER_CHANNEL_QUIT);
+	notify_channel_pipe(ctx, NULL, -1, CONSUMER_CHANNEL_QUIT);
 
 	/* Cleaning up possibly open sockets. */
 	if (sock >= 0) {
@@ -3075,12 +3052,15 @@ void lttng_consumer_init(void)
  */
 int consumer_add_relayd_socket(int net_seq_idx, int sock_type,
 		struct lttng_consumer_local_data *ctx, int sock,
-		struct pollfd *consumer_sockpoll, struct lttcomm_sock *relayd_sock,
-		unsigned int sessiond_id)
+		struct pollfd *consumer_sockpoll,
+		struct lttcomm_relayd_sock *relayd_sock, unsigned int sessiond_id)
 {
 	int fd = -1, ret = -1, relayd_created = 0;
 	enum lttng_error_code ret_code = LTTNG_OK;
-	struct consumer_relayd_sock_pair *relayd;
+	struct consumer_relayd_sock_pair *relayd = NULL;
+
+	assert(ctx);
+	assert(relayd_sock);
 
 	DBG("Consumer adding relayd socket (idx: %d)", net_seq_idx);
 
@@ -3131,11 +3111,11 @@ int consumer_add_relayd_socket(int net_seq_idx, int sock_type,
 	switch (sock_type) {
 	case LTTNG_STREAM_CONTROL:
 		/* Copy received lttcomm socket */
-		lttcomm_copy_sock(&relayd->control_sock, relayd_sock);
-		ret = lttcomm_create_sock(&relayd->control_sock);
+		lttcomm_copy_sock(&relayd->control_sock.sock, &relayd_sock->sock);
+		ret = lttcomm_create_sock(&relayd->control_sock.sock);
 		/* Immediately try to close the created socket if valid. */
-		if (relayd->control_sock.fd >= 0) {
-			if (close(relayd->control_sock.fd)) {
+		if (relayd->control_sock.sock.fd >= 0) {
+			if (close(relayd->control_sock.sock.fd)) {
 				PERROR("close relayd control socket");
 			}
 		}
@@ -3145,7 +3125,10 @@ int consumer_add_relayd_socket(int net_seq_idx, int sock_type,
 		}
 
 		/* Assign new file descriptor */
-		relayd->control_sock.fd = fd;
+		relayd->control_sock.sock.fd = fd;
+		/* Assign version values. */
+		relayd->control_sock.major = relayd_sock->major;
+		relayd->control_sock.minor = relayd_sock->minor;
 
 		/*
 		 * Create a session on the relayd and store the returned id. Lock the
@@ -3173,11 +3156,11 @@ int consumer_add_relayd_socket(int net_seq_idx, int sock_type,
 		break;
 	case LTTNG_STREAM_DATA:
 		/* Copy received lttcomm socket */
-		lttcomm_copy_sock(&relayd->data_sock, relayd_sock);
-		ret = lttcomm_create_sock(&relayd->data_sock);
+		lttcomm_copy_sock(&relayd->data_sock.sock, &relayd_sock->sock);
+		ret = lttcomm_create_sock(&relayd->data_sock.sock);
 		/* Immediately try to close the created socket if valid. */
-		if (relayd->data_sock.fd >= 0) {
-			if (close(relayd->data_sock.fd)) {
+		if (relayd->data_sock.sock.fd >= 0) {
+			if (close(relayd->data_sock.sock.fd)) {
 				PERROR("close relayd data socket");
 			}
 		}
@@ -3187,7 +3170,10 @@ int consumer_add_relayd_socket(int net_seq_idx, int sock_type,
 		}
 
 		/* Assign new file descriptor */
-		relayd->data_sock.fd = fd;
+		relayd->data_sock.sock.fd = fd;
+		/* Assign version values. */
+		relayd->data_sock.major = relayd_sock->major;
+		relayd->data_sock.minor = relayd_sock->minor;
 		break;
 	default:
 		ERR("Unknown relayd socket type (%d)", sock_type);