X-Git-Url: http://git.efficios.com/?p=lttng-tools.git;a=blobdiff_plain;f=src%2Fbin%2Flttng-sessiond%2Fust-consumer.c;h=8a3533ba10c360097a8fb2cb0099a4f3d986f324;hp=72b68bb7f7bed39d03ed5860381ef70d94df5ad8;hb=d7ba13889c8692b14f99238ddf2721ed78df89d2;hpb=2bba9e532ca1910822005ff7f67400a2e871467c diff --git a/src/bin/lttng-sessiond/ust-consumer.c b/src/bin/lttng-sessiond/ust-consumer.c index 72b68bb7f..8a3533ba1 100644 --- a/src/bin/lttng-sessiond/ust-consumer.c +++ b/src/bin/lttng-sessiond/ust-consumer.c @@ -16,6 +16,7 @@ */ #define _GNU_SOURCE +#define _LGPL_SOURCE #include #include #include @@ -28,7 +29,7 @@ #include #include "consumer.h" -#include "health.h" +#include "health-sessiond.h" #include "ust-consumer.h" #include "buffer-registry.h" #include "session.h" @@ -102,12 +103,13 @@ 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 ust_registry_session *registry) { - int ret; + int ret, output; uint32_t chan_id; uint64_t key, chan_reg_key; char *pathname = NULL; struct lttcomm_consumer_msg msg; struct ust_registry_channel *chan_reg; + char shm_path[PATH_MAX] = ""; assert(ua_sess); assert(ua_chan); @@ -118,10 +120,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. */ @@ -133,10 +137,32 @@ static int ask_channel_creation(struct ust_app_session *ua_sess, if (ua_chan->attr.type == LTTNG_UST_CHAN_METADATA) { chan_id = -1U; + /* + * Metadata channels shm_path (buffers) are handled within + * session daemon. Consumer daemon should not try to create + * those buffer files. + */ } else { chan_reg = ust_registry_channel_find(registry, chan_reg_key); assert(chan_reg); chan_id = chan_reg->chan_id; + if (ua_sess->shm_path[0]) { + strncpy(shm_path, ua_sess->shm_path, sizeof(shm_path)); + shm_path[sizeof(shm_path) - 1] = '\0'; + strncat(shm_path, "/", + sizeof(shm_path) - strlen(shm_path) - 1); + strncat(shm_path, ua_chan->name, + sizeof(shm_path) - strlen(shm_path) - 1); + strncat(shm_path, "_", + sizeof(shm_path) - strlen(shm_path) - 1); + } + } + + switch (ua_chan->attr.output) { + case LTTNG_UST_MMAP: + default: + output = LTTNG_EVENT_MMAP; + break; } consumer_init_ask_channel_comm_msg(&msg, @@ -145,7 +171,8 @@ 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, - (int) ua_chan->attr.output, + ua_sess->live_timer_interval, + output, (int) ua_chan->attr.type, ua_sess->tracing_id, pathname, @@ -158,11 +185,14 @@ static int ask_channel_creation(struct ust_app_session *ua_sess, chan_id, ua_chan->tracefile_size, ua_chan->tracefile_count, - ua_sess->output_traces); + ua_sess->id, + ua_sess->output_traces, + ua_sess->uid, + shm_path); 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; } @@ -175,7 +205,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); @@ -201,7 +233,6 @@ 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) { @@ -236,8 +267,8 @@ int ust_consumer_get_channel(struct consumer_socket *socket, assert(ua_chan); assert(socket); - assert(socket->fd >= 0); + memset(&msg, 0, sizeof(msg)); msg.cmd_type = LTTNG_CONSUMER_GET_CHANNEL; msg.u.get_channel.key = ua_chan->key; @@ -251,11 +282,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."); } @@ -274,7 +305,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) { @@ -284,7 +315,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."); } @@ -295,7 +326,7 @@ int ust_consumer_get_channel(struct consumer_socket *socket, cds_list_add_tail(&stream->list, &ua_chan->streams.head); ua_chan->streams.count++; - DBG2("UST app stream %d received succesfully", ua_chan->streams.count); + DBG2("UST app stream %d received successfully", ua_chan->streams.count); } /* This MUST match or else we have a synchronization problem. */ @@ -331,8 +362,8 @@ int ust_consumer_destroy_channel(struct consumer_socket *socket, assert(ua_chan); assert(socket); - assert(socket->fd >= 0); + memset(&msg, 0, sizeof(msg)); msg.cmd_type = LTTNG_CONSUMER_DESTROY_CHANNEL; msg.u.destroy_channel.key = ua_chan->key; @@ -370,8 +401,8 @@ int ust_consumer_send_stream_to_ust(struct ust_app *app, ret = ustctl_send_stream_to_ust(app->sock, channel->obj, stream->obj); if (ret < 0) { if (ret != -EPIPE && ret != -LTTNG_UST_ERR_EXITING) { - ERR("Error ustctl send stream %s to app pid: %d with ret %d", - stream->name, app->pid, ret); + ERR("ustctl send stream handle %d to app pid: %d with ret %d", + stream->obj->handle, app->pid, ret); } else { DBG3("UST app send stream to ust failed. Application is dead."); } @@ -434,19 +465,17 @@ int ust_consumer_metadata_request(struct consumer_socket *socket) 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; + pthread_mutex_lock(socket->lock); + ret = consumer_socket_recv(socket, &request, sizeof(request)); + pthread_mutex_unlock(socket->lock); + if (ret < 0) { goto end; } - DBG("Metadata request received for session %u, key %" PRIu64, + DBG("Metadata request received for session %" PRIu64 ", key %" PRIu64, request.session_id, request.key); reg_uid = buffer_reg_uid_find(request.session_id, @@ -455,11 +484,12 @@ int ust_consumer_metadata_request(struct consumer_socket *socket) ust_reg = reg_uid->registry->reg.ust; } else { struct buffer_reg_pid *reg_pid = - buffer_reg_pid_find(request.session_id); + buffer_reg_pid_find(request.session_id_per_pid); if (!reg_pid) { - DBG("PID registry not found for session id %u", - request.session_id); + DBG("PID registry not found for session id %" PRIu64, + request.session_id_per_pid); + memset(&msg, 0, sizeof(msg)); msg.cmd_type = LTTNG_ERR_UND; (void) consumer_send_msg(socket, &msg); /* @@ -475,7 +505,9 @@ int ust_consumer_metadata_request(struct consumer_socket *socket) } assert(ust_reg); + pthread_mutex_lock(&ust_reg->lock); ret_push = ust_app_push_metadata(ust_reg, socket, 1); + pthread_mutex_unlock(&ust_reg->lock); if (ret_push < 0) { ERR("Pushing metadata"); ret = -1; @@ -485,7 +517,6 @@ int ust_consumer_metadata_request(struct consumer_socket *socket) ret = 0; end: - pthread_mutex_unlock(socket->lock); rcu_read_unlock(); return ret; }