Fix: make sure no index is in flight before using inactivity beacons

[lttng-tools.git] / src / bin / lttng-relayd / main.c

diff --git a/src/bin/lttng-relayd/main.c b/src/bin/lttng-relayd/main.c
index d82a3412d7b8578be8f5b39abf0a9fd949ff07bc..fa5a7db86c393640c15fe9e8b3bf5f652da98111 100644 (file)
--- a/src/bin/lttng-relayd/main.c
+++ b/src/bin/lttng-relayd/main.c
@@ -44,6 +44,7 @@
 #include <common/common.h>
 #include <common/compat/poll.h>
 #include <common/compat/socket.h>
+#include <common/compat/endian.h>
 #include <common/defaults.h>
 #include <common/daemonize.h>
 #include <common/futex.h>
@@ -250,7 +251,10 @@ int set_option(int opt, const char *arg, const char *optname)
                if (arg) {
                        lttng_opt_verbose = config_parse_value(arg);
                } else {
-                       lttng_opt_verbose += 1;
+                       /* Only 3 level of verbosity (-vvv). */
+                       if (lttng_opt_verbose < 3) {
+                               lttng_opt_verbose += 1;
+                       }
                }
                break;
        default:
@@ -878,11 +882,12 @@ restart:
                                new_conn->sock = newsock;
 
                                /* Enqueue request for the dispatcher thread. */
-                               cds_wfq_enqueue(&relay_conn_queue.queue, &new_conn->qnode);
+                               cds_wfcq_enqueue(&relay_conn_queue.head, &relay_conn_queue.tail,
+                                                &new_conn->qnode);
 
                                /*
                                 * Wake the dispatch queue futex. Implicit memory barrier with
-                                * the exchange in cds_wfq_enqueue.
+                                * the exchange in cds_wfcq_enqueue.
                                 */
                                futex_nto1_wake(&relay_conn_queue.futex);
                        }
@@ -929,7 +934,7 @@ void *relay_thread_dispatcher(void *data)
 {
        int err = -1;
        ssize_t ret;
-       struct cds_wfq_node *node;
+       struct cds_wfcq_node *node;
        struct relay_connection *new_conn = NULL;
 
        DBG("[thread] Relay dispatcher started");
@@ -952,7 +957,8 @@ void *relay_thread_dispatcher(void *data)
                        health_code_update();
 
                        /* Dequeue commands */
-                       node = cds_wfq_dequeue_blocking(&relay_conn_queue.queue);
+                       node = cds_wfcq_dequeue_blocking(&relay_conn_queue.head,
+                                                        &relay_conn_queue.tail);
                        if (node == NULL) {
                                DBG("Woken up but nothing in the relay command queue");
                                /* Continue thread execution */
@@ -1201,6 +1207,7 @@ int relay_add_stream(struct lttcomm_relayd_hdr *recv_hdr,
        stream->session_id = session->id;
        stream->index_fd = -1;
        stream->read_index_fd = -1;
+       stream->ctf_stream_id = -1ULL;
        lttng_ht_node_init_u64(&stream->node, stream->stream_handle);
        pthread_mutex_init(&stream->lock, NULL);
 
@@ -1340,18 +1347,6 @@ int relay_close_stream(struct lttcomm_relayd_hdr *recv_hdr,
        session->stream_count--;
        assert(session->stream_count >= 0);
 
-       /*
-        * Remove the stream from the connection recv list since we are about to
-        * flag it invalid and thus might be freed. This has to be done here since
-        * only the control thread can do actions on that list.
-        *
-        * Note that this stream might NOT be in the list but we have to try to
-        * remove it here else this can race with the stream destruction freeing
-        * the object and the connection destroy doing a use after free when
-        * deleting the remaining nodes in this list.
-        */
-       cds_list_del(&stream->recv_list);
-
        /* Check if we can close it or else the data will do it. */
        try_close_stream(session, stream);
 
@@ -1949,9 +1944,10 @@ int relay_recv_index(struct lttcomm_relayd_hdr *recv_hdr,
                DBG("Received live beacon for stream %" PRIu64, stream->stream_handle);
 
                /*
-                * Only flag a stream inactive when it has already received data.
+                * Only flag a stream inactive when it has already received data
+                * and no indexes are in flight.
                 */
-               if (stream->total_index_received > 0) {
+               if (stream->total_index_received > 0 && stream->indexes_in_flight == 0) {
                        stream->beacon_ts_end = be64toh(index_info.timestamp_end);
                }
                ret = 0;
@@ -1968,9 +1964,13 @@ int relay_recv_index(struct lttcomm_relayd_hdr *recv_hdr,
                        goto end_rcu_unlock;
                }
                index_created = 1;
+               stream->indexes_in_flight++;
        }
 
        copy_index_control_data(index, &index_info);
+       if (stream->ctf_stream_id == -1ULL) {
+               stream->ctf_stream_id = be64toh(index_info.stream_id);
+       }
 
        if (index_created) {
                /*
@@ -1995,6 +1995,8 @@ int relay_recv_index(struct lttcomm_relayd_hdr *recv_hdr,
                        goto end_rcu_unlock;
                }
                stream->total_index_received++;
+               stream->indexes_in_flight--;
+               assert(stream->indexes_in_flight >= 0);
        }
 
 end_rcu_unlock:
@@ -2158,6 +2160,7 @@ static int handle_index_data(struct relay_stream *stream, uint64_t net_seq_num,
                        goto error;
                }
                index_created = 1;
+               stream->indexes_in_flight++;
        }
 
        if (rotate_index || stream->index_fd < 0) {
@@ -2200,6 +2203,8 @@ static int handle_index_data(struct relay_stream *stream, uint64_t net_seq_num,
                        goto error;
                }
                stream->total_index_received++;
+               stream->indexes_in_flight--;
+               assert(stream->indexes_in_flight >= 0);
        }
 
 error:
@@ -2766,7 +2771,7 @@ int main(int argc, char **argv)
        }
 
        /* Init relay command queue. */
-       cds_wfq_init(&relay_conn_queue.queue);
+       cds_wfcq_init(&relay_conn_queue.head, &relay_conn_queue.tail);
 
        /* Set up max poll set size */
        lttng_poll_set_max_size();