2 * Copyright (C) 2017 Jérémie Galarneau <jeremie.galarneau@efficios.com>
4 * SPDX-License-Identifier: GPL-2.0-only
9 #include "health-sessiond.hpp"
11 #include "lttng-sessiond.hpp"
12 #include "notification-thread-commands.hpp"
13 #include "notification-thread-events.hpp"
14 #include "notification-thread.hpp"
15 #include "testpoint.hpp"
18 #include <common/align.hpp>
19 #include <common/config/session-config.hpp>
20 #include <common/defaults.hpp>
21 #include <common/error.hpp>
22 #include <common/kernel-ctl/kernel-ctl.hpp>
23 #include <common/time.hpp>
24 #include <common/utils.hpp>
26 #include <lttng/condition/buffer-usage-internal.hpp>
27 #include <lttng/condition/condition-internal.hpp>
28 #include <lttng/notification/channel-internal.hpp>
29 #include <lttng/notification/notification-internal.hpp>
30 #include <lttng/trigger/trigger.h>
33 #include <sys/eventfd.h>
37 #include <urcu/list.h>
38 #include <urcu/rculfhash.h>
41 * Flag used to temporarily pause data consumption from testpoints.
43 * This variable is dlsym-ed from a test, so needs to be exported.
45 LTTNG_EXPORT
int notifier_consumption_paused
;
48 * Destroy the thread data previously created by the init function.
50 void notification_thread_handle_destroy(struct notification_thread_handle
*handle
)
58 LTTNG_ASSERT(cds_list_empty(&handle
->cmd_queue
.list
));
59 pthread_mutex_destroy(&handle
->cmd_queue
.lock
);
60 sem_destroy(&handle
->ready
);
62 if (handle
->cmd_queue
.event_fd
>= 0) {
63 ret
= close(handle
->cmd_queue
.event_fd
);
65 PERROR("Failed to close notification command queue event fd");
68 if (handle
->channel_monitoring_pipes
.ust32_consumer
>= 0) {
69 ret
= close(handle
->channel_monitoring_pipes
.ust32_consumer
);
71 PERROR("close 32-bit consumer channel monitoring pipe");
74 if (handle
->channel_monitoring_pipes
.ust64_consumer
>= 0) {
75 ret
= close(handle
->channel_monitoring_pipes
.ust64_consumer
);
77 PERROR("close 64-bit consumer channel monitoring pipe");
80 if (handle
->channel_monitoring_pipes
.kernel_consumer
>= 0) {
81 ret
= close(handle
->channel_monitoring_pipes
.kernel_consumer
);
83 PERROR("close kernel consumer channel monitoring pipe");
91 struct notification_thread_handle
*
92 notification_thread_handle_create(struct lttng_pipe
*ust32_channel_monitor_pipe
,
93 struct lttng_pipe
*ust64_channel_monitor_pipe
,
94 struct lttng_pipe
*kernel_channel_monitor_pipe
)
97 struct notification_thread_handle
*handle
;
100 handle
= zmalloc
<notification_thread_handle
>();
105 sem_init(&handle
->ready
, 0, 0);
107 event_fd
= eventfd(0, EFD_CLOEXEC
| EFD_SEMAPHORE
);
109 PERROR("event_fd creation");
113 handle
->cmd_queue
.event_fd
= event_fd
;
115 CDS_INIT_LIST_HEAD(&handle
->cmd_queue
.list
);
116 ret
= pthread_mutex_init(&handle
->cmd_queue
.lock
, nullptr);
121 if (ust32_channel_monitor_pipe
) {
122 handle
->channel_monitoring_pipes
.ust32_consumer
=
123 lttng_pipe_release_readfd(ust32_channel_monitor_pipe
);
124 if (handle
->channel_monitoring_pipes
.ust32_consumer
< 0) {
128 handle
->channel_monitoring_pipes
.ust32_consumer
= -1;
130 if (ust64_channel_monitor_pipe
) {
131 handle
->channel_monitoring_pipes
.ust64_consumer
=
132 lttng_pipe_release_readfd(ust64_channel_monitor_pipe
);
133 if (handle
->channel_monitoring_pipes
.ust64_consumer
< 0) {
137 handle
->channel_monitoring_pipes
.ust64_consumer
= -1;
139 if (kernel_channel_monitor_pipe
) {
140 handle
->channel_monitoring_pipes
.kernel_consumer
=
141 lttng_pipe_release_readfd(kernel_channel_monitor_pipe
);
142 if (handle
->channel_monitoring_pipes
.kernel_consumer
< 0) {
146 handle
->channel_monitoring_pipes
.kernel_consumer
= -1;
152 notification_thread_handle_destroy(handle
);
156 static char *get_notification_channel_sock_path()
159 bool is_root
= !getuid();
162 sock_path
= calloc
<char>(LTTNG_PATH_MAX
);
169 sock_path
, LTTNG_PATH_MAX
, DEFAULT_GLOBAL_NOTIFICATION_CHANNEL_UNIX_SOCK
);
174 const char *home_path
= utils_get_home_dir();
177 ERR("Can't get HOME directory for socket creation");
181 ret
= snprintf(sock_path
,
183 DEFAULT_HOME_NOTIFICATION_CHANNEL_UNIX_SOCK
,
196 static void notification_channel_socket_destroy(int fd
)
199 char *sock_path
= get_notification_channel_sock_path();
201 DBG("Destroying notification channel socket");
204 ret
= unlink(sock_path
);
207 PERROR("unlink notification channel socket");
213 PERROR("close notification channel socket");
217 static int notification_channel_socket_create()
220 char *sock_path
= get_notification_channel_sock_path();
222 DBG("Creating notification channel UNIX socket at %s", sock_path
);
224 ret
= lttcomm_create_unix_sock(sock_path
);
226 ERR("Failed to create notification socket");
231 ret
= chmod(sock_path
, S_IRUSR
| S_IWUSR
| S_IRGRP
| S_IWGRP
);
233 ERR("Set file permissions failed: %s", sock_path
);
234 PERROR("chmod notification channel socket");
241 ret
= utils_get_group_id(the_config
.tracing_group_name
.value
, true, &gid
);
243 /* Default to root group. */
247 ret
= chown(sock_path
, 0, gid
);
249 ERR("Failed to set the notification channel socket's group");
255 DBG("Notification channel UNIX socket created (fd = %i)", fd
);
259 if (fd
>= 0 && close(fd
) < 0) {
260 PERROR("close notification channel socket");
266 static int init_poll_set(struct lttng_poll_event
*poll_set
,
267 struct notification_thread_handle
*handle
,
268 int notification_channel_socket
)
273 * Create pollset with size 5:
274 * - notification channel socket (listen for new connections),
275 * - command queue event fd (internal sessiond commands),
276 * - consumerd (32-bit user space) channel monitor pipe,
277 * - consumerd (64-bit user space) channel monitor pipe,
278 * - consumerd (kernel) channel monitor pipe.
280 ret
= lttng_poll_create(poll_set
, 5, LTTNG_CLOEXEC
);
285 ret
= lttng_poll_add(poll_set
, notification_channel_socket
, LPOLLIN
| LPOLLRDHUP
);
287 ERR("Failed to add notification channel socket to pollset");
290 ret
= lttng_poll_add(poll_set
, handle
->cmd_queue
.event_fd
, LPOLLIN
);
292 ERR("Failed to add notification command queue event fd to pollset");
295 ret
= lttng_poll_add(poll_set
, handle
->channel_monitoring_pipes
.ust32_consumer
, LPOLLIN
);
297 ERR("Failed to add ust-32 channel monitoring pipe fd to pollset");
300 ret
= lttng_poll_add(poll_set
, handle
->channel_monitoring_pipes
.ust64_consumer
, LPOLLIN
);
302 ERR("Failed to add ust-64 channel monitoring pipe fd to pollset");
305 if (handle
->channel_monitoring_pipes
.kernel_consumer
< 0) {
308 ret
= lttng_poll_add(poll_set
, handle
->channel_monitoring_pipes
.kernel_consumer
, LPOLLIN
);
310 ERR("Failed to add kernel channel monitoring pipe fd to pollset");
316 lttng_poll_clean(poll_set
);
320 static void fini_thread_state(struct notification_thread_state
*state
)
324 if (state
->client_socket_ht
) {
325 ret
= handle_notification_thread_client_disconnect_all(state
);
327 ret
= cds_lfht_destroy(state
->client_socket_ht
, nullptr);
330 if (state
->client_id_ht
) {
331 ret
= cds_lfht_destroy(state
->client_id_ht
, nullptr);
334 if (state
->triggers_ht
) {
335 ret
= handle_notification_thread_trigger_unregister_all(state
);
337 ret
= cds_lfht_destroy(state
->triggers_ht
, nullptr);
340 if (state
->channel_triggers_ht
) {
341 ret
= cds_lfht_destroy(state
->channel_triggers_ht
, nullptr);
344 if (state
->channel_state_ht
) {
345 ret
= cds_lfht_destroy(state
->channel_state_ht
, nullptr);
348 if (state
->notification_trigger_clients_ht
) {
349 ret
= cds_lfht_destroy(state
->notification_trigger_clients_ht
, nullptr);
352 if (state
->channels_ht
) {
353 ret
= cds_lfht_destroy(state
->channels_ht
, nullptr);
356 if (state
->sessions_ht
) {
357 ret
= cds_lfht_destroy(state
->sessions_ht
, nullptr);
360 if (state
->triggers_by_name_uid_ht
) {
361 ret
= cds_lfht_destroy(state
->triggers_by_name_uid_ht
, nullptr);
364 if (state
->trigger_tokens_ht
) {
365 ret
= cds_lfht_destroy(state
->trigger_tokens_ht
, nullptr);
369 * Must be destroyed after all channels have been destroyed.
370 * See comment in struct lttng_session_trigger_list.
372 if (state
->session_triggers_ht
) {
373 ret
= cds_lfht_destroy(state
->session_triggers_ht
, nullptr);
376 if (state
->notification_channel_socket
>= 0) {
377 notification_channel_socket_destroy(state
->notification_channel_socket
);
380 LTTNG_ASSERT(cds_list_empty(&state
->tracer_event_sources_list
));
382 if (state
->executor
) {
383 action_executor_destroy(state
->executor
);
385 lttng_poll_clean(&state
->events
);
388 static void mark_thread_as_ready(struct notification_thread_handle
*handle
)
390 DBG("Marking notification thread as ready");
391 sem_post(&handle
->ready
);
394 static void wait_until_thread_is_ready(struct notification_thread_handle
*handle
)
396 DBG("Waiting for notification thread to be ready");
397 sem_wait(&handle
->ready
);
398 DBG("Notification thread is ready");
401 static int init_thread_state(struct notification_thread_handle
*handle
,
402 struct notification_thread_state
*state
)
406 memset(state
, 0, sizeof(*state
));
407 state
->notification_channel_socket
= -1;
408 state
->trigger_id
.next_tracer_token
= 1;
409 lttng_poll_init(&state
->events
);
411 ret
= notification_channel_socket_create();
415 state
->notification_channel_socket
= ret
;
417 ret
= init_poll_set(&state
->events
, handle
, state
->notification_channel_socket
);
422 DBG("Listening on notification channel socket");
423 ret
= lttcomm_listen_unix_sock(state
->notification_channel_socket
);
425 ERR("Listen failed on notification channel socket");
429 state
->client_socket_ht
= cds_lfht_new(
430 DEFAULT_HT_SIZE
, 1, 0, CDS_LFHT_AUTO_RESIZE
| CDS_LFHT_ACCOUNTING
, nullptr);
431 if (!state
->client_socket_ht
) {
435 state
->client_id_ht
= cds_lfht_new(
436 DEFAULT_HT_SIZE
, 1, 0, CDS_LFHT_AUTO_RESIZE
| CDS_LFHT_ACCOUNTING
, nullptr);
437 if (!state
->client_id_ht
) {
441 state
->channel_triggers_ht
= cds_lfht_new(
442 DEFAULT_HT_SIZE
, 1, 0, CDS_LFHT_AUTO_RESIZE
| CDS_LFHT_ACCOUNTING
, nullptr);
443 if (!state
->channel_triggers_ht
) {
447 state
->session_triggers_ht
= cds_lfht_new(
448 DEFAULT_HT_SIZE
, 1, 0, CDS_LFHT_AUTO_RESIZE
| CDS_LFHT_ACCOUNTING
, nullptr);
449 if (!state
->session_triggers_ht
) {
453 state
->channel_state_ht
= cds_lfht_new(
454 DEFAULT_HT_SIZE
, 1, 0, CDS_LFHT_AUTO_RESIZE
| CDS_LFHT_ACCOUNTING
, nullptr);
455 if (!state
->channel_state_ht
) {
459 state
->notification_trigger_clients_ht
= cds_lfht_new(
460 DEFAULT_HT_SIZE
, 1, 0, CDS_LFHT_AUTO_RESIZE
| CDS_LFHT_ACCOUNTING
, nullptr);
461 if (!state
->notification_trigger_clients_ht
) {
465 state
->channels_ht
= cds_lfht_new(
466 DEFAULT_HT_SIZE
, 1, 0, CDS_LFHT_AUTO_RESIZE
| CDS_LFHT_ACCOUNTING
, nullptr);
467 if (!state
->channels_ht
) {
470 state
->sessions_ht
= cds_lfht_new(
471 DEFAULT_HT_SIZE
, 1, 0, CDS_LFHT_AUTO_RESIZE
| CDS_LFHT_ACCOUNTING
, nullptr);
472 if (!state
->sessions_ht
) {
475 state
->triggers_ht
= cds_lfht_new(
476 DEFAULT_HT_SIZE
, 1, 0, CDS_LFHT_AUTO_RESIZE
| CDS_LFHT_ACCOUNTING
, nullptr);
477 if (!state
->triggers_ht
) {
480 state
->triggers_by_name_uid_ht
= cds_lfht_new(
481 DEFAULT_HT_SIZE
, 1, 0, CDS_LFHT_AUTO_RESIZE
| CDS_LFHT_ACCOUNTING
, nullptr);
482 if (!state
->triggers_by_name_uid_ht
) {
486 state
->trigger_tokens_ht
= cds_lfht_new(
487 DEFAULT_HT_SIZE
, 1, 0, CDS_LFHT_AUTO_RESIZE
| CDS_LFHT_ACCOUNTING
, nullptr);
488 if (!state
->trigger_tokens_ht
) {
492 CDS_INIT_LIST_HEAD(&state
->tracer_event_sources_list
);
494 state
->executor
= action_executor_create(handle
);
495 if (!state
->executor
) {
499 state
->restart_poll
= false;
501 mark_thread_as_ready(handle
);
505 fini_thread_state(state
);
509 static int handle_channel_monitoring_pipe(int fd
,
511 struct notification_thread_handle
*handle
,
512 struct notification_thread_state
*state
)
515 enum lttng_domain_type domain
;
517 if (fd
== handle
->channel_monitoring_pipes
.ust32_consumer
||
518 fd
== handle
->channel_monitoring_pipes
.ust64_consumer
) {
519 domain
= LTTNG_DOMAIN_UST
;
520 } else if (fd
== handle
->channel_monitoring_pipes
.kernel_consumer
) {
521 domain
= LTTNG_DOMAIN_KERNEL
;
526 if (revents
& (LPOLLERR
| LPOLLHUP
| LPOLLRDHUP
)) {
527 ret
= lttng_poll_del(&state
->events
, fd
);
529 ERR("Failed to remove consumer monitoring pipe from poll set");
534 ret
= handle_notification_thread_channel_sample(state
, fd
, domain
);
536 ERR("Consumer sample handling error occurred");
544 static int handle_event_notification_pipe(int event_source_fd
,
545 enum lttng_domain_type domain
,
547 struct notification_thread_state
*state
)
551 if (revents
& (LPOLLERR
| LPOLLHUP
| LPOLLRDHUP
)) {
552 ret
= handle_notification_thread_tracer_event_source_died(state
, event_source_fd
);
554 ERR("Failed to remove event notification pipe from poll set: fd = %d",
560 if (testpoint(sessiond_handle_notifier_event_pipe
)) {
565 if (caa_unlikely(notifier_consumption_paused
)) {
566 DBG("Event notifier notification consumption paused, sleeping...");
571 ret
= handle_notification_thread_event_notification(state
, event_source_fd
, domain
);
573 ERR("Event notification handling error occurred for fd: %d", event_source_fd
);
583 * Return the event source domain type via parameter.
585 static bool fd_is_event_notification_source(const struct notification_thread_state
*state
,
587 enum lttng_domain_type
*domain
)
589 struct notification_event_tracer_event_source_element
*source_element
;
591 LTTNG_ASSERT(domain
);
593 cds_list_for_each_entry (source_element
, &state
->tracer_event_sources_list
, node
) {
594 if (source_element
->fd
!= fd
) {
598 *domain
= source_element
->domain
;
606 * This thread services notification channel clients and commands received
607 * from various lttng-sessiond components over a command queue.
609 static void *thread_notification(void *data
)
612 struct notification_thread_handle
*handle
= (notification_thread_handle
*) data
;
613 struct notification_thread_state state
;
614 enum lttng_domain_type domain
;
616 DBG("Started notification thread");
618 health_register(the_health_sessiond
, HEALTH_SESSIOND_TYPE_NOTIFICATION
);
619 rcu_register_thread();
623 ERR("Invalid thread context provided");
627 health_code_update();
629 ret
= init_thread_state(handle
, &state
);
634 if (testpoint(sessiond_thread_notification
)) {
642 DBG("Entering poll wait");
643 ret
= lttng_poll_wait(&state
.events
, -1);
644 DBG("Poll wait returned (%i)", ret
);
648 * Restart interrupted system call.
650 if (errno
== EINTR
) {
653 ERR("Error encountered during lttng_poll_wait (%i)", ret
);
658 * Reset restart_poll flag so that calls below might turn it
661 state
.restart_poll
= false;
664 for (i
= 0; i
< fd_count
; i
++) {
665 int fd
= LTTNG_POLL_GETFD(&state
.events
, i
);
666 uint32_t revents
= LTTNG_POLL_GETEV(&state
.events
, i
);
668 DBG("Handling fd (%i) activity (%u)", fd
, revents
);
670 if (fd
== state
.notification_channel_socket
) {
671 if (revents
& LPOLLIN
) {
672 ret
= handle_notification_thread_client_connect(&state
);
676 } else if (revents
& (LPOLLERR
| LPOLLHUP
| LPOLLRDHUP
)) {
677 ERR("Notification socket poll error");
680 ERR("Unexpected poll events %u for notification socket %i",
685 } else if (fd
== handle
->cmd_queue
.event_fd
) {
686 ret
= handle_notification_thread_command(handle
, &state
);
688 DBG("Error encountered while servicing command queue");
690 } else if (ret
> 0) {
693 } else if (fd
== handle
->channel_monitoring_pipes
.ust32_consumer
||
694 fd
== handle
->channel_monitoring_pipes
.ust64_consumer
||
695 fd
== handle
->channel_monitoring_pipes
.kernel_consumer
) {
696 ret
= handle_channel_monitoring_pipe(fd
, revents
, handle
, &state
);
700 } else if (fd_is_event_notification_source(&state
, fd
, &domain
)) {
701 ret
= handle_event_notification_pipe(fd
, domain
, revents
, &state
);
706 /* Activity on a client's socket. */
707 if (revents
& (LPOLLERR
| LPOLLHUP
| LPOLLRDHUP
)) {
709 * It doesn't matter if a command was
710 * pending on the client socket at this
711 * point since it now has no way to
712 * receive the notifications to which
713 * it was subscribing or unsubscribing.
715 ret
= handle_notification_thread_client_disconnect(fd
,
721 if (revents
& LPOLLIN
) {
722 ret
= handle_notification_thread_client_in(&state
,
729 if (revents
& LPOLLOUT
) {
730 ret
= handle_notification_thread_client_out(&state
,
740 * Calls above might have changed the state of the
741 * FDs in `state.events`. Call _poll_wait() again to
742 * ensure we have a consistent state.
744 if (state
.restart_poll
) {
751 fini_thread_state(&state
);
753 rcu_thread_offline();
754 rcu_unregister_thread();
755 health_unregister(the_health_sessiond
);
759 static bool shutdown_notification_thread(void *thread_data
)
761 struct notification_thread_handle
*handle
= (notification_thread_handle
*) thread_data
;
763 notification_thread_command_quit(handle
);
767 struct lttng_thread
*launch_notification_thread(struct notification_thread_handle
*handle
)
769 struct lttng_thread
*thread
;
771 thread
= lttng_thread_create(
772 "Notification", thread_notification
, shutdown_notification_thread
, nullptr, handle
);
778 * Wait for the thread to be marked as "ready" before returning
779 * as other subsystems depend on the notification subsystem
780 * (e.g. rotation thread).
782 wait_until_thread_is_ready(handle
);