X-Git-Url: http://git.efficios.com/?p=lttng-tools.git;a=blobdiff_plain;f=src%2Fbin%2Flttng-sessiond%2Fkernel-consumer.c;h=0f59aa24f51decc781c760ddef1007399851355d;hp=70759fb36d90149d8964967d3b20608c60bcc9ff;hb=4a171f3e17f4799d52e13d385f195fd459b8d008;hpb=8782cc7477fae212607b9fd6395a4b2e2d3357ed diff --git a/src/bin/lttng-sessiond/kernel-consumer.c b/src/bin/lttng-sessiond/kernel-consumer.c index 70759fb36..0f59aa24f 100644 --- a/src/bin/lttng-sessiond/kernel-consumer.c +++ b/src/bin/lttng-sessiond/kernel-consumer.c @@ -15,19 +15,22 @@ * Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. */ -#define _GNU_SOURCE +#define _LGPL_SOURCE #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) @@ -47,12 +50,16 @@ static char *create_channel_path(struct consumer_output *consumer, PERROR("snprintf kernel channel path"); goto error; } - pathname = strndup(tmp_path, sizeof(tmp_path)); + pathname = lttng_strndup(tmp_path, sizeof(tmp_path)); + if (!pathname) { + 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; } @@ -64,7 +71,11 @@ static char *create_channel_path(struct consumer_output *consumer, PERROR("snprintf kernel metadata path"); goto error; } - pathname = strndup(tmp_path, sizeof(tmp_path)); + pathname = lttng_strndup(tmp_path, sizeof(tmp_path)); + if (!pathname) { + PERROR("lttng_strndup"); + goto error; + } DBG3("Kernel network consumer subdir path: %s", pathname); } @@ -79,43 +90,50 @@ 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); - if (!pathname) { - ret = -1; - goto error; - } + pathname = create_channel_path(consumer, ksession->uid, + ksession->gid); } else { /* Empty path. */ - pathname = ""; + pathname = strdup(""); + } + if (!pathname) { + ret = -1; + goto error; } /* Prep channel message structure */ consumer_init_channel_comm_msg(&lkm, LTTNG_CONSUMER_ADD_CHANNEL, channel->fd, - session->id, + ksession->id, pathname, - session->uid, - session->gid, + ksession->uid, + ksession->gid, consumer->net_seq_index, channel->channel->name, channel->stream_count, @@ -124,7 +142,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(); @@ -134,8 +153,26 @@ int kernel_consumer_add_channel(struct consumer_socket *sock, } health_code_update(); + rcu_read_lock(); + session = session_find_by_id(ksession->id); + assert(session); + + status = notification_thread_command_add_channel( + notification_thread_handle, session->name, + ksession->uid, ksession->gid, + channel->channel->name, channel->fd, + 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); return ret; } @@ -162,13 +199,13 @@ int kernel_consumer_add_metadata(struct consumer_socket *sock, if (monitor) { pathname = create_channel_path(consumer, session->uid, session->gid); - if (!pathname) { - ret = -1; - goto error; - } } else { /* Empty path. */ - pathname = ""; + pathname = strdup(""); + } + if (!pathname) { + ret = -1; + goto error; } /* Prep channel message structure */ @@ -185,7 +222,7 @@ int kernel_consumer_add_metadata(struct consumer_socket *sock, DEFAULT_KERNEL_CHANNEL_OUTPUT, CONSUMER_CHANNEL_TYPE_METADATA, 0, 0, - monitor, 0); + monitor, 0, 0); health_code_update(); @@ -215,6 +252,7 @@ int kernel_consumer_add_metadata(struct consumer_socket *sock, health_code_update(); error: + free(pathname); return ret; } @@ -262,6 +300,40 @@ error: return ret; } +/* + * Sending the notification that all streams were sent with STREAMS_SENT. + */ +int kernel_consumer_streams_sent(struct consumer_socket *sock, + struct ltt_kernel_session *session, uint64_t channel_key) +{ + int ret; + struct lttcomm_consumer_msg lkm; + struct consumer_output *consumer; + + assert(sock); + assert(session); + + DBG("Sending streams_sent"); + /* Get consumer output pointer */ + consumer = session->consumer; + + /* Prep stream consumer message */ + consumer_init_streams_sent_comm_msg(&lkm, + LTTNG_CONSUMER_STREAMS_SENT, + channel_key, consumer->net_seq_index); + + health_code_update(); + + /* Send stream and file descriptor */ + ret = consumer_send_msg(sock, &lkm); + if (ret < 0) { + goto error; + } + +error: + return ret; +} + /* * Send all stream fds of kernel channel to the consumer. */ @@ -269,7 +341,7 @@ int kernel_consumer_send_channel_stream(struct consumer_socket *sock, struct ltt_kernel_channel *channel, struct ltt_kernel_session *session, unsigned int monitor) { - int ret; + int ret = LTTNG_OK; struct ltt_kernel_stream *stream; /* Safety net */ @@ -287,14 +359,17 @@ 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, session, 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; } @@ -304,6 +379,7 @@ int kernel_consumer_send_channel_stream(struct consumer_socket *sock, if (ret < 0) { goto error; } + stream->sent_to_consumer = true; } error: @@ -337,14 +413,11 @@ int kernel_consumer_send_session(struct consumer_socket *sock, DBG("Sending session stream to kernel consumer"); - if (session->metadata_stream_fd >= 0) { + if (session->metadata_stream_fd >= 0 && session->metadata) { ret = kernel_consumer_add_metadata(sock, session, monitor); if (ret < 0) { goto error; } - - /* Flag that at least the metadata has been sent to the consumer. */ - session->consumer_fds_sent = 1; } /* Send channel and streams of it */ @@ -354,10 +427,21 @@ int kernel_consumer_send_session(struct consumer_socket *sock, if (ret < 0) { goto error; } + if (monitor) { + /* + * Inform the relay that all the streams for the + * channel were sent. + */ + ret = kernel_consumer_streams_sent(sock, session, chan->fd); + if (ret < 0) { + goto error; + } + } } DBG("Kernel consumer FDs of metadata and channel streams sent"); + session->consumer_fds_sent = 1; return 0; error: @@ -375,6 +459,7 @@ int kernel_consumer_destroy_channel(struct consumer_socket *socket, DBG("Sending kernel consumer destroy channel key %d", channel->fd); + memset(&msg, 0, sizeof(msg)); msg.cmd_type = LTTNG_CONSUMER_DESTROY_CHANNEL; msg.u.destroy_channel.key = channel->fd; @@ -403,6 +488,7 @@ int kernel_consumer_destroy_metadata(struct consumer_socket *socket, DBG("Sending kernel consumer destroy channel key %d", metadata->fd); + memset(&msg, 0, sizeof(msg)); msg.cmd_type = LTTNG_CONSUMER_DESTROY_CHANNEL; msg.u.destroy_channel.key = metadata->fd;