X-Git-Url: https://git.lttng.org/?p=lttng-tools.git;a=blobdiff_plain;f=src%2Fcommon%2Fust-consumer%2Fust-consumer.c;h=147fe8aafe899a092a91ee8be58357786d7f074f;hp=01d5ecf6505cc0667a41194d5cb1f07d06e3540b;hb=e9404c27e7cc9d841785e6c4292c1add19fbc1cc;hpb=83f4233dde73280d45fcfb799add96cce32098ce diff --git a/src/common/ust-consumer/ust-consumer.c b/src/common/ust-consumer/ust-consumer.c index 01d5ecf65..147fe8aaf 100644 --- a/src/common/ust-consumer/ust-consumer.c +++ b/src/common/ust-consumer/ust-consumer.c @@ -1,6 +1,7 @@ /* * Copyright (C) 2011 - Julien Desfossez * Mathieu Desnoyers + * Copyright (C) 2017 - Jérémie Galarneau * * This program is free software; you can redistribute it and/or modify * it under the terms of the GNU General Public License, version 2 only, @@ -510,7 +511,7 @@ error_open: } /* Try to rmdir all directories under shm_path root. */ if (channel->root_shm_path[0]) { - (void) run_as_recursive_rmdir(channel->root_shm_path, + (void) run_as_rmdir_recursive(channel->root_shm_path, channel->uid, channel->gid); } free(stream_fds); @@ -1501,8 +1502,17 @@ int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, consumer_timer_switch_start(channel, attr.switch_timer_interval); attr.switch_timer_interval = 0; } else { + int monitor_start_ret; + consumer_timer_live_start(channel, msg.u.ask_channel.live_timer_interval); + monitor_start_ret = consumer_timer_monitor_start( + channel, + msg.u.ask_channel.monitor_timer_interval); + if (monitor_start_ret < 0) { + ERR("Starting channel monitoring timer failed"); + goto end_channel_error; + } } health_code_update(); @@ -1525,6 +1535,9 @@ int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, if (channel->live_timer_enabled == 1) { consumer_timer_live_stop(channel); } + if (channel->monitor_timer_enabled == 1) { + consumer_timer_monitor_stop(channel); + } goto end_channel_error; } @@ -1857,6 +1870,51 @@ int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, break; } + case LTTNG_CONSUMER_SET_CHANNEL_MONITOR_PIPE: + { + int channel_monitor_pipe; + + ret_code = LTTCOMM_CONSUMERD_SUCCESS; + /* Successfully received the command's type. */ + ret = consumer_send_status_msg(sock, ret_code); + if (ret < 0) { + goto error_fatal; + } + + ret = lttcomm_recv_fds_unix_sock(sock, &channel_monitor_pipe, + 1); + if (ret != sizeof(channel_monitor_pipe)) { + ERR("Failed to receive channel monitor pipe"); + goto error_fatal; + } + + DBG("Received channel monitor pipe (%d)", channel_monitor_pipe); + ret = consumer_timer_thread_set_channel_monitor_pipe( + channel_monitor_pipe); + if (!ret) { + int flags; + + ret_code = LTTCOMM_CONSUMERD_SUCCESS; + /* Set the pipe as non-blocking. */ + ret = fcntl(channel_monitor_pipe, F_GETFL, 0); + if (ret == -1) { + PERROR("fcntl get flags of the channel monitoring pipe"); + goto error_fatal; + } + flags = ret; + + ret = fcntl(channel_monitor_pipe, F_SETFL, + flags | O_NONBLOCK); + if (ret == -1) { + PERROR("fcntl set O_NONBLOCK flag of the channel monitoring pipe"); + goto error_fatal; + } + DBG("Channel monitor pipe set as non-blocking"); + } else { + ret_code = LTTCOMM_CONSUMERD_ALREADY_SET; + } + goto end_msg_sessiond; + } default: break; } @@ -1938,7 +1996,7 @@ void *lttng_ustctl_get_mmap_base(struct lttng_consumer_stream *stream) } /* - * Take a snapshot for a specific fd + * Take a snapshot for a specific stream. * * Returns 0 on success, < 0 on error */ @@ -1950,6 +2008,20 @@ int lttng_ustconsumer_take_snapshot(struct lttng_consumer_stream *stream) return ustctl_snapshot(stream->ustream); } +/* + * Sample consumed and produced positions for a specific stream. + * + * Returns 0 on success, < 0 on error. + */ +int lttng_ustconsumer_sample_snapshot_positions( + struct lttng_consumer_stream *stream) +{ + assert(stream); + assert(stream->ustream); + + return ustctl_snapshot_sample_positions(stream->ustream); +} + /* * Get the produced position * @@ -2067,7 +2139,7 @@ void lttng_ustconsumer_free_channel(struct lttng_consumer_channel *chan) ustctl_destroy_channel(chan->uchan); /* Try to rmdir all directories under shm_path root. */ if (chan->root_shm_path[0]) { - (void) run_as_recursive_rmdir(chan->root_shm_path, + (void) run_as_rmdir_recursive(chan->root_shm_path, chan->uid, chan->gid); } free(chan->stream_fds); @@ -2273,10 +2345,10 @@ int lttng_ustconsumer_sync_metadata(struct lttng_consumer_local_data *ctx, * because we locked the metadata thread. */ ret = lttng_ustconsumer_request_metadata(ctx, metadata->chan, 0, 0); + pthread_mutex_lock(&metadata->lock); if (ret < 0) { goto end; } - pthread_mutex_lock(&metadata->lock); ret = commit_one_metadata_packet(metadata); if (ret <= 0) { @@ -2496,6 +2568,8 @@ retry: index.offset = htobe64(stream->out_fd_offset); ret = get_index_values(&index, ustream); if (ret < 0) { + err = ustctl_put_subbuf(ustream); + assert(err == 0); goto end; } @@ -2503,6 +2577,8 @@ retry: ret = update_stream_stats(stream); if (ret < 0) { PERROR("kernctl_get_events_discarded"); + err = ustctl_put_subbuf(ustream); + assert(err == 0); goto end; } } else { @@ -2620,14 +2696,18 @@ int lttng_ustconsumer_on_recv_stream(struct lttng_consumer_stream *stream) stream->tracefile_size_current = 0; if (!stream->metadata_flag) { - ret = index_create_file(stream->chan->pathname, + struct lttng_index_file *index_file; + + index_file = lttng_index_file_create(stream->chan->pathname, stream->name, stream->uid, stream->gid, stream->chan->tracefile_size, - stream->tracefile_count_current); - if (ret < 0) { + stream->tracefile_count_current, + CTF_INDEX_MAJOR, CTF_INDEX_MINOR); + if (!index_file) { goto error; } - stream->index_fd = ret; + assert(!stream->index_file); + stream->index_file = index_file; } } ret = 0;