2 * Copyright (C) 2017 Jérémie Galarneau <jeremie.galarneau@efficios.com>
4 * SPDX-License-Identifier: GPL-2.0-only
9 #include <lttng/trigger/trigger.h>
10 #include <lttng/notification/channel-internal.hpp>
11 #include <lttng/notification/notification-internal.hpp>
12 #include <lttng/condition/condition-internal.hpp>
13 #include <lttng/condition/buffer-usage-internal.hpp>
14 #include <common/error.hpp>
15 #include <common/config/session-config.hpp>
16 #include <common/defaults.hpp>
17 #include <common/utils.hpp>
18 #include <common/align.hpp>
19 #include <common/time.hpp>
21 #include <sys/eventfd.h>
25 #include "notification-thread.hpp"
26 #include "notification-thread-events.hpp"
27 #include "notification-thread-commands.hpp"
28 #include "lttng-sessiond.hpp"
29 #include "health-sessiond.hpp"
31 #include "testpoint.hpp"
34 #include <common/kernel-ctl/kernel-ctl.hpp>
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(
51 struct notification_thread_handle
*handle
)
59 LTTNG_ASSERT(cds_list_empty(&handle
->cmd_queue
.list
));
60 pthread_mutex_destroy(&handle
->cmd_queue
.lock
);
61 sem_destroy(&handle
->ready
);
63 if (handle
->cmd_queue
.event_fd
>= 0) {
64 ret
= close(handle
->cmd_queue
.event_fd
);
66 PERROR("Failed to close notification command queue event fd");
69 if (handle
->channel_monitoring_pipes
.ust32_consumer
>= 0) {
70 ret
= close(handle
->channel_monitoring_pipes
.ust32_consumer
);
72 PERROR("close 32-bit consumer channel monitoring pipe");
75 if (handle
->channel_monitoring_pipes
.ust64_consumer
>= 0) {
76 ret
= close(handle
->channel_monitoring_pipes
.ust64_consumer
);
78 PERROR("close 64-bit consumer channel monitoring pipe");
81 if (handle
->channel_monitoring_pipes
.kernel_consumer
>= 0) {
82 ret
= close(handle
->channel_monitoring_pipes
.kernel_consumer
);
84 PERROR("close kernel consumer channel monitoring pipe");
92 struct notification_thread_handle
*notification_thread_handle_create(
93 struct lttng_pipe
*ust32_channel_monitor_pipe
,
94 struct lttng_pipe
*ust64_channel_monitor_pipe
,
95 struct lttng_pipe
*kernel_channel_monitor_pipe
)
98 struct notification_thread_handle
*handle
;
101 handle
= zmalloc
<notification_thread_handle
>();
106 sem_init(&handle
->ready
, 0, 0);
108 event_fd
= eventfd(0, EFD_CLOEXEC
| EFD_SEMAPHORE
);
110 PERROR("event_fd creation");
114 handle
->cmd_queue
.event_fd
= event_fd
;
116 CDS_INIT_LIST_HEAD(&handle
->cmd_queue
.list
);
117 ret
= pthread_mutex_init(&handle
->cmd_queue
.lock
, NULL
);
122 if (ust32_channel_monitor_pipe
) {
123 handle
->channel_monitoring_pipes
.ust32_consumer
=
124 lttng_pipe_release_readfd(
125 ust32_channel_monitor_pipe
);
126 if (handle
->channel_monitoring_pipes
.ust32_consumer
< 0) {
130 handle
->channel_monitoring_pipes
.ust32_consumer
= -1;
132 if (ust64_channel_monitor_pipe
) {
133 handle
->channel_monitoring_pipes
.ust64_consumer
=
134 lttng_pipe_release_readfd(
135 ust64_channel_monitor_pipe
);
136 if (handle
->channel_monitoring_pipes
.ust64_consumer
< 0) {
140 handle
->channel_monitoring_pipes
.ust64_consumer
= -1;
142 if (kernel_channel_monitor_pipe
) {
143 handle
->channel_monitoring_pipes
.kernel_consumer
=
144 lttng_pipe_release_readfd(
145 kernel_channel_monitor_pipe
);
146 if (handle
->channel_monitoring_pipes
.kernel_consumer
< 0) {
150 handle
->channel_monitoring_pipes
.kernel_consumer
= -1;
156 notification_thread_handle_destroy(handle
);
161 char *get_notification_channel_sock_path(void)
164 bool is_root
= !getuid();
167 sock_path
= calloc
<char>(LTTNG_PATH_MAX
);
173 ret
= snprintf(sock_path
, LTTNG_PATH_MAX
,
174 DEFAULT_GLOBAL_NOTIFICATION_CHANNEL_UNIX_SOCK
);
179 const char *home_path
= utils_get_home_dir();
182 ERR("Can't get HOME directory for socket creation");
186 ret
= snprintf(sock_path
, LTTNG_PATH_MAX
,
187 DEFAULT_HOME_NOTIFICATION_CHANNEL_UNIX_SOCK
,
201 void notification_channel_socket_destroy(int fd
)
204 char *sock_path
= get_notification_channel_sock_path();
206 DBG("Destroying notification channel socket");
209 ret
= unlink(sock_path
);
212 PERROR("unlink notification channel socket");
218 PERROR("close notification channel socket");
223 int notification_channel_socket_create(void)
226 char *sock_path
= get_notification_channel_sock_path();
228 DBG("Creating notification channel UNIX socket at %s",
231 ret
= lttcomm_create_unix_sock(sock_path
);
233 ERR("Failed to create notification socket");
238 ret
= chmod(sock_path
, S_IRUSR
| S_IWUSR
| S_IRGRP
| S_IWGRP
);
240 ERR("Set file permissions failed: %s", sock_path
);
241 PERROR("chmod notification channel socket");
248 ret
= utils_get_group_id(the_config
.tracing_group_name
.value
,
251 /* Default to root group. */
255 ret
= chown(sock_path
, 0, gid
);
257 ERR("Failed to set the notification channel socket's group");
263 DBG("Notification channel UNIX socket created (fd = %i)",
268 if (fd
>= 0 && close(fd
) < 0) {
269 PERROR("close notification channel socket");
276 int init_poll_set(struct lttng_poll_event
*poll_set
,
277 struct notification_thread_handle
*handle
,
278 int notification_channel_socket
)
283 * Create pollset with size 5:
284 * - notification channel socket (listen for new connections),
285 * - command queue event fd (internal sessiond commands),
286 * - consumerd (32-bit user space) channel monitor pipe,
287 * - consumerd (64-bit user space) channel monitor pipe,
288 * - consumerd (kernel) channel monitor pipe.
290 ret
= lttng_poll_create(poll_set
, 5, LTTNG_CLOEXEC
);
295 ret
= lttng_poll_add(poll_set
, notification_channel_socket
,
296 LPOLLIN
| LPOLLERR
| LPOLLHUP
| LPOLLRDHUP
);
298 ERR("Failed to add notification channel socket to pollset");
301 ret
= lttng_poll_add(poll_set
, handle
->cmd_queue
.event_fd
,
304 ERR("Failed to add notification command queue event fd to pollset");
307 ret
= lttng_poll_add(poll_set
,
308 handle
->channel_monitoring_pipes
.ust32_consumer
,
311 ERR("Failed to add ust-32 channel monitoring pipe fd to pollset");
314 ret
= lttng_poll_add(poll_set
,
315 handle
->channel_monitoring_pipes
.ust64_consumer
,
318 ERR("Failed to add ust-64 channel monitoring pipe fd to pollset");
321 if (handle
->channel_monitoring_pipes
.kernel_consumer
< 0) {
324 ret
= lttng_poll_add(poll_set
,
325 handle
->channel_monitoring_pipes
.kernel_consumer
,
328 ERR("Failed to add kernel channel monitoring pipe fd to pollset");
334 lttng_poll_clean(poll_set
);
339 void fini_thread_state(struct notification_thread_state
*state
)
343 if (state
->client_socket_ht
) {
344 ret
= handle_notification_thread_client_disconnect_all(state
);
346 ret
= cds_lfht_destroy(state
->client_socket_ht
, NULL
);
349 if (state
->client_id_ht
) {
350 ret
= cds_lfht_destroy(state
->client_id_ht
, NULL
);
353 if (state
->triggers_ht
) {
354 ret
= handle_notification_thread_trigger_unregister_all(state
);
356 ret
= cds_lfht_destroy(state
->triggers_ht
, NULL
);
359 if (state
->channel_triggers_ht
) {
360 ret
= cds_lfht_destroy(state
->channel_triggers_ht
, NULL
);
363 if (state
->channel_state_ht
) {
364 ret
= cds_lfht_destroy(state
->channel_state_ht
, NULL
);
367 if (state
->notification_trigger_clients_ht
) {
368 ret
= cds_lfht_destroy(state
->notification_trigger_clients_ht
,
372 if (state
->channels_ht
) {
373 ret
= cds_lfht_destroy(state
->channels_ht
, NULL
);
376 if (state
->sessions_ht
) {
377 ret
= cds_lfht_destroy(state
->sessions_ht
, NULL
);
380 if (state
->triggers_by_name_uid_ht
) {
381 ret
= cds_lfht_destroy(state
->triggers_by_name_uid_ht
, NULL
);
384 if (state
->trigger_tokens_ht
) {
385 ret
= cds_lfht_destroy(state
->trigger_tokens_ht
, NULL
);
389 * Must be destroyed after all channels have been destroyed.
390 * See comment in struct lttng_session_trigger_list.
392 if (state
->session_triggers_ht
) {
393 ret
= cds_lfht_destroy(state
->session_triggers_ht
, NULL
);
396 if (state
->notification_channel_socket
>= 0) {
397 notification_channel_socket_destroy(
398 state
->notification_channel_socket
);
401 LTTNG_ASSERT(cds_list_empty(&state
->tracer_event_sources_list
));
403 if (state
->executor
) {
404 action_executor_destroy(state
->executor
);
406 lttng_poll_clean(&state
->events
);
410 void mark_thread_as_ready(struct notification_thread_handle
*handle
)
412 DBG("Marking notification thread as ready");
413 sem_post(&handle
->ready
);
417 void wait_until_thread_is_ready(struct notification_thread_handle
*handle
)
419 DBG("Waiting for notification thread to be ready");
420 sem_wait(&handle
->ready
);
421 DBG("Notification thread is ready");
425 int init_thread_state(struct notification_thread_handle
*handle
,
426 struct notification_thread_state
*state
)
430 memset(state
, 0, sizeof(*state
));
431 state
->notification_channel_socket
= -1;
432 state
->trigger_id
.next_tracer_token
= 1;
433 lttng_poll_init(&state
->events
);
435 ret
= notification_channel_socket_create();
439 state
->notification_channel_socket
= ret
;
441 ret
= init_poll_set(&state
->events
, handle
,
442 state
->notification_channel_socket
);
447 DBG("Listening on notification channel socket");
448 ret
= lttcomm_listen_unix_sock(state
->notification_channel_socket
);
450 ERR("Listen failed on notification channel socket");
454 state
->client_socket_ht
= cds_lfht_new(DEFAULT_HT_SIZE
, 1, 0,
455 CDS_LFHT_AUTO_RESIZE
| CDS_LFHT_ACCOUNTING
, NULL
);
456 if (!state
->client_socket_ht
) {
460 state
->client_id_ht
= cds_lfht_new(DEFAULT_HT_SIZE
, 1, 0,
461 CDS_LFHT_AUTO_RESIZE
| CDS_LFHT_ACCOUNTING
, NULL
);
462 if (!state
->client_id_ht
) {
466 state
->channel_triggers_ht
= cds_lfht_new(DEFAULT_HT_SIZE
, 1, 0,
467 CDS_LFHT_AUTO_RESIZE
| CDS_LFHT_ACCOUNTING
, NULL
);
468 if (!state
->channel_triggers_ht
) {
472 state
->session_triggers_ht
= cds_lfht_new(DEFAULT_HT_SIZE
, 1, 0,
473 CDS_LFHT_AUTO_RESIZE
| CDS_LFHT_ACCOUNTING
, NULL
);
474 if (!state
->session_triggers_ht
) {
478 state
->channel_state_ht
= cds_lfht_new(DEFAULT_HT_SIZE
, 1, 0,
479 CDS_LFHT_AUTO_RESIZE
| CDS_LFHT_ACCOUNTING
, NULL
);
480 if (!state
->channel_state_ht
) {
484 state
->notification_trigger_clients_ht
= cds_lfht_new(DEFAULT_HT_SIZE
,
485 1, 0, CDS_LFHT_AUTO_RESIZE
| CDS_LFHT_ACCOUNTING
, NULL
);
486 if (!state
->notification_trigger_clients_ht
) {
490 state
->channels_ht
= cds_lfht_new(DEFAULT_HT_SIZE
,
491 1, 0, CDS_LFHT_AUTO_RESIZE
| CDS_LFHT_ACCOUNTING
, NULL
);
492 if (!state
->channels_ht
) {
495 state
->sessions_ht
= cds_lfht_new(DEFAULT_HT_SIZE
,
496 1, 0, CDS_LFHT_AUTO_RESIZE
| CDS_LFHT_ACCOUNTING
, NULL
);
497 if (!state
->sessions_ht
) {
500 state
->triggers_ht
= cds_lfht_new(DEFAULT_HT_SIZE
,
501 1, 0, CDS_LFHT_AUTO_RESIZE
| CDS_LFHT_ACCOUNTING
, NULL
);
502 if (!state
->triggers_ht
) {
505 state
->triggers_by_name_uid_ht
= cds_lfht_new(DEFAULT_HT_SIZE
,
506 1, 0, CDS_LFHT_AUTO_RESIZE
| CDS_LFHT_ACCOUNTING
, NULL
);
507 if (!state
->triggers_by_name_uid_ht
) {
511 state
->trigger_tokens_ht
= cds_lfht_new(DEFAULT_HT_SIZE
,
512 1, 0, CDS_LFHT_AUTO_RESIZE
| CDS_LFHT_ACCOUNTING
, NULL
);
513 if (!state
->trigger_tokens_ht
) {
517 CDS_INIT_LIST_HEAD(&state
->tracer_event_sources_list
);
519 state
->executor
= action_executor_create(handle
);
520 if (!state
->executor
) {
524 state
->restart_poll
= false;
526 mark_thread_as_ready(handle
);
530 fini_thread_state(state
);
535 int handle_channel_monitoring_pipe(int fd
, uint32_t revents
,
536 struct notification_thread_handle
*handle
,
537 struct notification_thread_state
*state
)
540 enum lttng_domain_type domain
;
542 if (fd
== handle
->channel_monitoring_pipes
.ust32_consumer
||
543 fd
== handle
->channel_monitoring_pipes
.ust64_consumer
) {
544 domain
= LTTNG_DOMAIN_UST
;
545 } else if (fd
== handle
->channel_monitoring_pipes
.kernel_consumer
) {
546 domain
= LTTNG_DOMAIN_KERNEL
;
551 if (revents
& (LPOLLERR
| LPOLLHUP
| LPOLLRDHUP
)) {
552 ret
= lttng_poll_del(&state
->events
, fd
);
554 ERR("Failed to remove consumer monitoring pipe from poll set");
559 ret
= handle_notification_thread_channel_sample(
562 ERR("Consumer sample handling error occurred");
570 static int handle_event_notification_pipe(int event_source_fd
,
571 enum lttng_domain_type domain
,
573 struct notification_thread_state
*state
)
577 if (revents
& (LPOLLERR
| LPOLLHUP
| LPOLLRDHUP
)) {
578 ret
= handle_notification_thread_tracer_event_source_died(
579 state
, event_source_fd
);
581 ERR("Failed to remove event notification pipe from poll set: fd = %d",
587 if (testpoint(sessiond_handle_notifier_event_pipe
)) {
592 if (caa_unlikely(notifier_consumption_paused
)) {
593 DBG("Event notifier notification consumption paused, sleeping...");
598 ret
= handle_notification_thread_event_notification(
599 state
, event_source_fd
, domain
);
601 ERR("Event notification handling error occurred for fd: %d",
612 * Return the event source domain type via parameter.
614 static bool fd_is_event_notification_source(const struct notification_thread_state
*state
,
616 enum lttng_domain_type
*domain
)
618 struct notification_event_tracer_event_source_element
*source_element
;
620 LTTNG_ASSERT(domain
);
622 cds_list_for_each_entry(source_element
,
623 &state
->tracer_event_sources_list
, node
) {
624 if (source_element
->fd
!= fd
) {
628 *domain
= source_element
->domain
;
636 * This thread services notification channel clients and commands received
637 * from various lttng-sessiond components over a command queue.
640 void *thread_notification(void *data
)
643 struct notification_thread_handle
*handle
= (notification_thread_handle
*) data
;
644 struct notification_thread_state state
;
645 enum lttng_domain_type domain
;
647 DBG("Started notification thread");
649 health_register(the_health_sessiond
, HEALTH_SESSIOND_TYPE_NOTIFICATION
);
650 rcu_register_thread();
654 ERR("Invalid thread context provided");
658 health_code_update();
660 ret
= init_thread_state(handle
, &state
);
665 if (testpoint(sessiond_thread_notification
)) {
673 DBG("Entering poll wait");
674 ret
= lttng_poll_wait(&state
.events
, -1);
675 DBG("Poll wait returned (%i)", ret
);
679 * Restart interrupted system call.
681 if (errno
== EINTR
) {
684 ERR("Error encountered during lttng_poll_wait (%i)", ret
);
689 * Reset restart_poll flag so that calls below might turn it
692 state
.restart_poll
= false;
695 for (i
= 0; i
< fd_count
; i
++) {
696 int fd
= LTTNG_POLL_GETFD(&state
.events
, i
);
697 uint32_t revents
= LTTNG_POLL_GETEV(&state
.events
, i
);
699 DBG("Handling fd (%i) activity (%u)", fd
, revents
);
701 if (fd
== state
.notification_channel_socket
) {
702 if (revents
& LPOLLIN
) {
703 ret
= handle_notification_thread_client_connect(
709 (LPOLLERR
| LPOLLHUP
| LPOLLRDHUP
)) {
710 ERR("Notification socket poll error");
713 ERR("Unexpected poll events %u for notification socket %i", revents
, fd
);
716 } else if (fd
== handle
->cmd_queue
.event_fd
) {
717 ret
= handle_notification_thread_command(handle
,
720 DBG("Error encountered while servicing command queue");
722 } else if (ret
> 0) {
725 } else if (fd
== handle
->channel_monitoring_pipes
.ust32_consumer
||
726 fd
== handle
->channel_monitoring_pipes
.ust64_consumer
||
727 fd
== handle
->channel_monitoring_pipes
.kernel_consumer
) {
728 ret
= handle_channel_monitoring_pipe(fd
,
729 revents
, handle
, &state
);
733 } else if (fd_is_event_notification_source(&state
, fd
, &domain
)) {
734 ret
= handle_event_notification_pipe(fd
, domain
, revents
, &state
);
739 /* Activity on a client's socket. */
740 if (revents
& (LPOLLERR
| LPOLLHUP
| LPOLLRDHUP
)) {
742 * It doesn't matter if a command was
743 * pending on the client socket at this
744 * point since it now has no way to
745 * receive the notifications to which
746 * it was subscribing or unsubscribing.
748 ret
= handle_notification_thread_client_disconnect(
754 if (revents
& LPOLLIN
) {
755 ret
= handle_notification_thread_client_in(
762 if (revents
& LPOLLOUT
) {
763 ret
= handle_notification_thread_client_out(
773 * Calls above might have changed the state of the
774 * FDs in `state.events`. Call _poll_wait() again to
775 * ensure we have a consistent state.
777 if (state
.restart_poll
) {
784 fini_thread_state(&state
);
786 rcu_thread_offline();
787 rcu_unregister_thread();
788 health_unregister(the_health_sessiond
);
793 bool shutdown_notification_thread(void *thread_data
)
795 struct notification_thread_handle
*handle
= (notification_thread_handle
*) thread_data
;
797 notification_thread_command_quit(handle
);
801 struct lttng_thread
*launch_notification_thread(
802 struct notification_thread_handle
*handle
)
804 struct lttng_thread
*thread
;
806 thread
= lttng_thread_create("Notification",
808 shutdown_notification_thread
,
816 * Wait for the thread to be marked as "ready" before returning
817 * as other subsystems depend on the notification subsystem
818 * (e.g. rotation thread).
820 wait_until_thread_is_ready(handle
);