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"
32 #include <urcu/list.h>
33 #include <urcu/rculfhash.h>
36 * Destroy the thread data previously created by the init function.
38 void notification_thread_handle_destroy(
39 struct notification_thread_handle
*handle
)
47 assert(cds_list_empty(&handle
->cmd_queue
.list
));
48 pthread_mutex_destroy(&handle
->cmd_queue
.lock
);
49 sem_destroy(&handle
->ready
);
51 if (handle
->cmd_queue
.event_pipe
) {
52 lttng_pipe_destroy(handle
->cmd_queue
.event_pipe
);
54 if (handle
->channel_monitoring_pipes
.ust32_consumer
>= 0) {
55 ret
= close(handle
->channel_monitoring_pipes
.ust32_consumer
);
57 PERROR("close 32-bit consumer channel monitoring pipe");
60 if (handle
->channel_monitoring_pipes
.ust64_consumer
>= 0) {
61 ret
= close(handle
->channel_monitoring_pipes
.ust64_consumer
);
63 PERROR("close 64-bit consumer channel monitoring pipe");
66 if (handle
->channel_monitoring_pipes
.kernel_consumer
>= 0) {
67 ret
= close(handle
->channel_monitoring_pipes
.kernel_consumer
);
69 PERROR("close kernel consumer channel monitoring pipe");
76 struct notification_thread_handle
*notification_thread_handle_create(
77 struct lttng_pipe
*ust32_channel_monitor_pipe
,
78 struct lttng_pipe
*ust64_channel_monitor_pipe
,
79 struct lttng_pipe
*kernel_channel_monitor_pipe
)
82 struct notification_thread_handle
*handle
;
83 struct lttng_pipe
*event_pipe
= NULL
;
85 handle
= zmalloc(sizeof(*handle
));
90 sem_init(&handle
->ready
, 0, 0);
92 event_pipe
= lttng_pipe_open(FD_CLOEXEC
);
94 ERR("event_pipe creation");
98 handle
->cmd_queue
.event_pipe
= event_pipe
;
101 CDS_INIT_LIST_HEAD(&handle
->cmd_queue
.list
);
102 ret
= pthread_mutex_init(&handle
->cmd_queue
.lock
, NULL
);
107 if (ust32_channel_monitor_pipe
) {
108 handle
->channel_monitoring_pipes
.ust32_consumer
=
109 lttng_pipe_release_readfd(
110 ust32_channel_monitor_pipe
);
111 if (handle
->channel_monitoring_pipes
.ust32_consumer
< 0) {
115 handle
->channel_monitoring_pipes
.ust32_consumer
= -1;
117 if (ust64_channel_monitor_pipe
) {
118 handle
->channel_monitoring_pipes
.ust64_consumer
=
119 lttng_pipe_release_readfd(
120 ust64_channel_monitor_pipe
);
121 if (handle
->channel_monitoring_pipes
.ust64_consumer
< 0) {
125 handle
->channel_monitoring_pipes
.ust64_consumer
= -1;
127 if (kernel_channel_monitor_pipe
) {
128 handle
->channel_monitoring_pipes
.kernel_consumer
=
129 lttng_pipe_release_readfd(
130 kernel_channel_monitor_pipe
);
131 if (handle
->channel_monitoring_pipes
.kernel_consumer
< 0) {
135 handle
->channel_monitoring_pipes
.kernel_consumer
= -1;
140 lttng_pipe_destroy(event_pipe
);
141 notification_thread_handle_destroy(handle
);
146 char *get_notification_channel_sock_path(void)
149 bool is_root
= !getuid();
152 sock_path
= zmalloc(LTTNG_PATH_MAX
);
158 ret
= snprintf(sock_path
, LTTNG_PATH_MAX
,
159 DEFAULT_GLOBAL_NOTIFICATION_CHANNEL_UNIX_SOCK
);
164 const char *home_path
= utils_get_home_dir();
167 ERR("Can't get HOME directory for socket creation");
171 ret
= snprintf(sock_path
, LTTNG_PATH_MAX
,
172 DEFAULT_HOME_NOTIFICATION_CHANNEL_UNIX_SOCK
,
186 void notification_channel_socket_destroy(int fd
)
189 char *sock_path
= get_notification_channel_sock_path();
191 DBG("[notification-thread] Destroying notification channel socket");
194 ret
= unlink(sock_path
);
197 PERROR("unlink notification channel socket");
203 PERROR("close notification channel socket");
208 int notification_channel_socket_create(void)
211 char *sock_path
= get_notification_channel_sock_path();
213 DBG("[notification-thread] Creating notification channel UNIX socket at %s",
216 ret
= lttcomm_create_unix_sock(sock_path
);
218 ERR("[notification-thread] Failed to create notification socket");
223 ret
= chmod(sock_path
, S_IRUSR
| S_IWUSR
| S_IRGRP
| S_IWGRP
);
225 ERR("Set file permissions failed: %s", sock_path
);
226 PERROR("chmod notification channel socket");
233 ret
= utils_get_group_id(config
.tracing_group_name
.value
, true,
236 /* Default to root group. */
240 ret
= chown(sock_path
, 0, gid
);
242 ERR("Failed to set the notification channel socket's group");
248 DBG("[notification-thread] Notification channel UNIX socket created (fd = %i)",
253 if (fd
>= 0 && close(fd
) < 0) {
254 PERROR("close notification channel socket");
261 int init_poll_set(struct lttng_poll_event
*poll_set
,
262 struct notification_thread_handle
*handle
,
263 int notification_channel_socket
)
268 * Create pollset with size 5:
269 * - notification channel socket (listen for new connections),
270 * - command queue event fd (internal sessiond commands),
271 * - consumerd (32-bit user space) channel monitor pipe,
272 * - consumerd (64-bit user space) channel monitor pipe,
273 * - consumerd (kernel) channel monitor pipe.
275 ret
= lttng_poll_create(poll_set
, 5, LTTNG_CLOEXEC
);
280 ret
= lttng_poll_add(poll_set
, notification_channel_socket
,
281 LPOLLIN
| LPOLLERR
| LPOLLHUP
| LPOLLRDHUP
);
283 ERR("[notification-thread] Failed to add notification channel socket to pollset");
286 ret
= lttng_poll_add(poll_set
, lttng_pipe_get_readfd(handle
->cmd_queue
.event_pipe
),
289 ERR("[notification-thread] Failed to add notification command queue event fd to pollset");
292 ret
= lttng_poll_add(poll_set
,
293 handle
->channel_monitoring_pipes
.ust32_consumer
,
296 ERR("[notification-thread] Failed to add ust-32 channel monitoring pipe fd to pollset");
299 ret
= lttng_poll_add(poll_set
,
300 handle
->channel_monitoring_pipes
.ust64_consumer
,
303 ERR("[notification-thread] Failed to add ust-64 channel monitoring pipe fd to pollset");
306 if (handle
->channel_monitoring_pipes
.kernel_consumer
< 0) {
309 ret
= lttng_poll_add(poll_set
,
310 handle
->channel_monitoring_pipes
.kernel_consumer
,
313 ERR("[notification-thread] Failed to add kernel channel monitoring pipe fd to pollset");
319 lttng_poll_clean(poll_set
);
324 void fini_thread_state(struct notification_thread_state
*state
)
328 if (state
->client_socket_ht
) {
329 ret
= handle_notification_thread_client_disconnect_all(state
);
331 ret
= cds_lfht_destroy(state
->client_socket_ht
, NULL
);
334 if (state
->client_id_ht
) {
335 ret
= cds_lfht_destroy(state
->client_id_ht
, NULL
);
338 if (state
->triggers_ht
) {
339 ret
= handle_notification_thread_trigger_unregister_all(state
);
341 ret
= cds_lfht_destroy(state
->triggers_ht
, NULL
);
344 if (state
->channel_triggers_ht
) {
345 ret
= cds_lfht_destroy(state
->channel_triggers_ht
, NULL
);
348 if (state
->channel_state_ht
) {
349 ret
= cds_lfht_destroy(state
->channel_state_ht
, NULL
);
352 if (state
->notification_trigger_clients_ht
) {
353 ret
= cds_lfht_destroy(state
->notification_trigger_clients_ht
,
357 if (state
->channels_ht
) {
358 ret
= cds_lfht_destroy(state
->channels_ht
, NULL
);
361 if (state
->sessions_ht
) {
362 ret
= cds_lfht_destroy(state
->sessions_ht
, NULL
);
366 * Must be destroyed after all channels have been destroyed.
367 * See comment in struct lttng_session_trigger_list.
369 if (state
->session_triggers_ht
) {
370 ret
= cds_lfht_destroy(state
->session_triggers_ht
, NULL
);
373 if (state
->notification_channel_socket
>= 0) {
374 notification_channel_socket_destroy(
375 state
->notification_channel_socket
);
377 if (state
->executor
) {
378 action_executor_destroy(state
->executor
);
380 lttng_poll_clean(&state
->events
);
384 void mark_thread_as_ready(struct notification_thread_handle
*handle
)
386 DBG("Marking notification thread as ready");
387 sem_post(&handle
->ready
);
391 void wait_until_thread_is_ready(struct notification_thread_handle
*handle
)
393 DBG("Waiting for notification thread to be ready");
394 sem_wait(&handle
->ready
);
395 DBG("Notification thread is ready");
399 int init_thread_state(struct notification_thread_handle
*handle
,
400 struct notification_thread_state
*state
)
404 memset(state
, 0, sizeof(*state
));
405 state
->notification_channel_socket
= -1;
406 lttng_poll_init(&state
->events
);
408 ret
= notification_channel_socket_create();
412 state
->notification_channel_socket
= ret
;
414 ret
= init_poll_set(&state
->events
, handle
,
415 state
->notification_channel_socket
);
420 DBG("[notification-thread] Listening on notification channel socket");
421 ret
= lttcomm_listen_unix_sock(state
->notification_channel_socket
);
423 ERR("[notification-thread] Listen failed on notification channel socket");
427 state
->client_socket_ht
= cds_lfht_new(DEFAULT_HT_SIZE
, 1, 0,
428 CDS_LFHT_AUTO_RESIZE
| CDS_LFHT_ACCOUNTING
, NULL
);
429 if (!state
->client_socket_ht
) {
433 state
->client_id_ht
= cds_lfht_new(DEFAULT_HT_SIZE
, 1, 0,
434 CDS_LFHT_AUTO_RESIZE
| CDS_LFHT_ACCOUNTING
, NULL
);
435 if (!state
->client_id_ht
) {
439 state
->channel_triggers_ht
= cds_lfht_new(DEFAULT_HT_SIZE
, 1, 0,
440 CDS_LFHT_AUTO_RESIZE
| CDS_LFHT_ACCOUNTING
, NULL
);
441 if (!state
->channel_triggers_ht
) {
445 state
->session_triggers_ht
= cds_lfht_new(DEFAULT_HT_SIZE
, 1, 0,
446 CDS_LFHT_AUTO_RESIZE
| CDS_LFHT_ACCOUNTING
, NULL
);
447 if (!state
->session_triggers_ht
) {
451 state
->channel_state_ht
= cds_lfht_new(DEFAULT_HT_SIZE
, 1, 0,
452 CDS_LFHT_AUTO_RESIZE
| CDS_LFHT_ACCOUNTING
, NULL
);
453 if (!state
->channel_state_ht
) {
457 state
->notification_trigger_clients_ht
= cds_lfht_new(DEFAULT_HT_SIZE
,
458 1, 0, CDS_LFHT_AUTO_RESIZE
| CDS_LFHT_ACCOUNTING
, NULL
);
459 if (!state
->notification_trigger_clients_ht
) {
463 state
->channels_ht
= cds_lfht_new(DEFAULT_HT_SIZE
,
464 1, 0, CDS_LFHT_AUTO_RESIZE
| CDS_LFHT_ACCOUNTING
, NULL
);
465 if (!state
->channels_ht
) {
468 state
->sessions_ht
= cds_lfht_new(DEFAULT_HT_SIZE
,
469 1, 0, CDS_LFHT_AUTO_RESIZE
| CDS_LFHT_ACCOUNTING
, NULL
);
470 if (!state
->sessions_ht
) {
473 state
->triggers_ht
= cds_lfht_new(DEFAULT_HT_SIZE
,
474 1, 0, CDS_LFHT_AUTO_RESIZE
| CDS_LFHT_ACCOUNTING
, NULL
);
475 if (!state
->triggers_ht
) {
479 state
->executor
= action_executor_create(handle
);
480 if (!state
->executor
) {
483 mark_thread_as_ready(handle
);
487 fini_thread_state(state
);
492 int handle_channel_monitoring_pipe(int fd
, uint32_t revents
,
493 struct notification_thread_handle
*handle
,
494 struct notification_thread_state
*state
)
497 enum lttng_domain_type domain
;
499 if (fd
== handle
->channel_monitoring_pipes
.ust32_consumer
||
500 fd
== handle
->channel_monitoring_pipes
.ust64_consumer
) {
501 domain
= LTTNG_DOMAIN_UST
;
502 } else if (fd
== handle
->channel_monitoring_pipes
.kernel_consumer
) {
503 domain
= LTTNG_DOMAIN_KERNEL
;
508 if (revents
& (LPOLLERR
| LPOLLHUP
| LPOLLRDHUP
)) {
509 ret
= lttng_poll_del(&state
->events
, fd
);
511 ERR("[notification-thread] Failed to remove consumer monitoring pipe from poll set");
516 ret
= handle_notification_thread_channel_sample(
519 ERR("[notification-thread] Consumer sample handling error occurred");
528 * This thread services notification channel clients and commands received
529 * from various lttng-sessiond components over a command queue.
532 void *thread_notification(void *data
)
535 struct notification_thread_handle
*handle
= data
;
536 struct notification_thread_state state
;
538 DBG("[notification-thread] Started notification thread");
540 health_register(health_sessiond
, HEALTH_SESSIOND_TYPE_NOTIFICATION
);
541 rcu_register_thread();
545 ERR("[notification-thread] Invalid thread context provided");
549 health_code_update();
551 ret
= init_thread_state(handle
, &state
);
560 DBG("[notification-thread] Entering poll wait");
561 ret
= lttng_poll_wait(&state
.events
, -1);
562 DBG("[notification-thread] Poll wait returned (%i)", ret
);
566 * Restart interrupted system call.
568 if (errno
== EINTR
) {
571 ERR("[notification-thread] Error encountered during lttng_poll_wait (%i)", ret
);
576 for (i
= 0; i
< fd_count
; i
++) {
577 int fd
= LTTNG_POLL_GETFD(&state
.events
, i
);
578 uint32_t revents
= LTTNG_POLL_GETEV(&state
.events
, i
);
580 DBG("[notification-thread] Handling fd (%i) activity (%u)", fd
, revents
);
582 if (fd
== state
.notification_channel_socket
) {
583 if (revents
& LPOLLIN
) {
584 ret
= handle_notification_thread_client_connect(
590 (LPOLLERR
| LPOLLHUP
| LPOLLRDHUP
)) {
591 ERR("[notification-thread] Notification socket poll error");
594 ERR("[notification-thread] Unexpected poll events %u for notification socket %i", revents
, fd
);
597 } else if (fd
== lttng_pipe_get_readfd(handle
->cmd_queue
.event_pipe
)) {
598 ret
= handle_notification_thread_command(handle
,
601 DBG("[notification-thread] Error encountered while servicing command queue");
603 } else if (ret
> 0) {
606 } else if (fd
== handle
->channel_monitoring_pipes
.ust32_consumer
||
607 fd
== handle
->channel_monitoring_pipes
.ust64_consumer
||
608 fd
== handle
->channel_monitoring_pipes
.kernel_consumer
) {
609 ret
= handle_channel_monitoring_pipe(fd
,
610 revents
, handle
, &state
);
615 /* Activity on a client's socket. */
616 if (revents
& (LPOLLERR
| LPOLLHUP
| LPOLLRDHUP
)) {
618 * It doesn't matter if a command was
619 * pending on the client socket at this
620 * point since it now has no way to
621 * receive the notifications to which
622 * it was subscribing or unsubscribing.
624 ret
= handle_notification_thread_client_disconnect(
630 if (revents
& LPOLLIN
) {
631 ret
= handle_notification_thread_client_in(
638 if (revents
& LPOLLOUT
) {
639 ret
= handle_notification_thread_client_out(
651 fini_thread_state(&state
);
653 rcu_thread_offline();
654 rcu_unregister_thread();
655 health_unregister(health_sessiond
);
660 bool shutdown_notification_thread(void *thread_data
)
662 struct notification_thread_handle
*handle
= thread_data
;
664 notification_thread_command_quit(handle
);
668 struct lttng_thread
*launch_notification_thread(
669 struct notification_thread_handle
*handle
)
671 struct lttng_thread
*thread
;
673 thread
= lttng_thread_create("Notification",
675 shutdown_notification_thread
,
683 * Wait for the thread to be marked as "ready" before returning
684 * as other subsystems depend on the notification subsystem
685 * (e.g. rotation thread).
687 wait_until_thread_is_ready(handle
);