Start notification subsystem thread in sessiond initialization

Signed-off-by: Jérémie Galarneau <jeremie.galarneau@efficios.com>

diff --git a/src/bin/lttng-sessiond/main.c b/src/bin/lttng-sessiond/main.c
index deb2edf9ba1e5189cd1d9e40934abfc3da1e43ab..9cde946495436f04590339475e96f95b5ea7421c 100644 (file)
--- a/src/bin/lttng-sessiond/main.c
+++ b/src/bin/lttng-sessiond/main.c
@@ -71,6 +71,8 @@
 #include "agent-thread.h"
 #include "save.h"
 #include "load-session-thread.h"
+#include "notification-thread.h"
+#include "notification-thread-commands.h"
 #include "syscall.h"
 #include "agent.h"
 #include "ht-cleanup.h"
@@ -78,7 +80,7 @@
 #define CONSUMERD_FILE "lttng-consumerd"
 
 const char *progname;
-static const char *tracing_group_name = DEFAULT_TRACING_GROUP;
+const char *tracing_group_name = DEFAULT_TRACING_GROUP;
 static int tracing_group_name_override;
 static char *opt_pidfile;
 static int opt_sig_parent;
@@ -104,6 +106,7 @@ static struct consumer_data kconsumer_data = {
        .cmd_unix_sock_path = DEFAULT_KCONSUMERD_CMD_SOCK_PATH,
        .err_sock = -1,
        .cmd_sock = -1,
+       .channel_monitor_pipe = -1,
        .pid_mutex = PTHREAD_MUTEX_INITIALIZER,
        .lock = PTHREAD_MUTEX_INITIALIZER,
        .cond = PTHREAD_COND_INITIALIZER,
@@ -115,6 +118,7 @@ static struct consumer_data ustconsumer64_data = {
        .cmd_unix_sock_path = DEFAULT_USTCONSUMERD64_CMD_SOCK_PATH,
        .err_sock = -1,
        .cmd_sock = -1,
+       .channel_monitor_pipe = -1,
        .pid_mutex = PTHREAD_MUTEX_INITIALIZER,
        .lock = PTHREAD_MUTEX_INITIALIZER,
        .cond = PTHREAD_COND_INITIALIZER,
@@ -126,6 +130,7 @@ static struct consumer_data ustconsumer32_data = {
        .cmd_unix_sock_path = DEFAULT_USTCONSUMERD32_CMD_SOCK_PATH,
        .err_sock = -1,
        .cmd_sock = -1,
+       .channel_monitor_pipe = -1,
        .pid_mutex = PTHREAD_MUTEX_INITIALIZER,
        .lock = PTHREAD_MUTEX_INITIALIZER,
        .cond = PTHREAD_COND_INITIALIZER,
@@ -211,6 +216,7 @@ static pthread_t health_thread;
 static pthread_t ht_cleanup_thread;
 static pthread_t agent_reg_thread;
 static pthread_t load_session_thread;
+static pthread_t notification_thread;
 
 /*
  * UST registration command queue. This queue is tied with a futex and uses a N
@@ -305,15 +311,19 @@ const char * const config_section_name = "sessiond";
 /* Load session thread information to operate. */
 struct load_session_thread_data *load_info;
 
+/* Notification thread handle. */
+struct notification_thread_handle *notification_thread_handle;
+
 /* Global hash tables */
 struct lttng_ht *agent_apps_ht_by_sock = NULL;
 
 /*
- * Whether sessiond is ready for commands/health check requests.
+ * Whether sessiond is ready for commands/notification channel/health check
+ * requests.
  * NR_LTTNG_SESSIOND_READY must match the number of calls to
  * sessiond_notify_ready().
  */
-#define NR_LTTNG_SESSIOND_READY                3
+#define NR_LTTNG_SESSIOND_READY                4
 int lttng_sessiond_ready = NR_LTTNG_SESSIOND_READY;
 
 int sessiond_check_thread_quit_pipe(int fd, uint32_t events)
@@ -521,6 +531,24 @@ static void close_consumer_sockets(void)
                        PERROR("UST consumerd64 cmd_sock close");
                }
        }
+       if (kconsumer_data.channel_monitor_pipe >= 0) {
+               ret = close(kconsumer_data.channel_monitor_pipe);
+               if (ret < 0) {
+                       PERROR("kernel consumer channel monitor pipe close");
+               }
+       }
+       if (ustconsumer32_data.channel_monitor_pipe >= 0) {
+               ret = close(ustconsumer32_data.channel_monitor_pipe);
+               if (ret < 0) {
+                       PERROR("UST consumerd32 channel monitor pipe close");
+               }
+       }
+       if (ustconsumer64_data.channel_monitor_pipe >= 0) {
+               ret = close(ustconsumer64_data.channel_monitor_pipe);
+               if (ret < 0) {
+                       PERROR("UST consumerd64 channel monitor pipe close");
+               }
+       }
 }
 
 /*
@@ -1224,6 +1252,7 @@ static void *thread_manage_consumer(void *data)
        enum lttcomm_return_code code;
        struct lttng_poll_event events;
        struct consumer_data *consumer_data = data;
+       struct consumer_socket *cmd_socket_wrapper = NULL;
 
        DBG("[thread] Manage consumer started");
 
@@ -1333,39 +1362,43 @@ restart:
        }
 
        health_code_update();
-       if (code == LTTCOMM_CONSUMERD_COMMAND_SOCK_READY) {
-               /* Connect both socket, command and metadata. */
-               consumer_data->cmd_sock =
-                       lttcomm_connect_unix_sock(consumer_data->cmd_unix_sock_path);
-               consumer_data->metadata_fd =
-                       lttcomm_connect_unix_sock(consumer_data->cmd_unix_sock_path);
-               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_ptr = &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) {
-                       PERROR("zmalloc pthread mutex");
-                       goto error;
-               }
-               pthread_mutex_init(consumer_data->metadata_sock.lock, NULL);
-
-               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_fd);
-       } else {
+       if (code != LTTCOMM_CONSUMERD_COMMAND_SOCK_READY) {
                ERR("consumer error when waiting for SOCK_READY : %s",
                                lttcomm_get_readable_code(-code));
                goto error;
        }
 
-       /* Remove the consumerd error sock since we've established a connexion */
+       /* Connect both command and metadata sockets. */
+       consumer_data->cmd_sock =
+                       lttcomm_connect_unix_sock(
+                               consumer_data->cmd_unix_sock_path);
+       consumer_data->metadata_fd =
+                       lttcomm_connect_unix_sock(
+                               consumer_data->cmd_unix_sock_path);
+       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_ptr = &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) {
+               PERROR("zmalloc pthread mutex");
+               goto error;
+       }
+       pthread_mutex_init(consumer_data->metadata_sock.lock, NULL);
+
+       DBG("Consumer command socket ready (fd: %d", consumer_data->cmd_sock);
+       DBG("Consumer metadata socket ready (fd: %d)",
+                       consumer_data->metadata_fd);
+
+       /*
+        * Remove the consumerd error sock since we've established a connection.
+        */
        ret = lttng_poll_del(&events, consumer_data->err_sock);
        if (ret < 0) {
                goto error;
@@ -1386,6 +1419,27 @@ restart:
 
        health_code_update();
 
+       /*
+        * Transfer the write-end of the channel monitoring pipe to the
+        * by issuing a SET_CHANNEL_MONITOR_PIPE command.
+        */
+       cmd_socket_wrapper = consumer_allocate_socket(&consumer_data->cmd_sock);
+       if (!cmd_socket_wrapper) {
+               goto error;
+       }
+
+       ret = consumer_send_channel_monitor_pipe(cmd_socket_wrapper,
+                       consumer_data->channel_monitor_pipe);
+       if (ret) {
+               goto error;
+       }
+       /* Discard the socket wrapper as it is no longer needed. */
+       consumer_destroy_socket(cmd_socket_wrapper);
+       cmd_socket_wrapper = NULL;
+
+       /* The thread is completely initialized, signal that it is ready. */
+       signal_consumer_condition(consumer_data, 1);
+
        /* Infinite blocking call, waiting for transmission */
 restart_poll:
        while (1) {
@@ -1529,6 +1583,10 @@ error:
                free(consumer_data->metadata_sock.lock);
        }
        lttng_poll_clean(&events);
+
+       if (cmd_socket_wrapper) {
+               consumer_destroy_socket(cmd_socket_wrapper);
+       }
 error_poll:
        if (err) {
                health_error();
@@ -3004,6 +3062,8 @@ static int process_client_msg(struct command_ctx *cmd_ctx, int sock,
        case LTTNG_SET_SESSION_SHM_PATH:
        case LTTNG_REGENERATE_METADATA:
        case LTTNG_REGENERATE_STATEDUMP:
+       case LTTNG_REGISTER_TRIGGER:
+       case LTTNG_UNREGISTER_TRIGGER:
                need_domain = 0;
                break;
        default:
@@ -3066,6 +3126,8 @@ static int process_client_msg(struct command_ctx *cmd_ctx, int sock,
        case LTTNG_LIST_SYSCALLS:
        case LTTNG_LIST_TRACEPOINT_FIELDS:
        case LTTNG_SAVE_SESSION:
+       case LTTNG_REGISTER_TRIGGER:
+       case LTTNG_UNREGISTER_TRIGGER:
                need_tracing_session = 0;
                break;
        default:
@@ -3449,8 +3511,11 @@ error_add_context:
        }
        case LTTNG_ENABLE_CHANNEL:
        {
+               cmd_ctx->lsm->u.channel.chan.attr.extended.ptr =
+                               (struct lttng_channel_extended *) &cmd_ctx->lsm->u.channel.extended;
                ret = cmd_enable_channel(cmd_ctx->session, &cmd_ctx->lsm->domain,
-                               &cmd_ctx->lsm->u.channel.chan, kernel_poll_pipe[1]);
+                               &cmd_ctx->lsm->u.channel.chan,
+                               kernel_poll_pipe[1]);
                break;
        }
        case LTTNG_TRACK_PID:
@@ -4117,6 +4182,18 @@ error_add_context:
                ret = cmd_regenerate_statedump(cmd_ctx->session);
                break;
        }
+       case LTTNG_REGISTER_TRIGGER:
+       {
+               ret = cmd_register_trigger(cmd_ctx, sock,
+                               notification_thread_handle);
+               break;
+       }
+       case LTTNG_UNREGISTER_TRIGGER:
+       {
+               ret = cmd_unregister_trigger(cmd_ctx, sock,
+                               notification_thread_handle);
+               break;
+       }
        default:
                ret = LTTNG_ERR_UND;
                break;
@@ -5531,6 +5608,9 @@ int main(int argc, char **argv)
        int ret = 0, retval = 0;
        void *status;
        const char *home_path, *env_app_timeout;
+       struct lttng_pipe *ust32_channel_monitor_pipe = NULL,
+                       *ust64_channel_monitor_pipe = NULL,
+                       *kernel_channel_monitor_pipe = NULL;
 
        init_kernel_workarounds();
 
@@ -5689,6 +5769,19 @@ int main(int argc, char **argv)
                                kconsumer_data.err_unix_sock_path);
                DBG2("Kernel consumer cmd path: %s",
                                kconsumer_data.cmd_unix_sock_path);
+               kernel_channel_monitor_pipe = lttng_pipe_open(0);
+               if (!kernel_channel_monitor_pipe) {
+                       ERR("Failed to create kernel consumer channel monitor pipe");
+                       retval = -1;
+                       goto exit_init_data;
+               }
+               kconsumer_data.channel_monitor_pipe =
+                               lttng_pipe_release_writefd(
+                                       kernel_channel_monitor_pipe);
+               if (kconsumer_data.channel_monitor_pipe < 0) {
+                       retval = -1;
+                       goto exit_init_data;
+               }
        } else {
                home_path = utils_get_home_dir();
                if (home_path == NULL) {
@@ -5793,6 +5886,18 @@ int main(int argc, char **argv)
                        ustconsumer32_data.err_unix_sock_path);
        DBG2("UST consumer 32 bits cmd path: %s",
                        ustconsumer32_data.cmd_unix_sock_path);
+       ust32_channel_monitor_pipe = lttng_pipe_open(0);
+       if (!ust32_channel_monitor_pipe) {
+               ERR("Failed to create 32-bit user space consumer channel monitor pipe");
+               retval = -1;
+               goto exit_init_data;
+       }
+       ustconsumer32_data.channel_monitor_pipe = lttng_pipe_release_writefd(
+                       ust32_channel_monitor_pipe);
+       if (ustconsumer32_data.channel_monitor_pipe < 0) {
+               retval = -1;
+               goto exit_init_data;
+       }
 
        /* 64 bits consumerd path setup */
        ret = snprintf(ustconsumer64_data.err_unix_sock_path, PATH_MAX,
@@ -5814,6 +5919,18 @@ int main(int argc, char **argv)
                        ustconsumer64_data.err_unix_sock_path);
        DBG2("UST consumer 64 bits cmd path: %s",
                        ustconsumer64_data.cmd_unix_sock_path);
+       ust64_channel_monitor_pipe = lttng_pipe_open(0);
+       if (!ust64_channel_monitor_pipe) {
+               ERR("Failed to create 64-bit user space consumer channel monitor pipe");
+               retval = -1;
+               goto exit_init_data;
+       }
+       ustconsumer64_data.channel_monitor_pipe = lttng_pipe_release_writefd(
+                       ust64_channel_monitor_pipe);
+       if (ustconsumer64_data.channel_monitor_pipe < 0) {
+               retval = -1;
+               goto exit_init_data;
+       }
 
        /*
         * See if daemon already exist.
@@ -5983,6 +6100,29 @@ int main(int argc, char **argv)
                goto exit_health;
        }
 
+       /* notification_thread_data acquires the pipes' read side. */
+       notification_thread_handle = notification_thread_handle_create(
+                       ust32_channel_monitor_pipe,
+                       ust64_channel_monitor_pipe,
+                       kernel_channel_monitor_pipe);
+       if (!notification_thread_handle) {
+               retval = -1;
+               ERR("Failed to create notification thread shared data");
+               stop_threads();
+               goto exit_notification;
+       }
+
+       /* Create notification thread. */
+       ret = pthread_create(&notification_thread, default_pthread_attr(),
+                       thread_notification, notification_thread_handle);
+       if (ret) {
+               errno = ret;
+               PERROR("pthread_create notification");
+               retval = -1;
+               stop_threads();
+               goto exit_notification;
+       }
+
        /* Create thread to manage the client socket */
        ret = pthread_create(&client_thread, default_pthread_attr(),
                        thread_manage_clients, (void *) NULL);
@@ -5990,6 +6130,7 @@ int main(int argc, char **argv)
                errno = ret;
                PERROR("pthread_create clients");
                retval = -1;
+               stop_threads();
                goto exit_client;
        }
 
@@ -6000,6 +6141,7 @@ int main(int argc, char **argv)
                errno = ret;
                PERROR("pthread_create dispatch");
                retval = -1;
+               stop_threads();
                goto exit_dispatch;
        }
 
@@ -6010,6 +6152,7 @@ int main(int argc, char **argv)
                errno = ret;
                PERROR("pthread_create registration");
                retval = -1;
+               stop_threads();
                goto exit_reg_apps;
        }
 
@@ -6020,6 +6163,7 @@ int main(int argc, char **argv)
                errno = ret;
                PERROR("pthread_create apps");
                retval = -1;
+               stop_threads();
                goto exit_apps;
        }
 
@@ -6030,6 +6174,7 @@ int main(int argc, char **argv)
                errno = ret;
                PERROR("pthread_create notify");
                retval = -1;
+               stop_threads();
                goto exit_apps_notify;
        }
 
@@ -6040,6 +6185,7 @@ int main(int argc, char **argv)
                errno = ret;
                PERROR("pthread_create agent");
                retval = -1;
+               stop_threads();
                goto exit_agent_reg;
        }
 
@@ -6052,6 +6198,7 @@ int main(int argc, char **argv)
                        errno = ret;
                        PERROR("pthread_create kernel");
                        retval = -1;
+                       stop_threads();
                        goto exit_kernel;
                }
        }
@@ -6063,6 +6210,7 @@ int main(int argc, char **argv)
                errno = ret;
                PERROR("pthread_create load_session_thread");
                retval = -1;
+               stop_threads();
                goto exit_load_session;
        }
 
@@ -6139,16 +6287,17 @@ exit_dispatch:
                PERROR("pthread_join");
                retval = -1;
        }
-exit_client:
 
+exit_client:
+exit_notification:
        ret = pthread_join(health_thread, &status);
        if (ret) {
                errno = ret;
                PERROR("pthread_join health thread");
                retval = -1;
        }
-exit_health:
 
+exit_health:
 exit_init_data:
        /*
         * Wait for all pending call_rcu work to complete before tearing
@@ -6162,8 +6311,6 @@ exit_init_data:
         */
        rcu_thread_online();
        sessiond_cleanup();
-       rcu_thread_offline();
-       rcu_unregister_thread();
 
        /*
         * Ensure all prior call_rcu are done. call_rcu callbacks may push
@@ -6172,10 +6319,33 @@ exit_init_data:
         */
        rcu_barrier();
 
+       /*
+        * The teardown of the notification system is performed after the
+        * session daemon's teardown in order to allow it to be notified
+        * of the active session and channels at the moment of the teardown.
+        */
+       if (notification_thread_handle) {
+               notification_thread_command_quit(notification_thread_handle);
+               notification_thread_handle_destroy(notification_thread_handle);
+       }
+
+       ret = pthread_join(notification_thread, &status);
+       if (ret) {
+               errno = ret;
+               PERROR("pthread_join notification thread");
+               retval = -1;
+       }
+
+       rcu_thread_offline();
+       rcu_unregister_thread();
+
        ret = fini_ht_cleanup_thread(&ht_cleanup_thread);
        if (ret) {
                retval = -1;
        }
+       lttng_pipe_destroy(ust32_channel_monitor_pipe);
+       lttng_pipe_destroy(ust64_channel_monitor_pipe);
+       lttng_pipe_destroy(kernel_channel_monitor_pipe);
 exit_ht_cleanup:
 
        health_app_destroy(health_sessiond);
@@ -6186,7 +6356,6 @@ exit_options:
        sessiond_cleanup_options();
 
 exit_set_signal_handler:
-
        if (!retval) {
                exit(EXIT_SUCCESS);
        } else {

diff --git a/src/common/consumer/consumer.h b/src/common/consumer/consumer.h
index 37adecbfe689e28a79303c601cefbe2ea6311677..527331298c0440ece041d2a73a5296352aa8304b 100644 (file)
--- a/src/common/consumer/consumer.h
+++ b/src/common/consumer/consumer.h
@@ -61,6 +61,7 @@ enum lttng_consumer_command {
        LTTNG_CONSUMER_DISCARDED_EVENTS,
        LTTNG_CONSUMER_LOST_PACKETS,
        LTTNG_CONSUMER_CLEAR_QUIESCENT_CHANNEL,
+       LTTNG_CONSUMER_SET_CHANNEL_MONITOR_PIPE,
 };
 
 /* State of each fd in consumer */
@@ -170,6 +171,10 @@ struct lttng_consumer_channel {
        timer_t live_timer;
        int live_timer_error;
 
+       /* For channel monitoring timer. */
+       int monitor_timer_enabled;
+       timer_t monitor_timer;
+
        /* On-disk circular buffer */
        uint64_t tracefile_size;
        uint64_t tracefile_count;
@@ -539,6 +544,11 @@ struct lttng_consumer_local_data {
        int consumer_should_quit[2];
        /* Metadata poll thread pipe. Transfer metadata stream to it */
        struct lttng_pipe *consumer_metadata_pipe;
+       /*
+        * Pipe used by the channel monitoring timers to provide state samples
+        * to the session daemon (write-only).
+        */
+       int channel_monitor_pipe;
 };
 
 /*

diff --git a/src/common/kernel-consumer/kernel-consumer.c b/src/common/kernel-consumer/kernel-consumer.c
index d5418dcdabb0217873baf94dc11c15c8af96566d..1a786352d75adbfaedf93783ab4d9b91f559c0d2 100644 (file)
--- a/src/common/kernel-consumer/kernel-consumer.c
+++ b/src/common/kernel-consumer/kernel-consumer.c
@@ -1,6 +1,7 @@
 /*
  * Copyright (C) 2011 - Julien Desfossez <julien.desfossez@polymtl.ca>
  *                      Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
+ * Copyright (C) 2017 - Jérémie Galarneau <jeremie.galarneau@efficios.com>
  *
  * This program is free software; you can redistribute it and/or modify
  * it under the terms of the GNU General Public License, version 2 only,
@@ -1012,6 +1013,55 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
 
                break;
        }
+       case LTTNG_CONSUMER_SET_CHANNEL_MONITOR_PIPE:
+       {
+               int channel_monitor_pipe;
+
+               ret_code = LTTCOMM_CONSUMERD_SUCCESS;
+               /* Successfully received the command's type. */
+               ret = consumer_send_status_msg(sock, ret_code);
+               if (ret < 0) {
+                       goto error_fatal;
+               }
+
+               ret = lttcomm_recv_fds_unix_sock(sock, &channel_monitor_pipe,
+                               1);
+               if (ret != sizeof(channel_monitor_pipe)) {
+                       ERR("Failed to receive channel monitor pipe");
+                       goto error_fatal;
+               }
+
+               DBG("Received channel monitor pipe (%d)", channel_monitor_pipe);
+               ret = consumer_timer_thread_set_channel_monitor_pipe(
+                               channel_monitor_pipe);
+               if (!ret) {
+                       int flags;
+
+                       ret_code = LTTCOMM_CONSUMERD_SUCCESS;
+                       /* Set the pipe as non-blocking. */
+                       ret = fcntl(channel_monitor_pipe, F_GETFL, 0);
+                       if (ret == -1) {
+                               PERROR("fcntl get flags of the channel monitoring pipe");
+                               goto error_fatal;
+                       }
+                       flags = ret;
+
+                       ret = fcntl(channel_monitor_pipe, F_SETFL,
+                                       flags | O_NONBLOCK);
+                       if (ret == -1) {
+                               PERROR("fcntl set O_NONBLOCK flag of the channel monitoring pipe");
+                               goto error_fatal;
+                       }
+                       DBG("Channel monitor pipe set as non-blocking");
+               } else {
+                       ret_code = LTTCOMM_CONSUMERD_ALREADY_SET;
+               }
+               ret = consumer_send_status_msg(sock, ret_code);
+               if (ret < 0) {
+                       goto error_fatal;
+               }
+               break;
+       }
        default:
                goto end_nosignal;
        }

diff --git a/src/common/ust-consumer/ust-consumer.c b/src/common/ust-consumer/ust-consumer.c
index 97f0497cb12b14a71da7289c5030b256a8848ebd..7006e1cb435fb47a4b628e863d24aa338d1728ae 100644 (file)
--- a/src/common/ust-consumer/ust-consumer.c
+++ b/src/common/ust-consumer/ust-consumer.c
@@ -1,6 +1,7 @@
 /*
  * Copyright (C) 2011 - Julien Desfossez <julien.desfossez@polymtl.ca>
  *                      Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
+ * Copyright (C) 2017 - Jérémie Galarneau <jeremie.galarneau@efficios.com>
  *
  * This program is free software; you can redistribute it and/or modify
  * it under the terms of the GNU General Public License, version 2 only,
@@ -1857,6 +1858,51 @@ int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
 
                break;
        }
+       case LTTNG_CONSUMER_SET_CHANNEL_MONITOR_PIPE:
+       {
+               int channel_monitor_pipe;
+
+               ret_code = LTTCOMM_CONSUMERD_SUCCESS;
+               /* Successfully received the command's type. */
+               ret = consumer_send_status_msg(sock, ret_code);
+               if (ret < 0) {
+                       goto error_fatal;
+               }
+
+               ret = lttcomm_recv_fds_unix_sock(sock, &channel_monitor_pipe,
+                               1);
+               if (ret != sizeof(channel_monitor_pipe)) {
+                       ERR("Failed to receive channel monitor pipe");
+                       goto error_fatal;
+               }
+
+               DBG("Received channel monitor pipe (%d)", channel_monitor_pipe);
+               ret = consumer_timer_thread_set_channel_monitor_pipe(
+                               channel_monitor_pipe);
+               if (!ret) {
+                       int flags;
+
+                       ret_code = LTTCOMM_CONSUMERD_SUCCESS;
+                       /* Set the pipe as non-blocking. */
+                       ret = fcntl(channel_monitor_pipe, F_GETFL, 0);
+                       if (ret == -1) {
+                               PERROR("fcntl get flags of the channel monitoring pipe");
+                               goto error_fatal;
+                       }
+                       flags = ret;
+
+                       ret = fcntl(channel_monitor_pipe, F_SETFL,
+                                       flags | O_NONBLOCK);
+                       if (ret == -1) {
+                               PERROR("fcntl set O_NONBLOCK flag of the channel monitoring pipe");
+                               goto error_fatal;
+                       }
+                       DBG("Channel monitor pipe set as non-blocking");
+               } else {
+                       ret_code = LTTCOMM_CONSUMERD_ALREADY_SET;
+               }
+               goto end_msg_sessiond;
+       }
        default:
                break;
        }