X-Git-Url: https://git.liburcu.org/?a=blobdiff_plain;f=src%2Fcommon%2Fconsumer%2Fconsumer.cpp;h=8326a8e391d64d7506fcab19a981558955fcfd0b;hb=c715ddc950bf653d9456d92c6ead2e3cbd3c54ae;hp=685d55f838120a4b94cf407d9e420e03cf294e32;hpb=c9e313bc594f40a86eed237dce222c0fc99c957f;p=lttng-tools.git diff --git a/src/common/consumer/consumer.cpp b/src/common/consumer/consumer.cpp index 685d55f83..8326a8e39 100644 --- a/src/common/consumer/consumer.cpp +++ b/src/common/consumer/consumer.cpp @@ -52,12 +52,22 @@ enum consumer_channel_action { CONSUMER_CHANNEL_QUIT, }; +namespace { struct consumer_channel_msg { enum consumer_channel_action action; struct lttng_consumer_channel *chan; /* add */ uint64_t key; /* del */ }; +/* + * Global hash table containing respectively metadata and data streams. The + * stream element in this ht should only be updated by the metadata poll thread + * for the metadata and the data poll thread for the data. + */ +struct lttng_ht *metadata_ht; +struct lttng_ht *data_ht; +} /* namespace */ + /* Flag used to temporarily pause data consumption from testpoints. */ int data_consumption_paused; @@ -69,14 +79,6 @@ int data_consumption_paused; */ int consumer_quit; -/* - * Global hash table containing respectively metadata and data streams. The - * stream element in this ht should only be updated by the metadata poll thread - * for the metadata and the data poll thread for the data. - */ -static struct lttng_ht *metadata_ht; -static struct lttng_ht *data_ht; - static const char *get_consumer_domain(void) { switch (the_consumer_data.type) { @@ -207,7 +209,7 @@ static struct lttng_consumer_stream *find_stream(uint64_t key, lttng_ht_lookup(ht, &key, &iter); node = lttng_ht_iter_get_node_u64(&iter); if (node != NULL) { - stream = caa_container_of(node, struct lttng_consumer_stream, node); + stream = lttng::utils::container_of(node, <tng_consumer_stream::node); } rcu_read_unlock(); @@ -255,7 +257,7 @@ struct lttng_consumer_channel *consumer_find_channel(uint64_t key) lttng_ht_lookup(the_consumer_data.channel_ht, &key, &iter); node = lttng_ht_iter_get_node_u64(&iter); if (node != NULL) { - channel = caa_container_of(node, struct lttng_consumer_channel, node); + channel = lttng::utils::container_of(node, <tng_consumer_channel::node); } return channel; @@ -290,9 +292,9 @@ static void steal_channel_key(uint64_t key) static void free_channel_rcu(struct rcu_head *head) { struct lttng_ht_node_u64 *node = - caa_container_of(head, struct lttng_ht_node_u64, head); + lttng::utils::container_of(head, <tng_ht_node_u64::head); struct lttng_consumer_channel *channel = - caa_container_of(node, struct lttng_consumer_channel, node); + lttng::utils::container_of(node, <tng_consumer_channel::node); switch (the_consumer_data.type) { case LTTNG_CONSUMER_KERNEL: @@ -314,9 +316,9 @@ static void free_channel_rcu(struct rcu_head *head) static void free_relayd_rcu(struct rcu_head *head) { struct lttng_ht_node_u64 *node = - caa_container_of(head, struct lttng_ht_node_u64, head); + lttng::utils::container_of(head, <tng_ht_node_u64::head); struct consumer_relayd_sock_pair *relayd = - caa_container_of(node, struct consumer_relayd_sock_pair, node); + lttng::utils::container_of(node, &consumer_relayd_sock_pair::node); /* * Close all sockets. This is done in the call RCU since we don't want the @@ -662,7 +664,7 @@ static struct consumer_relayd_sock_pair *consumer_allocate_relayd_sock_pair( goto error; } - obj = (consumer_relayd_sock_pair *) zmalloc(sizeof(struct consumer_relayd_sock_pair)); + obj = zmalloc(); if (obj == NULL) { PERROR("zmalloc relayd sock"); goto error; @@ -703,7 +705,7 @@ struct consumer_relayd_sock_pair *consumer_find_relayd(uint64_t key) lttng_ht_lookup(the_consumer_data.relayd_ht, &key, &iter); node = lttng_ht_iter_get_node_u64(&iter); if (node != NULL) { - relayd = caa_container_of(node, struct consumer_relayd_sock_pair, node); + relayd = lttng::utils::container_of(node, &consumer_relayd_sock_pair::node); } error: @@ -1029,7 +1031,7 @@ struct lttng_consumer_channel *consumer_allocate_channel(uint64_t key, } } - channel = (lttng_consumer_channel *) zmalloc(sizeof(*channel)); + channel = zmalloc(); if (channel == NULL) { PERROR("malloc struct lttng_consumer_channel"); goto end; @@ -1425,7 +1427,7 @@ struct lttng_consumer_local_data *lttng_consumer_create( the_consumer_data.type == type); the_consumer_data.type = type; - ctx = (lttng_consumer_local_data *) zmalloc(sizeof(struct lttng_consumer_local_data)); + ctx = zmalloc(); if (ctx == NULL) { PERROR("allocating context"); goto error; @@ -2561,7 +2563,7 @@ void *consumer_thread_data_poll(void *data) health_code_update(); - local_stream = (lttng_consumer_stream **) zmalloc(sizeof(struct lttng_consumer_stream *)); + local_stream = zmalloc(); if (local_stream == NULL) { PERROR("local_stream malloc"); goto end; @@ -2586,18 +2588,14 @@ void *consumer_thread_data_poll(void *data) local_stream = NULL; /* Allocate for all fds */ - pollfd = (struct pollfd *) zmalloc((the_consumer_data.stream_count + - nb_pipes_fd) * - sizeof(struct pollfd)); + pollfd = calloc(the_consumer_data.stream_count + nb_pipes_fd); if (pollfd == NULL) { PERROR("pollfd malloc"); pthread_mutex_unlock(&the_consumer_data.lock); goto end; } - local_stream = (lttng_consumer_stream **) zmalloc((the_consumer_data.stream_count + - nb_pipes_fd) * - sizeof(struct lttng_consumer_stream *)); + local_stream = calloc(the_consumer_data.stream_count + nb_pipes_fd); if (local_stream == NULL) { PERROR("local_stream malloc"); pthread_mutex_unlock(&the_consumer_data.lock); @@ -2715,7 +2713,7 @@ void *consumer_thread_data_poll(void *data) consumer_del_stream(local_stream[i], data_ht); local_stream[i] = NULL; } else if (len > 0) { - local_stream[i]->data_read = 1; + local_stream[i]->has_data_left_to_be_read_before_teardown = 1; } } } @@ -2746,7 +2744,7 @@ void *consumer_thread_data_poll(void *data) consumer_del_stream(local_stream[i], data_ht); local_stream[i] = NULL; } else if (len > 0) { - local_stream[i]->data_read = 1; + local_stream[i]->has_data_left_to_be_read_before_teardown = 1; } } } @@ -2766,37 +2764,45 @@ void *consumer_thread_data_poll(void *data) pollfd[i].fd); lttng_ustconsumer_on_stream_hangup(local_stream[i]); /* Attempt read again, for the data we just flushed. */ - local_stream[i]->data_read = 1; + local_stream[i]->has_data_left_to_be_read_before_teardown = 1; } /* + * When a stream's pipe dies (hup/err/nval), an "inactive producer" flush is + * performed. This type of flush ensures that a new packet is produced no + * matter the consumed/produced positions are. + * + * This, in turn, causes the next pass to see that data available for the + * stream. When we come back here, we can be assured that all available + * data has been consumed and we can finally destroy the stream. + * * If the poll flag is HUP/ERR/NVAL and we have * read no data in this pass, we can remove the * stream from its hash table. */ if ((pollfd[i].revents & POLLHUP)) { DBG("Polling fd %d tells it has hung up.", pollfd[i].fd); - if (!local_stream[i]->data_read) { + if (!local_stream[i]->has_data_left_to_be_read_before_teardown) { consumer_del_stream(local_stream[i], data_ht); local_stream[i] = NULL; num_hup++; } } else if (pollfd[i].revents & POLLERR) { ERR("Error returned in polling fd %d.", pollfd[i].fd); - if (!local_stream[i]->data_read) { + if (!local_stream[i]->has_data_left_to_be_read_before_teardown) { consumer_del_stream(local_stream[i], data_ht); local_stream[i] = NULL; num_hup++; } } else if (pollfd[i].revents & POLLNVAL) { ERR("Polling fd %d tells fd is not open.", pollfd[i].fd); - if (!local_stream[i]->data_read) { + if (!local_stream[i]->has_data_left_to_be_read_before_teardown) { consumer_del_stream(local_stream[i], data_ht); local_stream[i] = NULL; num_hup++; } } if (local_stream[i] != NULL) { - local_stream[i]->data_read = 0; + local_stream[i]->has_data_left_to_be_read_before_teardown = 0; } } } @@ -4699,7 +4705,7 @@ end: enum lttcomm_return_code lttng_consumer_init_command( struct lttng_consumer_local_data *ctx, - const lttng_uuid sessiond_uuid) + const lttng_uuid& sessiond_uuid) { enum lttcomm_return_code ret; char uuid_str[LTTNG_UUID_STR_LEN]; @@ -4710,7 +4716,7 @@ enum lttcomm_return_code lttng_consumer_init_command( } ctx->sessiond_uuid.is_set = true; - memcpy(ctx->sessiond_uuid.value, sessiond_uuid, sizeof(lttng_uuid)); + ctx->sessiond_uuid.value = sessiond_uuid; ret = LTTCOMM_CONSUMERD_SUCCESS; lttng_uuid_to_str(sessiond_uuid, uuid_str); DBG("Received session daemon UUID: %s", uuid_str);