Merge branch 'master' of git://git.lttng.org/lttng-tools

diff --combined src/common/consumer.c
index e7d51078e64b9b5209d4e8208a61a25d9e0bef38,d7b319452a87088eab5d6cd648fd7515e5b387f5..05bf85b3dc8d8e73e0bbe62497a64bf0f0bc3e26
--- 1/src/common/consumer.c
--- 2/src/common/consumer.c
+++ b/src/common/consumer.c
@@@ -19,6 -19,7 +19,6 @@@
  
  #define _GNU_SOURCE
  #include <assert.h>
 -#include <fcntl.h>
  #include <poll.h>
  #include <pthread.h>
  #include <stdlib.h>
@@@ -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++;
                                }
                        }