X-Git-Url: http://git.efficios.com/?a=blobdiff_plain;f=src%2Fcommon%2Fconsumer%2Fconsumer.c;h=07de6757849f5a01e4c19d4bde771cca11ffe462;hb=15055ce5780557ef1053fcb4bd91a0746ffa30a0;hp=a8c369904ec35688a3467ab383c9c8f05cc23e7e;hpb=2527bf859786192dd530b408717960495f48d208;p=lttng-tools.git diff --git a/src/common/consumer/consumer.c b/src/common/consumer/consumer.c index a8c369904..07de67578 100644 --- a/src/common/consumer/consumer.c +++ b/src/common/consumer/consumer.c @@ -541,6 +541,16 @@ void consumer_del_stream_for_metadata(struct lttng_consumer_stream *stream) consumer_stream_destroy(stream, metadata_ht); } +void consumer_stream_update_channel_attributes( + struct lttng_consumer_stream *stream, + struct lttng_consumer_channel *channel) +{ + stream->channel_read_only_attributes.tracefile_size = + channel->tracefile_size; + memcpy(stream->channel_read_only_attributes.path, channel->pathname, + sizeof(stream->channel_read_only_attributes.path)); +} + struct lttng_consumer_stream *consumer_allocate_stream(uint64_t channel_key, uint64_t stream_key, enum lttng_consumer_stream_state state, @@ -630,10 +640,9 @@ end: /* * Add a stream to the global list protected by a mutex. */ -int consumer_add_data_stream(struct lttng_consumer_stream *stream) +void consumer_add_data_stream(struct lttng_consumer_stream *stream) { struct lttng_ht *ht = data_ht; - int ret = 0; assert(stream); assert(ht); @@ -683,8 +692,6 @@ int consumer_add_data_stream(struct lttng_consumer_stream *stream) 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) @@ -1077,7 +1084,7 @@ int consumer_add_channel(struct lttng_consumer_channel *channel, */ static int update_poll_array(struct lttng_consumer_local_data *ctx, struct pollfd **pollfd, struct lttng_consumer_stream **local_stream, - struct lttng_ht *ht) + struct lttng_ht *ht, int *nb_inactive_fd) { int i = 0; struct lttng_ht_iter iter; @@ -1089,6 +1096,7 @@ static int update_poll_array(struct lttng_consumer_local_data *ctx, assert(local_stream); DBG("Updating poll fd array"); + *nb_inactive_fd = 0; rcu_read_lock(); cds_lfht_for_each_entry(ht->ht, &iter.iter, stream, node.node) { /* @@ -1099,9 +1107,14 @@ static int update_poll_array(struct lttng_consumer_local_data *ctx, * just after the check. However, this is OK since the stream(s) will * be deleted once the thread is notified that the end point state has * changed where this function will be called back again. + * + * We track the number of inactive FDs because they still need to be + * closed by the polling thread after a wakeup on the data_pipe or + * metadata_pipe. */ if (stream->state != LTTNG_CONSUMER_ACTIVE_STREAM || stream->endpoint_status == CONSUMER_ENDPOINT_INACTIVE) { + (*nb_inactive_fd)++; continue; } /* @@ -1958,6 +1971,25 @@ end: return written; } +/* + * Sample the snapshot positions for a specific fd + * + * Returns 0 on success, < 0 on error + */ +int lttng_consumer_sample_snapshot_positions(struct lttng_consumer_stream *stream) +{ + switch (consumer_data.type) { + case LTTNG_CONSUMER_KERNEL: + return lttng_kconsumer_sample_snapshot_positions(stream); + case LTTNG_CONSUMER32_UST: + case LTTNG_CONSUMER64_UST: + return lttng_ustconsumer_sample_snapshot_positions(stream); + default: + ERR("Unknown consumer_data type"); + assert(0); + return -ENOSYS; + } +} /* * Take a snapshot for a specific fd * @@ -1999,6 +2031,27 @@ int lttng_consumer_get_produced_snapshot(struct lttng_consumer_stream *stream, } } +/* + * Get the consumed position (free-running counter position in bytes). + * + * Returns 0 on success, < 0 on error + */ +int lttng_consumer_get_consumed_snapshot(struct lttng_consumer_stream *stream, + unsigned long *pos) +{ + switch (consumer_data.type) { + case LTTNG_CONSUMER_KERNEL: + return lttng_kconsumer_get_consumed_snapshot(stream, pos); + case LTTNG_CONSUMER32_UST: + case LTTNG_CONSUMER64_UST: + return lttng_ustconsumer_get_consumed_snapshot(stream, pos); + default: + ERR("Unknown consumer_data type"); + assert(0); + return -ENOSYS; + } +} + int lttng_consumer_recv_cmd(struct lttng_consumer_local_data *ctx, int sock, struct pollfd *consumer_sockpoll) { @@ -2106,10 +2159,9 @@ void consumer_del_metadata_stream(struct lttng_consumer_stream *stream, * Action done with the metadata stream when adding it to the consumer internal * data structures to handle it. */ -int consumer_add_metadata_stream(struct lttng_consumer_stream *stream) +void 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; @@ -2153,7 +2205,7 @@ int 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); /* @@ -2169,7 +2221,6 @@ int consumer_add_metadata_stream(struct lttng_consumer_stream *stream) pthread_mutex_unlock(&stream->chan->lock); pthread_mutex_unlock(&stream->chan->timer_lock); pthread_mutex_unlock(&consumer_data.lock); - return ret; } /* @@ -2456,7 +2507,9 @@ 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; ssize_t len; @@ -2495,17 +2548,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"); @@ -2513,7 +2568,7 @@ void *consumer_thread_data_poll(void *data) goto end; } ret = update_poll_array(ctx, &pollfd, local_stream, - data_ht); + data_ht, &nb_inactive_fd); if (ret < 0) { ERR("Error in allocating pollfd or local_outfds"); lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_POLL_ERROR); @@ -2526,18 +2581,19 @@ void *consumer_thread_data_poll(void *data) pthread_mutex_unlock(&consumer_data.lock); /* No FDs and consumer_quit, consumer_cleanup the thread */ - if (nb_fd == 0 && CMM_LOAD_SHARED(consumer_quit) == 1) { + if (nb_fd == 0 && nb_inactive_fd == 0 && + CMM_LOAD_SHARED(consumer_quit) == 1) { err = 0; /* All is OK */ goto end; } /* 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) { @@ -3779,3 +3835,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); + } +}