X-Git-Url: https://git.lttng.org/?a=blobdiff_plain;f=src%2Fbin%2Flttng-sessiond%2Fkernel-consumer.c;h=6cda088d9e5e7600174476d29962adf2bd532189;hb=88f06f1562c6ecd3a5666baa93a57273fbe10437;hp=ef6dccb538219c86b70615b5323c00697f5a510d;hpb=2bba9e532ca1910822005ff7f67400a2e871467c;p=lttng-tools.git diff --git a/src/bin/lttng-sessiond/kernel-consumer.c b/src/bin/lttng-sessiond/kernel-consumer.c index ef6dccb53..6cda088d9 100644 --- a/src/bin/lttng-sessiond/kernel-consumer.c +++ b/src/bin/lttng-sessiond/kernel-consumer.c @@ -16,6 +16,7 @@ */ #define _GNU_SOURCE +#define _LGPL_SOURCE #include #include #include @@ -26,7 +27,7 @@ #include #include "consumer.h" -#include "health.h" +#include "health-sessiond.h" #include "kernel-consumer.h" static char *create_channel_path(struct consumer_output *consumer, @@ -48,6 +49,10 @@ static char *create_channel_path(struct consumer_output *consumer, goto error; } pathname = strndup(tmp_path, sizeof(tmp_path)); + if (!pathname) { + PERROR("strndup"); + goto error; + } /* Create directory */ ret = run_as_mkdir_recursive(pathname, S_IRWXU | S_IRWXG, uid, gid); @@ -65,6 +70,10 @@ static char *create_channel_path(struct consumer_output *consumer, goto error; } pathname = strndup(tmp_path, sizeof(tmp_path)); + if (!pathname) { + PERROR("strndup"); + goto error; + } DBG3("Kernel network consumer subdir path: %s", pathname); } @@ -99,13 +108,13 @@ int kernel_consumer_add_channel(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 */ @@ -123,7 +132,8 @@ int kernel_consumer_add_channel(struct consumer_socket *sock, CONSUMER_CHANNEL_TYPE_DATA, channel->channel->attr.tracefile_size, channel->channel->attr.tracefile_count, - monitor); + monitor, + channel->channel->attr.live_timer_interval); health_code_update(); @@ -135,6 +145,7 @@ int kernel_consumer_add_channel(struct consumer_socket *sock, health_code_update(); error: + free(pathname); return ret; } @@ -161,13 +172,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 */ @@ -184,7 +195,7 @@ int kernel_consumer_add_metadata(struct consumer_socket *sock, DEFAULT_KERNEL_CHANNEL_OUTPUT, CONSUMER_CHANNEL_TYPE_METADATA, 0, 0, - monitor); + monitor, 0); health_code_update(); @@ -202,14 +213,6 @@ int kernel_consumer_add_metadata(struct consumer_socket *sock, session->metadata_stream_fd, 0); /* CPU: 0 for metadata. */ - /* - * Set the no monitor flag. If set to 1, it indicates the consumer to NOT - * monitor the stream but rather add it to a special list in the associated - * channel. This is used to handle ephemeral stream used by the snapshot - * command or store streams for the flight recorder mode. - */ - lkm.u.stream.no_monitor = no_monitor; - health_code_update(); /* Send stream and file descriptor */ @@ -222,6 +225,7 @@ int kernel_consumer_add_metadata(struct consumer_socket *sock, health_code_update(); error: + free(pathname); return ret; } @@ -269,6 +273,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. */ @@ -344,14 +382,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 */ @@ -361,12 +396,81 @@ 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: return ret; } + +int kernel_consumer_destroy_channel(struct consumer_socket *socket, + struct ltt_kernel_channel *channel) +{ + int ret; + struct lttcomm_consumer_msg msg; + + assert(channel); + assert(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; + + pthread_mutex_lock(socket->lock); + health_code_update(); + + ret = consumer_send_msg(socket, &msg); + if (ret < 0) { + goto error; + } + +error: + health_code_update(); + pthread_mutex_unlock(socket->lock); + return ret; +} + +int kernel_consumer_destroy_metadata(struct consumer_socket *socket, + struct ltt_kernel_metadata *metadata) +{ + int ret; + struct lttcomm_consumer_msg msg; + + assert(metadata); + assert(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; + + pthread_mutex_lock(socket->lock); + health_code_update(); + + ret = consumer_send_msg(socket, &msg); + if (ret < 0) { + goto error; + } + +error: + health_code_update(); + pthread_mutex_unlock(socket->lock); + return ret; +}