From mboxrd@z Thu Jan 1 00:00:00 1970 From: mathieu.desnoyers@efficios.com (Mathieu Desnoyers) Date: Mon, 15 Oct 2012 13:57:54 -0400 Subject: [lttng-dev] [PATCH lttng-tools 3/4] Make stream hash tables global to the consumer In-Reply-To: <507C3013.2050405@efficios.com> References: <1350052235-12198-1-git-send-email-dgoulet@efficios.com> <1350052235-12198-3-git-send-email-dgoulet@efficios.com> <20121013155615.GC29985@Krystal> <507C3013.2050405@efficios.com> Message-ID: <20121015175754.GB9423@Krystal> * 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 > >> --- > >> 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