Fix: Missing rcu read side lock in consumer

The metadata thread was not using rcu read side lock for its operations
on the internal metadata hash table.

This led to faulty free() when destroying the hash table and possible
corrupted data when it was resized.

Also change some static definition of calls inside consumer.c.

Signed-off-by: David Goulet <dgoulet@efficios.com>

diff --git a/src/common/consumer.c b/src/common/consumer.c
index 72c820d90a11ec11d61e2e9024cb8f44f1ad009e..a2980e77d29c33b57c4a12889cb82db019956fa0 100644 (file)
--- a/src/common/consumer.c
+++ b/src/common/consumer.c
@@ -175,7 +175,7 @@ static void consumer_rcu_free_relayd(struct rcu_head *head)
  *
  * This function MUST be called with the consumer_data lock acquired.
  */
-void consumer_destroy_relayd(struct consumer_relayd_sock_pair *relayd)
+static void destroy_relayd(struct consumer_relayd_sock_pair *relayd)
 {
        int ret;
        struct lttng_ht_iter iter;
@@ -218,7 +218,7 @@ void consumer_flag_relayd_for_destroy(struct consumer_relayd_sock_pair *relayd)
 
        /* Destroy the relayd if refcount is 0 */
        if (uatomic_read(&relayd->refcount) == 0) {
-               consumer_destroy_relayd(relayd);
+               destroy_relayd(relayd);
        }
 }
 
@@ -314,7 +314,7 @@ void consumer_del_stream(struct lttng_consumer_stream *stream)
                /* Both conditions are met, we destroy the relayd. */
                if (uatomic_read(&relayd->refcount) == 0 &&
                                uatomic_read(&relayd->destroy_flag)) {
-                       consumer_destroy_relayd(relayd);
+                       destroy_relayd(relayd);
                }
        }
        rcu_read_unlock();
@@ -452,8 +452,7 @@ end:
  * Add relayd socket to global consumer data hashtable. RCU read side lock MUST
  * be acquired before calling this.
  */
-
-int consumer_add_relayd(struct consumer_relayd_sock_pair *relayd)
+static int add_relayd(struct consumer_relayd_sock_pair *relayd)
 {
        int ret = 0;
        struct lttng_ht_node_ulong *node;
@@ -1503,12 +1502,14 @@ static void destroy_stream_ht(struct lttng_ht *ht)
                return;
        }
 
+       rcu_read_lock();
        cds_lfht_for_each_entry(ht->ht, &iter.iter, stream, node.node) {
                ret = lttng_ht_del(ht, &iter);
                assert(!ret);
 
                free(stream);
        }
+       rcu_read_unlock();
 
        lttng_ht_destroy(ht);
 }
@@ -1594,7 +1595,7 @@ static void consumer_del_metadata_stream(struct lttng_consumer_stream *stream)
                /* Both conditions are met, we destroy the relayd. */
                if (uatomic_read(&relayd->refcount) == 0 &&
                                uatomic_read(&relayd->destroy_flag)) {
-                       consumer_destroy_relayd(relayd);
+                       destroy_relayd(relayd);
                }
        }
        rcu_read_unlock();
@@ -1730,9 +1731,11 @@ restart:
                                        DBG("Adding metadata stream %d to poll set",
                                                        stream->wait_fd);
 
+                                       rcu_read_lock();
                                        /* The node should be init at this point */
                                        lttng_ht_add_unique_ulong(metadata_ht,
                                                        &stream->waitfd_node);
+                                       rcu_read_unlock();
 
                                        /* Add metadata stream to the global poll events list */
                                        lttng_poll_add(&events, stream->wait_fd,
@@ -1747,11 +1750,13 @@ restart:
 
                        /* From here, the event is a metadata wait fd */
 
+                       rcu_read_lock();
                        lttng_ht_lookup(metadata_ht, (void *)((unsigned long) pollfd),
                                        &iter);
                        node = lttng_ht_iter_get_node_ulong(&iter);
                        if (node == NULL) {
                                /* FD not found, continue loop */
+                               rcu_read_unlock();
                                continue;
                        }
 
@@ -1766,6 +1771,7 @@ restart:
                                len = ctx->on_buffer_ready(stream, ctx);
                                /* It's ok to have an unavailable sub-buffer */
                                if (len < 0 && len != -EAGAIN) {
+                                       rcu_read_unlock();
                                        goto end;
                                } else if (len > 0) {
                                        stream->data_read = 1;
@@ -1788,15 +1794,18 @@ restart:
                                        len = ctx->on_buffer_ready(stream, ctx);
                                        /* It's ok to have an unavailable sub-buffer */
                                        if (len < 0 && len != -EAGAIN) {
+                                               rcu_read_unlock();
                                                goto end;
                                        }
                                }
 
                                /* Removing it from hash table, poll set and free memory */
                                lttng_ht_del(metadata_ht, &iter);
+
                                lttng_poll_del(&events, stream->wait_fd);
                                consumer_del_metadata_stream(stream);
                        }
+                       rcu_read_unlock();
                }
        }
 
@@ -2290,7 +2299,7 @@ int consumer_add_relayd_socket(int net_seq_idx, int sock_type,
         * Add relayd socket pair to consumer data hashtable. If object already
         * exists or on error, the function gracefully returns.
         */
-       consumer_add_relayd(relayd);
+       add_relayd(relayd);
 
        /* All good! */
        ret = 0;

diff --git a/src/common/consumer.h b/src/common/consumer.h
index 4da4b70d1496647e17c65afe06ae2d9889297ce4..dba7765772859bd703adb5ba9e0889759d5fb757 100644 (file)
--- a/src/common/consumer.h
+++ b/src/common/consumer.h
@@ -346,13 +346,11 @@ extern struct lttng_consumer_channel *consumer_allocate_channel(
 int consumer_add_channel(struct lttng_consumer_channel *channel);
 
 /* lttng-relayd consumer command */
-int consumer_add_relayd(struct consumer_relayd_sock_pair *relayd);
 struct consumer_relayd_sock_pair *consumer_allocate_relayd_sock_pair(
                int net_seq_idx);
 struct consumer_relayd_sock_pair *consumer_find_relayd(int key);
 int consumer_handle_stream_before_relayd(struct lttng_consumer_stream *stream,
                size_t data_size);
-void consumer_destroy_relayd(struct consumer_relayd_sock_pair *relayd);
 
 extern struct lttng_consumer_local_data *lttng_consumer_create(
                enum lttng_consumer_type type,