X-Git-Url: https://git.lttng.org/?a=blobdiff_plain;f=src%2Fcommon%2Fconsumer.c;h=5574c5f0e1d545691f6837d2d4100a10c2a59e48;hb=4b29f1cec9f8e93d4dde0c982f30e3201f0f4e65;hp=37a4e01978dc05324fb5739e3e681e84e82d1ad8;hpb=0d08d75e73e92edb002f3d480004ccb687cf1eeb;p=lttng-tools.git diff --git a/src/common/consumer.c b/src/common/consumer.c index 37a4e0197..5574c5f0e 100644 --- a/src/common/consumer.c +++ b/src/common/consumer.c @@ -41,6 +41,7 @@ #include #include "consumer.h" +#include "consumer-stream.h" struct lttng_consumer_global_data consumer_data = { .stream_count = 0, @@ -254,10 +255,8 @@ static void free_relayd_rcu(struct rcu_head *head) /* * Destroy and free relayd socket pair object. - * - * This function MUST be called with the consumer_data lock acquired. */ -static void destroy_relayd(struct consumer_relayd_sock_pair *relayd) +void consumer_destroy_relayd(struct consumer_relayd_sock_pair *relayd) { int ret; struct lttng_ht_iter iter; @@ -313,6 +312,25 @@ void consumer_del_channel(struct lttng_consumer_channel *channel) goto end; } + /* Empty no monitor streams list. */ + if (!channel->monitor) { + struct lttng_consumer_stream *stream, *stmp; + + /* + * So, these streams are not visible to any data thread. This is why we + * close and free them because they were never added to any data + * structure apart from this one. + */ + cds_list_for_each_entry_safe(stream, stmp, + &channel->stream_no_monitor_list.head, no_monitor_node) { + cds_list_del(&stream->no_monitor_node); + /* Close everything in that stream. */ + consumer_stream_close(stream); + /* Free the ressource. */ + consumer_stream_free(stream); + } + } + rcu_read_lock(); iter.iter.node = &channel->node.node; ret = lttng_ht_del(consumer_data.channel_ht, &iter); @@ -337,7 +355,7 @@ static void cleanup_relayd_ht(void) cds_lfht_for_each_entry(consumer_data.relayd_ht->ht, &iter.iter, relayd, node.node) { - destroy_relayd(relayd); + consumer_destroy_relayd(relayd); } rcu_read_unlock(); @@ -404,7 +422,7 @@ static void cleanup_relayd(struct consumer_relayd_sock_pair *relayd, * Delete the relayd from the relayd hash table, close the sockets and free * the object in a RCU call. */ - destroy_relayd(relayd); + consumer_destroy_relayd(relayd); /* Set inactive endpoint to all streams */ update_endpoint_status_by_netidx(netidx, CONSUMER_ENDPOINT_INACTIVE); @@ -436,130 +454,20 @@ void consumer_flag_relayd_for_destroy(struct consumer_relayd_sock_pair *relayd) /* Destroy the relayd if refcount is 0 */ if (uatomic_read(&relayd->refcount) == 0) { - destroy_relayd(relayd); + consumer_destroy_relayd(relayd); } } /* - * Remove a stream from the global list protected by a mutex. This - * function is also responsible for freeing its data structures. + * Completly destroy stream from every visiable data structure and the given + * hash table if one. + * + * One this call returns, the stream object is not longer usable nor visible. */ void consumer_del_stream(struct lttng_consumer_stream *stream, struct lttng_ht *ht) { - int ret; - struct lttng_ht_iter iter; - struct lttng_consumer_channel *free_chan = NULL; - struct consumer_relayd_sock_pair *relayd; - - assert(stream); - - DBG("Consumer del stream %d", stream->wait_fd); - - if (ht == NULL) { - /* Means the stream was allocated but not successfully added */ - goto free_stream_rcu; - } - - pthread_mutex_lock(&consumer_data.lock); - pthread_mutex_lock(&stream->lock); - - switch (consumer_data.type) { - case LTTNG_CONSUMER_KERNEL: - if (stream->mmap_base != NULL) { - ret = munmap(stream->mmap_base, stream->mmap_len); - if (ret != 0) { - PERROR("munmap"); - } - } - - if (stream->wait_fd >= 0) { - ret = close(stream->wait_fd); - if (ret) { - PERROR("close"); - } - } - break; - case LTTNG_CONSUMER32_UST: - case LTTNG_CONSUMER64_UST: - lttng_ustconsumer_del_stream(stream); - break; - default: - ERR("Unknown consumer_data type"); - assert(0); - goto end; - } - - rcu_read_lock(); - iter.iter.node = &stream->node.node; - ret = lttng_ht_del(ht, &iter); - assert(!ret); - - iter.iter.node = &stream->node_channel_id.node; - ret = lttng_ht_del(consumer_data.stream_per_chan_id_ht, &iter); - assert(!ret); - - iter.iter.node = &stream->node_session_id.node; - ret = lttng_ht_del(consumer_data.stream_list_ht, &iter); - assert(!ret); - rcu_read_unlock(); - - assert(consumer_data.stream_count > 0); - consumer_data.stream_count--; - - if (stream->out_fd >= 0) { - ret = close(stream->out_fd); - if (ret) { - PERROR("close"); - } - } - - /* Check and cleanup relayd */ - rcu_read_lock(); - relayd = consumer_find_relayd(stream->net_seq_idx); - if (relayd != NULL) { - uatomic_dec(&relayd->refcount); - assert(uatomic_read(&relayd->refcount) >= 0); - - /* Closing streams requires to lock the control socket. */ - pthread_mutex_lock(&relayd->ctrl_sock_mutex); - ret = relayd_send_close_stream(&relayd->control_sock, - stream->relayd_stream_id, - stream->next_net_seq_num - 1); - pthread_mutex_unlock(&relayd->ctrl_sock_mutex); - if (ret < 0) { - DBG("Unable to close stream on the relayd. Continuing"); - /* - * Continue here. There is nothing we can do for the relayd. - * Chances are that the relayd has closed the socket so we just - * continue cleaning up. - */ - } - - /* Both conditions are met, we destroy the relayd. */ - if (uatomic_read(&relayd->refcount) == 0 && - uatomic_read(&relayd->destroy_flag)) { - destroy_relayd(relayd); - } - } - rcu_read_unlock(); - - if (!uatomic_sub_return(&stream->chan->refcount, 1) - && !uatomic_read(&stream->chan->nb_init_stream_left)) { - free_chan = stream->chan; - } - -end: - consumer_data.need_update = 1; - pthread_mutex_unlock(&stream->lock); - pthread_mutex_unlock(&consumer_data.lock); - - if (free_chan) { - consumer_del_channel(free_chan); - } - -free_stream_rcu: - call_rcu(&stream->node.head, free_stream_rcu); + consumer_stream_destroy(stream, ht); } struct lttng_consumer_stream *consumer_allocate_stream(uint64_t channel_key, @@ -568,7 +476,7 @@ struct lttng_consumer_stream *consumer_allocate_stream(uint64_t channel_key, const char *channel_name, uid_t uid, gid_t gid, - int relayd_id, + uint64_t relayd_id, uint64_t session_id, int cpu, int *alloc_ret, @@ -620,8 +528,10 @@ struct lttng_consumer_stream *consumer_allocate_stream(uint64_t channel_key, /* Init session id node with the stream session id */ lttng_ht_node_init_u64(&stream->node_session_id, stream->session_id); - DBG3("Allocated stream %s (key %" PRIu64 ", chan_key %" PRIu64 " relayd_id %" PRIu64 ", session_id %" PRIu64, - stream->name, stream->key, channel_key, stream->net_seq_idx, stream->session_id); + DBG3("Allocated stream %s (key %" PRIu64 ", chan_key %" PRIu64 + " relayd_id %" PRIu64 ", session_id %" PRIu64, + stream->name, stream->key, channel_key, + stream->net_seq_idx, stream->session_id); rcu_read_unlock(); return stream; @@ -855,10 +765,11 @@ struct lttng_consumer_channel *consumer_allocate_channel(uint64_t key, const char *name, uid_t uid, gid_t gid, - int relayd_id, + uint64_t relayd_id, enum lttng_event_output output, uint64_t tracefile_size, - uint64_t tracefile_count) + uint64_t tracefile_count, + unsigned int monitor) { struct lttng_consumer_channel *channel; @@ -877,6 +788,34 @@ struct lttng_consumer_channel *consumer_allocate_channel(uint64_t key, channel->output = output; channel->tracefile_size = tracefile_size; channel->tracefile_count = tracefile_count; + channel->monitor = monitor; + + /* + * In monitor mode, the streams associated with the channel will be put in + * a special list ONLY owned by this channel. So, the refcount is set to 1 + * here meaning that the channel itself has streams that are referenced. + * + * On a channel deletion, once the channel is no longer visible, the + * refcount is decremented and checked for a zero value to delete it. With + * streams in no monitor mode, it will now be safe to destroy the channel. + */ + if (!channel->monitor) { + channel->refcount = 1; + } + + switch (output) { + case LTTNG_EVENT_SPLICE: + channel->output = CONSUMER_CHANNEL_SPLICE; + break; + case LTTNG_EVENT_MMAP: + channel->output = CONSUMER_CHANNEL_MMAP; + break; + default: + ERR("Allocate channel output unknown %d", output); + free(channel); + channel = NULL; + goto end; + } strncpy(channel->pathname, pathname, sizeof(channel->pathname)); channel->pathname[sizeof(channel->pathname) - 1] = '\0'; @@ -889,6 +828,7 @@ struct lttng_consumer_channel *consumer_allocate_channel(uint64_t key, channel->wait_fd = -1; CDS_INIT_LIST_HEAD(&channel->streams.head); + CDS_INIT_LIST_HEAD(&channel->stream_no_monitor_list.head); DBG("Allocated channel (key %" PRIu64 ")", channel->key) @@ -1894,7 +1834,6 @@ void consumer_del_metadata_stream(struct lttng_consumer_stream *stream, PERROR("munmap metadata stream"); } } - if (stream->wait_fd >= 0) { ret = close(stream->wait_fd); if (ret < 0) { @@ -1957,7 +1896,7 @@ void consumer_del_metadata_stream(struct lttng_consumer_stream *stream, /* Both conditions are met, we destroy the relayd. */ if (uatomic_read(&relayd->refcount) == 0 && uatomic_read(&relayd->destroy_flag)) { - destroy_relayd(relayd); + consumer_destroy_relayd(relayd); } } rcu_read_unlock(); @@ -3168,6 +3107,7 @@ int consumer_add_relayd_socket(int net_seq_idx, int sock_type, /* Assign new file descriptor */ relayd->control_sock.sock.fd = fd; + fd = -1; /* For error path */ /* Assign version values. */ relayd->control_sock.major = relayd_sock->major; relayd->control_sock.minor = relayd_sock->minor; @@ -3213,6 +3153,7 @@ int consumer_add_relayd_socket(int net_seq_idx, int sock_type, /* Assign new file descriptor */ relayd->data_sock.sock.fd = fd; + fd = -1; /* for eventual error paths */ /* Assign version values. */ relayd->data_sock.major = relayd_sock->major; relayd->data_sock.minor = relayd_sock->minor;