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 int notifier_consumption_paused
;
42 * Destroy the thread data previously created by the init function.
44 void notification_thread_handle_destroy(
45 struct notification_thread_handle
*handle
)
53 assert(cds_list_empty(&handle
->cmd_queue
.list
));
54 pthread_mutex_destroy(&handle
->cmd_queue
.lock
);
55 sem_destroy(&handle
->ready
);
57 if (handle
->cmd_queue
.event_pipe
) {
58 lttng_pipe_destroy(handle
->cmd_queue
.event_pipe
);
60 if (handle
->channel_monitoring_pipes
.ust32_consumer
>= 0) {
61 ret
= close(handle
->channel_monitoring_pipes
.ust32_consumer
);
63 PERROR("close 32-bit consumer channel monitoring pipe");
66 if (handle
->channel_monitoring_pipes
.ust64_consumer
>= 0) {
67 ret
= close(handle
->channel_monitoring_pipes
.ust64_consumer
);
69 PERROR("close 64-bit consumer channel monitoring pipe");
72 if (handle
->channel_monitoring_pipes
.kernel_consumer
>= 0) {
73 ret
= close(handle
->channel_monitoring_pipes
.kernel_consumer
);
75 PERROR("close kernel consumer channel monitoring pipe");
83 struct notification_thread_handle
*notification_thread_handle_create(
84 struct lttng_pipe
*ust32_channel_monitor_pipe
,
85 struct lttng_pipe
*ust64_channel_monitor_pipe
,
86 struct lttng_pipe
*kernel_channel_monitor_pipe
)
89 struct notification_thread_handle
*handle
;
90 struct lttng_pipe
*event_pipe
= NULL
;
92 handle
= zmalloc(sizeof(*handle
));
97 sem_init(&handle
->ready
, 0, 0);
99 event_pipe
= lttng_pipe_open(FD_CLOEXEC
);
101 ERR("event_pipe creation");
105 handle
->cmd_queue
.event_pipe
= event_pipe
;
108 CDS_INIT_LIST_HEAD(&handle
->cmd_queue
.list
);
109 ret
= pthread_mutex_init(&handle
->cmd_queue
.lock
, NULL
);
114 if (ust32_channel_monitor_pipe
) {
115 handle
->channel_monitoring_pipes
.ust32_consumer
=
116 lttng_pipe_release_readfd(
117 ust32_channel_monitor_pipe
);
118 if (handle
->channel_monitoring_pipes
.ust32_consumer
< 0) {
122 handle
->channel_monitoring_pipes
.ust32_consumer
= -1;
124 if (ust64_channel_monitor_pipe
) {
125 handle
->channel_monitoring_pipes
.ust64_consumer
=
126 lttng_pipe_release_readfd(
127 ust64_channel_monitor_pipe
);
128 if (handle
->channel_monitoring_pipes
.ust64_consumer
< 0) {
132 handle
->channel_monitoring_pipes
.ust64_consumer
= -1;
134 if (kernel_channel_monitor_pipe
) {
135 handle
->channel_monitoring_pipes
.kernel_consumer
=
136 lttng_pipe_release_readfd(
137 kernel_channel_monitor_pipe
);
138 if (handle
->channel_monitoring_pipes
.kernel_consumer
< 0) {
142 handle
->channel_monitoring_pipes
.kernel_consumer
= -1;
148 lttng_pipe_destroy(event_pipe
);
149 notification_thread_handle_destroy(handle
);
154 char *get_notification_channel_sock_path(void)
157 bool is_root
= !getuid();
160 sock_path
= zmalloc(LTTNG_PATH_MAX
);
166 ret
= snprintf(sock_path
, LTTNG_PATH_MAX
,
167 DEFAULT_GLOBAL_NOTIFICATION_CHANNEL_UNIX_SOCK
);
172 const char *home_path
= utils_get_home_dir();
175 ERR("Can't get HOME directory for socket creation");
179 ret
= snprintf(sock_path
, LTTNG_PATH_MAX
,
180 DEFAULT_HOME_NOTIFICATION_CHANNEL_UNIX_SOCK
,
194 void notification_channel_socket_destroy(int fd
)
197 char *sock_path
= get_notification_channel_sock_path();
199 DBG("[notification-thread] Destroying notification channel socket");
202 ret
= unlink(sock_path
);
205 PERROR("unlink notification channel socket");
211 PERROR("close notification channel socket");
216 int notification_channel_socket_create(void)
219 char *sock_path
= get_notification_channel_sock_path();
221 DBG("[notification-thread] Creating notification channel UNIX socket at %s",
224 ret
= lttcomm_create_unix_sock(sock_path
);
226 ERR("[notification-thread] 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(config
.tracing_group_name
.value
, true,
244 /* Default to root group. */
248 ret
= chown(sock_path
, 0, gid
);
250 ERR("Failed to set the notification channel socket's group");
256 DBG("[notification-thread] Notification channel UNIX socket created (fd = %i)",
261 if (fd
>= 0 && close(fd
) < 0) {
262 PERROR("close notification channel socket");
269 int init_poll_set(struct lttng_poll_event
*poll_set
,
270 struct notification_thread_handle
*handle
,
271 int notification_channel_socket
)
276 * Create pollset with size 5:
277 * - notification channel socket (listen for new connections),
278 * - command queue event fd (internal sessiond commands),
279 * - consumerd (32-bit user space) channel monitor pipe,
280 * - consumerd (64-bit user space) channel monitor pipe,
281 * - consumerd (kernel) channel monitor pipe.
283 ret
= lttng_poll_create(poll_set
, 5, LTTNG_CLOEXEC
);
288 ret
= lttng_poll_add(poll_set
, notification_channel_socket
,
289 LPOLLIN
| LPOLLERR
| LPOLLHUP
| LPOLLRDHUP
);
291 ERR("[notification-thread] Failed to add notification channel socket to pollset");
294 ret
= lttng_poll_add(poll_set
, lttng_pipe_get_readfd(handle
->cmd_queue
.event_pipe
),
297 ERR("[notification-thread] Failed to add notification command queue event fd to pollset");
300 ret
= lttng_poll_add(poll_set
,
301 handle
->channel_monitoring_pipes
.ust32_consumer
,
304 ERR("[notification-thread] Failed to add ust-32 channel monitoring pipe fd to pollset");
307 ret
= lttng_poll_add(poll_set
,
308 handle
->channel_monitoring_pipes
.ust64_consumer
,
311 ERR("[notification-thread] Failed to add ust-64 channel monitoring pipe fd to pollset");
314 if (handle
->channel_monitoring_pipes
.kernel_consumer
< 0) {
317 ret
= lttng_poll_add(poll_set
,
318 handle
->channel_monitoring_pipes
.kernel_consumer
,
321 ERR("[notification-thread] Failed to add kernel channel monitoring pipe fd to pollset");
327 lttng_poll_clean(poll_set
);
332 void fini_thread_state(struct notification_thread_state
*state
)
336 if (state
->client_socket_ht
) {
337 ret
= handle_notification_thread_client_disconnect_all(state
);
339 ret
= cds_lfht_destroy(state
->client_socket_ht
, NULL
);
342 if (state
->client_id_ht
) {
343 ret
= cds_lfht_destroy(state
->client_id_ht
, NULL
);
346 if (state
->triggers_ht
) {
347 ret
= handle_notification_thread_trigger_unregister_all(state
);
349 ret
= cds_lfht_destroy(state
->triggers_ht
, NULL
);
352 if (state
->channel_triggers_ht
) {
353 ret
= cds_lfht_destroy(state
->channel_triggers_ht
, NULL
);
356 if (state
->channel_state_ht
) {
357 ret
= cds_lfht_destroy(state
->channel_state_ht
, NULL
);
360 if (state
->notification_trigger_clients_ht
) {
361 ret
= cds_lfht_destroy(state
->notification_trigger_clients_ht
,
365 if (state
->channels_ht
) {
366 ret
= cds_lfht_destroy(state
->channels_ht
, NULL
);
369 if (state
->sessions_ht
) {
370 ret
= cds_lfht_destroy(state
->sessions_ht
, NULL
);
373 if (state
->triggers_by_name_uid_ht
) {
374 ret
= cds_lfht_destroy(state
->triggers_by_name_uid_ht
, NULL
);
377 if (state
->trigger_tokens_ht
) {
378 ret
= cds_lfht_destroy(state
->trigger_tokens_ht
, NULL
);
382 * Must be destroyed after all channels have been destroyed.
383 * See comment in struct lttng_session_trigger_list.
385 if (state
->session_triggers_ht
) {
386 ret
= cds_lfht_destroy(state
->session_triggers_ht
, NULL
);
389 if (state
->notification_channel_socket
>= 0) {
390 notification_channel_socket_destroy(
391 state
->notification_channel_socket
);
394 assert(cds_list_empty(&state
->tracer_event_sources_list
));
396 if (state
->executor
) {
397 action_executor_destroy(state
->executor
);
399 lttng_poll_clean(&state
->events
);
403 void mark_thread_as_ready(struct notification_thread_handle
*handle
)
405 DBG("Marking notification thread as ready");
406 sem_post(&handle
->ready
);
410 void wait_until_thread_is_ready(struct notification_thread_handle
*handle
)
412 DBG("Waiting for notification thread to be ready");
413 sem_wait(&handle
->ready
);
414 DBG("Notification thread is ready");
418 int init_thread_state(struct notification_thread_handle
*handle
,
419 struct notification_thread_state
*state
)
423 memset(state
, 0, sizeof(*state
));
424 state
->notification_channel_socket
= -1;
425 state
->trigger_id
.next_tracer_token
= 1;
426 lttng_poll_init(&state
->events
);
428 ret
= notification_channel_socket_create();
432 state
->notification_channel_socket
= ret
;
434 ret
= init_poll_set(&state
->events
, handle
,
435 state
->notification_channel_socket
);
440 DBG("[notification-thread] Listening on notification channel socket");
441 ret
= lttcomm_listen_unix_sock(state
->notification_channel_socket
);
443 ERR("[notification-thread] Listen failed on notification channel socket");
447 state
->client_socket_ht
= cds_lfht_new(DEFAULT_HT_SIZE
, 1, 0,
448 CDS_LFHT_AUTO_RESIZE
| CDS_LFHT_ACCOUNTING
, NULL
);
449 if (!state
->client_socket_ht
) {
453 state
->client_id_ht
= cds_lfht_new(DEFAULT_HT_SIZE
, 1, 0,
454 CDS_LFHT_AUTO_RESIZE
| CDS_LFHT_ACCOUNTING
, NULL
);
455 if (!state
->client_id_ht
) {
459 state
->channel_triggers_ht
= cds_lfht_new(DEFAULT_HT_SIZE
, 1, 0,
460 CDS_LFHT_AUTO_RESIZE
| CDS_LFHT_ACCOUNTING
, NULL
);
461 if (!state
->channel_triggers_ht
) {
465 state
->session_triggers_ht
= cds_lfht_new(DEFAULT_HT_SIZE
, 1, 0,
466 CDS_LFHT_AUTO_RESIZE
| CDS_LFHT_ACCOUNTING
, NULL
);
467 if (!state
->session_triggers_ht
) {
471 state
->channel_state_ht
= cds_lfht_new(DEFAULT_HT_SIZE
, 1, 0,
472 CDS_LFHT_AUTO_RESIZE
| CDS_LFHT_ACCOUNTING
, NULL
);
473 if (!state
->channel_state_ht
) {
477 state
->notification_trigger_clients_ht
= cds_lfht_new(DEFAULT_HT_SIZE
,
478 1, 0, CDS_LFHT_AUTO_RESIZE
| CDS_LFHT_ACCOUNTING
, NULL
);
479 if (!state
->notification_trigger_clients_ht
) {
483 state
->channels_ht
= cds_lfht_new(DEFAULT_HT_SIZE
,
484 1, 0, CDS_LFHT_AUTO_RESIZE
| CDS_LFHT_ACCOUNTING
, NULL
);
485 if (!state
->channels_ht
) {
489 state
->sessions_ht
= cds_lfht_new(DEFAULT_HT_SIZE
,
490 1, 0, CDS_LFHT_AUTO_RESIZE
| CDS_LFHT_ACCOUNTING
, NULL
);
491 if (!state
->sessions_ht
) {
494 state
->triggers_ht
= cds_lfht_new(DEFAULT_HT_SIZE
,
495 1, 0, CDS_LFHT_AUTO_RESIZE
| CDS_LFHT_ACCOUNTING
, NULL
);
496 if (!state
->triggers_ht
) {
499 state
->triggers_by_name_uid_ht
= cds_lfht_new(DEFAULT_HT_SIZE
,
500 1, 0, CDS_LFHT_AUTO_RESIZE
| CDS_LFHT_ACCOUNTING
, NULL
);
501 if (!state
->triggers_by_name_uid_ht
) {
505 state
->trigger_tokens_ht
= cds_lfht_new(DEFAULT_HT_SIZE
,
506 1, 0, CDS_LFHT_AUTO_RESIZE
| CDS_LFHT_ACCOUNTING
, NULL
);
507 if (!state
->trigger_tokens_ht
) {
511 CDS_INIT_LIST_HEAD(&state
->tracer_event_sources_list
);
513 state
->executor
= action_executor_create(handle
);
514 if (!state
->executor
) {
518 state
->restart_poll
= false;
520 mark_thread_as_ready(handle
);
524 fini_thread_state(state
);
529 int handle_channel_monitoring_pipe(int fd
, uint32_t revents
,
530 struct notification_thread_handle
*handle
,
531 struct notification_thread_state
*state
)
534 enum lttng_domain_type domain
;
536 if (fd
== handle
->channel_monitoring_pipes
.ust32_consumer
||
537 fd
== handle
->channel_monitoring_pipes
.ust64_consumer
) {
538 domain
= LTTNG_DOMAIN_UST
;
539 } else if (fd
== handle
->channel_monitoring_pipes
.kernel_consumer
) {
540 domain
= LTTNG_DOMAIN_KERNEL
;
545 if (revents
& (LPOLLERR
| LPOLLHUP
| LPOLLRDHUP
)) {
546 ret
= lttng_poll_del(&state
->events
, fd
);
548 ERR("[notification-thread] Failed to remove consumer monitoring pipe from poll set");
553 ret
= handle_notification_thread_channel_sample(
556 ERR("[notification-thread] Consumer sample handling error occurred");
564 static int handle_event_notification_pipe(int event_source_fd
,
565 enum lttng_domain_type domain
,
567 struct notification_thread_state
*state
)
571 if (revents
& (LPOLLERR
| LPOLLHUP
| LPOLLRDHUP
)) {
572 ret
= handle_notification_thread_remove_tracer_event_source_no_result(
573 state
, event_source_fd
);
575 ERR("[notification-thread] Failed to remove event notification pipe from poll set: fd = %d",
581 if (testpoint(sessiond_handle_notifier_event_pipe
)) {
586 if (caa_unlikely(notifier_consumption_paused
)) {
587 DBG("Event notifier notification consumption paused, sleeping...");
592 ret
= handle_notification_thread_event_notification(
593 state
, event_source_fd
, domain
);
595 ERR("[notification-thread] Event notification handling error occurred for fd: %d",
606 * Return the event source domain type via parameter.
608 static bool fd_is_event_notification_source(const struct notification_thread_state
*state
,
610 enum lttng_domain_type
*domain
)
612 struct notification_event_tracer_event_source_element
*source_element
;
616 cds_list_for_each_entry(source_element
,
617 &state
->tracer_event_sources_list
, node
) {
618 if (source_element
->fd
!= fd
) {
622 *domain
= source_element
->domain
;
630 * This thread services notification channel clients and commands received
631 * from various lttng-sessiond components over a command queue.
634 void *thread_notification(void *data
)
637 struct notification_thread_handle
*handle
= data
;
638 struct notification_thread_state state
;
639 enum lttng_domain_type domain
;
641 DBG("[notification-thread] Started notification thread");
643 health_register(health_sessiond
, HEALTH_SESSIOND_TYPE_NOTIFICATION
);
644 rcu_register_thread();
648 ERR("[notification-thread] Invalid thread context provided");
652 health_code_update();
654 ret
= init_thread_state(handle
, &state
);
659 if (testpoint(sessiond_thread_notification
)) {
667 DBG("[notification-thread] Entering poll wait");
668 ret
= lttng_poll_wait(&state
.events
, -1);
669 DBG("[notification-thread] Poll wait returned (%i)", ret
);
673 * Restart interrupted system call.
675 if (errno
== EINTR
) {
678 ERR("[notification-thread] Error encountered during lttng_poll_wait (%i)", ret
);
683 * Reset restart_poll flag so that calls below might turn it
686 state
.restart_poll
= false;
689 for (i
= 0; i
< fd_count
; i
++) {
690 int fd
= LTTNG_POLL_GETFD(&state
.events
, i
);
691 uint32_t revents
= LTTNG_POLL_GETEV(&state
.events
, i
);
693 DBG("[notification-thread] Handling fd (%i) activity (%u)", fd
, revents
);
695 if (fd
== state
.notification_channel_socket
) {
696 if (revents
& LPOLLIN
) {
697 ret
= handle_notification_thread_client_connect(
703 (LPOLLERR
| LPOLLHUP
| LPOLLRDHUP
)) {
704 ERR("[notification-thread] Notification socket poll error");
707 ERR("[notification-thread] Unexpected poll events %u for notification socket %i", revents
, fd
);
710 } else if (fd
== lttng_pipe_get_readfd(handle
->cmd_queue
.event_pipe
)) {
711 ret
= handle_notification_thread_command(handle
,
714 DBG("[notification-thread] Error encountered while servicing command queue");
716 } else if (ret
> 0) {
719 } else if (fd
== handle
->channel_monitoring_pipes
.ust32_consumer
||
720 fd
== handle
->channel_monitoring_pipes
.ust64_consumer
||
721 fd
== handle
->channel_monitoring_pipes
.kernel_consumer
) {
722 ret
= handle_channel_monitoring_pipe(fd
,
723 revents
, handle
, &state
);
727 } else if (fd_is_event_notification_source(&state
, fd
, &domain
)) {
728 ret
= handle_event_notification_pipe(fd
, domain
, revents
, &state
);
733 /* Activity on a client's socket. */
734 if (revents
& (LPOLLERR
| LPOLLHUP
| LPOLLRDHUP
)) {
736 * It doesn't matter if a command was
737 * pending on the client socket at this
738 * point since it now has no way to
739 * receive the notifications to which
740 * it was subscribing or unsubscribing.
742 ret
= handle_notification_thread_client_disconnect(
748 if (revents
& LPOLLIN
) {
749 ret
= handle_notification_thread_client_in(
756 if (revents
& LPOLLOUT
) {
757 ret
= handle_notification_thread_client_out(
767 * Calls above might have changed the state of the
768 * FDs in `state.events`. Call _poll_wait() again to
769 * ensure we have a consistent state.
771 if (state
.restart_poll
) {
778 fini_thread_state(&state
);
780 rcu_thread_offline();
781 rcu_unregister_thread();
782 health_unregister(health_sessiond
);
787 bool shutdown_notification_thread(void *thread_data
)
789 struct notification_thread_handle
*handle
= thread_data
;
791 notification_thread_command_quit(handle
);
795 struct lttng_thread
*launch_notification_thread(
796 struct notification_thread_handle
*handle
)
798 struct lttng_thread
*thread
;
800 thread
= lttng_thread_create("Notification",
802 shutdown_notification_thread
,
810 * Wait for the thread to be marked as "ready" before returning
811 * as other subsystems depend on the notification subsystem
812 * (e.g. rotation thread).
814 wait_until_thread_is_ready(handle
);