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"
42 #include <urcu/list.h>
43 #include <urcu/rculfhash.h>
46 * Destroy the thread data previously created by the init function.
48 void notification_thread_handle_destroy(
49 struct notification_thread_handle
*handle
)
57 assert(cds_list_empty(&handle
->cmd_queue
.list
));
58 pthread_mutex_destroy(&handle
->cmd_queue
.lock
);
60 if (handle
->cmd_queue
.event_pipe
) {
61 lttng_pipe_destroy(handle
->cmd_queue
.event_pipe
);
63 if (handle
->channel_monitoring_pipes
.ust32_consumer
>= 0) {
64 ret
= close(handle
->channel_monitoring_pipes
.ust32_consumer
);
66 PERROR("close 32-bit consumer channel monitoring pipe");
69 if (handle
->channel_monitoring_pipes
.ust64_consumer
>= 0) {
70 ret
= close(handle
->channel_monitoring_pipes
.ust64_consumer
);
72 PERROR("close 64-bit consumer channel monitoring pipe");
75 if (handle
->channel_monitoring_pipes
.kernel_consumer
>= 0) {
76 ret
= close(handle
->channel_monitoring_pipes
.kernel_consumer
);
78 PERROR("close kernel consumer channel monitoring pipe");
85 struct notification_thread_handle
*notification_thread_handle_create(
86 struct lttng_pipe
*ust32_channel_monitor_pipe
,
87 struct lttng_pipe
*ust64_channel_monitor_pipe
,
88 struct lttng_pipe
*kernel_channel_monitor_pipe
)
91 struct notification_thread_handle
*handle
;
92 struct lttng_pipe
*event_pipe
= NULL
;
94 handle
= zmalloc(sizeof(*handle
));
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;
147 lttng_pipe_destroy(event_pipe
);
148 notification_thread_handle_destroy(handle
);
153 char *get_notification_channel_sock_path(void)
156 bool is_root
= !getuid();
159 sock_path
= zmalloc(LTTNG_PATH_MAX
);
165 ret
= snprintf(sock_path
, LTTNG_PATH_MAX
,
166 DEFAULT_GLOBAL_NOTIFICATION_CHANNEL_UNIX_SOCK
);
171 char *home_path
= utils_get_home_dir();
174 ERR("Can't get HOME directory for socket creation");
178 ret
= snprintf(sock_path
, LTTNG_PATH_MAX
,
179 DEFAULT_HOME_NOTIFICATION_CHANNEL_UNIX_SOCK
,
193 void notification_channel_socket_destroy(int fd
)
196 char *sock_path
= get_notification_channel_sock_path();
198 DBG("[notification-thread] Destroying notification channel socket");
201 ret
= unlink(sock_path
);
204 PERROR("unlink notification channel socket");
210 PERROR("close notification channel socket");
215 int notification_channel_socket_create(void)
218 char *sock_path
= get_notification_channel_sock_path();
220 DBG("[notification-thread] Creating notification channel UNIX socket at %s",
223 ret
= lttcomm_create_unix_sock(sock_path
);
225 ERR("[notification-thread] Failed to create notification socket");
230 ret
= chmod(sock_path
, S_IRUSR
| S_IWUSR
| S_IRGRP
| S_IWGRP
);
232 ERR("Set file permissions failed: %s", sock_path
);
233 PERROR("chmod notification channel socket");
240 ret
= utils_get_group_id(config
.tracing_group_name
.value
, true,
243 /* Default to root group. */
247 ret
= chown(sock_path
, 0, gid
);
249 ERR("Failed to set the notification channel socket's group");
255 DBG("[notification-thread] Notification channel UNIX socket created (fd = %i)",
260 if (fd
>= 0 && close(fd
) < 0) {
261 PERROR("close notification channel socket");
268 int init_poll_set(struct lttng_poll_event
*poll_set
,
269 struct notification_thread_handle
*handle
,
270 int notification_channel_socket
)
275 * Create pollset with size 5:
276 * - notification channel socket (listen for new connections),
277 * - command queue event fd (internal sessiond commands),
278 * - consumerd (32-bit user space) channel monitor pipe,
279 * - consumerd (64-bit user space) channel monitor pipe,
280 * - consumerd (kernel) channel monitor pipe.
282 ret
= lttng_poll_create(poll_set
, 5, LTTNG_CLOEXEC
);
287 ret
= lttng_poll_add(poll_set
, notification_channel_socket
,
288 LPOLLIN
| LPOLLERR
| LPOLLHUP
| LPOLLRDHUP
);
290 ERR("[notification-thread] Failed to add notification channel socket to pollset");
293 ret
= lttng_poll_add(poll_set
, lttng_pipe_get_readfd(handle
->cmd_queue
.event_pipe
),
296 ERR("[notification-thread] Failed to add notification command queue event fd to pollset");
299 ret
= lttng_poll_add(poll_set
,
300 handle
->channel_monitoring_pipes
.ust32_consumer
,
303 ERR("[notification-thread] Failed to add ust-32 channel monitoring pipe fd to pollset");
306 ret
= lttng_poll_add(poll_set
,
307 handle
->channel_monitoring_pipes
.ust64_consumer
,
310 ERR("[notification-thread] Failed to add ust-64 channel monitoring pipe fd to pollset");
313 if (handle
->channel_monitoring_pipes
.kernel_consumer
< 0) {
316 ret
= lttng_poll_add(poll_set
,
317 handle
->channel_monitoring_pipes
.kernel_consumer
,
320 ERR("[notification-thread] Failed to add kernel channel monitoring pipe fd to pollset");
326 lttng_poll_clean(poll_set
);
331 void fini_thread_state(struct notification_thread_state
*state
)
335 if (state
->client_socket_ht
) {
336 ret
= handle_notification_thread_client_disconnect_all(state
);
338 ret
= cds_lfht_destroy(state
->client_socket_ht
, NULL
);
341 if (state
->triggers_ht
) {
342 ret
= handle_notification_thread_trigger_unregister_all(state
);
344 ret
= cds_lfht_destroy(state
->triggers_ht
, NULL
);
347 if (state
->channel_triggers_ht
) {
348 ret
= cds_lfht_destroy(state
->channel_triggers_ht
, NULL
);
351 if (state
->channel_state_ht
) {
352 ret
= cds_lfht_destroy(state
->channel_state_ht
, NULL
);
355 if (state
->notification_trigger_clients_ht
) {
356 ret
= cds_lfht_destroy(state
->notification_trigger_clients_ht
,
360 if (state
->channels_ht
) {
361 ret
= cds_lfht_destroy(state
->channels_ht
,
366 if (state
->notification_channel_socket
>= 0) {
367 notification_channel_socket_destroy(
368 state
->notification_channel_socket
);
370 lttng_poll_clean(&state
->events
);
374 int init_thread_state(struct notification_thread_handle
*handle
,
375 struct notification_thread_state
*state
)
379 memset(state
, 0, sizeof(*state
));
380 state
->notification_channel_socket
= -1;
381 lttng_poll_init(&state
->events
);
383 ret
= notification_channel_socket_create();
387 state
->notification_channel_socket
= ret
;
389 ret
= init_poll_set(&state
->events
, handle
,
390 state
->notification_channel_socket
);
395 DBG("[notification-thread] Listening on notification channel socket");
396 ret
= lttcomm_listen_unix_sock(state
->notification_channel_socket
);
398 ERR("[notification-thread] Listen failed on notification channel socket");
402 state
->client_socket_ht
= cds_lfht_new(DEFAULT_HT_SIZE
, 1, 0,
403 CDS_LFHT_AUTO_RESIZE
| CDS_LFHT_ACCOUNTING
, NULL
);
404 if (!state
->client_socket_ht
) {
408 state
->channel_triggers_ht
= cds_lfht_new(DEFAULT_HT_SIZE
, 1, 0,
409 CDS_LFHT_AUTO_RESIZE
| CDS_LFHT_ACCOUNTING
, NULL
);
410 if (!state
->channel_triggers_ht
) {
414 state
->channel_state_ht
= cds_lfht_new(DEFAULT_HT_SIZE
, 1, 0,
415 CDS_LFHT_AUTO_RESIZE
| CDS_LFHT_ACCOUNTING
, NULL
);
416 if (!state
->channel_state_ht
) {
420 state
->notification_trigger_clients_ht
= cds_lfht_new(DEFAULT_HT_SIZE
,
421 1, 0, CDS_LFHT_AUTO_RESIZE
| CDS_LFHT_ACCOUNTING
, NULL
);
422 if (!state
->notification_trigger_clients_ht
) {
426 state
->channels_ht
= cds_lfht_new(DEFAULT_HT_SIZE
,
427 1, 0, CDS_LFHT_AUTO_RESIZE
| CDS_LFHT_ACCOUNTING
, NULL
);
428 if (!state
->channels_ht
) {
432 state
->triggers_ht
= cds_lfht_new(DEFAULT_HT_SIZE
,
433 1, 0, CDS_LFHT_AUTO_RESIZE
| CDS_LFHT_ACCOUNTING
, NULL
);
434 if (!state
->triggers_ht
) {
440 fini_thread_state(state
);
445 int handle_channel_monitoring_pipe(int fd
, uint32_t revents
,
446 struct notification_thread_handle
*handle
,
447 struct notification_thread_state
*state
)
450 enum lttng_domain_type domain
;
452 if (fd
== handle
->channel_monitoring_pipes
.ust32_consumer
||
453 fd
== handle
->channel_monitoring_pipes
.ust64_consumer
) {
454 domain
= LTTNG_DOMAIN_UST
;
455 } else if (fd
== handle
->channel_monitoring_pipes
.kernel_consumer
) {
456 domain
= LTTNG_DOMAIN_KERNEL
;
461 if (revents
& (LPOLLERR
| LPOLLHUP
| LPOLLRDHUP
)) {
462 ret
= lttng_poll_del(&state
->events
, fd
);
464 ERR("[notification-thread] Failed to remove consumer monitoring pipe from poll set");
469 ret
= handle_notification_thread_channel_sample(
472 ERR("[notification-thread] Consumer sample handling error occurred");
481 * This thread services notification channel clients and commands received
482 * from various lttng-sessiond components over a command queue.
484 void *thread_notification(void *data
)
487 struct notification_thread_handle
*handle
= data
;
488 struct notification_thread_state state
;
490 DBG("[notification-thread] Started notification thread");
493 ERR("[notification-thread] Invalid thread context provided");
497 rcu_register_thread();
500 health_register(health_sessiond
, HEALTH_SESSIOND_TYPE_NOTIFICATION
);
501 health_code_update();
503 ret
= init_thread_state(handle
, &state
);
508 /* Ready to handle client connections. */
509 sessiond_notify_ready();
515 DBG("[notification-thread] Entering poll wait");
516 ret
= lttng_poll_wait(&state
.events
, -1);
517 DBG("[notification-thread] Poll wait returned (%i)", ret
);
521 * Restart interrupted system call.
523 if (errno
== EINTR
) {
526 ERR("[notification-thread] Error encountered during lttng_poll_wait (%i)", ret
);
531 for (i
= 0; i
< fd_count
; i
++) {
532 int fd
= LTTNG_POLL_GETFD(&state
.events
, i
);
533 uint32_t revents
= LTTNG_POLL_GETEV(&state
.events
, i
);
538 DBG("[notification-thread] Handling fd (%i) activity (%u)", fd
, revents
);
540 if (fd
== state
.notification_channel_socket
) {
541 if (revents
& LPOLLIN
) {
542 ret
= handle_notification_thread_client_connect(
548 (LPOLLERR
| LPOLLHUP
| LPOLLRDHUP
)) {
549 ERR("[notification-thread] Notification socket poll error");
552 ERR("[notification-thread] Unexpected poll events %u for notification socket %i", revents
, fd
);
555 } else if (fd
== lttng_pipe_get_readfd(handle
->cmd_queue
.event_pipe
)) {
556 ret
= handle_notification_thread_command(handle
,
559 DBG("[notification-thread] Error encountered while servicing command queue");
561 } else if (ret
> 0) {
564 } else if (fd
== handle
->channel_monitoring_pipes
.ust32_consumer
||
565 fd
== handle
->channel_monitoring_pipes
.ust64_consumer
||
566 fd
== handle
->channel_monitoring_pipes
.kernel_consumer
) {
567 ret
= handle_channel_monitoring_pipe(fd
,
568 revents
, handle
, &state
);
573 /* Activity on a client's socket. */
574 if (revents
& (LPOLLERR
| LPOLLHUP
| LPOLLRDHUP
)) {
576 * It doesn't matter if a command was
577 * pending on the client socket at this
578 * point since it now has no way to
579 * receive the notifications to which
580 * it was subscribing or unsubscribing.
582 ret
= handle_notification_thread_client_disconnect(
588 if (revents
& LPOLLIN
) {
589 ret
= handle_notification_thread_client_in(
596 if (revents
& LPOLLOUT
) {
597 ret
= handle_notification_thread_client_out(
609 fini_thread_state(&state
);
610 health_unregister(health_sessiond
);
611 rcu_thread_offline();
612 rcu_unregister_thread();