#include "session.h"
#include "lttng-sessiond.h"
#include "notification-thread-commands.h"
+#include "rotate.h"
static
int ust_app_flush_app_session(struct ust_app *app, struct ust_app_session *ua_sess);
* Alloc new UST app session.
*/
static
-struct ust_app_session *alloc_ust_app_session(struct ust_app *app)
+struct ust_app_session *alloc_ust_app_session(void)
{
struct ust_app_session *ua_sess;
* Returns 0 on success or else a negative code which is either -ENOMEM or
* -ENOTCONN which is the default code if the ustctl_create_session fails.
*/
-static int create_ust_app_session(struct ltt_ust_session *usess,
+static int find_or_create_ust_app_session(struct ltt_ust_session *usess,
struct ust_app *app, struct ust_app_session **ua_sess_ptr,
int *is_created)
{
if (ua_sess == NULL) {
DBG2("UST app pid: %d session id %" PRIu64 " not found, creating it",
app->pid, usess->id);
- ua_sess = alloc_ust_app_session(app);
+ ua_sess = alloc_ust_app_session();
if (ua_sess == NULL) {
/* Only malloc can failed so something is really wrong */
ret = -ENOMEM;
* Called with UST app session lock held and a RCU read side lock.
*/
static
-int create_ust_app_channel_context(struct ust_app_session *ua_sess,
- struct ust_app_channel *ua_chan,
+int create_ust_app_channel_context(struct ust_app_channel *ua_chan,
struct lttng_ust_context_attr *uctx,
struct ust_app *app)
{
*/
static int do_consumer_create_channel(struct ltt_ust_session *usess,
struct ust_app_session *ua_sess, struct ust_app_channel *ua_chan,
- int bitness, struct ust_registry_session *registry)
+ int bitness, struct ust_registry_session *registry,
+ uint64_t trace_archive_id)
{
int ret;
unsigned int nb_fd = 0;
* stream we have to expect.
*/
ret = ust_consumer_ask_channel(ua_sess, ua_chan, usess->consumer, socket,
- registry);
+ registry, trace_archive_id);
if (ret < 0) {
goto error_ask;
}
* Create and send to the application the created buffers with per UID buffers.
*
* This MUST be called with a RCU read side lock acquired.
+ * The session list lock and the session's lock must be acquired.
*
* Return 0 on success else a negative value.
*/
int ret;
struct buffer_reg_uid *reg_uid;
struct buffer_reg_channel *reg_chan;
- bool created = false;
+ struct ltt_session *session;
+ enum lttng_error_code notification_ret;
+ struct ust_registry_channel *chan_reg;
assert(app);
assert(usess);
reg_chan = buffer_reg_channel_find(ua_chan->tracing_channel_id,
reg_uid);
- if (!reg_chan) {
- /* Create the buffer registry channel object. */
- ret = create_buffer_reg_channel(reg_uid->registry, ua_chan, ®_chan);
- if (ret < 0) {
- ERR("Error creating the UST channel \"%s\" registry instance",
- ua_chan->name);
- goto error;
- }
- assert(reg_chan);
+ if (reg_chan) {
+ goto send_channel;
+ }
- /*
- * Create the buffers on the consumer side. This call populates the
- * ust app channel object with all streams and data object.
- */
- ret = do_consumer_create_channel(usess, ua_sess, ua_chan,
- app->bits_per_long, reg_uid->registry->reg.ust);
- if (ret < 0) {
- ERR("Error creating UST channel \"%s\" on the consumer daemon",
+ /* Create the buffer registry channel object. */
+ ret = create_buffer_reg_channel(reg_uid->registry, ua_chan, ®_chan);
+ if (ret < 0) {
+ ERR("Error creating the UST channel \"%s\" registry instance",
ua_chan->name);
+ goto error;
+ }
- /*
- * Let's remove the previously created buffer registry channel so
- * it's not visible anymore in the session registry.
- */
- ust_registry_channel_del_free(reg_uid->registry->reg.ust,
- ua_chan->tracing_channel_id, false);
- buffer_reg_channel_remove(reg_uid->registry, reg_chan);
- buffer_reg_channel_destroy(reg_chan, LTTNG_DOMAIN_UST);
- goto error;
- }
+ session = session_find_by_id(ua_sess->tracing_id);
+ assert(session);
+ assert(pthread_mutex_trylock(&session->lock));
+ assert(session_trylock_list());
+
+ /*
+ * Create the buffers on the consumer side. This call populates the
+ * ust app channel object with all streams and data object.
+ */
+ ret = do_consumer_create_channel(usess, ua_sess, ua_chan,
+ app->bits_per_long, reg_uid->registry->reg.ust,
+ session->current_archive_id);
+ if (ret < 0) {
+ ERR("Error creating UST channel \"%s\" on the consumer daemon",
+ ua_chan->name);
/*
- * Setup the streams and add it to the session registry.
+ * Let's remove the previously created buffer registry channel so
+ * it's not visible anymore in the session registry.
*/
- ret = setup_buffer_reg_channel(reg_uid->registry,
- ua_chan, reg_chan, app);
- if (ret < 0) {
- ERR("Error setting up UST channel \"%s\"",
- ua_chan->name);
- goto error;
- }
- created = true;
+ ust_registry_channel_del_free(reg_uid->registry->reg.ust,
+ ua_chan->tracing_channel_id, false);
+ buffer_reg_channel_remove(reg_uid->registry, reg_chan);
+ buffer_reg_channel_destroy(reg_chan, LTTNG_DOMAIN_UST);
+ goto error;
}
- if (created) {
- enum lttng_error_code cmd_ret;
- struct ltt_session *session;
- uint64_t chan_reg_key;
- struct ust_registry_channel *chan_reg;
+ /*
+ * Setup the streams and add it to the session registry.
+ */
+ ret = setup_buffer_reg_channel(reg_uid->registry,
+ ua_chan, reg_chan, app);
+ if (ret < 0) {
+ ERR("Error setting up UST channel \"%s\"", ua_chan->name);
+ goto error;
+ }
- chan_reg_key = ua_chan->tracing_channel_id;
+ /* Notify the notification subsystem of the channel's creation. */
+ pthread_mutex_lock(®_uid->registry->reg.ust->lock);
+ chan_reg = ust_registry_channel_find(reg_uid->registry->reg.ust,
+ ua_chan->tracing_channel_id);
+ assert(chan_reg);
+ chan_reg->consumer_key = ua_chan->key;
+ chan_reg = NULL;
+ pthread_mutex_unlock(®_uid->registry->reg.ust->lock);
- pthread_mutex_lock(®_uid->registry->reg.ust->lock);
- chan_reg = ust_registry_channel_find(reg_uid->registry->reg.ust,
- chan_reg_key);
- assert(chan_reg);
- chan_reg->consumer_key = ua_chan->key;
- chan_reg = NULL;
- pthread_mutex_unlock(®_uid->registry->reg.ust->lock);
-
- session = session_find_by_id(ua_sess->tracing_id);
- assert(session);
-
- cmd_ret = notification_thread_command_add_channel(
- notification_thread_handle, session->name,
- ua_sess->euid, ua_sess->egid,
- ua_chan->name,
- ua_chan->key,
- LTTNG_DOMAIN_UST,
- ua_chan->attr.subbuf_size * ua_chan->attr.num_subbuf);
- if (cmd_ret != LTTNG_OK) {
- ret = - (int) cmd_ret;
- ERR("Failed to add channel to notification thread");
- goto error;
- }
+ notification_ret = notification_thread_command_add_channel(
+ notification_thread_handle, session->name,
+ ua_sess->euid, ua_sess->egid,
+ ua_chan->name,
+ ua_chan->key,
+ LTTNG_DOMAIN_UST,
+ ua_chan->attr.subbuf_size * ua_chan->attr.num_subbuf);
+ if (notification_ret != LTTNG_OK) {
+ ret = - (int) notification_ret;
+ ERR("Failed to add channel to notification thread");
+ goto error;
}
+send_channel:
/* Send buffers to the application. */
ret = send_channel_uid_to_ust(reg_chan, app, ua_sess, ua_chan);
if (ret < 0) {
* Create and send to the application the created buffers with per PID buffers.
*
* Called with UST app session lock held.
+ * The session list lock and the session's lock must be acquired.
*
* Return 0 on success else a negative value.
*/
goto error;
}
+ session = session_find_by_id(ua_sess->tracing_id);
+ assert(session);
+
+ assert(pthread_mutex_trylock(&session->lock));
+ assert(session_trylock_list());
+
/* Create and get channel on the consumer side. */
ret = do_consumer_create_channel(usess, ua_sess, ua_chan,
- app->bits_per_long, registry);
+ app->bits_per_long, registry,
+ session->current_archive_id);
if (ret < 0) {
ERR("Error creating UST channel \"%s\" on the consumer daemon",
ua_chan->name);
goto error;
}
- session = session_find_by_id(ua_sess->tracing_id);
- assert(session);
-
chan_reg_key = ua_chan->key;
pthread_mutex_lock(®istry->lock);
chan_reg = ust_registry_channel_find(registry, chan_reg_key);
struct ust_app_channel *metadata;
struct consumer_socket *socket;
struct ust_registry_session *registry;
+ struct ltt_session *session;
assert(ua_sess);
assert(app);
*/
registry->metadata_key = metadata->key;
+ session = session_find_by_id(ua_sess->tracing_id);
+ assert(session);
+
+ assert(pthread_mutex_trylock(&session->lock));
+ assert(session_trylock_list());
+
/*
* Ask the metadata channel creation to the consumer. The metadata object
* will be created by the consumer and kept their. However, the stream is
* consumer.
*/
ret = ust_consumer_ask_channel(ua_sess, metadata, consumer, socket,
- registry);
+ registry, session->current_archive_id);
if (ret < 0) {
/* Nullify the metadata key so we don't try to close it later on. */
registry->metadata_key = 0;
* that if session exist, it will simply return a pointer to the ust
* app session.
*/
- ret = create_ust_app_session(usess, app, &ua_sess, &created);
+ ret = find_or_create_ust_app_session(usess, app, &ua_sess, &created);
if (ret < 0) {
switch (ret) {
case -ENOTCONN:
usess->consumer->dst.session_root_path[0] != '\0') {
char *tmp_path;
- tmp_path = zmalloc(PATH_MAX * sizeof(char));
+ tmp_path = zmalloc(LTTNG_PATH_MAX);
if (!tmp_path) {
ERR("Alloc tmp_path");
goto error_unlock;
}
- ret = snprintf(tmp_path, PATH_MAX, "%s%s%s",
+ ret = snprintf(tmp_path, LTTNG_PATH_MAX, "%s%s%s",
usess->consumer->dst.session_root_path,
usess->consumer->chunk_path,
usess->consumer->subdir);
- if (ret >= PATH_MAX) {
+ if (ret >= LTTNG_PATH_MAX) {
ERR("Local destination path exceeds the maximal allowed length of %i bytes (needs %i bytes) with path = \"%s%s%s\"",
- PATH_MAX, ret,
+ LTTNG_PATH_MAX, ret,
usess->consumer->dst.session_root_path,
usess->consumer->chunk_path,
usess->consumer->subdir);
+ free(tmp_path);
goto error_unlock;
}
struct ust_app_ctx *ua_ctx;
int is_created = 0;
- ret = create_ust_app_session(usess, app, &ua_sess, &is_created);
+ ret = find_or_create_ust_app_session(usess, app, &ua_sess, &is_created);
if (ret < 0) {
/* Tracer is probably gone or ENOMEM. */
goto error;
}
ua_chan = caa_container_of(ua_chan_node, struct ust_app_channel,
node);
- ret = create_ust_app_channel_context(ua_sess, ua_chan, &uctx->ctx, app);
+ ret = create_ust_app_channel_context(ua_chan, &uctx->ctx, app);
if (ret < 0) {
goto next_app;
}
*
* On success 0 is returned else a negative value.
*/
-static int reply_ust_register_channel(int sock, int sobjd, int cobjd,
+static int reply_ust_register_channel(int sock, int cobjd,
size_t nr_fields, struct ustctl_field *fields)
{
int ret, ret_code = 0;
* that if needed it will be freed. After this, it's invalid to access
* fields or clean it up.
*/
- ret = reply_ust_register_channel(sock, sobjd, cobjd, nr_fields,
+ ret = reply_ust_register_channel(sock, cobjd, nr_fields,
fields);
if (ret < 0) {
goto error;
struct lttng_ht_iter iter;
struct ust_app *app;
char pathname[PATH_MAX];
+ struct ltt_session *session;
+ uint64_t trace_archive_id;
assert(usess);
assert(output);
rcu_read_lock();
+ session = session_find_by_id(usess->id);
+ assert(session);
+ assert(pthread_mutex_trylock(&session->lock));
+ assert(session_trylock_list());
+ trace_archive_id = session->current_archive_id;
+
switch (usess->buffer_type) {
case LTTNG_BUFFER_PER_UID:
{
struct buffer_reg_channel *reg_chan;
struct consumer_socket *socket;
+ if (!reg->registry->reg.ust->metadata_key) {
+ /* Skip since no metadata is present */
+ continue;
+ }
+
/* Get consumer socket to use to push the metadata.*/
socket = consumer_find_socket_by_bitness(reg->bits_per_long,
usess->consumer);
/* Add the UST default trace dir to path. */
cds_lfht_for_each_entry(reg->registry->channels->ht, &iter.iter,
reg_chan, node.node) {
- ret = consumer_snapshot_channel(socket, reg_chan->consumer_key,
- output, 0, usess->uid, usess->gid, pathname, wait,
- nb_packets_per_stream);
+ ret = consumer_snapshot_channel(socket,
+ reg_chan->consumer_key,
+ output, 0, usess->uid,
+ usess->gid, pathname, wait,
+ nb_packets_per_stream,
+ trace_archive_id);
if (ret < 0) {
goto error;
}
}
ret = consumer_snapshot_channel(socket,
reg->registry->reg.ust->metadata_key, output, 1,
- usess->uid, usess->gid, pathname, wait, 0);
+ usess->uid, usess->gid, pathname, wait, 0,
+ trace_archive_id);
if (ret < 0) {
goto error;
}
cds_lfht_for_each_entry(ua_sess->channels->ht, &chan_iter.iter,
ua_chan, node.node) {
- ret = consumer_snapshot_channel(socket, ua_chan->key, output,
- 0, ua_sess->euid, ua_sess->egid, pathname, wait,
- nb_packets_per_stream);
+ ret = consumer_snapshot_channel(socket,
+ ua_chan->key, output,
+ 0, ua_sess->euid, ua_sess->egid,
+ pathname, wait,
+ nb_packets_per_stream,
+ trace_archive_id);
if (ret < 0) {
goto error;
}
ret = -1;
goto error;
}
- ret = consumer_snapshot_channel(socket, registry->metadata_key, output,
- 1, ua_sess->euid, ua_sess->egid, pathname, wait, 0);
+ ret = consumer_snapshot_channel(socket,
+ registry->metadata_key, output,
+ 1, ua_sess->euid, ua_sess->egid,
+ pathname, wait, 0,
+ trace_archive_id);
if (ret < 0) {
goto error;
}
*lost = 0;
ret = buffer_reg_uid_consumer_channel_key(
- buffer_reg_uid_list, ust_session_id,
- uchan_id, &consumer_chan_key);
+ buffer_reg_uid_list, uchan_id, &consumer_chan_key);
if (ret < 0) {
/* Not found */
ret = 0;
return 0;
}
+
+/*
+ * Rotate all the channels of a session.
+ *
+ * Return 0 on success or else a negative value.
+ */
+int ust_app_rotate_session(struct ltt_session *session, bool *ust_active)
+{
+ int ret = 0;
+ struct lttng_ht_iter iter;
+ struct ust_app *app;
+ struct ltt_ust_session *usess = session->ust_session;
+ char pathname[LTTNG_PATH_MAX];
+
+ assert(usess);
+
+ rcu_read_lock();
+
+ switch (usess->buffer_type) {
+ case LTTNG_BUFFER_PER_UID:
+ {
+ struct buffer_reg_uid *reg;
+
+ cds_list_for_each_entry(reg, &usess->buffer_reg_uid_list, lnode) {
+ struct buffer_reg_channel *reg_chan;
+ struct consumer_socket *socket;
+
+ /* Get consumer socket to use to push the metadata.*/
+ socket = consumer_find_socket_by_bitness(reg->bits_per_long,
+ usess->consumer);
+ if (!socket) {
+ ret = -EINVAL;
+ goto error;
+ }
+
+ /*
+ * Account the metadata channel first to make sure the
+ * number of channels waiting for a rotation cannot
+ * reach 0 before we complete the iteration over all
+ * the channels.
+ */
+ ret = rotate_add_channel_pending(
+ reg->registry->reg.ust->metadata_key,
+ LTTNG_DOMAIN_UST, session);
+ if (ret < 0) {
+ ret = reg->bits_per_long == 32 ?
+ -LTTNG_ERR_UST_CONSUMER32_FAIL :
+ -LTTNG_ERR_UST_CONSUMER64_FAIL;
+ goto error;
+ }
+
+ ret = snprintf(pathname, sizeof(pathname),
+ DEFAULT_UST_TRACE_DIR "/" DEFAULT_UST_TRACE_UID_PATH,
+ reg->uid, reg->bits_per_long);
+ if (ret < 0 || ret == sizeof(pathname)) {
+ PERROR("Failed to format rotation path");
+ goto error;
+ }
+
+ /* Rotate the data channels. */
+ cds_lfht_for_each_entry(reg->registry->channels->ht, &iter.iter,
+ reg_chan, node.node) {
+ ret = rotate_add_channel_pending(
+ reg_chan->consumer_key,
+ LTTNG_DOMAIN_UST, session);
+ if (ret < 0) {
+ ret = reg->bits_per_long == 32 ?
+ -LTTNG_ERR_UST_CONSUMER32_FAIL :
+ -LTTNG_ERR_UST_CONSUMER64_FAIL;
+ goto error;
+ }
+ ret = consumer_rotate_channel(socket,
+ reg_chan->consumer_key,
+ usess->uid, usess->gid,
+ usess->consumer, pathname,
+ /* is_metadata_channel */ false,
+ session->current_archive_id,
+ &session->rotate_pending_relay);
+ if (ret < 0) {
+ goto error;
+ }
+ }
+
+ (void) push_metadata(reg->registry->reg.ust, usess->consumer);
+
+ ret = consumer_rotate_channel(socket,
+ reg->registry->reg.ust->metadata_key,
+ usess->uid, usess->gid,
+ usess->consumer, pathname,
+ /* is_metadata_channel */ true,
+ session->current_archive_id,
+ &session->rotate_pending_relay);
+ if (ret < 0) {
+ goto error;
+ }
+ *ust_active = true;
+ }
+ break;
+ }
+ case LTTNG_BUFFER_PER_PID:
+ {
+ cds_lfht_for_each_entry(ust_app_ht->ht, &iter.iter, app, pid_n.node) {
+ struct consumer_socket *socket;
+ struct lttng_ht_iter chan_iter;
+ struct ust_app_channel *ua_chan;
+ struct ust_app_session *ua_sess;
+ struct ust_registry_session *registry;
+
+ ua_sess = lookup_session_by_app(usess, app);
+ if (!ua_sess) {
+ /* Session not associated with this app. */
+ continue;
+ }
+ ret = snprintf(pathname, sizeof(pathname),
+ DEFAULT_UST_TRACE_DIR "/%s",
+ ua_sess->path);
+ if (ret < 0 || ret == sizeof(pathname)) {
+ PERROR("Failed to format rotation path");
+ goto error;
+ }
+
+ /* Get the right consumer socket for the application. */
+ socket = consumer_find_socket_by_bitness(app->bits_per_long,
+ usess->consumer);
+ if (!socket) {
+ ret = -EINVAL;
+ goto error;
+ }
+
+ registry = get_session_registry(ua_sess);
+ if (!registry) {
+ DBG("Application session is being torn down. Abort snapshot record.");
+ ret = -1;
+ goto error;
+ }
+
+ /*
+ * Account the metadata channel first to make sure the
+ * number of channels waiting for a rotation cannot
+ * reach 0 before we complete the iteration over all
+ * the channels.
+ */
+ ret = rotate_add_channel_pending(registry->metadata_key,
+ LTTNG_DOMAIN_UST, session);
+ if (ret < 0) {
+ ret = app->bits_per_long == 32 ?
+ -LTTNG_ERR_UST_CONSUMER32_FAIL :
+ -LTTNG_ERR_UST_CONSUMER64_FAIL;
+ goto error;
+ }
+
+ /* Rotate the data channels. */
+ cds_lfht_for_each_entry(ua_sess->channels->ht, &chan_iter.iter,
+ ua_chan, node.node) {
+ ret = rotate_add_channel_pending(
+ ua_chan->key, LTTNG_DOMAIN_UST,
+ session);
+ if (ret < 0) {
+ ret = app->bits_per_long == 32 ?
+ -LTTNG_ERR_UST_CONSUMER32_FAIL :
+ -LTTNG_ERR_UST_CONSUMER64_FAIL;
+ goto error;
+ }
+ ret = consumer_rotate_channel(socket, ua_chan->key,
+ ua_sess->euid, ua_sess->egid,
+ ua_sess->consumer, pathname,
+ /* is_metadata_channel */ false,
+ session->current_archive_id,
+ &session->rotate_pending_relay);
+ if (ret < 0) {
+ goto error;
+ }
+ }
+
+ /* Rotate the metadata channel. */
+ (void) push_metadata(registry, usess->consumer);
+ ret = consumer_rotate_channel(socket, registry->metadata_key,
+ ua_sess->euid, ua_sess->egid,
+ ua_sess->consumer, pathname,
+ /* is_metadata_channel */ true,
+ session->current_archive_id,
+ &session->rotate_pending_relay);
+ if (ret < 0) {
+ goto error;
+ }
+ *ust_active = true;
+ }
+ break;
+ }
+ default:
+ assert(0);
+ break;
+ }
+
+ ret = LTTNG_OK;
+
+error:
+ rcu_read_unlock();
+ return ret;
+}
projects / lttng-tools.git / blobdiff