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.h>
11 #include <lttng/notification/notification-internal.h>
12 #include <lttng/condition/condition-internal.h>
13 #include <lttng/condition/buffer-usage-internal.h>
14 #include <common/error.h>
15 #include <common/config/session-config.h>
16 #include <common/defaults.h>
17 #include <common/utils.h>
18 #include <common/align.h>
19 #include <common/time.h>
24 #include "notification-thread.h"
25 #include "notification-thread-events.h"
26 #include "notification-thread-commands.h"
27 #include "lttng-sessiond.h"
28 #include "health-sessiond.h"
30 #include "testpoint.h"
33 #include <common/kernel-ctl/kernel-ctl.h>
36 #include <urcu/list.h>
37 #include <urcu/rculfhash.h>
40 * Flag used to temporarily pause data consumption from testpoints.
42 * This variable is dlsym-ed from a test, so needs to be exported.
44 LTTNG_EXPORT
int notifier_consumption_paused
;
47 * Destroy the thread data previously created by the init function.
49 void notification_thread_handle_destroy(
50 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_pipe
) {
63 lttng_pipe_destroy(handle
->cmd_queue
.event_pipe
);
65 if (handle
->channel_monitoring_pipes
.ust32_consumer
>= 0) {
66 ret
= close(handle
->channel_monitoring_pipes
.ust32_consumer
);
68 PERROR("close 32-bit consumer channel monitoring pipe");
71 if (handle
->channel_monitoring_pipes
.ust64_consumer
>= 0) {
72 ret
= close(handle
->channel_monitoring_pipes
.ust64_consumer
);
74 PERROR("close 64-bit consumer channel monitoring pipe");
77 if (handle
->channel_monitoring_pipes
.kernel_consumer
>= 0) {
78 ret
= close(handle
->channel_monitoring_pipes
.kernel_consumer
);
80 PERROR("close kernel consumer channel monitoring pipe");
88 struct notification_thread_handle
*notification_thread_handle_create(
89 struct lttng_pipe
*ust32_channel_monitor_pipe
,
90 struct lttng_pipe
*ust64_channel_monitor_pipe
,
91 struct lttng_pipe
*kernel_channel_monitor_pipe
)
94 struct notification_thread_handle
*handle
;
95 struct lttng_pipe
*event_pipe
= NULL
;
97 handle
= (notification_thread_handle
*) zmalloc(sizeof(*handle
));
102 sem_init(&handle
->ready
, 0, 0);
104 event_pipe
= lttng_pipe_open(FD_CLOEXEC
);
106 ERR("event_pipe creation");
110 handle
->cmd_queue
.event_pipe
= event_pipe
;
113 CDS_INIT_LIST_HEAD(&handle
->cmd_queue
.list
);
114 ret
= pthread_mutex_init(&handle
->cmd_queue
.lock
, NULL
);
119 if (ust32_channel_monitor_pipe
) {
120 handle
->channel_monitoring_pipes
.ust32_consumer
=
121 lttng_pipe_release_readfd(
122 ust32_channel_monitor_pipe
);
123 if (handle
->channel_monitoring_pipes
.ust32_consumer
< 0) {
127 handle
->channel_monitoring_pipes
.ust32_consumer
= -1;
129 if (ust64_channel_monitor_pipe
) {
130 handle
->channel_monitoring_pipes
.ust64_consumer
=
131 lttng_pipe_release_readfd(
132 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(
142 kernel_channel_monitor_pipe
);
143 if (handle
->channel_monitoring_pipes
.kernel_consumer
< 0) {
147 handle
->channel_monitoring_pipes
.kernel_consumer
= -1;
153 lttng_pipe_destroy(event_pipe
);
154 notification_thread_handle_destroy(handle
);
159 char *get_notification_channel_sock_path(void)
162 bool is_root
= !getuid();
165 sock_path
= (char *) zmalloc(LTTNG_PATH_MAX
);
171 ret
= snprintf(sock_path
, LTTNG_PATH_MAX
,
172 DEFAULT_GLOBAL_NOTIFICATION_CHANNEL_UNIX_SOCK
);
177 const char *home_path
= utils_get_home_dir();
180 ERR("Can't get HOME directory for socket creation");
184 ret
= snprintf(sock_path
, LTTNG_PATH_MAX
,
185 DEFAULT_HOME_NOTIFICATION_CHANNEL_UNIX_SOCK
,
199 void notification_channel_socket_destroy(int fd
)
202 char *sock_path
= get_notification_channel_sock_path();
204 DBG("Destroying notification channel socket");
207 ret
= unlink(sock_path
);
210 PERROR("unlink notification channel socket");
216 PERROR("close notification channel socket");
221 int notification_channel_socket_create(void)
224 char *sock_path
= get_notification_channel_sock_path();
226 DBG("Creating notification channel UNIX socket at %s",
229 ret
= lttcomm_create_unix_sock(sock_path
);
231 ERR("Failed to create notification socket");
236 ret
= chmod(sock_path
, S_IRUSR
| S_IWUSR
| S_IRGRP
| S_IWGRP
);
238 ERR("Set file permissions failed: %s", sock_path
);
239 PERROR("chmod notification channel socket");
246 ret
= utils_get_group_id(the_config
.tracing_group_name
.value
,
249 /* Default to root group. */
253 ret
= chown(sock_path
, 0, gid
);
255 ERR("Failed to set the notification channel socket's group");
261 DBG("Notification channel UNIX socket created (fd = %i)",
266 if (fd
>= 0 && close(fd
) < 0) {
267 PERROR("close notification channel socket");
274 int init_poll_set(struct lttng_poll_event
*poll_set
,
275 struct notification_thread_handle
*handle
,
276 int notification_channel_socket
)
281 * Create pollset with size 5:
282 * - notification channel socket (listen for new connections),
283 * - command queue event fd (internal sessiond commands),
284 * - consumerd (32-bit user space) channel monitor pipe,
285 * - consumerd (64-bit user space) channel monitor pipe,
286 * - consumerd (kernel) channel monitor pipe.
288 ret
= lttng_poll_create(poll_set
, 5, LTTNG_CLOEXEC
);
293 ret
= lttng_poll_add(poll_set
, notification_channel_socket
,
294 LPOLLIN
| LPOLLERR
| LPOLLHUP
| LPOLLRDHUP
);
296 ERR("Failed to add notification channel socket to pollset");
299 ret
= lttng_poll_add(poll_set
, lttng_pipe_get_readfd(handle
->cmd_queue
.event_pipe
),
302 ERR("Failed to add notification command queue event fd to pollset");
305 ret
= lttng_poll_add(poll_set
,
306 handle
->channel_monitoring_pipes
.ust32_consumer
,
309 ERR("Failed to add ust-32 channel monitoring pipe fd to pollset");
312 ret
= lttng_poll_add(poll_set
,
313 handle
->channel_monitoring_pipes
.ust64_consumer
,
316 ERR("Failed to add ust-64 channel monitoring pipe fd to pollset");
319 if (handle
->channel_monitoring_pipes
.kernel_consumer
< 0) {
322 ret
= lttng_poll_add(poll_set
,
323 handle
->channel_monitoring_pipes
.kernel_consumer
,
326 ERR("Failed to add kernel channel monitoring pipe fd to pollset");
332 lttng_poll_clean(poll_set
);
337 void fini_thread_state(struct notification_thread_state
*state
)
341 if (state
->client_socket_ht
) {
342 ret
= handle_notification_thread_client_disconnect_all(state
);
344 ret
= cds_lfht_destroy(state
->client_socket_ht
, NULL
);
347 if (state
->client_id_ht
) {
348 ret
= cds_lfht_destroy(state
->client_id_ht
, NULL
);
351 if (state
->triggers_ht
) {
352 ret
= handle_notification_thread_trigger_unregister_all(state
);
354 ret
= cds_lfht_destroy(state
->triggers_ht
, NULL
);
357 if (state
->channel_triggers_ht
) {
358 ret
= cds_lfht_destroy(state
->channel_triggers_ht
, NULL
);
361 if (state
->channel_state_ht
) {
362 ret
= cds_lfht_destroy(state
->channel_state_ht
, NULL
);
365 if (state
->notification_trigger_clients_ht
) {
366 ret
= cds_lfht_destroy(state
->notification_trigger_clients_ht
,
370 if (state
->channels_ht
) {
371 ret
= cds_lfht_destroy(state
->channels_ht
, NULL
);
374 if (state
->sessions_ht
) {
375 ret
= cds_lfht_destroy(state
->sessions_ht
, NULL
);
378 if (state
->triggers_by_name_uid_ht
) {
379 ret
= cds_lfht_destroy(state
->triggers_by_name_uid_ht
, NULL
);
382 if (state
->trigger_tokens_ht
) {
383 ret
= cds_lfht_destroy(state
->trigger_tokens_ht
, NULL
);
387 * Must be destroyed after all channels have been destroyed.
388 * See comment in struct lttng_session_trigger_list.
390 if (state
->session_triggers_ht
) {
391 ret
= cds_lfht_destroy(state
->session_triggers_ht
, NULL
);
394 if (state
->notification_channel_socket
>= 0) {
395 notification_channel_socket_destroy(
396 state
->notification_channel_socket
);
399 LTTNG_ASSERT(cds_list_empty(&state
->tracer_event_sources_list
));
401 if (state
->executor
) {
402 action_executor_destroy(state
->executor
);
404 lttng_poll_clean(&state
->events
);
408 void mark_thread_as_ready(struct notification_thread_handle
*handle
)
410 DBG("Marking notification thread as ready");
411 sem_post(&handle
->ready
);
415 void wait_until_thread_is_ready(struct notification_thread_handle
*handle
)
417 DBG("Waiting for notification thread to be ready");
418 sem_wait(&handle
->ready
);
419 DBG("Notification thread is ready");
423 int init_thread_state(struct notification_thread_handle
*handle
,
424 struct notification_thread_state
*state
)
428 memset(state
, 0, sizeof(*state
));
429 state
->notification_channel_socket
= -1;
430 state
->trigger_id
.next_tracer_token
= 1;
431 lttng_poll_init(&state
->events
);
433 ret
= notification_channel_socket_create();
437 state
->notification_channel_socket
= ret
;
439 ret
= init_poll_set(&state
->events
, handle
,
440 state
->notification_channel_socket
);
445 DBG("Listening on notification channel socket");
446 ret
= lttcomm_listen_unix_sock(state
->notification_channel_socket
);
448 ERR("Listen failed on notification channel socket");
452 state
->client_socket_ht
= cds_lfht_new(DEFAULT_HT_SIZE
, 1, 0,
453 CDS_LFHT_AUTO_RESIZE
| CDS_LFHT_ACCOUNTING
, NULL
);
454 if (!state
->client_socket_ht
) {
458 state
->client_id_ht
= cds_lfht_new(DEFAULT_HT_SIZE
, 1, 0,
459 CDS_LFHT_AUTO_RESIZE
| CDS_LFHT_ACCOUNTING
, NULL
);
460 if (!state
->client_id_ht
) {
464 state
->channel_triggers_ht
= cds_lfht_new(DEFAULT_HT_SIZE
, 1, 0,
465 CDS_LFHT_AUTO_RESIZE
| CDS_LFHT_ACCOUNTING
, NULL
);
466 if (!state
->channel_triggers_ht
) {
470 state
->session_triggers_ht
= cds_lfht_new(DEFAULT_HT_SIZE
, 1, 0,
471 CDS_LFHT_AUTO_RESIZE
| CDS_LFHT_ACCOUNTING
, NULL
);
472 if (!state
->session_triggers_ht
) {
476 state
->channel_state_ht
= cds_lfht_new(DEFAULT_HT_SIZE
, 1, 0,
477 CDS_LFHT_AUTO_RESIZE
| CDS_LFHT_ACCOUNTING
, NULL
);
478 if (!state
->channel_state_ht
) {
482 state
->notification_trigger_clients_ht
= cds_lfht_new(DEFAULT_HT_SIZE
,
483 1, 0, CDS_LFHT_AUTO_RESIZE
| CDS_LFHT_ACCOUNTING
, NULL
);
484 if (!state
->notification_trigger_clients_ht
) {
488 state
->channels_ht
= cds_lfht_new(DEFAULT_HT_SIZE
,
489 1, 0, CDS_LFHT_AUTO_RESIZE
| CDS_LFHT_ACCOUNTING
, NULL
);
490 if (!state
->channels_ht
) {
493 state
->sessions_ht
= cds_lfht_new(DEFAULT_HT_SIZE
,
494 1, 0, CDS_LFHT_AUTO_RESIZE
| CDS_LFHT_ACCOUNTING
, NULL
);
495 if (!state
->sessions_ht
) {
498 state
->triggers_ht
= cds_lfht_new(DEFAULT_HT_SIZE
,
499 1, 0, CDS_LFHT_AUTO_RESIZE
| CDS_LFHT_ACCOUNTING
, NULL
);
500 if (!state
->triggers_ht
) {
503 state
->triggers_by_name_uid_ht
= cds_lfht_new(DEFAULT_HT_SIZE
,
504 1, 0, CDS_LFHT_AUTO_RESIZE
| CDS_LFHT_ACCOUNTING
, NULL
);
505 if (!state
->triggers_by_name_uid_ht
) {
509 state
->trigger_tokens_ht
= cds_lfht_new(DEFAULT_HT_SIZE
,
510 1, 0, CDS_LFHT_AUTO_RESIZE
| CDS_LFHT_ACCOUNTING
, NULL
);
511 if (!state
->trigger_tokens_ht
) {
515 CDS_INIT_LIST_HEAD(&state
->tracer_event_sources_list
);
517 state
->executor
= action_executor_create(handle
);
518 if (!state
->executor
) {
522 state
->restart_poll
= false;
524 mark_thread_as_ready(handle
);
528 fini_thread_state(state
);
533 int handle_channel_monitoring_pipe(int fd
, uint32_t revents
,
534 struct notification_thread_handle
*handle
,
535 struct notification_thread_state
*state
)
538 enum lttng_domain_type domain
;
540 if (fd
== handle
->channel_monitoring_pipes
.ust32_consumer
||
541 fd
== handle
->channel_monitoring_pipes
.ust64_consumer
) {
542 domain
= LTTNG_DOMAIN_UST
;
543 } else if (fd
== handle
->channel_monitoring_pipes
.kernel_consumer
) {
544 domain
= LTTNG_DOMAIN_KERNEL
;
549 if (revents
& (LPOLLERR
| LPOLLHUP
| LPOLLRDHUP
)) {
550 ret
= lttng_poll_del(&state
->events
, fd
);
552 ERR("Failed to remove consumer monitoring pipe from poll set");
557 ret
= handle_notification_thread_channel_sample(
560 ERR("Consumer sample handling error occurred");
568 static int handle_event_notification_pipe(int event_source_fd
,
569 enum lttng_domain_type domain
,
571 struct notification_thread_state
*state
)
575 if (revents
& (LPOLLERR
| LPOLLHUP
| LPOLLRDHUP
)) {
576 ret
= handle_notification_thread_tracer_event_source_died(
577 state
, event_source_fd
);
579 ERR("Failed to remove event notification pipe from poll set: fd = %d",
585 if (testpoint(sessiond_handle_notifier_event_pipe
)) {
590 if (caa_unlikely(notifier_consumption_paused
)) {
591 DBG("Event notifier notification consumption paused, sleeping...");
596 ret
= handle_notification_thread_event_notification(
597 state
, event_source_fd
, domain
);
599 ERR("Event notification handling error occurred for fd: %d",
610 * Return the event source domain type via parameter.
612 static bool fd_is_event_notification_source(const struct notification_thread_state
*state
,
614 enum lttng_domain_type
*domain
)
616 struct notification_event_tracer_event_source_element
*source_element
;
618 LTTNG_ASSERT(domain
);
620 cds_list_for_each_entry(source_element
,
621 &state
->tracer_event_sources_list
, node
) {
622 if (source_element
->fd
!= fd
) {
626 *domain
= source_element
->domain
;
634 * This thread services notification channel clients and commands received
635 * from various lttng-sessiond components over a command queue.
638 void *thread_notification(void *data
)
641 struct notification_thread_handle
*handle
= (notification_thread_handle
*) data
;
642 struct notification_thread_state state
;
643 enum lttng_domain_type domain
;
645 DBG("Started notification thread");
647 health_register(the_health_sessiond
, HEALTH_SESSIOND_TYPE_NOTIFICATION
);
648 rcu_register_thread();
652 ERR("Invalid thread context provided");
656 health_code_update();
658 ret
= init_thread_state(handle
, &state
);
663 if (testpoint(sessiond_thread_notification
)) {
671 DBG("Entering poll wait");
672 ret
= lttng_poll_wait(&state
.events
, -1);
673 DBG("Poll wait returned (%i)", ret
);
677 * Restart interrupted system call.
679 if (errno
== EINTR
) {
682 ERR("Error encountered during lttng_poll_wait (%i)", ret
);
687 * Reset restart_poll flag so that calls below might turn it
690 state
.restart_poll
= false;
693 for (i
= 0; i
< fd_count
; i
++) {
694 int fd
= LTTNG_POLL_GETFD(&state
.events
, i
);
695 uint32_t revents
= LTTNG_POLL_GETEV(&state
.events
, i
);
697 DBG("Handling fd (%i) activity (%u)", fd
, revents
);
699 if (fd
== state
.notification_channel_socket
) {
700 if (revents
& LPOLLIN
) {
701 ret
= handle_notification_thread_client_connect(
707 (LPOLLERR
| LPOLLHUP
| LPOLLRDHUP
)) {
708 ERR("Notification socket poll error");
711 ERR("Unexpected poll events %u for notification socket %i", revents
, fd
);
714 } else if (fd
== lttng_pipe_get_readfd(handle
->cmd_queue
.event_pipe
)) {
715 ret
= handle_notification_thread_command(handle
,
718 DBG("Error encountered while servicing command queue");
720 } else if (ret
> 0) {
723 } else if (fd
== handle
->channel_monitoring_pipes
.ust32_consumer
||
724 fd
== handle
->channel_monitoring_pipes
.ust64_consumer
||
725 fd
== handle
->channel_monitoring_pipes
.kernel_consumer
) {
726 ret
= handle_channel_monitoring_pipe(fd
,
727 revents
, handle
, &state
);
731 } else if (fd_is_event_notification_source(&state
, fd
, &domain
)) {
732 ret
= handle_event_notification_pipe(fd
, domain
, revents
, &state
);
737 /* Activity on a client's socket. */
738 if (revents
& (LPOLLERR
| LPOLLHUP
| LPOLLRDHUP
)) {
740 * It doesn't matter if a command was
741 * pending on the client socket at this
742 * point since it now has no way to
743 * receive the notifications to which
744 * it was subscribing or unsubscribing.
746 ret
= handle_notification_thread_client_disconnect(
752 if (revents
& LPOLLIN
) {
753 ret
= handle_notification_thread_client_in(
760 if (revents
& LPOLLOUT
) {
761 ret
= handle_notification_thread_client_out(
771 * Calls above might have changed the state of the
772 * FDs in `state.events`. Call _poll_wait() again to
773 * ensure we have a consistent state.
775 if (state
.restart_poll
) {
782 fini_thread_state(&state
);
784 rcu_thread_offline();
785 rcu_unregister_thread();
786 health_unregister(the_health_sessiond
);
791 bool shutdown_notification_thread(void *thread_data
)
793 struct notification_thread_handle
*handle
= (notification_thread_handle
*) thread_data
;
795 notification_thread_command_quit(handle
);
799 struct lttng_thread
*launch_notification_thread(
800 struct notification_thread_handle
*handle
)
802 struct lttng_thread
*thread
;
804 thread
= lttng_thread_create("Notification",
806 shutdown_notification_thread
,
814 * Wait for the thread to be marked as "ready" before returning
815 * as other subsystems depend on the notification subsystem
816 * (e.g. rotation thread).
818 wait_until_thread_is_ready(handle
);