2 * Copyright (C) 2017 - Jérémie Galarneau <jeremie.galarneau@efficios.com>
4 * This program is free software; you can redistribute it and/or modify it
5 * under the terms of the GNU General Public License, version 2 only, as
6 * published by the Free Software Foundation.
8 * This program is distributed in the hope that it will be useful, but WITHOUT
9 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
10 * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
13 * You should have received a copy of the GNU General Public License along with
14 * this program; if not, write to the Free Software Foundation, Inc., 51
15 * Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
19 #include <lttng/trigger/trigger.h>
20 #include <lttng/notification/channel-internal.h>
21 #include <lttng/notification/notification-internal.h>
22 #include <lttng/condition/condition-internal.h>
23 #include <lttng/condition/buffer-usage-internal.h>
24 #include <common/error.h>
25 #include <common/config/session-config.h>
26 #include <common/defaults.h>
27 #include <common/utils.h>
28 #include <common/align.h>
29 #include <common/time.h>
30 #include <sys/eventfd.h>
35 #include "notification-thread.h"
36 #include "notification-thread-events.h"
37 #include "notification-thread-commands.h"
38 #include "lttng-sessiond.h"
39 #include "health-sessiond.h"
43 #include <urcu/list.h>
44 #include <urcu/rculfhash.h>
47 * Destroy the thread data previously created by the init function.
49 void notification_thread_handle_destroy(
50 struct notification_thread_handle
*handle
)
58 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");
87 struct notification_thread_handle
*notification_thread_handle_create(
88 struct lttng_pipe
*ust32_channel_monitor_pipe
,
89 struct lttng_pipe
*ust64_channel_monitor_pipe
,
90 struct lttng_pipe
*kernel_channel_monitor_pipe
)
93 struct notification_thread_handle
*handle
;
94 struct lttng_pipe
*event_pipe
= NULL
;
96 handle
= zmalloc(sizeof(*handle
));
101 sem_init(&handle
->ready
, 0, 0);
103 event_pipe
= lttng_pipe_open(FD_CLOEXEC
);
105 ERR("event_pipe creation");
109 handle
->cmd_queue
.event_pipe
= event_pipe
;
112 CDS_INIT_LIST_HEAD(&handle
->cmd_queue
.list
);
113 ret
= pthread_mutex_init(&handle
->cmd_queue
.lock
, NULL
);
118 if (ust32_channel_monitor_pipe
) {
119 handle
->channel_monitoring_pipes
.ust32_consumer
=
120 lttng_pipe_release_readfd(
121 ust32_channel_monitor_pipe
);
122 if (handle
->channel_monitoring_pipes
.ust32_consumer
< 0) {
126 handle
->channel_monitoring_pipes
.ust32_consumer
= -1;
128 if (ust64_channel_monitor_pipe
) {
129 handle
->channel_monitoring_pipes
.ust64_consumer
=
130 lttng_pipe_release_readfd(
131 ust64_channel_monitor_pipe
);
132 if (handle
->channel_monitoring_pipes
.ust64_consumer
< 0) {
136 handle
->channel_monitoring_pipes
.ust64_consumer
= -1;
138 if (kernel_channel_monitor_pipe
) {
139 handle
->channel_monitoring_pipes
.kernel_consumer
=
140 lttng_pipe_release_readfd(
141 kernel_channel_monitor_pipe
);
142 if (handle
->channel_monitoring_pipes
.kernel_consumer
< 0) {
146 handle
->channel_monitoring_pipes
.kernel_consumer
= -1;
151 lttng_pipe_destroy(event_pipe
);
152 notification_thread_handle_destroy(handle
);
157 char *get_notification_channel_sock_path(void)
160 bool is_root
= !getuid();
163 sock_path
= zmalloc(LTTNG_PATH_MAX
);
169 ret
= snprintf(sock_path
, LTTNG_PATH_MAX
,
170 DEFAULT_GLOBAL_NOTIFICATION_CHANNEL_UNIX_SOCK
);
175 char *home_path
= utils_get_home_dir();
178 ERR("Can't get HOME directory for socket creation");
182 ret
= snprintf(sock_path
, LTTNG_PATH_MAX
,
183 DEFAULT_HOME_NOTIFICATION_CHANNEL_UNIX_SOCK
,
197 void notification_channel_socket_destroy(int fd
)
200 char *sock_path
= get_notification_channel_sock_path();
202 DBG("[notification-thread] Destroying notification channel socket");
205 ret
= unlink(sock_path
);
208 PERROR("unlink notification channel socket");
214 PERROR("close notification channel socket");
219 int notification_channel_socket_create(void)
222 char *sock_path
= get_notification_channel_sock_path();
224 DBG("[notification-thread] Creating notification channel UNIX socket at %s",
227 ret
= lttcomm_create_unix_sock(sock_path
);
229 ERR("[notification-thread] Failed to create notification socket");
234 ret
= chmod(sock_path
, S_IRUSR
| S_IWUSR
| S_IRGRP
| S_IWGRP
);
236 ERR("Set file permissions failed: %s", sock_path
);
237 PERROR("chmod notification channel socket");
242 ret
= chown(sock_path
, 0,
243 utils_get_group_id(config
.tracing_group_name
.value
));
245 ERR("Failed to set the notification channel socket's group");
251 DBG("[notification-thread] Notification channel UNIX socket created (fd = %i)",
256 if (fd
>= 0 && close(fd
) < 0) {
257 PERROR("close notification channel socket");
264 int init_poll_set(struct lttng_poll_event
*poll_set
,
265 struct notification_thread_handle
*handle
,
266 int notification_channel_socket
)
271 * Create pollset with size 5:
272 * - notification channel socket (listen for new connections),
273 * - command queue event fd (internal sessiond commands),
274 * - consumerd (32-bit user space) channel monitor pipe,
275 * - consumerd (64-bit user space) channel monitor pipe,
276 * - consumerd (kernel) channel monitor pipe.
278 ret
= lttng_poll_create(poll_set
, 5, LTTNG_CLOEXEC
);
283 ret
= lttng_poll_add(poll_set
, notification_channel_socket
,
284 LPOLLIN
| LPOLLERR
| LPOLLHUP
| LPOLLRDHUP
);
286 ERR("[notification-thread] Failed to add notification channel socket to pollset");
289 ret
= lttng_poll_add(poll_set
, lttng_pipe_get_readfd(handle
->cmd_queue
.event_pipe
),
292 ERR("[notification-thread] Failed to add notification command queue event fd to pollset");
295 ret
= lttng_poll_add(poll_set
,
296 handle
->channel_monitoring_pipes
.ust32_consumer
,
299 ERR("[notification-thread] Failed to add ust-32 channel monitoring pipe fd to pollset");
302 ret
= lttng_poll_add(poll_set
,
303 handle
->channel_monitoring_pipes
.ust64_consumer
,
306 ERR("[notification-thread] Failed to add ust-64 channel monitoring pipe fd to pollset");
309 if (handle
->channel_monitoring_pipes
.kernel_consumer
< 0) {
312 ret
= lttng_poll_add(poll_set
,
313 handle
->channel_monitoring_pipes
.kernel_consumer
,
316 ERR("[notification-thread] Failed to add kernel channel monitoring pipe fd to pollset");
322 lttng_poll_clean(poll_set
);
327 void fini_thread_state(struct notification_thread_state
*state
)
331 if (state
->client_socket_ht
) {
332 ret
= handle_notification_thread_client_disconnect_all(state
);
334 ret
= cds_lfht_destroy(state
->client_socket_ht
, NULL
);
337 if (state
->triggers_ht
) {
338 ret
= handle_notification_thread_trigger_unregister_all(state
);
340 ret
= cds_lfht_destroy(state
->triggers_ht
, NULL
);
343 if (state
->channel_triggers_ht
) {
344 ret
= cds_lfht_destroy(state
->channel_triggers_ht
, NULL
);
347 if (state
->channel_state_ht
) {
348 ret
= cds_lfht_destroy(state
->channel_state_ht
, NULL
);
351 if (state
->notification_trigger_clients_ht
) {
352 ret
= cds_lfht_destroy(state
->notification_trigger_clients_ht
,
356 if (state
->channels_ht
) {
357 ret
= cds_lfht_destroy(state
->channels_ht
, NULL
);
360 if (state
->sessions_ht
) {
361 ret
= cds_lfht_destroy(state
->sessions_ht
, NULL
);
365 * Must be destroyed after all channels have been destroyed.
366 * See comment in struct lttng_session_trigger_list.
368 if (state
->session_triggers_ht
) {
369 ret
= cds_lfht_destroy(state
->session_triggers_ht
, NULL
);
372 if (state
->notification_channel_socket
>= 0) {
373 notification_channel_socket_destroy(
374 state
->notification_channel_socket
);
376 lttng_poll_clean(&state
->events
);
380 void mark_thread_as_ready(struct notification_thread_handle
*handle
)
382 DBG("Marking notification thread as ready");
383 sem_post(&handle
->ready
);
387 void wait_until_thread_is_ready(struct notification_thread_handle
*handle
)
389 DBG("Waiting for notification thread to be ready");
390 sem_wait(&handle
->ready
);
391 DBG("Notification thread is ready");
395 int init_thread_state(struct notification_thread_handle
*handle
,
396 struct notification_thread_state
*state
)
400 memset(state
, 0, sizeof(*state
));
401 state
->notification_channel_socket
= -1;
402 lttng_poll_init(&state
->events
);
404 ret
= notification_channel_socket_create();
408 state
->notification_channel_socket
= ret
;
410 ret
= init_poll_set(&state
->events
, handle
,
411 state
->notification_channel_socket
);
416 DBG("[notification-thread] Listening on notification channel socket");
417 ret
= lttcomm_listen_unix_sock(state
->notification_channel_socket
);
419 ERR("[notification-thread] Listen failed on notification channel socket");
423 state
->client_socket_ht
= cds_lfht_new(DEFAULT_HT_SIZE
, 1, 0,
424 CDS_LFHT_AUTO_RESIZE
| CDS_LFHT_ACCOUNTING
, NULL
);
425 if (!state
->client_socket_ht
) {
429 state
->channel_triggers_ht
= cds_lfht_new(DEFAULT_HT_SIZE
, 1, 0,
430 CDS_LFHT_AUTO_RESIZE
| CDS_LFHT_ACCOUNTING
, NULL
);
431 if (!state
->channel_triggers_ht
) {
435 state
->session_triggers_ht
= cds_lfht_new(DEFAULT_HT_SIZE
, 1, 0,
436 CDS_LFHT_AUTO_RESIZE
| CDS_LFHT_ACCOUNTING
, NULL
);
437 if (!state
->session_triggers_ht
) {
441 state
->channel_state_ht
= cds_lfht_new(DEFAULT_HT_SIZE
, 1, 0,
442 CDS_LFHT_AUTO_RESIZE
| CDS_LFHT_ACCOUNTING
, NULL
);
443 if (!state
->channel_state_ht
) {
447 state
->notification_trigger_clients_ht
= cds_lfht_new(DEFAULT_HT_SIZE
,
448 1, 0, CDS_LFHT_AUTO_RESIZE
| CDS_LFHT_ACCOUNTING
, NULL
);
449 if (!state
->notification_trigger_clients_ht
) {
453 state
->channels_ht
= cds_lfht_new(DEFAULT_HT_SIZE
,
454 1, 0, CDS_LFHT_AUTO_RESIZE
| CDS_LFHT_ACCOUNTING
, NULL
);
455 if (!state
->channels_ht
) {
458 state
->sessions_ht
= cds_lfht_new(DEFAULT_HT_SIZE
,
459 1, 0, CDS_LFHT_AUTO_RESIZE
| CDS_LFHT_ACCOUNTING
, NULL
);
460 if (!state
->sessions_ht
) {
463 state
->triggers_ht
= cds_lfht_new(DEFAULT_HT_SIZE
,
464 1, 0, CDS_LFHT_AUTO_RESIZE
| CDS_LFHT_ACCOUNTING
, NULL
);
465 if (!state
->triggers_ht
) {
468 mark_thread_as_ready(handle
);
472 fini_thread_state(state
);
477 int handle_channel_monitoring_pipe(int fd
, uint32_t revents
,
478 struct notification_thread_handle
*handle
,
479 struct notification_thread_state
*state
)
482 enum lttng_domain_type domain
;
484 if (fd
== handle
->channel_monitoring_pipes
.ust32_consumer
||
485 fd
== handle
->channel_monitoring_pipes
.ust64_consumer
) {
486 domain
= LTTNG_DOMAIN_UST
;
487 } else if (fd
== handle
->channel_monitoring_pipes
.kernel_consumer
) {
488 domain
= LTTNG_DOMAIN_KERNEL
;
493 if (revents
& (LPOLLERR
| LPOLLHUP
| LPOLLRDHUP
)) {
494 ret
= lttng_poll_del(&state
->events
, fd
);
496 ERR("[notification-thread] Failed to remove consumer monitoring pipe from poll set");
501 ret
= handle_notification_thread_channel_sample(
504 ERR("[notification-thread] Consumer sample handling error occurred");
513 * This thread services notification channel clients and commands received
514 * from various lttng-sessiond components over a command queue.
517 void *thread_notification(void *data
)
520 struct notification_thread_handle
*handle
= data
;
521 struct notification_thread_state state
;
523 DBG("[notification-thread] Started notification thread");
526 ERR("[notification-thread] Invalid thread context provided");
530 rcu_register_thread();
533 health_register(health_sessiond
, HEALTH_SESSIOND_TYPE_NOTIFICATION
);
534 health_code_update();
536 ret
= init_thread_state(handle
, &state
);
545 DBG("[notification-thread] Entering poll wait");
546 ret
= lttng_poll_wait(&state
.events
, -1);
547 DBG("[notification-thread] Poll wait returned (%i)", ret
);
551 * Restart interrupted system call.
553 if (errno
== EINTR
) {
556 ERR("[notification-thread] Error encountered during lttng_poll_wait (%i)", ret
);
561 for (i
= 0; i
< fd_count
; i
++) {
562 int fd
= LTTNG_POLL_GETFD(&state
.events
, i
);
563 uint32_t revents
= LTTNG_POLL_GETEV(&state
.events
, i
);
568 DBG("[notification-thread] Handling fd (%i) activity (%u)", fd
, revents
);
570 if (fd
== state
.notification_channel_socket
) {
571 if (revents
& LPOLLIN
) {
572 ret
= handle_notification_thread_client_connect(
578 (LPOLLERR
| LPOLLHUP
| LPOLLRDHUP
)) {
579 ERR("[notification-thread] Notification socket poll error");
582 ERR("[notification-thread] Unexpected poll events %u for notification socket %i", revents
, fd
);
585 } else if (fd
== lttng_pipe_get_readfd(handle
->cmd_queue
.event_pipe
)) {
586 ret
= handle_notification_thread_command(handle
,
589 DBG("[notification-thread] Error encountered while servicing command queue");
591 } else if (ret
> 0) {
594 } else if (fd
== handle
->channel_monitoring_pipes
.ust32_consumer
||
595 fd
== handle
->channel_monitoring_pipes
.ust64_consumer
||
596 fd
== handle
->channel_monitoring_pipes
.kernel_consumer
) {
597 ret
= handle_channel_monitoring_pipe(fd
,
598 revents
, handle
, &state
);
603 /* Activity on a client's socket. */
604 if (revents
& (LPOLLERR
| LPOLLHUP
| LPOLLRDHUP
)) {
606 * It doesn't matter if a command was
607 * pending on the client socket at this
608 * point since it now has no way to
609 * receive the notifications to which
610 * it was subscribing or unsubscribing.
612 ret
= handle_notification_thread_client_disconnect(
618 if (revents
& LPOLLIN
) {
619 ret
= handle_notification_thread_client_in(
626 if (revents
& LPOLLOUT
) {
627 ret
= handle_notification_thread_client_out(
639 fini_thread_state(&state
);
640 health_unregister(health_sessiond
);
641 rcu_thread_offline();
642 rcu_unregister_thread();
648 bool shutdown_notification_thread(void *thread_data
)
650 struct notification_thread_handle
*handle
= thread_data
;
652 notification_thread_command_quit(handle
);
656 struct lttng_thread
*launch_notification_thread(
657 struct notification_thread_handle
*handle
)
659 struct lttng_thread
*thread
;
661 thread
= lttng_thread_create("Notification",
663 shutdown_notification_thread
,
671 * Wait for the thread to be marked as "ready" before returning
672 * as other subsystems depend on the notification subsystem
673 * (e.g. rotation thread).
675 wait_until_thread_is_ready(handle
);