917cec5a2500f56f8f12350b74db94e423dc6928
[lttng-tools.git] / src / bin / lttng-sessiond / notification-thread.c
1 /*
2 * Copyright (C) 2017 Jérémie Galarneau <jeremie.galarneau@efficios.com>
3 *
4 * SPDX-License-Identifier: GPL-2.0-only
5 *
6 */
7
8 #define _LGPL_SOURCE
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>
20 #include <sys/stat.h>
21 #include <time.h>
22 #include <signal.h>
23
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"
29 #include "thread.h"
30
31 #include <urcu.h>
32 #include <urcu/list.h>
33 #include <urcu/rculfhash.h>
34
35 /*
36 * Destroy the thread data previously created by the init function.
37 */
38 void notification_thread_handle_destroy(
39 struct notification_thread_handle *handle)
40 {
41 int ret;
42
43 if (!handle) {
44 goto end;
45 }
46
47 assert(cds_list_empty(&handle->cmd_queue.list));
48 pthread_mutex_destroy(&handle->cmd_queue.lock);
49 sem_destroy(&handle->ready);
50
51 if (handle->cmd_queue.event_pipe) {
52 lttng_pipe_destroy(handle->cmd_queue.event_pipe);
53 }
54 if (handle->channel_monitoring_pipes.ust32_consumer >= 0) {
55 ret = close(handle->channel_monitoring_pipes.ust32_consumer);
56 if (ret) {
57 PERROR("close 32-bit consumer channel monitoring pipe");
58 }
59 }
60 if (handle->channel_monitoring_pipes.ust64_consumer >= 0) {
61 ret = close(handle->channel_monitoring_pipes.ust64_consumer);
62 if (ret) {
63 PERROR("close 64-bit consumer channel monitoring pipe");
64 }
65 }
66 if (handle->channel_monitoring_pipes.kernel_consumer >= 0) {
67 ret = close(handle->channel_monitoring_pipes.kernel_consumer);
68 if (ret) {
69 PERROR("close kernel consumer channel monitoring pipe");
70 }
71 }
72 end:
73 free(handle);
74 }
75
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)
80 {
81 int ret;
82 struct notification_thread_handle *handle;
83 struct lttng_pipe *event_pipe = NULL;
84
85 handle = zmalloc(sizeof(*handle));
86 if (!handle) {
87 goto end;
88 }
89
90 sem_init(&handle->ready, 0, 0);
91
92 event_pipe = lttng_pipe_open(FD_CLOEXEC);
93 if (!event_pipe) {
94 ERR("event_pipe creation");
95 goto error;
96 }
97
98 handle->cmd_queue.event_pipe = event_pipe;
99 event_pipe = NULL;
100
101 CDS_INIT_LIST_HEAD(&handle->cmd_queue.list);
102 ret = pthread_mutex_init(&handle->cmd_queue.lock, NULL);
103 if (ret) {
104 goto error;
105 }
106
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) {
112 goto error;
113 }
114 } else {
115 handle->channel_monitoring_pipes.ust32_consumer = -1;
116 }
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) {
122 goto error;
123 }
124 } else {
125 handle->channel_monitoring_pipes.ust64_consumer = -1;
126 }
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) {
132 goto error;
133 }
134 } else {
135 handle->channel_monitoring_pipes.kernel_consumer = -1;
136 }
137
138 end:
139 return handle;
140 error:
141 lttng_pipe_destroy(event_pipe);
142 notification_thread_handle_destroy(handle);
143 return NULL;
144 }
145
146 static
147 char *get_notification_channel_sock_path(void)
148 {
149 int ret;
150 bool is_root = !getuid();
151 char *sock_path;
152
153 sock_path = zmalloc(LTTNG_PATH_MAX);
154 if (!sock_path) {
155 goto error;
156 }
157
158 if (is_root) {
159 ret = snprintf(sock_path, LTTNG_PATH_MAX,
160 DEFAULT_GLOBAL_NOTIFICATION_CHANNEL_UNIX_SOCK);
161 if (ret < 0) {
162 goto error;
163 }
164 } else {
165 const char *home_path = utils_get_home_dir();
166
167 if (!home_path) {
168 ERR("Can't get HOME directory for socket creation");
169 goto error;
170 }
171
172 ret = snprintf(sock_path, LTTNG_PATH_MAX,
173 DEFAULT_HOME_NOTIFICATION_CHANNEL_UNIX_SOCK,
174 home_path);
175 if (ret < 0) {
176 goto error;
177 }
178 }
179
180 return sock_path;
181 error:
182 free(sock_path);
183 return NULL;
184 }
185
186 static
187 void notification_channel_socket_destroy(int fd)
188 {
189 int ret;
190 char *sock_path = get_notification_channel_sock_path();
191
192 DBG("[notification-thread] Destroying notification channel socket");
193
194 if (sock_path) {
195 ret = unlink(sock_path);
196 free(sock_path);
197 if (ret < 0) {
198 PERROR("unlink notification channel socket");
199 }
200 }
201
202 ret = close(fd);
203 if (ret) {
204 PERROR("close notification channel socket");
205 }
206 }
207
208 static
209 int notification_channel_socket_create(void)
210 {
211 int fd = -1, ret;
212 char *sock_path = get_notification_channel_sock_path();
213
214 DBG("[notification-thread] Creating notification channel UNIX socket at %s",
215 sock_path);
216
217 ret = lttcomm_create_unix_sock(sock_path);
218 if (ret < 0) {
219 ERR("[notification-thread] Failed to create notification socket");
220 goto error;
221 }
222 fd = ret;
223
224 ret = chmod(sock_path, S_IRUSR | S_IWUSR | S_IRGRP | S_IWGRP);
225 if (ret < 0) {
226 ERR("Set file permissions failed: %s", sock_path);
227 PERROR("chmod notification channel socket");
228 goto error;
229 }
230
231 if (getuid() == 0) {
232 gid_t gid;
233
234 ret = utils_get_group_id(config.tracing_group_name.value, true,
235 &gid);
236 if (ret) {
237 /* Default to root group. */
238 gid = 0;
239 }
240
241 ret = chown(sock_path, 0, gid);
242 if (ret) {
243 ERR("Failed to set the notification channel socket's group");
244 ret = -1;
245 goto error;
246 }
247 }
248
249 DBG("[notification-thread] Notification channel UNIX socket created (fd = %i)",
250 fd);
251 free(sock_path);
252 return fd;
253 error:
254 if (fd >= 0 && close(fd) < 0) {
255 PERROR("close notification channel socket");
256 }
257 free(sock_path);
258 return ret;
259 }
260
261 static
262 int init_poll_set(struct lttng_poll_event *poll_set,
263 struct notification_thread_handle *handle,
264 int notification_channel_socket)
265 {
266 int ret;
267
268 /*
269 * Create pollset with size 5:
270 * - notification channel socket (listen for new connections),
271 * - command queue event fd (internal sessiond commands),
272 * - consumerd (32-bit user space) channel monitor pipe,
273 * - consumerd (64-bit user space) channel monitor pipe,
274 * - consumerd (kernel) channel monitor pipe.
275 */
276 ret = lttng_poll_create(poll_set, 5, LTTNG_CLOEXEC);
277 if (ret < 0) {
278 goto end;
279 }
280
281 ret = lttng_poll_add(poll_set, notification_channel_socket,
282 LPOLLIN | LPOLLERR | LPOLLHUP | LPOLLRDHUP);
283 if (ret < 0) {
284 ERR("[notification-thread] Failed to add notification channel socket to pollset");
285 goto error;
286 }
287 ret = lttng_poll_add(poll_set, lttng_pipe_get_readfd(handle->cmd_queue.event_pipe),
288 LPOLLIN | LPOLLERR);
289 if (ret < 0) {
290 ERR("[notification-thread] Failed to add notification command queue event fd to pollset");
291 goto error;
292 }
293 ret = lttng_poll_add(poll_set,
294 handle->channel_monitoring_pipes.ust32_consumer,
295 LPOLLIN | LPOLLERR);
296 if (ret < 0) {
297 ERR("[notification-thread] Failed to add ust-32 channel monitoring pipe fd to pollset");
298 goto error;
299 }
300 ret = lttng_poll_add(poll_set,
301 handle->channel_monitoring_pipes.ust64_consumer,
302 LPOLLIN | LPOLLERR);
303 if (ret < 0) {
304 ERR("[notification-thread] Failed to add ust-64 channel monitoring pipe fd to pollset");
305 goto error;
306 }
307 if (handle->channel_monitoring_pipes.kernel_consumer < 0) {
308 goto end;
309 }
310 ret = lttng_poll_add(poll_set,
311 handle->channel_monitoring_pipes.kernel_consumer,
312 LPOLLIN | LPOLLERR);
313 if (ret < 0) {
314 ERR("[notification-thread] Failed to add kernel channel monitoring pipe fd to pollset");
315 goto error;
316 }
317 end:
318 return ret;
319 error:
320 lttng_poll_clean(poll_set);
321 return ret;
322 }
323
324 static
325 void fini_thread_state(struct notification_thread_state *state)
326 {
327 int ret;
328
329 if (state->client_socket_ht) {
330 ret = handle_notification_thread_client_disconnect_all(state);
331 assert(!ret);
332 ret = cds_lfht_destroy(state->client_socket_ht, NULL);
333 assert(!ret);
334 }
335 if (state->client_id_ht) {
336 ret = cds_lfht_destroy(state->client_id_ht, NULL);
337 assert(!ret);
338 }
339 if (state->triggers_ht) {
340 ret = handle_notification_thread_trigger_unregister_all(state);
341 assert(!ret);
342 ret = cds_lfht_destroy(state->triggers_ht, NULL);
343 assert(!ret);
344 }
345 if (state->channel_triggers_ht) {
346 ret = cds_lfht_destroy(state->channel_triggers_ht, NULL);
347 assert(!ret);
348 }
349 if (state->channel_state_ht) {
350 ret = cds_lfht_destroy(state->channel_state_ht, NULL);
351 assert(!ret);
352 }
353 if (state->notification_trigger_clients_ht) {
354 ret = cds_lfht_destroy(state->notification_trigger_clients_ht,
355 NULL);
356 assert(!ret);
357 }
358 if (state->channels_ht) {
359 ret = cds_lfht_destroy(state->channels_ht, NULL);
360 assert(!ret);
361 }
362 if (state->sessions_ht) {
363 ret = cds_lfht_destroy(state->sessions_ht, NULL);
364 assert(!ret);
365 }
366 if (state->triggers_by_name_uid_ht) {
367 ret = cds_lfht_destroy(state->triggers_by_name_uid_ht, NULL);
368 assert(!ret);
369 }
370 /*
371 * Must be destroyed after all channels have been destroyed.
372 * See comment in struct lttng_session_trigger_list.
373 */
374 if (state->session_triggers_ht) {
375 ret = cds_lfht_destroy(state->session_triggers_ht, NULL);
376 assert(!ret);
377 }
378 if (state->notification_channel_socket >= 0) {
379 notification_channel_socket_destroy(
380 state->notification_channel_socket);
381 }
382
383 assert(cds_list_empty(&state->tracer_event_sources_list));
384
385 if (state->executor) {
386 action_executor_destroy(state->executor);
387 }
388 lttng_poll_clean(&state->events);
389 }
390
391 static
392 void mark_thread_as_ready(struct notification_thread_handle *handle)
393 {
394 DBG("Marking notification thread as ready");
395 sem_post(&handle->ready);
396 }
397
398 static
399 void wait_until_thread_is_ready(struct notification_thread_handle *handle)
400 {
401 DBG("Waiting for notification thread to be ready");
402 sem_wait(&handle->ready);
403 DBG("Notification thread is ready");
404 }
405
406 static
407 int init_thread_state(struct notification_thread_handle *handle,
408 struct notification_thread_state *state)
409 {
410 int ret;
411
412 memset(state, 0, sizeof(*state));
413 state->notification_channel_socket = -1;
414 state->trigger_id.next_tracer_token = 1;
415 lttng_poll_init(&state->events);
416
417 ret = notification_channel_socket_create();
418 if (ret < 0) {
419 goto end;
420 }
421 state->notification_channel_socket = ret;
422
423 ret = init_poll_set(&state->events, handle,
424 state->notification_channel_socket);
425 if (ret) {
426 goto end;
427 }
428
429 DBG("[notification-thread] Listening on notification channel socket");
430 ret = lttcomm_listen_unix_sock(state->notification_channel_socket);
431 if (ret < 0) {
432 ERR("[notification-thread] Listen failed on notification channel socket");
433 goto error;
434 }
435
436 state->client_socket_ht = cds_lfht_new(DEFAULT_HT_SIZE, 1, 0,
437 CDS_LFHT_AUTO_RESIZE | CDS_LFHT_ACCOUNTING, NULL);
438 if (!state->client_socket_ht) {
439 goto error;
440 }
441
442 state->client_id_ht = cds_lfht_new(DEFAULT_HT_SIZE, 1, 0,
443 CDS_LFHT_AUTO_RESIZE | CDS_LFHT_ACCOUNTING, NULL);
444 if (!state->client_id_ht) {
445 goto error;
446 }
447
448 state->channel_triggers_ht = cds_lfht_new(DEFAULT_HT_SIZE, 1, 0,
449 CDS_LFHT_AUTO_RESIZE | CDS_LFHT_ACCOUNTING, NULL);
450 if (!state->channel_triggers_ht) {
451 goto error;
452 }
453
454 state->session_triggers_ht = cds_lfht_new(DEFAULT_HT_SIZE, 1, 0,
455 CDS_LFHT_AUTO_RESIZE | CDS_LFHT_ACCOUNTING, NULL);
456 if (!state->session_triggers_ht) {
457 goto error;
458 }
459
460 state->channel_state_ht = cds_lfht_new(DEFAULT_HT_SIZE, 1, 0,
461 CDS_LFHT_AUTO_RESIZE | CDS_LFHT_ACCOUNTING, NULL);
462 if (!state->channel_state_ht) {
463 goto error;
464 }
465
466 state->notification_trigger_clients_ht = cds_lfht_new(DEFAULT_HT_SIZE,
467 1, 0, CDS_LFHT_AUTO_RESIZE | CDS_LFHT_ACCOUNTING, NULL);
468 if (!state->notification_trigger_clients_ht) {
469 goto error;
470 }
471
472 state->channels_ht = cds_lfht_new(DEFAULT_HT_SIZE,
473 1, 0, CDS_LFHT_AUTO_RESIZE | CDS_LFHT_ACCOUNTING, NULL);
474 if (!state->channels_ht) {
475 goto error;
476 }
477 state->sessions_ht = cds_lfht_new(DEFAULT_HT_SIZE,
478 1, 0, CDS_LFHT_AUTO_RESIZE | CDS_LFHT_ACCOUNTING, NULL);
479 if (!state->sessions_ht) {
480 goto error;
481 }
482 state->triggers_ht = cds_lfht_new(DEFAULT_HT_SIZE,
483 1, 0, CDS_LFHT_AUTO_RESIZE | CDS_LFHT_ACCOUNTING, NULL);
484 if (!state->triggers_ht) {
485 goto error;
486 }
487 state->triggers_by_name_uid_ht = cds_lfht_new(DEFAULT_HT_SIZE,
488 1, 0, CDS_LFHT_AUTO_RESIZE | CDS_LFHT_ACCOUNTING, NULL);
489 if (!state->triggers_by_name_uid_ht) {
490 goto error;
491 }
492
493 CDS_INIT_LIST_HEAD(&state->tracer_event_sources_list);
494
495 state->executor = action_executor_create(handle);
496 if (!state->executor) {
497 goto error;
498 }
499 mark_thread_as_ready(handle);
500 end:
501 return 0;
502 error:
503 fini_thread_state(state);
504 return -1;
505 }
506
507 static
508 int handle_channel_monitoring_pipe(int fd, uint32_t revents,
509 struct notification_thread_handle *handle,
510 struct notification_thread_state *state)
511 {
512 int ret = 0;
513 enum lttng_domain_type domain;
514
515 if (fd == handle->channel_monitoring_pipes.ust32_consumer ||
516 fd == handle->channel_monitoring_pipes.ust64_consumer) {
517 domain = LTTNG_DOMAIN_UST;
518 } else if (fd == handle->channel_monitoring_pipes.kernel_consumer) {
519 domain = LTTNG_DOMAIN_KERNEL;
520 } else {
521 abort();
522 }
523
524 if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
525 ret = lttng_poll_del(&state->events, fd);
526 if (ret) {
527 ERR("[notification-thread] Failed to remove consumer monitoring pipe from poll set");
528 }
529 goto end;
530 }
531
532 ret = handle_notification_thread_channel_sample(
533 state, fd, domain);
534 if (ret) {
535 ERR("[notification-thread] Consumer sample handling error occurred");
536 ret = -1;
537 goto end;
538 }
539 end:
540 return ret;
541 }
542
543 /*
544 * This thread services notification channel clients and commands received
545 * from various lttng-sessiond components over a command queue.
546 */
547 static
548 void *thread_notification(void *data)
549 {
550 int ret;
551 struct notification_thread_handle *handle = data;
552 struct notification_thread_state state;
553
554 DBG("[notification-thread] Started notification thread");
555
556 health_register(health_sessiond, HEALTH_SESSIOND_TYPE_NOTIFICATION);
557 rcu_register_thread();
558 rcu_thread_online();
559
560 if (!handle) {
561 ERR("[notification-thread] Invalid thread context provided");
562 goto end;
563 }
564
565 health_code_update();
566
567 ret = init_thread_state(handle, &state);
568 if (ret) {
569 goto end;
570 }
571
572 while (true) {
573 int fd_count, i;
574
575 health_poll_entry();
576 DBG("[notification-thread] Entering poll wait");
577 ret = lttng_poll_wait(&state.events, -1);
578 DBG("[notification-thread] Poll wait returned (%i)", ret);
579 health_poll_exit();
580 if (ret < 0) {
581 /*
582 * Restart interrupted system call.
583 */
584 if (errno == EINTR) {
585 continue;
586 }
587 ERR("[notification-thread] Error encountered during lttng_poll_wait (%i)", ret);
588 goto error;
589 }
590
591 fd_count = ret;
592 for (i = 0; i < fd_count; i++) {
593 int fd = LTTNG_POLL_GETFD(&state.events, i);
594 uint32_t revents = LTTNG_POLL_GETEV(&state.events, i);
595
596 DBG("[notification-thread] Handling fd (%i) activity (%u)", fd, revents);
597
598 if (fd == state.notification_channel_socket) {
599 if (revents & LPOLLIN) {
600 ret = handle_notification_thread_client_connect(
601 &state);
602 if (ret < 0) {
603 goto error;
604 }
605 } else if (revents &
606 (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
607 ERR("[notification-thread] Notification socket poll error");
608 goto error;
609 } else {
610 ERR("[notification-thread] Unexpected poll events %u for notification socket %i", revents, fd);
611 goto error;
612 }
613 } else if (fd == lttng_pipe_get_readfd(handle->cmd_queue.event_pipe)) {
614 ret = handle_notification_thread_command(handle,
615 &state);
616 if (ret < 0) {
617 DBG("[notification-thread] Error encountered while servicing command queue");
618 goto error;
619 } else if (ret > 0) {
620 goto exit;
621 }
622 } else if (fd == handle->channel_monitoring_pipes.ust32_consumer ||
623 fd == handle->channel_monitoring_pipes.ust64_consumer ||
624 fd == handle->channel_monitoring_pipes.kernel_consumer) {
625 ret = handle_channel_monitoring_pipe(fd,
626 revents, handle, &state);
627 if (ret) {
628 goto error;
629 }
630 } else {
631 /* Activity on a client's socket. */
632 if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
633 /*
634 * It doesn't matter if a command was
635 * pending on the client socket at this
636 * point since it now has no way to
637 * receive the notifications to which
638 * it was subscribing or unsubscribing.
639 */
640 ret = handle_notification_thread_client_disconnect(
641 fd, &state);
642 if (ret) {
643 goto error;
644 }
645 } else {
646 if (revents & LPOLLIN) {
647 ret = handle_notification_thread_client_in(
648 &state, fd);
649 if (ret) {
650 goto error;
651 }
652 }
653
654 if (revents & LPOLLOUT) {
655 ret = handle_notification_thread_client_out(
656 &state, fd);
657 if (ret) {
658 goto error;
659 }
660 }
661 }
662 }
663 }
664 }
665 exit:
666 error:
667 fini_thread_state(&state);
668 end:
669 rcu_thread_offline();
670 rcu_unregister_thread();
671 health_unregister(health_sessiond);
672 return NULL;
673 }
674
675 static
676 bool shutdown_notification_thread(void *thread_data)
677 {
678 struct notification_thread_handle *handle = thread_data;
679
680 notification_thread_command_quit(handle);
681 return true;
682 }
683
684 struct lttng_thread *launch_notification_thread(
685 struct notification_thread_handle *handle)
686 {
687 struct lttng_thread *thread;
688
689 thread = lttng_thread_create("Notification",
690 thread_notification,
691 shutdown_notification_thread,
692 NULL,
693 handle);
694 if (!thread) {
695 goto error;
696 }
697
698 /*
699 * Wait for the thread to be marked as "ready" before returning
700 * as other subsystems depend on the notification subsystem
701 * (e.g. rotation thread).
702 */
703 wait_until_thread_is_ready(handle);
704 return thread;
705 error:
706 return NULL;
707 }
This page took 0.066052 seconds and 3 git commands to generate.