#include <unistd.h>
#include <urcu/compiler.h>
#include <lttng/ust-error.h>
+#include <signal.h>
#include <common/common.h>
#include <common/sessiond-comm/sessiond-comm.h>
free(ua_chan);
}
+/*
+ * Push metadata to consumer socket. The socket lock MUST be acquired.
+ *
+ * On success, return the len of metadata pushed or else a negative value.
+ */
+ssize_t ust_app_push_metadata(struct ust_registry_session *registry,
+ struct consumer_socket *socket, int send_zero_data)
+{
+ int ret;
+ char *metadata_str = NULL;
+ size_t len, offset;
+ ssize_t ret_val;
+
+ assert(registry);
+ assert(socket);
+ /* Should never be 0 which is the initial state. */
+ assert(registry->metadata_key);
+
+ pthread_mutex_lock(®istry->lock);
+
+ offset = registry->metadata_len_sent;
+ len = registry->metadata_len - registry->metadata_len_sent;
+ if (len == 0) {
+ DBG3("No metadata to push for metadata key %" PRIu64,
+ registry->metadata_key);
+ ret_val = len;
+ if (send_zero_data) {
+ DBG("No metadata to push");
+ goto push_data;
+ }
+ goto end;
+ }
+
+ /* Allocate only what we have to send. */
+ metadata_str = zmalloc(len);
+ if (!metadata_str) {
+ PERROR("zmalloc ust app metadata string");
+ ret_val = -ENOMEM;
+ goto error;
+ }
+ /* Copy what we haven't send out. */
+ memcpy(metadata_str, registry->metadata + offset, len);
+ registry->metadata_len_sent += len;
+
+push_data:
+ pthread_mutex_unlock(®istry->lock);
+ ret = consumer_push_metadata(socket, registry->metadata_key,
+ metadata_str, len, offset);
+ if (ret < 0) {
+ ret_val = ret;
+ goto error_push;
+ }
+
+ free(metadata_str);
+ return len;
+
+end:
+error:
+ pthread_mutex_unlock(®istry->lock);
+error_push:
+ free(metadata_str);
+ return ret_val;
+}
+
/*
* For a given application and session, push metadata to consumer. The session
* lock MUST be acquired here before calling this.
+ * Either sock or consumer is required : if sock is NULL, the default
+ * socket to send the metadata is retrieved from consumer, if sock
+ * is not NULL we use it to send the metadata.
*
* Return 0 on success else a negative error.
*/
static int push_metadata(struct ust_registry_session *registry,
struct consumer_output *consumer)
{
- int ret;
- char *metadata_str = NULL;
- size_t len, offset;
+ int ret_val;
+ ssize_t ret;
struct consumer_socket *socket;
assert(registry);
* no start has been done previously.
*/
if (!registry->metadata_key) {
- ret = 0;
+ ret_val = 0;
goto error_rcu_unlock;
}
socket = consumer_find_socket_by_bitness(registry->bits_per_long,
consumer);
if (!socket) {
- ret = -1;
+ ret_val = -1;
goto error_rcu_unlock;
}
* ability to reorder the metadata it receives.
*/
pthread_mutex_lock(socket->lock);
- pthread_mutex_lock(®istry->lock);
-
- offset = registry->metadata_len_sent;
- len = registry->metadata_len - registry->metadata_len_sent;
- if (len == 0) {
- DBG3("No metadata to push for metadata key %" PRIu64,
- registry->metadata_key);
- ret = 0;
- goto error_reg_unlock;
- }
- assert(len > 0);
-
- /* Allocate only what we have to send. */
- metadata_str = zmalloc(len);
- if (!metadata_str) {
- PERROR("zmalloc ust app metadata string");
- ret = -ENOMEM;
- goto error_reg_unlock;
- }
- /* Copy what we haven't send out. */
- memcpy(metadata_str, registry->metadata + offset, len);
-
- pthread_mutex_unlock(®istry->lock);
-
- ret = consumer_push_metadata(socket, registry->metadata_key,
- metadata_str, len, offset);
+ ret = ust_app_push_metadata(registry, socket, 0);
+ pthread_mutex_unlock(socket->lock);
if (ret < 0) {
- pthread_mutex_unlock(socket->lock);
+ ret_val = ret;
goto error_rcu_unlock;
}
- /* Update len sent of the registry. */
- pthread_mutex_lock(®istry->lock);
- registry->metadata_len_sent += len;
- pthread_mutex_unlock(®istry->lock);
- pthread_mutex_unlock(socket->lock);
-
rcu_read_unlock();
- free(metadata_str);
return 0;
-error_reg_unlock:
- pthread_mutex_unlock(®istry->lock);
- pthread_mutex_unlock(socket->lock);
error_rcu_unlock:
rcu_read_unlock();
- free(metadata_str);
- return ret;
+ return ret_val;
}
/*
strncpy(ua_chan->name, uchan->name, sizeof(ua_chan->name));
ua_chan->name[sizeof(ua_chan->name) - 1] = '\0';
+ ua_chan->tracefile_size = uchan->tracefile_size;
+ ua_chan->tracefile_count = uchan->tracefile_count;
+
/* Copy event attributes since the layout is different. */
ua_chan->attr.subbuf_size = uchan->attr.subbuf_size;
ua_chan->attr.num_subbuf = uchan->attr.num_subbuf;
switch (ua_sess->buffer_type) {
case LTTNG_BUFFER_PER_PID:
ret = snprintf(ua_sess->path, sizeof(ua_sess->path),
- DEFAULT_UST_TRACE_PID_PATH "/%s-%d-%s/", app->name, app->pid,
+ DEFAULT_UST_TRACE_PID_PATH "/%s-%d-%s", app->name, app->pid,
datetime);
break;
case LTTNG_BUFFER_PER_UID:
ret = ust_registry_session_init(®_pid->registry->reg.ust, app,
app->bits_per_long, app->uint8_t_alignment,
app->uint16_t_alignment, app->uint32_t_alignment,
- app->uint64_t_alignment, app->long_alignment, app->byte_order);
+ app->uint64_t_alignment, app->long_alignment,
+ app->byte_order, app->version.major,
+ app->version.minor);
if (ret < 0) {
goto error;
}
}
/* Initialize registry. */
- ret = ust_registry_session_init(®_uid->registry->reg.ust, app,
+ ret = ust_registry_session_init(®_uid->registry->reg.ust, NULL,
app->bits_per_long, app->uint8_t_alignment,
app->uint16_t_alignment, app->uint32_t_alignment,
- app->uint64_t_alignment, app->long_alignment, app->byte_order);
+ app->uint64_t_alignment, app->long_alignment,
+ app->byte_order, app->version.major,
+ app->version.minor);
if (ret < 0) {
goto error;
}
goto error;
}
+ /*
+ * Keep metadata key so we can identify it on the consumer side. Assign it
+ * to the registry *before* we ask the consumer so we avoid the race of the
+ * consumer requesting the metadata and the ask_channel call on our side
+ * did not returned yet.
+ */
+ registry->metadata_key = metadata->key;
+
/*
* Ask the metadata channel creation to the consumer. The metadata object
* will be created by the consumer and kept their. However, the stream is
goto error_consumer;
}
- /* Keep metadata key so we can identify it on the consumer side. */
- registry->metadata_key = metadata->key;
-
DBG2("UST metadata with key %" PRIu64 " created for app pid %d",
metadata->key, app->pid);
/*
* Start tracing for a specific UST session and app.
*/
+static
int ust_app_start_trace(struct ltt_ust_session *usess, struct ust_app *app)
{
int ret = 0;
/*
* Stop tracing for a specific UST session and app.
*/
+static
int ust_app_stop_trace(struct ltt_ust_session *usess, struct ust_app *app)
{
int ret = 0;
- struct lttng_ht_iter iter;
struct ust_app_session *ua_sess;
- struct ust_app_channel *ua_chan;
struct ust_registry_session *registry;
DBG("Stopping tracing for ust app pid %d", app->pid);
health_code_update();
+ registry = get_session_registry(ua_sess);
+ assert(registry);
+ /* Push metadata for application before freeing the application. */
+ (void) push_metadata(registry, ua_sess->consumer);
+
+ pthread_mutex_unlock(&ua_sess->lock);
+end_no_session:
+ rcu_read_unlock();
+ health_code_update();
+ return 0;
+
+error_rcu_unlock:
+ pthread_mutex_unlock(&ua_sess->lock);
+ rcu_read_unlock();
+ health_code_update();
+ return -1;
+}
+
+/*
+ * Flush buffers for a specific UST session and app.
+ */
+static
+int ust_app_flush_trace(struct ltt_ust_session *usess, struct ust_app *app)
+{
+ int ret = 0;
+ struct lttng_ht_iter iter;
+ struct ust_app_session *ua_sess;
+ struct ust_app_channel *ua_chan;
+
+ DBG("Flushing buffers for ust app pid %d", app->pid);
+
+ rcu_read_lock();
+
+ if (!app->compatible) {
+ goto end_no_session;
+ }
+
+ ua_sess = lookup_session_by_app(usess, app);
+ if (ua_sess == NULL) {
+ goto end_no_session;
+ }
+
+ pthread_mutex_lock(&ua_sess->lock);
+
+ health_code_update();
+
/* Flushing buffers */
cds_lfht_for_each_entry(ua_sess->channels->ht, &iter.iter, ua_chan,
node.node) {
health_code_update();
- registry = get_session_registry(ua_sess);
- assert(registry);
- /* Push metadata for application before freeing the application. */
- (void) push_metadata(registry, ua_sess->consumer);
-
pthread_mutex_unlock(&ua_sess->lock);
end_no_session:
rcu_read_unlock();
health_code_update();
return 0;
-
-error_rcu_unlock:
- pthread_mutex_unlock(&ua_sess->lock);
- rcu_read_unlock();
- health_code_update();
- return -1;
}
/*
rcu_read_lock();
- /* Flush all per UID buffers associated to that session. */
- if (usess->buffer_type == LTTNG_BUFFER_PER_UID) {
+ cds_lfht_for_each_entry(ust_app_ht->ht, &iter.iter, app, pid_n.node) {
+ ret = ust_app_stop_trace(usess, app);
+ if (ret < 0) {
+ /* Continue to next apps even on error */
+ continue;
+ }
+ }
+
+ /* Flush buffers */
+ switch (usess->buffer_type) {
+ case LTTNG_BUFFER_PER_UID:
+ {
struct buffer_reg_uid *reg;
+
+ /* Flush all per UID buffers associated to that session. */
cds_list_for_each_entry(reg, &usess->buffer_reg_uid_list, lnode) {
struct buffer_reg_channel *reg_chan;
struct consumer_socket *socket;
(void) consumer_flush_channel(socket, reg_chan->consumer_key);
}
}
+ break;
}
-
- cds_lfht_for_each_entry(ust_app_ht->ht, &iter.iter, app, pid_n.node) {
- ret = ust_app_stop_trace(usess, app);
- if (ret < 0) {
- /* Continue to next apps even on error */
- continue;
+ case LTTNG_BUFFER_PER_PID:
+ cds_lfht_for_each_entry(ust_app_ht->ht, &iter.iter, app, pid_n.node) {
+ ret = ust_app_flush_trace(usess, app);
+ if (ret < 0) {
+ /* Continue to next apps even on error */
+ continue;
+ }
}
+ break;
+ default:
+ assert(0);
+ break;
}
rcu_read_unlock();
sizeof(ua_chan->name))) {
ret = create_ust_app_metadata(ua_sess, app, usess->consumer,
&ua_chan->attr);
+ if (ret < 0) {
+ goto error_unlock;
+ }
/* Remove it from the hash table and continue!. */
ret = lttng_ht_del(ua_sess->channels, &iter);
assert(!ret);
continue;
} else {
ret = do_create_channel(app, usess, ua_sess, ua_chan);
- }
- if (ret < 0) {
- /*
- * Stop everything. On error, the application failed, no more
- * file descriptor are available or ENOMEM so stopping here is
- * the only thing we can do for now.
- */
- goto error_unlock;
+ if (ret < 0) {
+ /*
+ * Stop everything. On error, the application failed, no more
+ * file descriptor are available or ENOMEM so stopping here is
+ * the only thing we can do for now.
+ */
+ goto error_unlock;
+ }
}
cds_lfht_for_each_entry(ua_chan->ctx->ht, &iter_ctx.iter, ua_ctx,