X-Git-Url: http://git.efficios.com/?a=blobdiff_plain;f=src%2Fcommon%2Fconsumer%2Fconsumer.c;h=6b920b6b1ef63ee5d6e50cc4092b89bc498b7a9d;hb=261de6373c70dcd52421642db5486747e9c10bae;hp=3755932ab59cf8ad177607b5ad64c95a9c76bcae;hpb=9a2fcf78ea85b9ad74405db4605338896d70b46f;p=lttng-tools.git

diff --git a/src/common/consumer/consumer.c b/src/common/consumer/consumer.c
index 3755932ab..6b920b6b1 100644
--- a/src/common/consumer/consumer.c
+++ b/src/common/consumer/consumer.c
@@ -2155,7 +2155,7 @@ void consumer_add_metadata_stream(struct lttng_consumer_stream *stream)
 
 	lttng_ht_add_unique_u64(ht, &stream->node);
 
-	lttng_ht_add_unique_u64(consumer_data.stream_per_chan_id_ht,
+	lttng_ht_add_u64(consumer_data.stream_per_chan_id_ht,
 		&stream->node_channel_id);
 
 	/*
@@ -2457,7 +2457,7 @@ void *consumer_thread_data_poll(void *data)
 	/* local view of the streams */
 	struct lttng_consumer_stream **local_stream = NULL, *new_stream = NULL;
 	/* local view of consumer_data.fds_count */
-	int nb_fd = 0;
+	int nb_fd = 0, nb_pipes_fd;
 	/* Number of FDs with CONSUMER_ENDPOINT_INACTIVE but still open. */
 	int nb_inactive_fd = 0;
 	struct lttng_consumer_local_data *ctx = data;
@@ -2498,17 +2498,19 @@ void *consumer_thread_data_poll(void *data)
 			local_stream = NULL;
 
 			/*
-			 * Allocate for all fds +1 for the consumer_data_pipe and +1 for
-			 * wake up pipe.
+			 * Allocate for all fds + 2:
+			 *   +1 for the consumer_data_pipe
+			 *   +1 for wake up pipe
 			 */
-			pollfd = zmalloc((consumer_data.stream_count + 2) * sizeof(struct pollfd));
+			nb_pipes_fd = 2;
+			pollfd = zmalloc((consumer_data.stream_count + nb_pipes_fd) * sizeof(struct pollfd));
 			if (pollfd == NULL) {
 				PERROR("pollfd malloc");
 				pthread_mutex_unlock(&consumer_data.lock);
 				goto end;
 			}
 
-			local_stream = zmalloc((consumer_data.stream_count + 2) *
+			local_stream = zmalloc((consumer_data.stream_count + nb_pipes_fd) *
 					sizeof(struct lttng_consumer_stream *));
 			if (local_stream == NULL) {
 				PERROR("local_stream malloc");
@@ -2536,12 +2538,12 @@ void *consumer_thread_data_poll(void *data)
 		}
 		/* poll on the array of fds */
 	restart:
-		DBG("polling on %d fd", nb_fd + 2);
+		DBG("polling on %d fd", nb_fd + nb_pipes_fd);
 		if (testpoint(consumerd_thread_data_poll)) {
 			goto end;
 		}
 		health_poll_entry();
-		num_rdy = poll(pollfd, nb_fd + 2, -1);
+		num_rdy = poll(pollfd, nb_fd + nb_pipes_fd, -1);
 		health_poll_exit();
 		DBG("poll num_rdy : %d", num_rdy);
 		if (num_rdy == -1) {
@@ -3783,3 +3785,108 @@ unsigned long consumer_get_consume_start_pos(unsigned long consumed_pos,
 	}
 	return start_pos;
 }
+
+static
+int rotate_rename_local(const char *old_path, const char *new_path,
+		uid_t uid, gid_t gid)
+{
+	int ret;
+
+	assert(old_path);
+	assert(new_path);
+
+	ret = utils_mkdir_recursive(new_path, S_IRWXU | S_IRWXG, uid, gid);
+	if (ret < 0) {
+		ERR("Create directory on rotate");
+		goto end;
+	}
+
+	ret = rename(old_path, new_path);
+	if (ret < 0 && errno != ENOENT) {
+		PERROR("Rename completed rotation chunk");
+		goto end;
+	}
+
+	ret = 0;
+end:
+	return ret;
+}
+
+static
+int rotate_rename_relay(const char *old_path, const char *new_path,
+		uint64_t relayd_id)
+{
+	int ret;
+	struct consumer_relayd_sock_pair *relayd;
+
+	relayd = consumer_find_relayd(relayd_id);
+	if (!relayd) {
+		ERR("Failed to find relayd while running rotate_rename_relay command");
+		ret = -1;
+		goto end;
+	}
+
+	pthread_mutex_lock(&relayd->ctrl_sock_mutex);
+	ret = relayd_rotate_rename(&relayd->control_sock, old_path, new_path);
+	pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
+end:
+	return ret;
+}
+
+int lttng_consumer_rotate_rename(const char *old_path, const char *new_path,
+		uid_t uid, gid_t gid, uint64_t relayd_id)
+{
+	if (relayd_id != -1ULL) {
+		return rotate_rename_relay(old_path, new_path, relayd_id);
+	} else {
+		return rotate_rename_local(old_path, new_path, uid, gid);
+	}
+}
+
+static
+int mkdir_local(const char *path, uid_t uid, gid_t gid)
+{
+	int ret;
+
+	ret = utils_mkdir_recursive(path, S_IRWXU | S_IRWXG, uid, gid);
+	if (ret < 0) {
+		/* utils_mkdir_recursive logs an error. */
+		goto end;
+	}
+
+	ret = 0;
+end:
+	return ret;
+}
+
+static
+int mkdir_relay(const char *path, uint64_t relayd_id)
+{
+	int ret;
+	struct consumer_relayd_sock_pair *relayd;
+
+	relayd = consumer_find_relayd(relayd_id);
+	if (!relayd) {
+		ERR("Failed to find relayd");
+		ret = -1;
+		goto end;
+	}
+
+	pthread_mutex_lock(&relayd->ctrl_sock_mutex);
+	ret = relayd_mkdir(&relayd->control_sock, path);
+	pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
+
+end:
+	return ret;
+
+}
+
+int lttng_consumer_mkdir(const char *path, uid_t uid, gid_t gid,
+		uint64_t relayd_id)
+{
+	if (relayd_id != -1ULL) {
+		return mkdir_relay(path, relayd_id);
+	} else {
+		return mkdir_local(path, uid, gid);
+	}
+}