X-Git-Url: http://git.efficios.com/?p=lttng-tools.git;a=blobdiff_plain;f=src%2Fbin%2Flttng-sessiond%2Fkernel-consumer.c;h=05e447754bb2156a77b6ce3d8dc1555f84973864;hp=6cda088d9e5e7600174476d29962adf2bd532189;hb=ea9a44f03f3f6e144b5f7f312eee2a5f7a849ecb;hpb=bb3c4e70f3dc7550185398d88b07081e97c0ad22 diff --git a/src/bin/lttng-sessiond/kernel-consumer.c b/src/bin/lttng-sessiond/kernel-consumer.c index 6cda088d9..05e447754 100644 --- a/src/bin/lttng-sessiond/kernel-consumer.c +++ b/src/bin/lttng-sessiond/kernel-consumer.c @@ -15,20 +15,23 @@ * Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. */ -#define _GNU_SOURCE #define _LGPL_SOURCE #include #include -#include #include #include +#include #include #include +#include #include "consumer.h" #include "health-sessiond.h" #include "kernel-consumer.h" +#include "notification-thread-commands.h" +#include "session.h" +#include "lttng-sessiond.h" static char *create_channel_path(struct consumer_output *consumer, uid_t uid, gid_t gid) @@ -42,36 +45,53 @@ static char *create_channel_path(struct consumer_output *consumer, /* Get the right path name destination */ if (consumer->type == CONSUMER_DST_LOCAL) { /* Set application path to the destination path */ - ret = snprintf(tmp_path, sizeof(tmp_path), "%s%s", - consumer->dst.trace_path, consumer->subdir); + ret = snprintf(tmp_path, sizeof(tmp_path), "%s%s%s", + consumer->dst.session_root_path, + consumer->chunk_path, + consumer->subdir); if (ret < 0) { PERROR("snprintf kernel channel path"); goto error; + } else if (ret >= sizeof(tmp_path)) { + ERR("Kernel channel path exceeds the maximal allowed length of of %zu bytes (%i bytes required) with path \"%s%s%s\"", + sizeof(tmp_path), ret, + consumer->dst.session_root_path, + consumer->chunk_path, + consumer->subdir); + goto error; } - pathname = strndup(tmp_path, sizeof(tmp_path)); + pathname = lttng_strndup(tmp_path, sizeof(tmp_path)); if (!pathname) { - PERROR("strndup"); + PERROR("lttng_strndup"); goto error; } /* Create directory */ ret = run_as_mkdir_recursive(pathname, S_IRWXU | S_IRWXG, uid, gid); if (ret < 0) { - if (ret != -EEXIST) { + if (errno != EEXIST) { ERR("Trace directory creation error"); goto error; } } DBG3("Kernel local consumer tracefile path: %s", pathname); } else { - ret = snprintf(tmp_path, sizeof(tmp_path), "%s", consumer->subdir); + ret = snprintf(tmp_path, sizeof(tmp_path), "%s%s", + consumer->dst.net.base_dir, + consumer->subdir); if (ret < 0) { PERROR("snprintf kernel metadata path"); goto error; + } else if (ret >= sizeof(tmp_path)) { + ERR("Kernel channel path exceeds the maximal allowed length of of %zu bytes (%i bytes required) with path \"%s%s\"", + sizeof(tmp_path), ret, + consumer->dst.net.base_dir, + consumer->subdir); + goto error; } - pathname = strndup(tmp_path, sizeof(tmp_path)); + pathname = lttng_strndup(tmp_path, sizeof(tmp_path)); if (!pathname) { - PERROR("strndup"); + PERROR("lttng_strndup"); goto error; } DBG3("Kernel network consumer subdir path: %s", pathname); @@ -88,26 +108,33 @@ error: * Sending a single channel to the consumer with command ADD_CHANNEL. */ int kernel_consumer_add_channel(struct consumer_socket *sock, - struct ltt_kernel_channel *channel, struct ltt_kernel_session *session, + struct ltt_kernel_channel *channel, + struct ltt_kernel_session *ksession, unsigned int monitor) { int ret; char *pathname; struct lttcomm_consumer_msg lkm; struct consumer_output *consumer; + enum lttng_error_code status; + struct ltt_session *session; + struct lttng_channel_extended *channel_attr_extended; /* Safety net */ assert(channel); - assert(session); - assert(session->consumer); + assert(ksession); + assert(ksession->consumer); - consumer = session->consumer; + consumer = ksession->consumer; + channel_attr_extended = (struct lttng_channel_extended *) + channel->channel->attr.extended.ptr; DBG("Kernel consumer adding channel %s to kernel consumer", channel->channel->name); if (monitor) { - pathname = create_channel_path(consumer, session->uid, session->gid); + pathname = create_channel_path(consumer, ksession->uid, + ksession->gid); } else { /* Empty path. */ pathname = strdup(""); @@ -118,13 +145,12 @@ int kernel_consumer_add_channel(struct consumer_socket *sock, } /* Prep channel message structure */ - consumer_init_channel_comm_msg(&lkm, - LTTNG_CONSUMER_ADD_CHANNEL, - channel->fd, - session->id, + consumer_init_add_channel_comm_msg(&lkm, + channel->key, + ksession->id, pathname, - session->uid, - session->gid, + ksession->uid, + ksession->gid, consumer->net_seq_index, channel->channel->name, channel->stream_count, @@ -133,7 +159,8 @@ int kernel_consumer_add_channel(struct consumer_socket *sock, channel->channel->attr.tracefile_size, channel->channel->attr.tracefile_count, monitor, - channel->channel->attr.live_timer_interval); + channel->channel->attr.live_timer_interval, + channel_attr_extended->monitor_timer_interval); health_code_update(); @@ -143,6 +170,25 @@ int kernel_consumer_add_channel(struct consumer_socket *sock, } health_code_update(); + rcu_read_lock(); + session = session_find_by_id(ksession->id); + assert(session); + assert(pthread_mutex_trylock(&session->lock)); + assert(session_trylock_list()); + + status = notification_thread_command_add_channel( + notification_thread_handle, session->name, + ksession->uid, ksession->gid, + channel->channel->name, channel->key, + LTTNG_DOMAIN_KERNEL, + channel->channel->attr.subbuf_size * channel->channel->attr.num_subbuf); + rcu_read_unlock(); + if (status != LTTNG_OK) { + ret = -1; + goto error; + } + + channel->published_to_notification_thread = true; error: free(pathname); @@ -151,27 +197,34 @@ error: /* * Sending metadata to the consumer with command ADD_CHANNEL and ADD_STREAM. + * + * The consumer socket lock must be held by the caller. */ int kernel_consumer_add_metadata(struct consumer_socket *sock, - struct ltt_kernel_session *session, unsigned int monitor) + struct ltt_kernel_session *ksession, unsigned int monitor) { int ret; char *pathname; struct lttcomm_consumer_msg lkm; struct consumer_output *consumer; + struct ltt_session *session; + + rcu_read_lock(); /* Safety net */ - assert(session); - assert(session->consumer); + assert(ksession); + assert(ksession->consumer); assert(sock); - DBG("Sending metadata %d to kernel consumer", session->metadata_stream_fd); + DBG("Sending metadata %d to kernel consumer", + ksession->metadata_stream_fd); /* Get consumer output pointer */ - consumer = session->consumer; + consumer = ksession->consumer; if (monitor) { - pathname = create_channel_path(consumer, session->uid, session->gid); + pathname = create_channel_path(consumer, + ksession->uid, ksession->gid); } else { /* Empty path. */ pathname = strdup(""); @@ -181,21 +234,25 @@ int kernel_consumer_add_metadata(struct consumer_socket *sock, goto error; } + session = session_find_by_id(ksession->id); + assert(session); + assert(pthread_mutex_trylock(&session->lock)); + assert(session_trylock_list()); + /* Prep channel message structure */ - consumer_init_channel_comm_msg(&lkm, - LTTNG_CONSUMER_ADD_CHANNEL, - session->metadata->fd, - session->id, + consumer_init_add_channel_comm_msg(&lkm, + ksession->metadata->key, + ksession->id, pathname, - session->uid, - session->gid, + ksession->uid, + ksession->gid, consumer->net_seq_index, DEFAULT_METADATA_NAME, 1, DEFAULT_KERNEL_CHANNEL_OUTPUT, CONSUMER_CHANNEL_TYPE_METADATA, 0, 0, - monitor, 0); + monitor, 0, 0); health_code_update(); @@ -207,17 +264,17 @@ int kernel_consumer_add_metadata(struct consumer_socket *sock, health_code_update(); /* Prep stream message structure */ - consumer_init_stream_comm_msg(&lkm, - LTTNG_CONSUMER_ADD_STREAM, - session->metadata->fd, - session->metadata_stream_fd, - 0); /* CPU: 0 for metadata. */ + consumer_init_add_stream_comm_msg(&lkm, + ksession->metadata->key, + ksession->metadata_stream_fd, + 0 /* CPU: 0 for metadata. */, + session->current_archive_id); health_code_update(); /* Send stream and file descriptor */ ret = consumer_send_stream(sock, consumer, &lkm, - &session->metadata_stream_fd, 1); + &ksession->metadata_stream_fd, 1); if (ret < 0) { goto error; } @@ -225,6 +282,7 @@ int kernel_consumer_add_metadata(struct consumer_socket *sock, health_code_update(); error: + rcu_read_unlock(); free(pathname); return ret; } @@ -232,9 +290,12 @@ error: /* * Sending a single stream to the consumer with command ADD_STREAM. */ +static int kernel_consumer_add_stream(struct consumer_socket *sock, - struct ltt_kernel_channel *channel, struct ltt_kernel_stream *stream, - struct ltt_kernel_session *session, unsigned int monitor) + struct ltt_kernel_channel *channel, + struct ltt_kernel_stream *stream, + struct ltt_kernel_session *session, unsigned int monitor, + uint64_t trace_archive_id) { int ret; struct lttcomm_consumer_msg lkm; @@ -253,11 +314,11 @@ int kernel_consumer_add_stream(struct consumer_socket *sock, consumer = session->consumer; /* Prep stream consumer message */ - consumer_init_stream_comm_msg(&lkm, - LTTNG_CONSUMER_ADD_STREAM, - channel->fd, + consumer_init_add_stream_comm_msg(&lkm, + channel->key, stream->fd, - stream->cpu); + stream->cpu, + trace_archive_id); health_code_update(); @@ -309,22 +370,32 @@ error: /* * Send all stream fds of kernel channel to the consumer. + * + * The consumer socket lock must be held by the caller. */ -int kernel_consumer_send_channel_stream(struct consumer_socket *sock, - struct ltt_kernel_channel *channel, struct ltt_kernel_session *session, +int kernel_consumer_send_channel_streams(struct consumer_socket *sock, + struct ltt_kernel_channel *channel, struct ltt_kernel_session *ksession, unsigned int monitor) { - int ret; + int ret = LTTNG_OK; struct ltt_kernel_stream *stream; + struct ltt_session *session; /* Safety net */ assert(channel); - assert(session); - assert(session->consumer); + assert(ksession); + assert(ksession->consumer); assert(sock); + rcu_read_lock(); + + session = session_find_by_id(ksession->id); + assert(session); + assert(pthread_mutex_trylock(&session->lock)); + assert(session_trylock_list()); + /* Bail out if consumer is disabled */ - if (!session->consumer->enabled) { + if (!ksession->consumer->enabled) { ret = LTTNG_OK; goto error; } @@ -332,31 +403,38 @@ int kernel_consumer_send_channel_stream(struct consumer_socket *sock, DBG("Sending streams of channel %s to kernel consumer", channel->channel->name); - ret = kernel_consumer_add_channel(sock, channel, session, monitor); - if (ret < 0) { - goto error; + if (!channel->sent_to_consumer) { + ret = kernel_consumer_add_channel(sock, channel, ksession, monitor); + if (ret < 0) { + goto error; + } + channel->sent_to_consumer = true; } /* Send streams */ cds_list_for_each_entry(stream, &channel->stream_list.head, list) { - if (!stream->fd) { + if (!stream->fd || stream->sent_to_consumer) { continue; } /* Add stream on the kernel consumer side. */ - ret = kernel_consumer_add_stream(sock, channel, stream, session, - monitor); + ret = kernel_consumer_add_stream(sock, channel, stream, + ksession, monitor, session->current_archive_id); if (ret < 0) { goto error; } + stream->sent_to_consumer = true; } error: + rcu_read_unlock(); return ret; } /* * Send all stream fds of the kernel session to the consumer. + * + * The consumer socket lock must be held by the caller. */ int kernel_consumer_send_session(struct consumer_socket *sock, struct ltt_kernel_session *session) @@ -391,7 +469,7 @@ int kernel_consumer_send_session(struct consumer_socket *sock, /* Send channel and streams of it */ cds_list_for_each_entry(chan, &session->channel_list.head, list) { - ret = kernel_consumer_send_channel_stream(sock, chan, session, + ret = kernel_consumer_send_channel_streams(sock, chan, session, monitor); if (ret < 0) { goto error; @@ -401,7 +479,7 @@ int kernel_consumer_send_session(struct consumer_socket *sock, * Inform the relay that all the streams for the * channel were sent. */ - ret = kernel_consumer_streams_sent(sock, session, chan->fd); + ret = kernel_consumer_streams_sent(sock, session, chan->key); if (ret < 0) { goto error; } @@ -426,11 +504,11 @@ int kernel_consumer_destroy_channel(struct consumer_socket *socket, assert(channel); assert(socket); - DBG("Sending kernel consumer destroy channel key %d", channel->fd); + DBG("Sending kernel consumer destroy channel key %" PRIu64, channel->key); memset(&msg, 0, sizeof(msg)); msg.cmd_type = LTTNG_CONSUMER_DESTROY_CHANNEL; - msg.u.destroy_channel.key = channel->fd; + msg.u.destroy_channel.key = channel->key; pthread_mutex_lock(socket->lock); health_code_update(); @@ -455,11 +533,11 @@ int kernel_consumer_destroy_metadata(struct consumer_socket *socket, assert(metadata); assert(socket); - DBG("Sending kernel consumer destroy channel key %d", metadata->fd); + DBG("Sending kernel consumer destroy channel key %" PRIu64, metadata->key); memset(&msg, 0, sizeof(msg)); msg.cmd_type = LTTNG_CONSUMER_DESTROY_CHANNEL; - msg.u.destroy_channel.key = metadata->fd; + msg.u.destroy_channel.key = metadata->key; pthread_mutex_lock(socket->lock); health_code_update();