X-Git-Url: http://git.efficios.com/?p=lttng-tools.git;a=blobdiff_plain;f=src%2Fbin%2Flttng-sessiond%2Fust-consumer.c;h=3bb54f03983178f035fdcb345f8e1d7670557847;hp=33cb6ff376c16c92db8bf605963054b364b59555;hb=c8fea79c745d42ea8143b7020ae11b4fc2da0d8a;hpb=d88aee689d5bd0067f362a323cb69c37717df59f diff --git a/src/bin/lttng-sessiond/ust-consumer.c b/src/bin/lttng-sessiond/ust-consumer.c index 33cb6ff37..3bb54f039 100644 --- a/src/bin/lttng-sessiond/ust-consumer.c +++ b/src/bin/lttng-sessiond/ust-consumer.c @@ -15,7 +15,7 @@ * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. */ -#define _GNU_SOURCE +#define _LGPL_SOURCE #include #include #include @@ -24,11 +24,14 @@ #include #include -#include +#include #include #include "consumer.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 @@ -58,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"); @@ -66,16 +69,16 @@ 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) { + if (errno != EEXIST) { ERR("Trace directory creation error"); goto error; } } } 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"); @@ -93,29 +96,75 @@ 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; - uint64_t key; + 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] = ""; + char root_shm_path[PATH_MAX] = ""; assert(ua_sess); assert(ua_chan); assert(socket); assert(consumer); + assert(registry); 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. */ + 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; + /* + * 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); + } + strncpy(root_shm_path, ua_sess->root_shm_path, sizeof(root_shm_path)); + root_shm_path[sizeof(root_shm_path) - 1] = '\0'; + } + + switch (ua_chan->attr.output) { + case LTTNG_UST_MMAP: + default: + output = LTTNG_EVENT_MMAP; + break; } consumer_init_ask_channel_comm_msg(&msg, @@ -124,20 +173,28 @@ 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->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->registry.uuid); + registry->uuid, + chan_id, + ua_chan->tracefile_size, + ua_chan->tracefile_count, + ua_sess->id, + ua_sess->output_traces, + ua_sess->uid, + root_shm_path, 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; } @@ -150,7 +207,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); @@ -168,7 +227,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,17 +235,22 @@ int ust_consumer_ask_channel(struct ust_app_session *ua_sess, assert(ua_chan); assert(consumer); assert(socket); - assert(socket->fd >= 0); + assert(registry); - pthread_mutex_lock(socket->lock); + if (!consumer->enabled) { + ret = -LTTNG_ERR_NO_CONSUMER; + DBG3("Consumer is disabled"); + goto error; + } - ret = ask_channel_creation(ua_sess, ua_chan, consumer, socket); + pthread_mutex_lock(socket->lock); + ret = ask_channel_creation(ua_sess, ua_chan, consumer, socket, registry); + pthread_mutex_unlock(socket->lock); if (ret < 0) { goto error; } error: - pthread_mutex_unlock(socket->lock); return ret; } @@ -204,8 +268,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; @@ -219,11 +283,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."); } @@ -242,7 +306,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) { @@ -252,7 +316,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."); } @@ -263,7 +327,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. */ @@ -299,8 +363,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; @@ -335,11 +399,13 @@ int ust_consumer_send_stream_to_ust(struct ust_app *app, DBG2("UST consumer send stream to app %d", app->sock); /* Relay stream to application. */ + pthread_mutex_lock(&app->sock_lock); ret = ustctl_send_stream_to_ust(app->sock, channel->obj, stream->obj); + pthread_mutex_unlock(&app->sock_lock); 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."); } @@ -366,11 +432,13 @@ 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. */ + pthread_mutex_lock(&app->sock_lock); ret = ustctl_send_channel_to_ust(app->sock, ua_sess->handle, channel->obj); + pthread_mutex_unlock(&app->sock_lock); if (ret < 0) { if (ret != -EPIPE && ret != -LTTNG_UST_ERR_EXITING) { ERR("Error ustctl send channel %s to app pid: %d with ret %d", @@ -386,129 +454,77 @@ error: } /* - * Send metadata string to consumer. + * Handle the metadata requests from the UST consumer * * Return 0 on success else a negative value. */ -int ust_consumer_push_metadata(struct consumer_socket *socket, - struct ust_app_session *ua_sess, char *metadata_str, - size_t len, size_t target_offset) +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); - assert(socket->fd >= 0); - assert(ua_sess); - assert(ua_sess->metadata); - - DBG2("UST consumer push metadata to consumer socket %d", socket->fd); - - msg.cmd_type = LTTNG_CONSUMER_PUSH_METADATA; - msg.u.push_metadata.key = ua_sess->metadata->key; - msg.u.push_metadata.target_offset = target_offset; - msg.u.push_metadata.len = len; - - /* - * TODO: reenable these locks when the consumerd gets the ability to - * reorder the metadata it receives. This fits with locking in - * src/bin/lttng-sessiond/ust-app.c:push_metadata() - * - * pthread_mutex_lock(socket->lock); - */ + rcu_read_lock(); health_code_update(); - ret = consumer_send_msg(socket, &msg); - if (ret < 0) { - goto error; - } - - DBG3("UST consumer push metadata on sock %d of len %lu", socket->fd, len); - - ret = lttcomm_send_unix_sock(socket->fd, metadata_str, len); - if (ret < 0) { - fprintf(stderr, "send error: %d\n", ret); - goto error; - } - - health_code_update(); - ret = consumer_recv_status_reply(socket); - if (ret < 0) { - goto error; - } - -error: - health_code_update(); - /* - * pthread_mutex_unlock(socket->lock); - */ - return ret; -} - -/* - * Send a close metdata command to consumer using the given channel key. - * - * Return 0 on success else a negative value. - */ -int ust_consumer_close_metadata(struct consumer_socket *socket, - struct ust_app_channel *ua_chan) -{ - int ret; - struct lttcomm_consumer_msg msg; - - assert(ua_chan); - assert(socket); - assert(socket->fd >= 0); - - DBG2("UST consumer close metadata channel key %lu", ua_chan->key); - - msg.cmd_type = LTTNG_CONSUMER_CLOSE_METADATA; - msg.u.close_metadata.key = ua_chan->key; + /* Wait for a metadata request */ pthread_mutex_lock(socket->lock); - health_code_update(); - - ret = consumer_send_msg(socket, &msg); + ret = consumer_socket_recv(socket, &request, sizeof(request)); + pthread_mutex_unlock(socket->lock); if (ret < 0) { - goto error; + goto end; } -error: - health_code_update(); - pthread_mutex_unlock(socket->lock); - return ret; -} + DBG("Metadata request received for session %" PRIu64 ", key %" PRIu64, + request.session_id, request.key); -/* - * Send a setup metdata command to consumer using the given channel key. - * - * Return 0 on success else a negative value. - */ -int ust_consumer_setup_metadata(struct consumer_socket *socket, - struct ust_app_channel *ua_chan) -{ - int ret; - struct lttcomm_consumer_msg msg; - - assert(ua_chan); - assert(socket); - assert(socket->fd >= 0); - - DBG2("UST consumer setup metadata channel key %lu", ua_chan->key); - - msg.cmd_type = LTTNG_CONSUMER_SETUP_METADATA; - msg.u.setup_metadata.key = ua_chan->key; - - pthread_mutex_lock(socket->lock); - health_code_update(); - - ret = consumer_send_msg(socket, &msg); - if (ret < 0) { - goto error; + 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); + + memset(&msg, 0, sizeof(msg)); + 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); + + 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 == -EPIPE) { + DBG("Application or relay closed while pushing metadata"); + } else if (ret_push < 0) { + ERR("Pushing metadata"); + ret = -1; + goto end; + } else { + DBG("UST Consumer metadata pushed successfully"); + } + ret = 0; -error: - health_code_update(); - pthread_mutex_unlock(socket->lock); +end: + rcu_read_unlock(); return ret; }