X-Git-Url: http://git.efficios.com/?a=blobdiff_plain;f=src%2Fcommon%2Fconsumer.c;h=6fd716bdb4cd34ff881659702ab88518097175b7;hb=c7260a81b6c99282da7c59b6124381f92c637658;hp=c47c0ff08239f9c80ccb9a8fae286034def197fc;hpb=36b588eddce05ef840bd247f6a58316925b9a0a2;p=lttng-tools.git

diff --git a/src/common/consumer.c b/src/common/consumer.c
index c47c0ff08..6fd716bdb 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 */
 };
 
 /*
@@ -75,36 +77,46 @@ static struct lttng_ht *metadata_ht;
 static struct lttng_ht *data_ht;
 
 /*
- * Notify a thread pipe to poll back again. This usually means that some global
- * state has changed so we just send back the thread in a poll wait call.
+ * Notify a thread lttng pipe to poll back again. This usually means that some
+ * global state has changed so we just send back the thread in a poll wait
+ * call.
  */
-static void notify_thread_pipe(int wpipe)
+static void notify_thread_lttng_pipe(struct lttng_pipe *pipe)
 {
-	int ret;
+	struct lttng_consumer_stream *null_stream = NULL;
 
-	do {
-		struct lttng_consumer_stream *null_stream = NULL;
+	assert(pipe);
 
-		ret = write(wpipe, &null_stream, sizeof(null_stream));
-	} while (ret < 0 && errno == EINTR);
+	(void) lttng_pipe_write(pipe, &null_stream, sizeof(null_stream));
 }
 
 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;
+	msg.key = key;
 	do {
 		ret = write(ctx->consumer_channel_pipe[1], &msg, sizeof(msg));
 	} 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;
 }
@@ -395,8 +408,8 @@ static void cleanup_relayd(struct consumer_relayd_sock_pair *relayd,
 	 * read of this status which happens AFTER receiving this notify.
 	 */
 	if (ctx) {
-		notify_thread_pipe(ctx->consumer_data_pipe[1]);
-		notify_thread_pipe(ctx->consumer_metadata_pipe[1]);
+		notify_thread_lttng_pipe(ctx->consumer_data_pipe);
+		notify_thread_lttng_pipe(ctx->consumer_metadata_pipe);
 	}
 }
 
@@ -451,6 +464,13 @@ void consumer_del_stream(struct lttng_consumer_stream *stream,
 				PERROR("munmap");
 			}
 		}
+
+		if (stream->wait_fd >= 0) {
+			ret = close(stream->wait_fd);
+			if (ret) {
+				PERROR("close");
+			}
+		}
 		break;
 	case LTTNG_CONSUMER32_UST:
 	case LTTNG_CONSUMER64_UST:
@@ -647,9 +667,6 @@ static int add_stream(struct lttng_consumer_stream *stream,
 		uatomic_inc(&relayd->refcount);
 	}
 
-	/* Update channel refcount once added without error(s). */
-	uatomic_inc(&stream->chan->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
@@ -720,6 +737,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);
 
@@ -888,7 +907,7 @@ int consumer_add_channel(struct lttng_consumer_channel *channel,
 		/* Channel already exist. Ignore the insertion */
 		ERR("Consumer add channel key %" PRIu64 " already exists!",
 			channel->key);
-		ret = -1;
+		ret = LTTNG_ERR_KERN_CHAN_EXIST;
 		goto end;
 	}
 
@@ -900,7 +919,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;
 }
@@ -958,7 +977,7 @@ static int update_poll_array(struct lttng_consumer_local_data *ctx,
 	 * Insert the consumer_data_pipe at the end of the array and don't
 	 * increment i so nb_fd is the number of real FD.
 	 */
-	(*pollfd)[i].fd = ctx->consumer_data_pipe[0];
+	(*pollfd)[i].fd = lttng_pipe_get_readfd(ctx->consumer_data_pipe);
 	(*pollfd)[i].events = POLLIN | POLLPRI;
 	return i;
 }
@@ -1154,26 +1173,11 @@ struct lttng_consumer_local_data *lttng_consumer_create(
 	ctx->on_recv_stream = recv_stream;
 	ctx->on_update_stream = update_stream;
 
-	ret = pipe(ctx->consumer_data_pipe);
-	if (ret < 0) {
-		PERROR("Error creating poll pipe");
+	ctx->consumer_data_pipe = lttng_pipe_open(0);
+	if (!ctx->consumer_data_pipe) {
 		goto error_poll_pipe;
 	}
 
-	/* set read end of the pipe to non-blocking */
-	ret = fcntl(ctx->consumer_data_pipe[0], F_SETFL, O_NONBLOCK);
-	if (ret < 0) {
-		PERROR("fcntl O_NONBLOCK");
-		goto error_poll_fcntl;
-	}
-
-	/* set write end of the pipe to non-blocking */
-	ret = fcntl(ctx->consumer_data_pipe[1], F_SETFL, O_NONBLOCK);
-	if (ret < 0) {
-		PERROR("fcntl O_NONBLOCK");
-		goto error_poll_fcntl;
-	}
-
 	ret = pipe(ctx->consumer_should_quit);
 	if (ret < 0) {
 		PERROR("Error creating recv pipe");
@@ -1192,8 +1196,8 @@ struct lttng_consumer_local_data *lttng_consumer_create(
 		goto error_channel_pipe;
 	}
 
-	ret = utils_create_pipe(ctx->consumer_metadata_pipe);
-	if (ret < 0) {
+	ctx->consumer_metadata_pipe = lttng_pipe_open(0);
+	if (!ctx->consumer_metadata_pipe) {
 		goto error_metadata_pipe;
 	}
 
@@ -1205,16 +1209,15 @@ struct lttng_consumer_local_data *lttng_consumer_create(
 	return ctx;
 
 error_splice_pipe:
-	utils_close_pipe(ctx->consumer_metadata_pipe);
+	lttng_pipe_destroy(ctx->consumer_metadata_pipe);
 error_metadata_pipe:
 	utils_close_pipe(ctx->consumer_channel_pipe);
 error_channel_pipe:
 	utils_close_pipe(ctx->consumer_thread_pipe);
 error_thread_pipe:
 	utils_close_pipe(ctx->consumer_should_quit);
-error_poll_fcntl:
 error_quit_pipe:
-	utils_close_pipe(ctx->consumer_data_pipe);
+	lttng_pipe_destroy(ctx->consumer_data_pipe);
 error_poll_pipe:
 	free(ctx);
 error:
@@ -1240,7 +1243,8 @@ void lttng_consumer_destroy(struct lttng_consumer_local_data *ctx)
 	}
 	utils_close_pipe(ctx->consumer_thread_pipe);
 	utils_close_pipe(ctx->consumer_channel_pipe);
-	utils_close_pipe(ctx->consumer_data_pipe);
+	lttng_pipe_destroy(ctx->consumer_data_pipe);
+	lttng_pipe_destroy(ctx->consumer_metadata_pipe);
 	utils_close_pipe(ctx->consumer_should_quit);
 	utils_close_pipe(ctx->consumer_splice_metadata_pipe);
 
@@ -1411,6 +1415,8 @@ ssize_t lttng_consumer_on_read_subbuffer_mmap(
 				goto end;
 			}
 			outfd = stream->out_fd = ret;
+			/* Reset current size because we just perform a rotation. */
+			stream->tracefile_size_current = 0;
 		}
 		stream->tracefile_size_current += len;
 	}
@@ -1591,6 +1597,8 @@ ssize_t lttng_consumer_on_read_subbuffer_splice(
 				goto end;
 			}
 			outfd = stream->out_fd = ret;
+			/* Reset current size because we just perform a rotation. */
+			stream->tracefile_size_current = 0;
 		}
 		stream->tracefile_size_current += len;
 	}
@@ -1876,6 +1884,13 @@ void consumer_del_metadata_stream(struct lttng_consumer_stream *stream,
 				PERROR("munmap metadata stream");
 			}
 		}
+
+		if (stream->wait_fd >= 0) {
+			ret = close(stream->wait_fd);
+			if (ret < 0) {
+				PERROR("close kernel metadata wait_fd");
+			}
+		}
 		break;
 	case LTTNG_CONSUMER32_UST:
 	case LTTNG_CONSUMER64_UST:
@@ -1945,6 +1960,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.
+	 */
+	stream->chan->metadata_stream = NULL;
+
 	pthread_mutex_unlock(&stream->lock);
 	pthread_mutex_unlock(&consumer_data.lock);
 
@@ -1997,9 +2019,6 @@ static int add_metadata_stream(struct lttng_consumer_stream *stream,
 		uatomic_inc(&relayd->refcount);
 	}
 
-	/* Update channel refcount once added without error(s). */
-	uatomic_inc(&stream->chan->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
@@ -2117,7 +2136,8 @@ void *consumer_thread_metadata_poll(void *data)
 		goto end_poll;
 	}
 
-	ret = lttng_poll_add(&events, ctx->consumer_metadata_pipe[0], LPOLLIN);
+	ret = lttng_poll_add(&events,
+			lttng_pipe_get_readfd(ctx->consumer_metadata_pipe), LPOLLIN);
 	if (ret < 0) {
 		goto end;
 	}
@@ -2155,30 +2175,26 @@ restart:
 				continue;
 			}
 
-			if (pollfd == ctx->consumer_metadata_pipe[0]) {
+			if (pollfd == lttng_pipe_get_readfd(ctx->consumer_metadata_pipe)) {
 				if (revents & (LPOLLERR | LPOLLHUP )) {
 					DBG("Metadata thread pipe hung up");
 					/*
 					 * Remove the pipe from the poll set and continue the loop
 					 * since their might be data to consume.
 					 */
-					lttng_poll_del(&events, ctx->consumer_metadata_pipe[0]);
-					ret = close(ctx->consumer_metadata_pipe[0]);
-					if (ret < 0) {
-						PERROR("close metadata pipe");
-					}
+					lttng_poll_del(&events,
+							lttng_pipe_get_readfd(ctx->consumer_metadata_pipe));
+					lttng_pipe_read_close(ctx->consumer_metadata_pipe);
 					continue;
 				} else if (revents & LPOLLIN) {
-					do {
-						/* Get the stream pointer received */
-						ret = read(pollfd, &stream, sizeof(stream));
-					} while (ret < 0 && errno == EINTR);
-					if (ret < 0 ||
-							ret < sizeof(struct lttng_consumer_stream *)) {
-						PERROR("read metadata stream");
+					ssize_t pipe_len;
+
+					pipe_len = lttng_pipe_read(ctx->consumer_metadata_pipe,
+							&stream, sizeof(stream));
+					if (pipe_len < 0) {
+						ERR("read metadata stream, ret: %ld", pipe_len);
 						/*
-						 * Let's continue here and hope we can still work
-						 * without stopping the consumer. XXX: Should we?
+						 * Continue here to handle the rest of the streams.
 						 */
 						continue;
 					}
@@ -2385,13 +2401,10 @@ void *consumer_thread_data_poll(void *data)
 			ssize_t pipe_readlen;
 
 			DBG("consumer_data_pipe wake up");
-			/* Consume 1 byte of pipe data */
-			do {
-				pipe_readlen = read(ctx->consumer_data_pipe[0], &new_stream,
-						sizeof(new_stream));
-			} while (pipe_readlen == -1 && errno == EINTR);
+			pipe_readlen = lttng_pipe_read(ctx->consumer_data_pipe,
+					&new_stream, sizeof(new_stream));
 			if (pipe_readlen < 0) {
-				PERROR("read consumer data pipe");
+				ERR("Consumer data pipe ret %ld", pipe_readlen);
 				/* Continue so we can at least handle the current stream(s). */
 				continue;
 			}
@@ -2529,10 +2542,7 @@ end:
 	 * only tracked fd in the poll set. The thread will take care of closing
 	 * the read side.
 	 */
-	ret = close(ctx->consumer_metadata_pipe[1]);
-	if (ret < 0) {
-		PERROR("close data pipe");
-	}
+	(void) lttng_pipe_write_close(ctx->consumer_metadata_pipe);
 
 	destroy_data_stream_ht(data_ht);
 
@@ -2691,8 +2701,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;
@@ -2705,12 +2716,38 @@ restart:
 
 						lttng_ht_node_init_u64(&chan->wait_fd_node,
 							chan->wait_fd);
+						rcu_read_lock();
 						lttng_ht_add_unique_u64(channel_ht,
 								&chan->wait_fd_node);
+						rcu_read_unlock();
 						/* Add channel to the global poll events list */
 						lttng_poll_add(&events, chan->wait_fd,
 								LPOLLIN | LPOLLPRI);
 						break;
+					case CONSUMER_CHANNEL_DEL:
+					{
+						rcu_read_lock();
+						chan = consumer_find_channel(key);
+						if (!chan) {
+							rcu_read_unlock();
+							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);
+						}
+						rcu_read_unlock();
+						goto restart;
+					}
 					case CONSUMER_CHANNEL_QUIT:
 						/*
 						 * Remove the pipe from the poll set and continue the loop
@@ -2929,9 +2966,9 @@ end:
 	 * Notify the data poll thread to poll back again and test the
 	 * consumer_quit state that we just set so to quit gracefully.
 	 */
-	notify_thread_pipe(ctx->consumer_data_pipe[1]);
+	notify_thread_lttng_pipe(ctx->consumer_data_pipe);
 
-	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) {
@@ -2941,7 +2978,7 @@ end:
 		}
 	}
 	if (client_socket >= 0) {
-		ret = close(sock);
+		ret = close(client_socket);
 		if (ret < 0) {
 			PERROR("close client_socket sessiond poll");
 		}