From: David Goulet Date: Fri, 2 Mar 2012 20:18:33 +0000 (-0500) Subject: Merge branch 'master' of git://git.lttng.org/lttng-tools X-Git-Tag: v2.0.0-rc3~18 X-Git-Url: http://git.liburcu.org/?a=commitdiff_plain;h=a186a15913c34e8adc83ea71565d3b0eec296774;hp=-c;p=lttng-tools.git Merge branch 'master' of git://git.lttng.org/lttng-tools --- a186a15913c34e8adc83ea71565d3b0eec296774 diff --combined src/common/consumer.c index e7d51078e,d7b319452..05bf85b3d --- a/src/common/consumer.c +++ b/src/common/consumer.c @@@ -19,6 -19,7 +19,6 @@@ #define _GNU_SOURCE #include -#include #include #include #include @@@ -123,6 -124,17 +123,17 @@@ static void consumer_steal_channel_key( channel->key = -1; } + static + void consumer_free_stream(struct rcu_head *head) + { + struct lttng_ht_node_ulong *node = + caa_container_of(head, struct lttng_ht_node_ulong, head); + struct lttng_consumer_stream *stream = + caa_container_of(node, struct lttng_consumer_stream, node); + + free(stream); + } + /* * Remove a stream from the global list protected by a mutex. This * function is also responsible for freeing its data structures. @@@ -159,9 -171,11 +170,11 @@@ void consumer_del_stream(struct lttng_c /* Get stream node from hash table */ lttng_ht_lookup(consumer_data.stream_ht, (void *)((unsigned long) stream->key), &iter); - /* Remove stream node from hash table */ - ret = lttng_ht_del(consumer_data.stream_ht, &iter); - assert(!ret); + /* + * Remove stream node from hash table. It can fail if it's been + * replaced due to key reuse. + */ + (void) lttng_ht_del(consumer_data.stream_ht, &iter); rcu_read_unlock(); @@@ -192,7 -206,8 +205,8 @@@ } if (!--stream->chan->refcount) free_chan = stream->chan; - free(stream); + + call_rcu(&stream->node.head, consumer_free_stream); end: consumer_data.need_update = 1; pthread_mutex_unlock(&consumer_data.lock); @@@ -201,16 -216,6 +215,6 @@@ consumer_del_channel(free_chan); } - static void consumer_del_stream_rcu(struct rcu_head *head) - { - struct lttng_ht_node_ulong *node = - caa_container_of(head, struct lttng_ht_node_ulong, head); - struct lttng_consumer_stream *stream = - caa_container_of(node, struct lttng_consumer_stream, node); - - consumer_del_stream(stream); - } - struct lttng_consumer_stream *consumer_allocate_stream( int channel_key, int stream_key, int shm_fd, int wait_fd, @@@ -288,7 -293,12 +292,12 @@@ int consumer_add_stream(struct lttng_co /* Steal stream identifier, for UST */ consumer_steal_stream_key(stream->key); rcu_read_lock(); - lttng_ht_add_unique_ulong(consumer_data.stream_ht, &stream->node); + /* + * We simply remove the old channel from the hash table. It's + * ok, since we know for sure the sessiond wants to replace it + * with the new version, because the key has been reused. + */ + (void) lttng_ht_add_replace_ulong(consumer_data.stream_ht, &stream->node); rcu_read_unlock(); consumer_data.stream_count++; consumer_data.need_update = 1; @@@ -309,6 -319,7 +318,7 @@@ end: pthread_mutex_unlock(&consumer_data.lock); + return ret; } @@@ -329,6 -340,17 +339,17 @@@ void consumer_change_stream_state(int s pthread_mutex_unlock(&consumer_data.lock); } + static + void consumer_free_channel(struct rcu_head *head) + { + struct lttng_ht_node_ulong *node = + caa_container_of(head, struct lttng_ht_node_ulong, head); + struct lttng_consumer_channel *channel = + caa_container_of(node, struct lttng_consumer_channel, node); + + free(channel); + } + /* * Remove a channel from the global list protected by a mutex. This * function is also responsible for freeing its data structures. @@@ -357,8 -379,12 +378,12 @@@ void consumer_del_channel(struct lttng_ lttng_ht_lookup(consumer_data.channel_ht, (void *)((unsigned long) channel->key), &iter); - ret = lttng_ht_del(consumer_data.channel_ht, &iter); - assert(!ret); + + /* + * Remove channel node from hash table. It can fail if it's been + * replaced due to key reuse. + */ + (void) lttng_ht_del(consumer_data.channel_ht, &iter); rcu_read_unlock(); @@@ -380,21 -406,12 +405,12 @@@ PERROR("close"); } } - free(channel); + + call_rcu(&channel->node.head, consumer_free_channel); end: pthread_mutex_unlock(&consumer_data.lock); } - static void consumer_del_channel_rcu(struct rcu_head *head) - { - struct lttng_ht_node_ulong *node = - caa_container_of(head, struct lttng_ht_node_ulong, head); - struct lttng_consumer_channel *channel= - caa_container_of(node, struct lttng_consumer_channel, node); - - consumer_del_channel(channel); - } - struct lttng_consumer_channel *consumer_allocate_channel( int channel_key, int shm_fd, int wait_fd, @@@ -455,9 -472,15 +471,15 @@@ int consumer_add_channel(struct lttng_c /* Steal channel identifier, for UST */ consumer_steal_channel_key(channel->key); rcu_read_lock(); - lttng_ht_add_unique_ulong(consumer_data.channel_ht, &channel->node); + /* + * We simply remove the old channel from the hash table. It's + * ok, since we know for sure the sessiond wants to replace it + * with the new version, because the key has been reused. + */ + (void) lttng_ht_add_replace_ulong(consumer_data.channel_ht, &channel->node); rcu_read_unlock(); pthread_mutex_unlock(&consumer_data.lock); + return 0; } @@@ -568,7 -591,6 +590,6 @@@ int lttng_consumer_send_error */ void lttng_consumer_cleanup(void) { - int ret; struct lttng_ht_iter iter; struct lttng_ht_node_ulong *node; @@@ -580,16 -602,16 +601,16 @@@ */ cds_lfht_for_each_entry(consumer_data.stream_ht->ht, &iter.iter, node, node) { - ret = lttng_ht_del(consumer_data.stream_ht, &iter); - assert(!ret); - call_rcu(&node->head, consumer_del_stream_rcu); + struct lttng_consumer_stream *stream = + caa_container_of(node, struct lttng_consumer_stream, node); + consumer_del_stream(stream); } cds_lfht_for_each_entry(consumer_data.channel_ht->ht, &iter.iter, node, node) { - ret = lttng_ht_del(consumer_data.channel_ht, &iter); - assert(!ret); - call_rcu(&node->head, consumer_del_channel_rcu); + struct lttng_consumer_channel *channel = + caa_container_of(node, struct lttng_consumer_channel, node); + consumer_del_channel(channel); } rcu_read_unlock(); @@@ -622,7 -644,7 +643,7 @@@ void lttng_consumer_sync_trace_file if (orig_offset < stream->chan->max_sb_size) { return; } - sync_file_range(outfd, orig_offset - stream->chan->max_sb_size, + lttng_sync_file_range(outfd, orig_offset - stream->chan->max_sb_size, stream->chan->max_sb_size, SYNC_FILE_RANGE_WAIT_BEFORE | SYNC_FILE_RANGE_WRITE @@@ -790,8 -812,6 +811,8 @@@ ssize_t lttng_consumer_on_read_subbuffe ERR("Unknown consumer_data type"); assert(0); } + + return 0; } /* @@@ -1048,25 -1068,19 +1069,19 @@@ void *lttng_consumer_thread_poll_fds(vo if ((pollfd[i].revents & POLLHUP)) { DBG("Polling fd %d tells it has hung up.", pollfd[i].fd); if (!local_stream[i]->data_read) { - rcu_read_lock(); - consumer_del_stream_rcu(&local_stream[i]->node.head); - rcu_read_unlock(); + consumer_del_stream(local_stream[i]); num_hup++; } } else if (pollfd[i].revents & POLLERR) { ERR("Error returned in polling fd %d.", pollfd[i].fd); if (!local_stream[i]->data_read) { - rcu_read_lock(); - consumer_del_stream_rcu(&local_stream[i]->node.head); - rcu_read_unlock(); + consumer_del_stream(local_stream[i]); 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) { - rcu_read_lock(); - consumer_del_stream_rcu(&local_stream[i]->node.head); - rcu_read_unlock(); + consumer_del_stream(local_stream[i]); num_hup++; } }