Use consumer fd reference in consumer socket obj

This is done so we don't copy the fd value of the consumer thus having a
central control. This is part of a set of patches that makes the session
daemon stop sending command to an invalid consumer (dead, closed).

Acked-by: Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
Signed-off-by: David Goulet <dgoulet@efficios.com>

diff --git a/src/bin/lttng-sessiond/cmd.c b/src/bin/lttng-sessiond/cmd.c
index 6a195e1bdd8352c752ab1b22de82bd08ac573faf..dcb5da5f44864a84af1004669c760b6a77e5f06a 100644 (file)
--- a/src/bin/lttng-sessiond/cmd.c
+++ b/src/bin/lttng-sessiond/cmd.c
@@ -463,7 +463,7 @@ static int init_kernel_tracing(struct ltt_kernel_session *session)
                cds_lfht_for_each_entry(session->consumer->socks->ht, &iter.iter,
                                socket, node.node) {
                        /* Code flow error */
-                       assert(socket->fd >= 0);
+                       assert(socket->fd);
 
                        pthread_mutex_lock(socket->lock);
                        ret = kernel_consumer_send_session(socket, session);
@@ -675,7 +675,7 @@ int cmd_setup_relayd(struct ltt_session *session)
                cds_lfht_for_each_entry(usess->consumer->socks->ht, &iter.iter,
                                socket, node.node) {
                        /* Code flow error */
-                       assert(socket->fd >= 0);
+                       assert(socket->fd);
 
                        pthread_mutex_lock(socket->lock);
                        ret = send_consumer_relayd_sockets(LTTNG_DOMAIN_UST, session->id,
@@ -694,7 +694,7 @@ int cmd_setup_relayd(struct ltt_session *session)
                cds_lfht_for_each_entry(ksess->consumer->socks->ht, &iter.iter,
                                socket, node.node) {
                        /* Code flow error */
-                       assert(socket->fd >= 0);
+                       assert(socket->fd);
 
                        pthread_mutex_lock(socket->lock);
                        ret = send_consumer_relayd_sockets(LTTNG_DOMAIN_KERNEL, session->id,
@@ -2027,13 +2027,15 @@ int cmd_register_consumer(struct ltt_session *session, int domain,
                        ret = LTTNG_ERR_CONNECT_FAIL;
                        goto error;
                }
+               cdata->cmd_sock = sock;
 
-               socket = consumer_allocate_socket(sock);
+               socket = consumer_allocate_socket(&cdata->cmd_sock);
                if (socket == NULL) {
                        ret = close(sock);
                        if (ret < 0) {
                                PERROR("close register consumer");
                        }
+                       cdata->cmd_sock = -1;
                        ret = LTTNG_ERR_FATAL;
                        goto error;
                }

diff --git a/src/bin/lttng-sessiond/consumer.c b/src/bin/lttng-sessiond/consumer.c
index 17fc919768fa98610ad4a0afddfdb58ef38db773..39a1946380120db04419809e52a147a08b999726 100644 (file)
--- a/src/bin/lttng-sessiond/consumer.c
+++ b/src/bin/lttng-sessiond/consumer.c
@@ -48,14 +48,14 @@ int consumer_recv_status_reply(struct consumer_socket *sock)
 
        assert(sock);
 
-       ret = lttcomm_recv_unix_sock(sock->fd, &reply, sizeof(reply));
+       ret = lttcomm_recv_unix_sock(*sock->fd, &reply, sizeof(reply));
        if (ret <= 0) {
                if (ret == 0) {
                        /* Orderly shutdown. Don't return 0 which means success. */
                        ret = -1;
                }
                /* The above call will print a PERROR on error. */
-               DBG("Fail to receive status reply on sock %d", sock->fd);
+               DBG("Fail to receive status reply on sock %d", *sock->fd);
                goto end;
        }
 
@@ -89,14 +89,14 @@ int consumer_recv_status_channel(struct consumer_socket *sock,
        assert(stream_count);
        assert(key);
 
-       ret = lttcomm_recv_unix_sock(sock->fd, &reply, sizeof(reply));
+       ret = lttcomm_recv_unix_sock(*sock->fd, &reply, sizeof(reply));
        if (ret <= 0) {
                if (ret == 0) {
                        /* Orderly shutdown. Don't return 0 which means success. */
                        ret = -1;
                }
                /* The above call will print a PERROR on error. */
-               DBG("Fail to receive status reply on sock %d", sock->fd);
+               DBG("Fail to receive status reply on sock %d", *sock->fd);
                goto end;
        }
 
@@ -127,7 +127,7 @@ int consumer_send_destroy_relayd(struct consumer_socket *sock,
        assert(consumer);
        assert(sock);
 
-       DBG2("Sending destroy relayd command to consumer sock %d", sock->fd);
+       DBG2("Sending destroy relayd command to consumer sock %d", *sock->fd);
 
        /* Bail out if consumer is disabled */
        if (!consumer->enabled) {
@@ -140,7 +140,7 @@ int consumer_send_destroy_relayd(struct consumer_socket *sock,
        msg.u.destroy_relayd.net_seq_idx = consumer->net_seq_index;
 
        pthread_mutex_lock(sock->lock);
-       ret = lttcomm_send_unix_sock(sock->fd, &msg, sizeof(msg));
+       ret = lttcomm_send_unix_sock(*sock->fd, &msg, sizeof(msg));
        if (ret < 0) {
                /* Indicate that the consumer is probably closing at this point. */
                DBG("send consumer destroy relayd command");
@@ -213,7 +213,7 @@ int consumer_create_socket(struct consumer_data *data,
        socket = consumer_find_socket(data->cmd_sock, output);
        rcu_read_unlock();
        if (socket == NULL) {
-               socket = consumer_allocate_socket(data->cmd_sock);
+               socket = consumer_allocate_socket(&data->cmd_sock);
                if (socket == NULL) {
                        ret = -1;
                        goto error;
@@ -300,10 +300,12 @@ struct consumer_socket *consumer_find_socket(int key,
 /*
  * Allocate a new consumer_socket and return the pointer.
  */
-struct consumer_socket *consumer_allocate_socket(int fd)
+struct consumer_socket *consumer_allocate_socket(int *fd)
 {
        struct consumer_socket *socket = NULL;
 
+       assert(fd);
+
        socket = zmalloc(sizeof(struct consumer_socket));
        if (socket == NULL) {
                PERROR("zmalloc consumer socket");
@@ -311,7 +313,7 @@ struct consumer_socket *consumer_allocate_socket(int fd)
        }
 
        socket->fd = fd;
-       lttng_ht_node_init_ulong(&socket->node, fd);
+       lttng_ht_node_init_ulong(&socket->node, *fd);
 
 error:
        return socket;
@@ -375,8 +377,8 @@ void consumer_destroy_socket(struct consumer_socket *sock)
         * consumer was registered,
         */
        if (sock->registered) {
-               DBG3("Consumer socket was registered. Closing fd %d", sock->fd);
-               lttcomm_close_unix_sock(sock->fd);
+               DBG3("Consumer socket was registered. Closing fd %d", *sock->fd);
+               lttcomm_close_unix_sock(*sock->fd);
        }
 
        call_rcu(&sock->node.head, destroy_socket_rcu);
@@ -507,7 +509,7 @@ int consumer_copy_sockets(struct consumer_output *dst,
        rcu_read_lock();
        cds_lfht_for_each_entry(src->socks->ht, &iter.iter, socket, node.node) {
                /* Ignore socket that are already there. */
-               copy_sock = consumer_find_socket(socket->fd, dst);
+               copy_sock = consumer_find_socket(*socket->fd, dst);
                if (copy_sock) {
                        continue;
                }
@@ -646,12 +648,13 @@ int consumer_send_fds(struct consumer_socket *sock, int *fds, size_t nb_fd)
 
        assert(fds);
        assert(sock);
+       assert(sock->fd);
        assert(nb_fd > 0);
 
-       ret = lttcomm_send_fds_unix_sock(sock->fd, fds, nb_fd);
+       ret = lttcomm_send_fds_unix_sock(*sock->fd, fds, nb_fd);
        if (ret < 0) {
                /* The above call will print a PERROR on error. */
-               DBG("Error when sending consumer fds on sock %d", sock->fd);
+               DBG("Error when sending consumer fds on sock %d", *sock->fd);
                goto error;
        }
 
@@ -671,13 +674,13 @@ int consumer_send_msg(struct consumer_socket *sock,
 
        assert(msg);
        assert(sock);
-       assert(sock->fd >= 0);
+       assert(sock->fd);
 
-       ret = lttcomm_send_unix_sock(sock->fd, msg,
+       ret = lttcomm_send_unix_sock(*sock->fd, msg,
                        sizeof(struct lttcomm_consumer_msg));
        if (ret < 0) {
                /* The above call will print a PERROR on error. */
-               DBG("Error when sending consumer channel on sock %d", sock->fd);
+               DBG("Error when sending consumer channel on sock %d", *sock->fd);
                goto error;
        }
 
@@ -697,13 +700,13 @@ int consumer_send_channel(struct consumer_socket *sock,
 
        assert(msg);
        assert(sock);
-       assert(sock->fd >= 0);
+       assert(sock->fd);
 
-       ret = lttcomm_send_unix_sock(sock->fd, msg,
+       ret = lttcomm_send_unix_sock(*sock->fd, msg,
                        sizeof(struct lttcomm_consumer_msg));
        if (ret < 0) {
                /* The above call will print a PERROR on error. */
-               DBG("Error when sending consumer channel on sock %d", sock->fd);
+               DBG("Error when sending consumer channel on sock %d", *sock->fd);
                goto error;
        }
 
@@ -854,14 +857,15 @@ int consumer_send_stream(struct consumer_socket *sock,
        assert(msg);
        assert(dst);
        assert(sock);
+       assert(sock->fd);
        assert(fds);
 
        /* Send on socket */
-       ret = lttcomm_send_unix_sock(sock->fd, msg,
+       ret = lttcomm_send_unix_sock(*sock->fd, msg,
                        sizeof(struct lttcomm_consumer_msg));
        if (ret < 0) {
                /* The above call will print a PERROR on error. */
-               DBG("Error when sending consumer stream on sock %d", sock->fd);
+               DBG("Error when sending consumer stream on sock %d", *sock->fd);
                goto error;
        }
 
@@ -895,6 +899,7 @@ int consumer_send_relayd_socket(struct consumer_socket *consumer_sock,
        assert(rsock);
        assert(consumer);
        assert(consumer_sock);
+       assert(consumer_sock->fd);
 
        /* Bail out if consumer is disabled */
        if (!consumer->enabled) {
@@ -913,8 +918,8 @@ int consumer_send_relayd_socket(struct consumer_socket *consumer_sock,
        msg.u.relayd_sock.session_id = session_id;
        memcpy(&msg.u.relayd_sock.sock, rsock, sizeof(msg.u.relayd_sock.sock));
 
-       DBG3("Sending relayd sock info to consumer on %d", consumer_sock->fd);
-       ret = lttcomm_send_unix_sock(consumer_sock->fd, &msg, sizeof(msg));
+       DBG3("Sending relayd sock info to consumer on %d", *consumer_sock->fd);
+       ret = lttcomm_send_unix_sock(*consumer_sock->fd, &msg, sizeof(msg));
        if (ret < 0) {
                /* The above call will print a PERROR on error. */
                DBG("Error when sending relayd sockets on sock %d", rsock->sock.fd);
@@ -1015,14 +1020,14 @@ int consumer_is_data_pending(uint64_t session_id,
        cds_lfht_for_each_entry(consumer->socks->ht, &iter.iter, socket,
                        node.node) {
                /* Code flow error */
-               assert(socket->fd >= 0);
+               assert(socket->fd);
 
                pthread_mutex_lock(socket->lock);
 
-               ret = lttcomm_send_unix_sock(socket->fd, &msg, sizeof(msg));
+               ret = lttcomm_send_unix_sock(*socket->fd, &msg, sizeof(msg));
                if (ret < 0) {
                        /* The above call will print a PERROR on error. */
-                       DBG("Error on consumer is data pending on sock %d", socket->fd);
+                       DBG("Error on consumer is data pending on sock %d", *socket->fd);
                        pthread_mutex_unlock(socket->lock);
                        goto error_unlock;
                }
@@ -1032,14 +1037,14 @@ int consumer_is_data_pending(uint64_t session_id,
                 * the reply status message.
                 */
 
-               ret = lttcomm_recv_unix_sock(socket->fd, &ret_code, sizeof(ret_code));
+               ret = lttcomm_recv_unix_sock(*socket->fd, &ret_code, sizeof(ret_code));
                if (ret <= 0) {
                        if (ret == 0) {
                                /* Orderly shutdown. Don't return 0 which means success. */
                                ret = -1;
                        }
                        /* The above call will print a PERROR on error. */
-                       DBG("Error on recv consumer is data pending on sock %d", socket->fd);
+                       DBG("Error on recv consumer is data pending on sock %d", *socket->fd);
                        pthread_mutex_unlock(socket->lock);
                        goto error_unlock;
                }
@@ -1072,7 +1077,7 @@ int consumer_flush_channel(struct consumer_socket *socket, uint64_t key)
        struct lttcomm_consumer_msg msg;
 
        assert(socket);
-       assert(socket->fd >= 0);
+       assert(socket->fd);
 
        DBG2("Consumer flush channel key %" PRIu64, key);
 
@@ -1105,7 +1110,7 @@ int consumer_close_metadata(struct consumer_socket *socket,
        struct lttcomm_consumer_msg msg;
 
        assert(socket);
-       assert(socket->fd >= 0);
+       assert(socket->fd);
 
        DBG2("Consumer close metadata channel key %" PRIu64, metadata_key);
 
@@ -1138,7 +1143,7 @@ int consumer_setup_metadata(struct consumer_socket *socket,
        struct lttcomm_consumer_msg msg;
 
        assert(socket);
-       assert(socket->fd >= 0);
+       assert(socket->fd);
 
        DBG2("Consumer setup metadata channel key %" PRIu64, metadata_key);
 
@@ -1172,9 +1177,9 @@ int consumer_push_metadata(struct consumer_socket *socket,
        struct lttcomm_consumer_msg msg;
 
        assert(socket);
-       assert(socket->fd >= 0);
+       assert(socket->fd);
 
-       DBG2("Consumer push metadata to consumer socket %d", socket->fd);
+       DBG2("Consumer push metadata to consumer socket %d", *socket->fd);
 
        msg.cmd_type = LTTNG_CONSUMER_PUSH_METADATA;
        msg.u.push_metadata.key = metadata_key;
@@ -1187,9 +1192,9 @@ int consumer_push_metadata(struct consumer_socket *socket,
                goto end;
        }
 
-       DBG3("Consumer pushing metadata on sock %d of len %zu", socket->fd, len);
+       DBG3("Consumer pushing metadata on sock %d of len %zu", *socket->fd, len);
 
-       ret = lttcomm_send_unix_sock(socket->fd, metadata_str, len);
+       ret = lttcomm_send_unix_sock(*socket->fd, metadata_str, len);
        if (ret < 0) {
                goto end;
        }
@@ -1218,7 +1223,7 @@ int consumer_snapshot_channel(struct consumer_socket *socket, uint64_t key,
        struct lttcomm_consumer_msg msg;
 
        assert(socket);
-       assert(socket->fd >= 0);
+       assert(socket->fd);
        assert(output);
        assert(output->consumer);
 

diff --git a/src/bin/lttng-sessiond/consumer.h b/src/bin/lttng-sessiond/consumer.h
index 2dd3b81ddbb40b537b1c314f131446a609d6d406..7ff3e5dcf70cbe06f28c371480ffd4391eec75a1 100644 (file)
--- a/src/bin/lttng-sessiond/consumer.h
+++ b/src/bin/lttng-sessiond/consumer.h
@@ -33,8 +33,12 @@ enum consumer_dst_type {
 };
 
 struct consumer_socket {
-       /* File descriptor */
-       int fd;
+       /*
+        * File descriptor. This is just a reference to the consumer data meaning
+        * that every access must be locked and check for a possible invalid value.
+        */
+       int *fd;
+
        /*
         * To use this socket (send/recv), this lock MUST be acquired.
         */
@@ -87,6 +91,13 @@ struct consumer_data {
        int err_sock;
        /* These two sockets uses the cmd_unix_sock_path. */
        int cmd_sock;
+       /*
+        * The metadata socket object is handled differently and only created
+        * locally in this object thus it's the only reference available in the
+        * session daemon. For that reason, a static variable for the fd is
+        * required and the metadata socket fd points to it.
+        */
+       int metadata_fd;
        struct consumer_socket metadata_sock;
 
        /* consumer error and command Unix socket path */
@@ -157,7 +168,7 @@ struct consumer_socket *consumer_find_socket(int key,
                struct consumer_output *consumer);
 struct consumer_socket *consumer_find_socket_by_bitness(int bits,
                struct consumer_output *consumer);
-struct consumer_socket *consumer_allocate_socket(int fd);
+struct consumer_socket *consumer_allocate_socket(int *fd);
 void consumer_add_socket(struct consumer_socket *sock,
                struct consumer_output *consumer);
 void consumer_del_socket(struct consumer_socket *sock,

diff --git a/src/bin/lttng-sessiond/kernel-consumer.c b/src/bin/lttng-sessiond/kernel-consumer.c
index bf3559a21fa021a55e5cebeb7a582b72e80b87de..1405554d97cc04e2a466c74a8930d2deb27de0de 100644 (file)
--- a/src/bin/lttng-sessiond/kernel-consumer.c
+++ b/src/bin/lttng-sessiond/kernel-consumer.c
@@ -371,7 +371,7 @@ int kernel_consumer_destroy_channel(struct consumer_socket *socket,
 
        assert(channel);
        assert(socket);
-       assert(socket->fd >= 0);
+       assert(socket->fd);
 
        DBG("Sending kernel consumer destroy channel key %d", channel->fd);
 
@@ -400,7 +400,7 @@ int kernel_consumer_destroy_metadata(struct consumer_socket *socket,
 
        assert(metadata);
        assert(socket);
-       assert(socket->fd >= 0);
+       assert(socket->fd);
 
        DBG("Sending kernel consumer destroy channel key %d", metadata->fd);
 

diff --git a/src/bin/lttng-sessiond/kernel.c b/src/bin/lttng-sessiond/kernel.c
index bf1a52171ace5fd378d66a0beae2df458839e87b..5d4f61451a53cf3d2d6cd8da21feeec086ad9e94 100644 (file)
--- a/src/bin/lttng-sessiond/kernel.c
+++ b/src/bin/lttng-sessiond/kernel.c
@@ -864,7 +864,7 @@ int kernel_snapshot_record(struct ltt_kernel_session *ksess,
                struct consumer_output *saved_output;
                struct ltt_kernel_channel *chan;
                /* Code flow error */
-               assert(socket->fd >= 0);
+               assert(socket->fd);
 
                /*
                 * Temporarly switch consumer output for our snapshot output. As long

diff --git a/src/bin/lttng-sessiond/main.c b/src/bin/lttng-sessiond/main.c
index 4ebd1e7d6c50d89da95aae89ff7e4d7b1ea77c59..b688a4c09e45f25c3d62bc82b86cc73bcb23fe16 100644 (file)
--- a/src/bin/lttng-sessiond/main.c
+++ b/src/bin/lttng-sessiond/main.c
@@ -90,7 +90,6 @@ static struct consumer_data kconsumer_data = {
        .cmd_unix_sock_path = DEFAULT_KCONSUMERD_CMD_SOCK_PATH,
        .err_sock = -1,
        .cmd_sock = -1,
-       .metadata_sock.fd = -1,
        .pid_mutex = PTHREAD_MUTEX_INITIALIZER,
        .lock = PTHREAD_MUTEX_INITIALIZER,
        .cond = PTHREAD_COND_INITIALIZER,
@@ -102,7 +101,6 @@ static struct consumer_data ustconsumer64_data = {
        .cmd_unix_sock_path = DEFAULT_USTCONSUMERD64_CMD_SOCK_PATH,
        .err_sock = -1,
        .cmd_sock = -1,
-       .metadata_sock.fd = -1,
        .pid_mutex = PTHREAD_MUTEX_INITIALIZER,
        .lock = PTHREAD_MUTEX_INITIALIZER,
        .cond = PTHREAD_COND_INITIALIZER,
@@ -114,7 +112,6 @@ static struct consumer_data ustconsumer32_data = {
        .cmd_unix_sock_path = DEFAULT_USTCONSUMERD32_CMD_SOCK_PATH,
        .err_sock = -1,
        .cmd_sock = -1,
-       .metadata_sock.fd = -1,
        .pid_mutex = PTHREAD_MUTEX_INITIALIZER,
        .lock = PTHREAD_MUTEX_INITIALIZER,
        .cond = PTHREAD_COND_INITIALIZER,
@@ -694,7 +691,7 @@ static int update_kernel_stream(struct consumer_data *consumer_data, int fd)
                                        cds_lfht_for_each_entry(ksess->consumer->socks->ht,
                                                        &iter.iter, socket, node.node) {
                                                /* Code flow error */
-                                               assert(socket->fd >= 0);
+                                               assert(socket->fd);
 
                                                pthread_mutex_lock(socket->lock);
                                                ret = kernel_consumer_send_channel_stream(socket,
@@ -1019,15 +1016,15 @@ restart:
                /* Connect both socket, command and metadata. */
                consumer_data->cmd_sock =
                        lttcomm_connect_unix_sock(consumer_data->cmd_unix_sock_path);
-               consumer_data->metadata_sock.fd =
+               consumer_data->metadata_fd =
                        lttcomm_connect_unix_sock(consumer_data->cmd_unix_sock_path);
-               if (consumer_data->cmd_sock < 0 ||
-                               consumer_data->metadata_sock.fd < 0) {
+               if (consumer_data->cmd_sock < 0 || consumer_data->metadata_fd < 0) {
                        PERROR("consumer connect cmd socket");
                        /* On error, signal condition and quit. */
                        signal_consumer_condition(consumer_data, -1);
                        goto error;
                }
+               consumer_data->metadata_sock.fd = &consumer_data->metadata_fd;
                /* Create metadata socket lock. */
                consumer_data->metadata_sock.lock = zmalloc(sizeof(pthread_mutex_t));
                if (consumer_data->metadata_sock.lock == NULL) {
@@ -1040,7 +1037,7 @@ restart:
                signal_consumer_condition(consumer_data, 1);
                DBG("Consumer command socket ready (fd: %d", consumer_data->cmd_sock);
                DBG("Consumer metadata socket ready (fd: %d)",
-                               consumer_data->metadata_sock.fd);
+                               consumer_data->metadata_fd);
        } else {
                ERR("consumer error when waiting for SOCK_READY : %s",
                                lttcomm_get_readable_code(-code));
@@ -1060,7 +1057,7 @@ restart:
        }
 
        /* Add metadata socket that is successfully connected. */
-       ret = lttng_poll_add(&events, consumer_data->metadata_sock.fd,
+       ret = lttng_poll_add(&events, consumer_data->metadata_fd,
                        LPOLLIN | LPOLLRDHUP);
        if (ret < 0) {
                goto error;
@@ -1119,7 +1116,7 @@ restart_poll:
                                                lttcomm_get_readable_code(-code));
 
                                goto exit;
-                       } else if (pollfd == consumer_data->metadata_sock.fd) {
+                       } else if (pollfd == consumer_data->metadata_fd) {
                                /* UST metadata requests */
                                ret = ust_consumer_metadata_request(
                                                &consumer_data->metadata_sock);
@@ -1163,8 +1160,8 @@ error:
                }
                consumer_data->cmd_sock = -1;
        }
-       if (consumer_data->metadata_sock.fd >= 0) {
-               ret = close(consumer_data->metadata_sock.fd);
+       if (*consumer_data->metadata_sock.fd >= 0) {
+               ret = close(*consumer_data->metadata_sock.fd);
                if (ret) {
                        PERROR("close");
                }

diff --git a/src/bin/lttng-sessiond/ust-consumer.c b/src/bin/lttng-sessiond/ust-consumer.c
index d6fe58cd971aa9f1f770f99860a65129c329ac76..2eb8cda253a096c89f03d0e7cfd6bd869ed675f2 100644 (file)
--- a/src/bin/lttng-sessiond/ust-consumer.c
+++ b/src/bin/lttng-sessiond/ust-consumer.c
@@ -166,7 +166,7 @@ static int ask_channel_creation(struct ust_app_session *ua_sess,
 
        health_code_update();
 
-       ret = lttcomm_send_unix_sock(socket->fd, &msg, sizeof(msg));
+       ret = lttcomm_send_unix_sock(*socket->fd, &msg, sizeof(msg));
        if (ret < 0) {
                goto error;
        }
@@ -207,7 +207,7 @@ int ust_consumer_ask_channel(struct ust_app_session *ua_sess,
        assert(ua_chan);
        assert(consumer);
        assert(socket);
-       assert(socket->fd >= 0);
+       assert(socket->fd);
        assert(registry);
 
        if (!consumer->enabled) {
@@ -242,7 +242,7 @@ int ust_consumer_get_channel(struct consumer_socket *socket,
 
        assert(ua_chan);
        assert(socket);
-       assert(socket->fd >= 0);
+       assert(socket->fd);
 
        msg.cmd_type = LTTNG_CONSUMER_GET_CHANNEL;
        msg.u.get_channel.key = ua_chan->key;
@@ -257,11 +257,11 @@ int ust_consumer_get_channel(struct consumer_socket *socket,
        }
 
        /* First, get the channel from consumer. */
-       ret = ustctl_recv_channel_from_consumer(socket->fd, &ua_chan->obj);
+       ret = ustctl_recv_channel_from_consumer(*socket->fd, &ua_chan->obj);
        if (ret < 0) {
                if (ret != -EPIPE) {
                        ERR("Error recv channel from consumer %d with ret %d",
-                                       socket->fd, ret);
+                                       *socket->fd, ret);
                } else {
                        DBG3("UST app recv channel from consumer. Consumer is dead.");
                }
@@ -280,7 +280,7 @@ int ust_consumer_get_channel(struct consumer_socket *socket,
                }
 
                /* Stream object is populated by this call if successful. */
-               ret = ustctl_recv_stream_from_consumer(socket->fd, &stream->obj);
+               ret = ustctl_recv_stream_from_consumer(*socket->fd, &stream->obj);
                if (ret < 0) {
                        free(stream);
                        if (ret == -LTTNG_UST_ERR_NOENT) {
@@ -290,7 +290,7 @@ int ust_consumer_get_channel(struct consumer_socket *socket,
                        }
                        if (ret != -EPIPE) {
                                ERR("Recv stream from consumer %d with ret %d",
-                                               socket->fd, ret);
+                                               *socket->fd, ret);
                        } else {
                                DBG3("UST app recv stream from consumer. Consumer is dead.");
                        }
@@ -337,7 +337,7 @@ int ust_consumer_destroy_channel(struct consumer_socket *socket,
 
        assert(ua_chan);
        assert(socket);
-       assert(socket->fd >= 0);
+       assert(socket->fd);
 
        msg.cmd_type = LTTNG_CONSUMER_DESTROY_CHANNEL;
        msg.u.destroy_channel.key = ua_chan->key;
@@ -445,7 +445,7 @@ int ust_consumer_metadata_request(struct consumer_socket *socket)
        health_code_update();
 
        /* Wait for a metadata request */
-       ret = lttcomm_recv_unix_sock(socket->fd, &request, sizeof(request));
+       ret = lttcomm_recv_unix_sock(*socket->fd, &request, sizeof(request));
        if (ret <= 0) {
                ERR("Consumer closed the metadata socket");
                ret = -1;