X-Git-Url: https://git.liburcu.org/?a=blobdiff_plain;f=src%2Fbin%2Flttng-sessiond%2Fconsumer.c;h=17fc919768fa98610ad4a0afddfdb58ef38db773;hb=567eb353c7f88e2fdaa106eb7e0a38dbb8717792;hp=d91c3e60bf0e16156232ca3a2f251d0d6ae779e6;hpb=6dc3064a30b0cc7cfa9fdd22da1963525dfb7388;p=lttng-tools.git diff --git a/src/bin/lttng-sessiond/consumer.c b/src/bin/lttng-sessiond/consumer.c index d91c3e60b..17fc91976 100644 --- a/src/bin/lttng-sessiond/consumer.c +++ b/src/bin/lttng-sessiond/consumer.c @@ -408,6 +408,28 @@ error: return output; } +/* + * Iterate over the consumer output socket hash table and destroy them. The + * socket file descriptor are only closed if the consumer output was + * registered meaning it's an external consumer. + */ +void consumer_destroy_output_sockets(struct consumer_output *obj) +{ + struct lttng_ht_iter iter; + struct consumer_socket *socket; + + if (!obj->socks) { + return; + } + + rcu_read_lock(); + cds_lfht_for_each_entry(obj->socks->ht, &iter.iter, socket, node.node) { + consumer_del_socket(socket, obj); + consumer_destroy_socket(socket); + } + rcu_read_unlock(); +} + /* * Delete the consumer_output object from the list and free the ptr. * @@ -419,17 +441,9 @@ void consumer_destroy_output(struct consumer_output *obj) return; } - if (obj->socks) { - struct lttng_ht_iter iter; - struct consumer_socket *socket; - - rcu_read_lock(); - cds_lfht_for_each_entry(obj->socks->ht, &iter.iter, socket, node.node) { - consumer_del_socket(socket, obj); - consumer_destroy_socket(socket); - } - rcu_read_unlock(); + consumer_destroy_output_sockets(obj); + if (obj->socks) { /* Finally destroy HT */ ht_cleanup_push(obj->socks); } @@ -721,7 +735,10 @@ void consumer_init_ask_channel_comm_msg(struct lttcomm_consumer_msg *msg, unsigned char *uuid, uint32_t chan_id, uint64_t tracefile_size, - uint64_t tracefile_count) + uint64_t tracefile_count, + uint64_t session_id_per_pid, + unsigned int monitor, + uint32_t ust_app_uid) { assert(msg); @@ -737,6 +754,7 @@ void consumer_init_ask_channel_comm_msg(struct lttcomm_consumer_msg *msg, msg->u.ask_channel.output = output; msg->u.ask_channel.type = type; msg->u.ask_channel.session_id = session_id; + msg->u.ask_channel.session_id_per_pid = session_id_per_pid; msg->u.ask_channel.uid = uid; msg->u.ask_channel.gid = gid; msg->u.ask_channel.relayd_id = relayd_id; @@ -744,12 +762,16 @@ void consumer_init_ask_channel_comm_msg(struct lttcomm_consumer_msg *msg, msg->u.ask_channel.chan_id = chan_id; msg->u.ask_channel.tracefile_size = tracefile_size; msg->u.ask_channel.tracefile_count = tracefile_count; + msg->u.ask_channel.monitor = monitor; + msg->u.ask_channel.ust_app_uid = ust_app_uid; memcpy(msg->u.ask_channel.uuid, uuid, sizeof(msg->u.ask_channel.uuid)); - strncpy(msg->u.ask_channel.pathname, pathname, - sizeof(msg->u.ask_channel.pathname)); - msg->u.ask_channel.pathname[sizeof(msg->u.ask_channel.pathname)-1] = '\0'; + if (pathname) { + strncpy(msg->u.ask_channel.pathname, pathname, + sizeof(msg->u.ask_channel.pathname)); + msg->u.ask_channel.pathname[sizeof(msg->u.ask_channel.pathname)-1] = '\0'; + } strncpy(msg->u.ask_channel.name, name, sizeof(msg->u.ask_channel.name)); msg->u.ask_channel.name[sizeof(msg->u.ask_channel.name) - 1] = '\0'; @@ -771,7 +793,8 @@ void consumer_init_channel_comm_msg(struct lttcomm_consumer_msg *msg, enum lttng_event_output output, int type, uint64_t tracefile_size, - uint64_t tracefile_count) + uint64_t tracefile_count, + unsigned int monitor) { assert(msg); @@ -790,6 +813,7 @@ void consumer_init_channel_comm_msg(struct lttcomm_consumer_msg *msg, msg->u.channel.type = type; msg->u.channel.tracefile_size = tracefile_size; msg->u.channel.tracefile_count = tracefile_count; + msg->u.channel.monitor = monitor; strncpy(msg->u.channel.pathname, pathname, sizeof(msg->u.channel.pathname)); @@ -1188,7 +1212,7 @@ end: */ int consumer_snapshot_channel(struct consumer_socket *socket, uint64_t key, struct snapshot_output *output, int metadata, uid_t uid, gid_t gid, - int wait) + const char *session_path, int wait, int max_stream_size) { int ret; struct lttcomm_consumer_msg msg; @@ -1201,30 +1225,34 @@ int consumer_snapshot_channel(struct consumer_socket *socket, uint64_t key, DBG("Consumer snapshot channel key %" PRIu64, key); memset(&msg, 0, sizeof(msg)); - msg.cmd_type = LTTNG_CONSUMER_SNAPSHOT_CHANNEL; msg.u.snapshot_channel.key = key; - msg.u.snapshot_channel.max_size = output->max_size; + msg.u.snapshot_channel.max_stream_size = max_stream_size; msg.u.snapshot_channel.metadata = metadata; if (output->consumer->type == CONSUMER_DST_NET) { msg.u.snapshot_channel.relayd_id = output->consumer->net_seq_index; msg.u.snapshot_channel.use_relayd = 1; ret = snprintf(msg.u.snapshot_channel.pathname, - sizeof(msg.u.snapshot_channel.pathname), "%s/%s", - output->consumer->subdir, DEFAULT_SNAPSHOT_NAME); + sizeof(msg.u.snapshot_channel.pathname), + "%s/%s-%s-%" PRIu64 "%s", output->consumer->subdir, + output->name, output->datetime, output->nb_snapshot, + session_path); if (ret < 0) { ret = -LTTNG_ERR_NOMEM; goto error; } } else { ret = snprintf(msg.u.snapshot_channel.pathname, - sizeof(msg.u.snapshot_channel.pathname), "%s/%s", - output->consumer->dst.trace_path, DEFAULT_SNAPSHOT_NAME); + sizeof(msg.u.snapshot_channel.pathname), + "%s/%s-%s-%" PRIu64 "%s", output->consumer->dst.trace_path, + output->name, output->datetime, output->nb_snapshot, + session_path); if (ret < 0) { ret = -LTTNG_ERR_NOMEM; goto error; } + msg.u.snapshot_channel.relayd_id = (uint64_t) -1ULL; /* Create directory. Ignore if exist. */ ret = run_as_mkdir_recursive(msg.u.snapshot_channel.pathname,