UST periodical metadata flush

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

diff --git a/src/common/consumer.c b/src/common/consumer.c
index 8730e39fbbda7db3fda89758cbd8b5f01385f9f7..5f87f4b5018fbec1a2de542797f46b4f6d17475f 100644 (file)
--- a/src/common/consumer.c
+++ b/src/common/consumer.c
@@ -28,6 +28,7 @@
 #include <sys/types.h>
 #include <unistd.h>
 #include <inttypes.h>
+#include <signal.h>
 
 #include <common/common.h>
 #include <common/utils.h>
@@ -876,8 +877,7 @@ int consumer_add_channel(struct lttng_consumer_channel *channel,
        pthread_mutex_lock(&consumer_data.lock);
        rcu_read_lock();
 
-       lttng_ht_lookup(consumer_data.channel_ht,
-                       &channel->key, &iter);
+       lttng_ht_lookup(consumer_data.channel_ht, &channel->key, &iter);
        node = lttng_ht_iter_get_node_u64(&iter);
        if (node != NULL) {
                /* Channel already exist. Ignore the insertion */
@@ -936,7 +936,12 @@ static int update_poll_array(struct lttng_consumer_local_data *ctx,
                                stream->endpoint_status == CONSUMER_ENDPOINT_INACTIVE) {
                        continue;
                }
-               DBG("Active FD %d", stream->wait_fd);
+               /*
+                * This clobbers way too much the debug output. Uncomment that if you
+                * need it for debugging purposes.
+                *
+                * DBG("Active FD %d", stream->wait_fd);
+                */
                (*pollfd)[i].fd = stream->wait_fd;
                (*pollfd)[i].events = POLLIN | POLLPRI;
                local_stream[i] = stream;
@@ -1137,6 +1142,7 @@ struct lttng_consumer_local_data *lttng_consumer_create(
        }
 
        ctx->consumer_error_socket = -1;
+       ctx->consumer_metadata_socket = -1;
        /* assign the callbacks */
        ctx->on_buffer_ready = buffer_ready;
        ctx->on_recv_channel = recv_channel;
@@ -1223,6 +1229,10 @@ void lttng_consumer_destroy(struct lttng_consumer_local_data *ctx)
        if (ret) {
                PERROR("close");
        }
+       ret = close(ctx->consumer_metadata_socket);
+       if (ret) {
+               PERROR("close");
+       }
        utils_close_pipe(ctx->consumer_thread_pipe);
        utils_close_pipe(ctx->consumer_channel_pipe);
        utils_close_pipe(ctx->consumer_data_pipe);
@@ -1324,6 +1334,7 @@ ssize_t lttng_consumer_on_read_subbuffer_mmap(
                        goto end;
                }
                ret = lttng_ustctl_get_mmap_read_offset(stream, &mmap_offset);
+
                break;
        default:
                ERR("Unknown consumer_data type");
@@ -2703,6 +2714,33 @@ end_ht:
        return NULL;
 }
 
+static int set_metadata_socket(struct lttng_consumer_local_data *ctx,
+               struct pollfd *sockpoll, int client_socket)
+{
+       int ret;
+
+       assert(ctx);
+       assert(sockpoll);
+
+       if (lttng_consumer_poll_socket(sockpoll) < 0) {
+               ret = -1;
+               goto error;
+       }
+       DBG("Metadata connection on client_socket");
+
+       /* Blocking call, waiting for transmission */
+       ctx->consumer_metadata_socket = lttcomm_accept_unix_sock(client_socket);
+       if (ctx->consumer_metadata_socket < 0) {
+               WARN("On accept metadata");
+               ret = -1;
+               goto error;
+       }
+       ret = 0;
+
+error:
+       return ret;
+}
+
 /*
  * This thread listens on the consumerd socket and receives the file
  * descriptors from the session daemon.
@@ -2769,6 +2807,15 @@ void *consumer_thread_sessiond_poll(void *data)
                goto end;
        }
 
+       /*
+        * Setup metadata socket which is the second socket connection on the
+        * command unix socket.
+        */
+       ret = set_metadata_socket(ctx, consumer_sockpoll, client_socket);
+       if (ret < 0) {
+               goto end;
+       }
+
        /* This socket is not useful anymore. */
        ret = close(client_socket);
        if (ret < 0) {