X-Git-Url: https://git.lttng.org/?p=lttng-tools.git;a=blobdiff_plain;f=src%2Fbin%2Flttng-sessiond%2Fust-consumer.c;h=93c1f713562b016f7147ba16e402dae18ce88e0d;hp=7c1ae402c8ded33c75532d76530ba5ddf195001d;hb=1624d5b7ba86633d36f3a5c86ea1df5d308c4360;hpb=ffe600149a7608221985751e1bf293234bf2545c diff --git a/src/bin/lttng-sessiond/ust-consumer.c b/src/bin/lttng-sessiond/ust-consumer.c index 7c1ae402c..93c1f7135 100644 --- a/src/bin/lttng-sessiond/ust-consumer.c +++ b/src/bin/lttng-sessiond/ust-consumer.c @@ -21,13 +21,17 @@ #include #include #include +#include #include #include #include #include "consumer.h" +#include "health.h" #include "ust-consumer.h" +#include "buffer-registry.h" +#include "session.h" /* * Return allocated full pathname of the session using the consumer trace path @@ -65,8 +69,8 @@ static char *setup_trace_path(struct consumer_output *consumer, } /* Create directory. Ignore if exist. */ - ret = run_as_mkdir_recursive(pathname, S_IRWXU | S_IRWXG, ua_sess->uid, - ua_sess->gid); + ret = run_as_mkdir_recursive(pathname, S_IRWXU | S_IRWXG, + ua_sess->euid, ua_sess->egid); if (ret < 0) { if (ret != -EEXIST) { ERR("Trace directory creation error"); @@ -92,21 +96,24 @@ error: /* * Send a single channel to the consumer using command ADD_CHANNEL. * - * Consumer socket MUST be acquired before calling this. + * Consumer socket lock MUST be acquired before calling this. */ static int ask_channel_creation(struct ust_app_session *ua_sess, struct ust_app_channel *ua_chan, struct consumer_output *consumer, - struct consumer_socket *socket) + struct consumer_socket *socket, struct ust_registry_session *registry) { int ret; - unsigned long key; + uint32_t chan_id; + uint64_t key, chan_reg_key; char *pathname = NULL; struct lttcomm_consumer_msg msg; + struct ust_registry_channel *chan_reg; assert(ua_sess); assert(ua_chan); assert(socket); assert(consumer); + assert(registry); DBG2("Asking UST consumer for channel"); @@ -117,6 +124,21 @@ static int ask_channel_creation(struct ust_app_session *ua_sess, goto error; } + /* Depending on the buffer type, a different channel key is used. */ + if (ua_sess->buffer_type == LTTNG_BUFFER_PER_UID) { + chan_reg_key = ua_chan->tracing_channel_id; + } else { + chan_reg_key = ua_chan->key; + } + + if (ua_chan->attr.type == LTTNG_UST_CHAN_METADATA) { + chan_id = -1U; + } else { + chan_reg = ust_registry_channel_find(registry, chan_reg_key); + assert(chan_reg); + chan_id = chan_reg->chan_id; + } + consumer_init_ask_channel_comm_msg(&msg, ua_chan->attr.subbuf_size, ua_chan->attr.num_subbuf, @@ -125,14 +147,17 @@ static int ask_channel_creation(struct ust_app_session *ua_sess, ua_chan->attr.read_timer_interval, (int) ua_chan->attr.output, (int) ua_chan->attr.type, - ua_sess->id, + ua_sess->tracing_id, pathname, ua_chan->name, - ua_sess->uid, - ua_sess->gid, + ua_sess->euid, + ua_sess->egid, consumer->net_seq_index, ua_chan->key, - ua_sess->uuid); + registry->uuid, + chan_id, + ua_chan->tracefile_size, + ua_chan->tracefile_count); health_code_update(); @@ -151,7 +176,7 @@ static int ask_channel_creation(struct ust_app_session *ua_sess, /* We need at least one where 1 stream for 1 cpu. */ assert(ua_chan->expected_stream_count > 0); - DBG2("UST ask channel %lu successfully done with %u stream(s)", key, + DBG2("UST ask channel %" PRIu64 " successfully done with %u stream(s)", key, ua_chan->expected_stream_count); error: @@ -167,7 +192,7 @@ error: */ int ust_consumer_ask_channel(struct ust_app_session *ua_sess, struct ust_app_channel *ua_chan, struct consumer_output *consumer, - struct consumer_socket *socket) + struct consumer_socket *socket, struct ust_registry_session *registry) { int ret; @@ -176,10 +201,11 @@ int ust_consumer_ask_channel(struct ust_app_session *ua_sess, assert(consumer); assert(socket); assert(socket->fd >= 0); + assert(registry); pthread_mutex_lock(socket->lock); - ret = ask_channel_creation(ua_sess, ua_chan, consumer, socket); + ret = ask_channel_creation(ua_sess, ua_chan, consumer, socket, registry); if (ret < 0) { goto error; } @@ -228,7 +254,6 @@ int ust_consumer_get_channel(struct consumer_socket *socket, } goto error; } - ua_chan->handle = ua_chan->obj->handle; /* Next, get all streams. */ while (1) { @@ -345,6 +370,7 @@ int ust_consumer_send_stream_to_ust(struct ust_app *app, } goto error; } + channel->handle = channel->obj->handle; error: return ret; @@ -365,8 +391,8 @@ int ust_consumer_send_channel_to_ust(struct ust_app *app, assert(channel); assert(channel->obj); - DBG2("UST app send channel to app sock %d pid %d (name: %s, key: %lu)", - app->sock, app->pid, channel->name, channel->key); + DBG2("UST app send channel to sock %d pid %d (name: %s, key: %" PRIu64 ")", + app->sock, app->pid, channel->name, channel->tracing_channel_id); /* Send stream to application. */ ret = ustctl_send_channel_to_ust(app->sock, ua_sess->handle, channel->obj); @@ -383,3 +409,76 @@ int ust_consumer_send_channel_to_ust(struct ust_app *app, error: return ret; } + +/* + * Handle the metadata requests from the UST consumer + * + * Return 0 on success else a negative value. + */ +int ust_consumer_metadata_request(struct consumer_socket *socket) +{ + int ret; + ssize_t ret_push; + struct lttcomm_metadata_request_msg request; + struct buffer_reg_uid *reg_uid; + struct ust_registry_session *ust_reg; + struct lttcomm_consumer_msg msg; + + assert(socket); + + rcu_read_lock(); + pthread_mutex_lock(socket->lock); + + health_code_update(); + + /* Wait for a metadata request */ + ret = lttcomm_recv_unix_sock(socket->fd, &request, sizeof(request)); + if (ret <= 0) { + ERR("Consumer closed the metadata socket"); + ret = -1; + goto end; + } + + DBG("Metadata request received for session %u, key %" PRIu64, + request.session_id, request.key); + + reg_uid = buffer_reg_uid_find(request.session_id, + request.bits_per_long, request.uid); + if (reg_uid) { + ust_reg = reg_uid->registry->reg.ust; + } else { + struct buffer_reg_pid *reg_pid = + buffer_reg_pid_find(request.session_id); + if (!reg_pid) { + DBG("PID registry not found for session id %u", + request.session_id); + + msg.cmd_type = LTTNG_ERR_UND; + (void) consumer_send_msg(socket, &msg); + /* + * This is possible since the session might have been destroyed + * during a consumer metadata request. So here, return gracefully + * because the destroy session will push the remaining metadata to + * the consumer. + */ + ret = 0; + goto end; + } + ust_reg = reg_pid->registry->reg.ust; + } + assert(ust_reg); + + ret_push = ust_app_push_metadata(ust_reg, socket, 1); + if (ret_push < 0) { + ERR("Pushing metadata"); + ret = -1; + goto end; + } + DBG("UST Consumer metadata pushed successfully"); + ret = 0; + +end: + pthread_mutex_unlock(socket->lock); + rcu_read_unlock(); + return ret; +}