Fix: reset current size for tracefile rotation

[lttng-tools.git] / src / common / consumer.c

diff --git a/src/common/consumer.c b/src/common/consumer.c
index 34385579741f6f0273aee213576bd0a188a95b0e..01266a700cb37199966c9257eba335689e877453 100644 (file)
--- a/src/common/consumer.c
+++ b/src/common/consumer.c
@@ -50,12 +50,14 @@ struct lttng_consumer_global_data consumer_data = {
 
 enum consumer_channel_action {
        CONSUMER_CHANNEL_ADD,
+       CONSUMER_CHANNEL_DEL,
        CONSUMER_CHANNEL_QUIT,
 };
 
 struct consumer_channel_msg {
        enum consumer_channel_action action;
-       struct lttng_consumer_channel *chan;
+       struct lttng_consumer_channel *chan;    /* add */
+       uint64_t key;                           /* del */
 };
 
 /*
@@ -91,6 +93,7 @@ static void notify_thread_pipe(int wpipe)
 
 static void notify_channel_pipe(struct lttng_consumer_local_data *ctx,
                struct lttng_consumer_channel *chan,
+               uint64_t key,
                enum consumer_channel_action action)
 {
        struct consumer_channel_msg msg;
@@ -103,8 +106,15 @@ static void notify_channel_pipe(struct lttng_consumer_local_data *ctx,
        } while (ret < 0 && errno == EINTR);
 }
 
+void notify_thread_del_channel(struct lttng_consumer_local_data *ctx,
+               uint64_t key)
+{
+       notify_channel_pipe(ctx, NULL, key, CONSUMER_CHANNEL_DEL);
+}
+
 static int read_channel_pipe(struct lttng_consumer_local_data *ctx,
                struct lttng_consumer_channel **chan,
+               uint64_t *key,
                enum consumer_channel_action *action)
 {
        struct consumer_channel_msg msg;
@@ -116,6 +126,7 @@ static int read_channel_pipe(struct lttng_consumer_local_data *ctx,
        if (ret > 0) {
                *action = msg.action;
                *chan = msg.chan;
+               *key = msg.key;
        }
        return ret;
 }
@@ -319,9 +330,9 @@ static void cleanup_relayd_ht(void)
                destroy_relayd(relayd);
        }
 
-       lttng_ht_destroy(consumer_data.relayd_ht);
-
        rcu_read_unlock();
+
+       lttng_ht_destroy(consumer_data.relayd_ht);
 }
 
 /*
@@ -720,6 +731,8 @@ struct consumer_relayd_sock_pair *consumer_allocate_relayd_sock_pair(
        obj->net_seq_idx = net_seq_idx;
        obj->refcount = 0;
        obj->destroy_flag = 0;
+       obj->control_sock.sock.fd = -1;
+       obj->data_sock.sock.fd = -1;
        lttng_ht_node_init_u64(&obj->node, obj->net_seq_idx);
        pthread_mutex_init(&obj->ctrl_sock_mutex, NULL);
 
@@ -900,7 +913,7 @@ end:
 
        if (!ret && channel->wait_fd != -1 &&
                        channel->metadata_stream == NULL) {
-               notify_channel_pipe(ctx, channel, CONSUMER_CHANNEL_ADD);
+               notify_channel_pipe(ctx, channel, -1, CONSUMER_CHANNEL_ADD);
        }
        return ret;
 }
@@ -1411,6 +1424,8 @@ ssize_t lttng_consumer_on_read_subbuffer_mmap(
                                goto end;
                        }
                        outfd = stream->out_fd = ret;
+                       /* Reset current size because we just perform a rotation. */
+                       stream->tracefile_size_current = 0;
                }
                stream->tracefile_size_current += len;
        }
@@ -1591,6 +1606,8 @@ ssize_t lttng_consumer_on_read_subbuffer_splice(
                                goto end;
                        }
                        outfd = stream->out_fd = ret;
+                       /* Reset current size because we just perform a rotation. */
+                       stream->tracefile_size_current = 0;
                }
                stream->tracefile_size_current += len;
        }
@@ -2691,8 +2708,9 @@ restart:
                                        continue;
                                } else if (revents & LPOLLIN) {
                                        enum consumer_channel_action action;
+                                       uint64_t key;
 
-                                       ret = read_channel_pipe(ctx, &chan, &action);
+                                       ret = read_channel_pipe(ctx, &chan, &key, &action);
                                        if (ret <= 0) {
                                                ERR("Error reading channel pipe");
                                                continue;
@@ -2711,6 +2729,27 @@ restart:
                                                lttng_poll_add(&events, chan->wait_fd,
                                                                LPOLLIN | LPOLLPRI);
                                                break;
+                                       case CONSUMER_CHANNEL_DEL:
+                                       {
+                                               chan = consumer_find_channel(key);
+                                               if (!chan) {
+                                                       ERR("UST consumer get channel key %" PRIu64 " not found for del channel", key);
+                                                       break;
+                                               }
+                                               lttng_poll_del(&events, chan->wait_fd);
+                                               ret = lttng_ht_del(channel_ht, &iter);
+                                               assert(ret == 0);
+                                               consumer_close_channel_streams(chan);
+
+                                               /*
+                                                * Release our own refcount. Force channel deletion even if
+                                                * streams were not initialized.
+                                                */
+                                               if (!uatomic_sub_return(&chan->refcount, 1)) {
+                                                       consumer_del_channel(chan);
+                                               }
+                                               goto restart;
+                                       }
                                        case CONSUMER_CHANNEL_QUIT:
                                                /*
                                                 * Remove the pipe from the poll set and continue the loop
@@ -2931,7 +2970,7 @@ end:
         */
        notify_thread_pipe(ctx->consumer_data_pipe[1]);
 
-       notify_channel_pipe(ctx, NULL, CONSUMER_CHANNEL_QUIT);
+       notify_channel_pipe(ctx, NULL, -1, CONSUMER_CHANNEL_QUIT);
 
        /* Cleaning up possibly open sockets. */
        if (sock >= 0) {