X-Git-Url: http://git.efficios.com/?p=lttng-tools.git;a=blobdiff_plain;f=src%2Fbin%2Flttng-sessiond%2Fust-consumer.c;h=ee56d6dea67b61406fd1d5ba946c03a23e6612a3;hp=ba74112fa72a35758cde4006320dbbfbfd15ce37;hb=8782cc7477fae212607b9fd6395a4b2e2d3357ed;hpb=7972aab22f74b18faa168c0482216a3dd711a075 diff --git a/src/bin/lttng-sessiond/ust-consumer.c b/src/bin/lttng-sessiond/ust-consumer.c index ba74112fa..ee56d6dea 100644 --- a/src/bin/lttng-sessiond/ust-consumer.c +++ b/src/bin/lttng-sessiond/ust-consumer.c @@ -28,8 +28,10 @@ #include #include "consumer.h" -#include "health.h" +#include "health-sessiond.h" #include "ust-consumer.h" +#include "buffer-registry.h" +#include "session.h" /* * Return allocated full pathname of the session using the consumer trace path @@ -59,7 +61,7 @@ static char *setup_trace_path(struct consumer_output *consumer, /* Get correct path name destination */ if (consumer->type == CONSUMER_DST_LOCAL) { /* Set application path to the destination path */ - ret = snprintf(pathname, PATH_MAX, "%s/%s/%s", + ret = snprintf(pathname, PATH_MAX, "%s%s%s", consumer->dst.trace_path, consumer->subdir, ua_sess->path); if (ret < 0) { PERROR("snprintf channel path"); @@ -76,7 +78,7 @@ static char *setup_trace_path(struct consumer_output *consumer, } } } else { - ret = snprintf(pathname, PATH_MAX, "%s/%s", consumer->subdir, + ret = snprintf(pathname, PATH_MAX, "%s%s", consumer->subdir, ua_sess->path); if (ret < 0) { PERROR("snprintf channel path"); @@ -116,10 +118,12 @@ static int ask_channel_creation(struct ust_app_session *ua_sess, DBG2("Asking UST consumer for channel"); /* Get and create full trace path of session. */ - pathname = setup_trace_path(consumer, ua_sess); - if (!pathname) { - ret = -1; - goto error; + if (ua_sess->output_traces) { + pathname = setup_trace_path(consumer, ua_sess); + if (!pathname) { + ret = -1; + goto error; + } } /* Depending on the buffer type, a different channel key is used. */ @@ -143,6 +147,7 @@ static int ask_channel_creation(struct ust_app_session *ua_sess, ua_chan->attr.overwrite, ua_chan->attr.switch_timer_interval, ua_chan->attr.read_timer_interval, + ua_sess->live_timer_interval, (int) ua_chan->attr.output, (int) ua_chan->attr.type, ua_sess->tracing_id, @@ -153,11 +158,16 @@ static int ask_channel_creation(struct ust_app_session *ua_sess, consumer->net_seq_index, ua_chan->key, registry->uuid, - chan_id); + chan_id, + ua_chan->tracefile_size, + ua_chan->tracefile_count, + ua_sess->id, + ua_sess->output_traces, + ua_sess->uid); health_code_update(); - ret = lttcomm_send_unix_sock(socket->fd, &msg, sizeof(msg)); + ret = consumer_socket_send(socket, &msg, sizeof(msg)); if (ret < 0) { goto error; } @@ -170,7 +180,9 @@ static int ask_channel_creation(struct ust_app_session *ua_sess, /* Communication protocol error. */ assert(key == ua_chan->key); /* We need at least one where 1 stream for 1 cpu. */ - assert(ua_chan->expected_stream_count > 0); + if (ua_sess->output_traces) { + assert(ua_chan->expected_stream_count > 0); + } DBG2("UST ask channel %" PRIu64 " successfully done with %u stream(s)", key, ua_chan->expected_stream_count); @@ -196,9 +208,14 @@ int ust_consumer_ask_channel(struct ust_app_session *ua_sess, assert(ua_chan); assert(consumer); assert(socket); - assert(socket->fd >= 0); assert(registry); + if (!consumer->enabled) { + ret = -LTTNG_ERR_NO_CONSUMER; + DBG3("Consumer is disabled"); + goto error; + } + pthread_mutex_lock(socket->lock); ret = ask_channel_creation(ua_sess, ua_chan, consumer, socket, registry); @@ -225,7 +242,6 @@ int ust_consumer_get_channel(struct consumer_socket *socket, assert(ua_chan); assert(socket); - assert(socket->fd >= 0); msg.cmd_type = LTTNG_CONSUMER_GET_CHANNEL; msg.u.get_channel.key = ua_chan->key; @@ -240,11 +256,11 @@ int ust_consumer_get_channel(struct consumer_socket *socket, } /* First, get the channel from consumer. */ - ret = ustctl_recv_channel_from_consumer(socket->fd, &ua_chan->obj); + ret = ustctl_recv_channel_from_consumer(*socket->fd_ptr, &ua_chan->obj); if (ret < 0) { if (ret != -EPIPE) { ERR("Error recv channel from consumer %d with ret %d", - socket->fd, ret); + *socket->fd_ptr, ret); } else { DBG3("UST app recv channel from consumer. Consumer is dead."); } @@ -263,7 +279,7 @@ int ust_consumer_get_channel(struct consumer_socket *socket, } /* Stream object is populated by this call if successful. */ - ret = ustctl_recv_stream_from_consumer(socket->fd, &stream->obj); + ret = ustctl_recv_stream_from_consumer(*socket->fd_ptr, &stream->obj); if (ret < 0) { free(stream); if (ret == -LTTNG_UST_ERR_NOENT) { @@ -273,7 +289,7 @@ int ust_consumer_get_channel(struct consumer_socket *socket, } if (ret != -EPIPE) { ERR("Recv stream from consumer %d with ret %d", - socket->fd, ret); + *socket->fd_ptr, ret); } else { DBG3("UST app recv stream from consumer. Consumer is dead."); } @@ -320,7 +336,6 @@ int ust_consumer_destroy_channel(struct consumer_socket *socket, assert(ua_chan); assert(socket); - assert(socket->fd >= 0); msg.cmd_type = LTTNG_CONSUMER_DESTROY_CHANNEL; msg.u.destroy_channel.key = ua_chan->key; @@ -405,3 +420,74 @@ 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 = consumer_socket_recv(socket, &request, sizeof(request)); + if (ret < 0) { + goto end; + } + + DBG("Metadata request received for session %" PRIu64 ", 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_per_pid); + if (!reg_pid) { + DBG("PID registry not found for session id %" PRIu64, + request.session_id_per_pid); + + 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; +}