[lttng-dev] [PATCH lttng-tools 3/4] Make stream hash tables global to the consumer
Mathieu Desnoyers
mathieu.desnoyers at efficios.com
Mon Oct 15 13:57:54 EDT 2012
* David Goulet (dgoulet at efficios.com) wrote:
>
>
> Mathieu Desnoyers:
> > * David Goulet (dgoulet at efficios.com) wrote:
> >> The data stream hash table is now global to the consumer and used in the
> >> data thread. The consumer_data stream_ht is no longer used to track the
> >> data streams but instead will be used (and possibly renamed) by the
> >> session daemon poll thread to keep track of streams on a per session id
> >> basis for the upcoming feature that check traced data availability.
> >>
> >> For now, in order to avoid mind bugging problems to access the streams,
> >> both hash table are now defined globally (metadata and data). However,
> >> stream update are still done in a single thread. Don't count on this to
> >> be guaranteed in the next commits.
> >>
> >> Signed-off-by: David Goulet <dgoulet at efficios.com>
> >> ---
> >> src/common/consumer.c | 91 +++++++++++++++++++++++++-------
> >> src/common/consumer.h | 9 ++--
> >> src/common/ust-consumer/ust-consumer.c | 2 -
> >> 3 files changed, 75 insertions(+), 27 deletions(-)
> >>
> >> diff --git a/src/common/consumer.c b/src/common/consumer.c
> >> index 1d2b1f7..1fb9960 100644
> >> --- a/src/common/consumer.c
> >> +++ b/src/common/consumer.c
> >> @@ -59,6 +59,17 @@ int consumer_poll_timeout = -1;
> >> volatile int consumer_quit = 0;
> >>
> >> /*
> >> + * The following two hash tables are visible by all threads which are separated
> >> + * in different source files.
> >> + *
> >> + * 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 = NULL;
> >> +struct lttng_ht *data_ht = NULL;
> >> +
> >> +/*
> >> * Find a stream. The consumer_data.lock must be locked during this
> >> * call.
> >> */
> >> @@ -433,19 +444,24 @@ end:
> >> /*
> >> * Add a stream to the global list protected by a mutex.
> >> */
> >> -int consumer_add_stream(struct lttng_consumer_stream *stream)
> >> +static int consumer_add_stream(struct lttng_consumer_stream *stream,
> >> + struct lttng_ht *ht)
> >> {
> >> int ret = 0;
> >> struct consumer_relayd_sock_pair *relayd;
> >>
> >> assert(stream);
> >> + assert(ht);
> >>
> >> DBG3("Adding consumer stream %d", stream->key);
> >>
> >> pthread_mutex_lock(&consumer_data.lock);
> >> rcu_read_lock();
> >>
> >> - lttng_ht_add_unique_ulong(consumer_data.stream_ht, &stream->node);
> >> + /* Steal stream identifier to avoid having streams with the same key */
> >> + consumer_steal_stream_key(stream->key, ht);
> >
> > I don't understand why suddenly this change is needed. Considering what
> > this patch should be doing (just moving a ht from per-thread to global),
> > it should not have any behavior impact.
>
> We move the steal stream key from the sessiond thread to the add_stream
> function call since we do not use the consumer_data hash table anymore
> (stream_ht) and uses per thread hashtable (global for now though).
>
> If you look below, you'll see that the steal stream key call is removed
> (using the consumer data stream_ht).
>
> This commit makes sure that both consumer_add_stream and
> add_metadata_stream steal the stream key if needed.
ok, makes sense.
Thanks!
Mathieu
>
> Thanks
> David
>
> >
> > Thanks,
> >
> > Mathieu
> >
> >> +
> >> + lttng_ht_add_unique_ulong(ht, &stream->node);
> >>
> >> /* Check and cleanup relayd */
> >> relayd = consumer_find_relayd(stream->net_seq_idx);
> >> @@ -783,9 +799,9 @@ end:
> >> *
> >> * Returns the number of fds in the structures.
> >> */
> >> -int consumer_update_poll_array(
> >> +static int consumer_update_poll_array(
> >> struct lttng_consumer_local_data *ctx, struct pollfd **pollfd,
> >> - struct lttng_consumer_stream **local_stream)
> >> + struct lttng_consumer_stream **local_stream, struct lttng_ht *ht)
> >> {
> >> int i = 0;
> >> struct lttng_ht_iter iter;
> >> @@ -793,8 +809,7 @@ int consumer_update_poll_array(
> >>
> >> DBG("Updating poll fd array");
> >> rcu_read_lock();
> >> - cds_lfht_for_each_entry(consumer_data.stream_ht->ht, &iter.iter, stream,
> >> - node.node) {
> >> + cds_lfht_for_each_entry(ht->ht, &iter.iter, stream, node.node) {
> >> if (stream->state != LTTNG_CONSUMER_ACTIVE_STREAM) {
> >> continue;
> >> }
> >> @@ -1523,6 +1538,33 @@ int lttng_consumer_recv_cmd(struct lttng_consumer_local_data *ctx,
> >> /*
> >> * Iterate over all streams of the hashtable and free them properly.
> >> *
> >> + * WARNING: *MUST* be used with data stream only.
> >> + */
> >> +static void destroy_data_stream_ht(struct lttng_ht *ht)
> >> +{
> >> + int ret;
> >> + struct lttng_ht_iter iter;
> >> + struct lttng_consumer_stream *stream;
> >> +
> >> + if (ht == NULL) {
> >> + return;
> >> + }
> >> +
> >> + rcu_read_lock();
> >> + cds_lfht_for_each_entry(ht->ht, &iter.iter, stream, node.node) {
> >> + ret = lttng_ht_del(ht, &iter);
> >> + assert(!ret);
> >> +
> >> + call_rcu(&stream->node.head, consumer_free_stream);
> >> + }
> >> + rcu_read_unlock();
> >> +
> >> + lttng_ht_destroy(ht);
> >> +}
> >> +
> >> +/*
> >> + * Iterate over all streams of the hashtable and free them properly.
> >> + *
> >> * XXX: Should not be only for metadata stream or else use an other name.
> >> */
> >> static void destroy_stream_ht(struct lttng_ht *ht)
> >> @@ -1711,6 +1753,9 @@ static int consumer_add_metadata_stream(struct lttng_consumer_stream *stream,
> >> uatomic_dec(&stream->chan->nb_init_streams);
> >> }
> >>
> >> + /* Steal stream identifier to avoid having streams with the same key */
> >> + consumer_steal_stream_key(stream->key, ht);
> >> +
> >> lttng_ht_add_unique_ulong(ht, &stream->waitfd_node);
> >> rcu_read_unlock();
> >>
> >> @@ -1729,7 +1774,6 @@ void *consumer_thread_metadata_poll(void *data)
> >> struct lttng_consumer_stream *stream = NULL;
> >> struct lttng_ht_iter iter;
> >> struct lttng_ht_node_ulong *node;
> >> - struct lttng_ht *metadata_ht = NULL;
> >> struct lttng_poll_event events;
> >> struct lttng_consumer_local_data *ctx = data;
> >> ssize_t len;
> >> @@ -1738,11 +1782,6 @@ void *consumer_thread_metadata_poll(void *data)
> >>
> >> DBG("Thread metadata poll started");
> >>
> >> - metadata_ht = lttng_ht_new(0, LTTNG_HT_TYPE_ULONG);
> >> - if (metadata_ht == NULL) {
> >> - goto end;
> >> - }
> >> -
> >> /* Size is set to 1 for the consumer_metadata pipe */
> >> ret = lttng_poll_create(&events, 2, LTTNG_CLOEXEC);
> >> if (ret < 0) {
> >> @@ -1918,6 +1957,11 @@ void *consumer_thread_data_poll(void *data)
> >>
> >> rcu_register_thread();
> >>
> >> + data_ht = lttng_ht_new(0, LTTNG_HT_TYPE_ULONG);
> >> + if (data_ht == NULL) {
> >> + goto end;
> >> + }
> >> +
> >> local_stream = zmalloc(sizeof(struct lttng_consumer_stream));
> >>
> >> while (1) {
> >> @@ -1955,7 +1999,8 @@ void *consumer_thread_data_poll(void *data)
> >> pthread_mutex_unlock(&consumer_data.lock);
> >> goto end;
> >> }
> >> - ret = consumer_update_poll_array(ctx, &pollfd, local_stream);
> >> + ret = consumer_update_poll_array(ctx, &pollfd, local_stream,
> >> + data_ht);
> >> if (ret < 0) {
> >> ERR("Error in allocating pollfd or local_outfds");
> >> lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_POLL_ERROR);
> >> @@ -2015,7 +2060,7 @@ void *consumer_thread_data_poll(void *data)
> >> continue;
> >> }
> >>
> >> - ret = consumer_add_stream(new_stream);
> >> + ret = consumer_add_stream(new_stream, data_ht);
> >> if (ret) {
> >> ERR("Consumer add stream %d failed. Continuing",
> >> new_stream->key);
> >> @@ -2088,22 +2133,19 @@ void *consumer_thread_data_poll(void *data)
> >> if ((pollfd[i].revents & POLLHUP)) {
> >> DBG("Polling fd %d tells it has hung up.", pollfd[i].fd);
> >> if (!local_stream[i]->data_read) {
> >> - consumer_del_stream(local_stream[i],
> >> - consumer_data.stream_ht);
> >> + consumer_del_stream(local_stream[i], data_ht);
> >> num_hup++;
> >> }
> >> } else if (pollfd[i].revents & POLLERR) {
> >> ERR("Error returned in polling fd %d.", pollfd[i].fd);
> >> if (!local_stream[i]->data_read) {
> >> - consumer_del_stream(local_stream[i],
> >> - consumer_data.stream_ht);
> >> + consumer_del_stream(local_stream[i], data_ht);
> >> 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) {
> >> - consumer_del_stream(local_stream[i],
> >> - consumer_data.stream_ht);
> >> + consumer_del_stream(local_stream[i], data_ht);
> >> num_hup++;
> >> }
> >> }
> >> @@ -2131,6 +2173,10 @@ end:
> >> */
> >> close(ctx->consumer_metadata_pipe[1]);
> >>
> >> + if (data_ht) {
> >> + destroy_data_stream_ht(data_ht);
> >> + }
> >> +
> >> rcu_unregister_thread();
> >> return NULL;
> >> }
> >> @@ -2299,6 +2345,11 @@ void lttng_consumer_init(void)
> >> consumer_data.stream_ht = lttng_ht_new(0, LTTNG_HT_TYPE_ULONG);
> >> consumer_data.channel_ht = lttng_ht_new(0, LTTNG_HT_TYPE_ULONG);
> >> consumer_data.relayd_ht = lttng_ht_new(0, LTTNG_HT_TYPE_ULONG);
> >> +
> >> + metadata_ht = lttng_ht_new(0, LTTNG_HT_TYPE_ULONG);
> >> + assert(metadata_ht);
> >> + data_ht = lttng_ht_new(0, LTTNG_HT_TYPE_ULONG);
> >> + assert(data_ht);
> >> }
> >>
> >> /*
> >> diff --git a/src/common/consumer.h b/src/common/consumer.h
> >> index 8e5891a..6bce96d 100644
> >> --- a/src/common/consumer.h
> >> +++ b/src/common/consumer.h
> >> @@ -275,6 +275,10 @@ struct lttng_consumer_global_data {
> >> struct lttng_ht *relayd_ht;
> >> };
> >>
> >> +/* Defined in consumer.c and coupled with explanations */
> >> +extern struct lttng_ht *metadata_ht;
> >> +extern struct lttng_ht *data_ht;
> >> +
> >> /*
> >> * Init consumer data structures.
> >> */
> >> @@ -324,10 +328,6 @@ extern void lttng_consumer_sync_trace_file(
> >> */
> >> extern int lttng_consumer_poll_socket(struct pollfd *kconsumer_sockpoll);
> >>
> >> -extern int consumer_update_poll_array(
> >> - struct lttng_consumer_local_data *ctx, struct pollfd **pollfd,
> >> - struct lttng_consumer_stream **local_consumer_streams);
> >> -
> >> extern struct lttng_consumer_stream *consumer_allocate_stream(
> >> int channel_key, int stream_key,
> >> int shm_fd, int wait_fd,
> >> @@ -340,7 +340,6 @@ extern struct lttng_consumer_stream *consumer_allocate_stream(
> >> int net_index,
> >> int metadata_flag,
> >> int *alloc_ret);
> >> -extern int consumer_add_stream(struct lttng_consumer_stream *stream);
> >> extern void consumer_del_stream(struct lttng_consumer_stream *stream,
> >> struct lttng_ht *ht);
> >> extern void consumer_del_metadata_stream(struct lttng_consumer_stream *stream,
> >> diff --git a/src/common/ust-consumer/ust-consumer.c b/src/common/ust-consumer/ust-consumer.c
> >> index 4ca4b84..3b41e55 100644
> >> --- a/src/common/ust-consumer/ust-consumer.c
> >> +++ b/src/common/ust-consumer/ust-consumer.c
> >> @@ -233,8 +233,6 @@ int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
> >> consumer_del_stream(new_stream, NULL);
> >> goto end_nosignal;
> >> }
> >> - /* Steal stream identifier to avoid having streams with the same key */
> >> - consumer_steal_stream_key(new_stream->key, consumer_data.stream_ht);
> >>
> >> /* The stream is not metadata. Get relayd reference if exists. */
> >> relayd = consumer_find_relayd(msg.u.stream.net_index);
> >> --
> >> 1.7.10.4
> >>
> >>
> >> _______________________________________________
> >> lttng-dev mailing list
> >> lttng-dev at lists.lttng.org
> >> http://lists.lttng.org/cgi-bin/mailman/listinfo/lttng-dev
> >
--
Mathieu Desnoyers
Operating System Efficiency R&D Consultant
EfficiOS Inc.
http://www.efficios.com
More information about the lttng-dev
mailing list