sessiond: notification: maintain an id to notification_client ht
[lttng-tools.git] / src / bin / lttng-sessiond / notification-thread.c
CommitLineData
ab0ee2ca 1/*
ab5be9fa 2 * Copyright (C) 2017 Jérémie Galarneau <jeremie.galarneau@efficios.com>
ab0ee2ca 3 *
ab5be9fa 4 * SPDX-License-Identifier: GPL-2.0-only
ab0ee2ca 5 *
ab0ee2ca
JG
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>
ab0ee2ca
JG
18#include <common/align.h>
19#include <common/time.h>
20#include <sys/eventfd.h>
21#include <sys/stat.h>
22#include <time.h>
23#include <signal.h>
24
25#include "notification-thread.h"
26#include "notification-thread-events.h"
27#include "notification-thread-commands.h"
28#include "lttng-sessiond.h"
29#include "health-sessiond.h"
c8a9de5a 30#include "thread.h"
ab0ee2ca
JG
31
32#include <urcu.h>
33#include <urcu/list.h>
34#include <urcu/rculfhash.h>
35
ab0ee2ca
JG
36/*
37 * Destroy the thread data previously created by the init function.
38 */
39void notification_thread_handle_destroy(
40 struct notification_thread_handle *handle)
41{
42 int ret;
ab0ee2ca
JG
43
44 if (!handle) {
45 goto end;
46 }
47
8ada111f 48 assert(cds_list_empty(&handle->cmd_queue.list));
ab0ee2ca 49 pthread_mutex_destroy(&handle->cmd_queue.lock);
c8a9de5a 50 sem_destroy(&handle->ready);
ab0ee2ca 51
814b4934
JR
52 if (handle->cmd_queue.event_pipe) {
53 lttng_pipe_destroy(handle->cmd_queue.event_pipe);
54 }
ab0ee2ca
JG
55 if (handle->channel_monitoring_pipes.ust32_consumer >= 0) {
56 ret = close(handle->channel_monitoring_pipes.ust32_consumer);
57 if (ret) {
58 PERROR("close 32-bit consumer channel monitoring pipe");
59 }
60 }
61 if (handle->channel_monitoring_pipes.ust64_consumer >= 0) {
62 ret = close(handle->channel_monitoring_pipes.ust64_consumer);
63 if (ret) {
64 PERROR("close 64-bit consumer channel monitoring pipe");
65 }
66 }
67 if (handle->channel_monitoring_pipes.kernel_consumer >= 0) {
68 ret = close(handle->channel_monitoring_pipes.kernel_consumer);
69 if (ret) {
70 PERROR("close kernel consumer channel monitoring pipe");
71 }
72 }
73end:
74 free(handle);
75}
76
77struct notification_thread_handle *notification_thread_handle_create(
78 struct lttng_pipe *ust32_channel_monitor_pipe,
79 struct lttng_pipe *ust64_channel_monitor_pipe,
c8a9de5a 80 struct lttng_pipe *kernel_channel_monitor_pipe)
ab0ee2ca
JG
81{
82 int ret;
83 struct notification_thread_handle *handle;
814b4934 84 struct lttng_pipe *event_pipe = NULL;
ab0ee2ca
JG
85
86 handle = zmalloc(sizeof(*handle));
87 if (!handle) {
88 goto end;
89 }
90
c8a9de5a
JG
91 sem_init(&handle->ready, 0, 0);
92
18d08850 93 event_pipe = lttng_pipe_open(FD_CLOEXEC);
814b4934
JR
94 if (!event_pipe) {
95 ERR("event_pipe creation");
ab0ee2ca
JG
96 goto error;
97 }
814b4934
JR
98
99 handle->cmd_queue.event_pipe = event_pipe;
100 event_pipe = NULL;
101
ab0ee2ca
JG
102 CDS_INIT_LIST_HEAD(&handle->cmd_queue.list);
103 ret = pthread_mutex_init(&handle->cmd_queue.lock, NULL);
104 if (ret) {
105 goto error;
106 }
107
108 if (ust32_channel_monitor_pipe) {
109 handle->channel_monitoring_pipes.ust32_consumer =
110 lttng_pipe_release_readfd(
111 ust32_channel_monitor_pipe);
112 if (handle->channel_monitoring_pipes.ust32_consumer < 0) {
113 goto error;
114 }
115 } else {
116 handle->channel_monitoring_pipes.ust32_consumer = -1;
117 }
118 if (ust64_channel_monitor_pipe) {
119 handle->channel_monitoring_pipes.ust64_consumer =
120 lttng_pipe_release_readfd(
121 ust64_channel_monitor_pipe);
122 if (handle->channel_monitoring_pipes.ust64_consumer < 0) {
123 goto error;
124 }
125 } else {
126 handle->channel_monitoring_pipes.ust64_consumer = -1;
127 }
128 if (kernel_channel_monitor_pipe) {
129 handle->channel_monitoring_pipes.kernel_consumer =
130 lttng_pipe_release_readfd(
131 kernel_channel_monitor_pipe);
132 if (handle->channel_monitoring_pipes.kernel_consumer < 0) {
133 goto error;
134 }
135 } else {
136 handle->channel_monitoring_pipes.kernel_consumer = -1;
137 }
138end:
139 return handle;
140error:
814b4934 141 lttng_pipe_destroy(event_pipe);
ab0ee2ca
JG
142 notification_thread_handle_destroy(handle);
143 return NULL;
144}
145
146static
147char *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 {
4f00620d 165 const char *home_path = utils_get_home_dir();
ab0ee2ca
JG
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;
181error:
182 free(sock_path);
183 return NULL;
184}
185
186static
187void 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
208static
209int 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) {
28ab59d0
JR
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);
ab0ee2ca
JG
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;
253error:
254 if (fd >= 0 && close(fd) < 0) {
255 PERROR("close notification channel socket");
256 }
257 free(sock_path);
258 return ret;
259}
260
261static
262int 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 }
814b4934 287 ret = lttng_poll_add(poll_set, lttng_pipe_get_readfd(handle->cmd_queue.event_pipe),
ab0ee2ca
JG
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 }
317end:
318 return ret;
319error:
320 lttng_poll_clean(poll_set);
321 return ret;
322}
323
324static
325void 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 }
ac1889bf
JG
335 if (state->client_id_ht) {
336 ret = cds_lfht_destroy(state->client_id_ht, NULL);
337 assert(!ret);
338 }
ab0ee2ca
JG
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) {
8abe313a
JG
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);
ab0ee2ca
JG
364 assert(!ret);
365 }
ea9a44f0
JG
366 /*
367 * Must be destroyed after all channels have been destroyed.
368 * See comment in struct lttng_session_trigger_list.
369 */
370 if (state->session_triggers_ht) {
371 ret = cds_lfht_destroy(state->session_triggers_ht, NULL);
372 assert(!ret);
373 }
ab0ee2ca
JG
374 if (state->notification_channel_socket >= 0) {
375 notification_channel_socket_destroy(
376 state->notification_channel_socket);
377 }
378 lttng_poll_clean(&state->events);
379}
380
c8a9de5a
JG
381static
382void mark_thread_as_ready(struct notification_thread_handle *handle)
383{
384 DBG("Marking notification thread as ready");
385 sem_post(&handle->ready);
386}
387
388static
389void wait_until_thread_is_ready(struct notification_thread_handle *handle)
390{
391 DBG("Waiting for notification thread to be ready");
392 sem_wait(&handle->ready);
393 DBG("Notification thread is ready");
394}
395
ab0ee2ca
JG
396static
397int init_thread_state(struct notification_thread_handle *handle,
398 struct notification_thread_state *state)
399{
400 int ret;
401
402 memset(state, 0, sizeof(*state));
403 state->notification_channel_socket = -1;
404 lttng_poll_init(&state->events);
405
406 ret = notification_channel_socket_create();
407 if (ret < 0) {
408 goto end;
409 }
410 state->notification_channel_socket = ret;
411
412 ret = init_poll_set(&state->events, handle,
413 state->notification_channel_socket);
414 if (ret) {
415 goto end;
416 }
417
418 DBG("[notification-thread] Listening on notification channel socket");
419 ret = lttcomm_listen_unix_sock(state->notification_channel_socket);
420 if (ret < 0) {
421 ERR("[notification-thread] Listen failed on notification channel socket");
422 goto error;
423 }
424
425 state->client_socket_ht = cds_lfht_new(DEFAULT_HT_SIZE, 1, 0,
426 CDS_LFHT_AUTO_RESIZE | CDS_LFHT_ACCOUNTING, NULL);
427 if (!state->client_socket_ht) {
428 goto error;
429 }
430
ac1889bf
JG
431 state->client_id_ht = cds_lfht_new(DEFAULT_HT_SIZE, 1, 0,
432 CDS_LFHT_AUTO_RESIZE | CDS_LFHT_ACCOUNTING, NULL);
433 if (!state->client_id_ht) {
434 goto error;
435 }
436
ab0ee2ca
JG
437 state->channel_triggers_ht = cds_lfht_new(DEFAULT_HT_SIZE, 1, 0,
438 CDS_LFHT_AUTO_RESIZE | CDS_LFHT_ACCOUNTING, NULL);
439 if (!state->channel_triggers_ht) {
440 goto error;
441 }
442
ea9a44f0
JG
443 state->session_triggers_ht = cds_lfht_new(DEFAULT_HT_SIZE, 1, 0,
444 CDS_LFHT_AUTO_RESIZE | CDS_LFHT_ACCOUNTING, NULL);
445 if (!state->session_triggers_ht) {
446 goto error;
447 }
448
ab0ee2ca
JG
449 state->channel_state_ht = cds_lfht_new(DEFAULT_HT_SIZE, 1, 0,
450 CDS_LFHT_AUTO_RESIZE | CDS_LFHT_ACCOUNTING, NULL);
451 if (!state->channel_state_ht) {
452 goto error;
453 }
454
455 state->notification_trigger_clients_ht = cds_lfht_new(DEFAULT_HT_SIZE,
456 1, 0, CDS_LFHT_AUTO_RESIZE | CDS_LFHT_ACCOUNTING, NULL);
457 if (!state->notification_trigger_clients_ht) {
458 goto error;
459 }
460
461 state->channels_ht = cds_lfht_new(DEFAULT_HT_SIZE,
462 1, 0, CDS_LFHT_AUTO_RESIZE | CDS_LFHT_ACCOUNTING, NULL);
463 if (!state->channels_ht) {
464 goto error;
465 }
8abe313a
JG
466 state->sessions_ht = cds_lfht_new(DEFAULT_HT_SIZE,
467 1, 0, CDS_LFHT_AUTO_RESIZE | CDS_LFHT_ACCOUNTING, NULL);
468 if (!state->sessions_ht) {
469 goto error;
470 }
ab0ee2ca
JG
471 state->triggers_ht = cds_lfht_new(DEFAULT_HT_SIZE,
472 1, 0, CDS_LFHT_AUTO_RESIZE | CDS_LFHT_ACCOUNTING, NULL);
473 if (!state->triggers_ht) {
474 goto error;
475 }
c8a9de5a 476 mark_thread_as_ready(handle);
ab0ee2ca
JG
477end:
478 return 0;
479error:
480 fini_thread_state(state);
481 return -1;
482}
483
484static
485int handle_channel_monitoring_pipe(int fd, uint32_t revents,
486 struct notification_thread_handle *handle,
487 struct notification_thread_state *state)
488{
489 int ret = 0;
490 enum lttng_domain_type domain;
491
492 if (fd == handle->channel_monitoring_pipes.ust32_consumer ||
493 fd == handle->channel_monitoring_pipes.ust64_consumer) {
494 domain = LTTNG_DOMAIN_UST;
495 } else if (fd == handle->channel_monitoring_pipes.kernel_consumer) {
496 domain = LTTNG_DOMAIN_KERNEL;
497 } else {
498 abort();
499 }
500
501 if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
502 ret = lttng_poll_del(&state->events, fd);
503 if (ret) {
504 ERR("[notification-thread] Failed to remove consumer monitoring pipe from poll set");
505 }
506 goto end;
507 }
508
509 ret = handle_notification_thread_channel_sample(
510 state, fd, domain);
511 if (ret) {
4149ace8 512 ERR("[notification-thread] Consumer sample handling error occurred");
ab0ee2ca
JG
513 ret = -1;
514 goto end;
515 }
516end:
517 return ret;
518}
519
520/*
521 * This thread services notification channel clients and commands received
522 * from various lttng-sessiond components over a command queue.
523 */
c8a9de5a 524static
ab0ee2ca
JG
525void *thread_notification(void *data)
526{
527 int ret;
528 struct notification_thread_handle *handle = data;
529 struct notification_thread_state state;
530
531 DBG("[notification-thread] Started notification thread");
532
f620cc28
JG
533 health_register(health_sessiond, HEALTH_SESSIOND_TYPE_NOTIFICATION);
534 rcu_register_thread();
535 rcu_thread_online();
536
ab0ee2ca
JG
537 if (!handle) {
538 ERR("[notification-thread] Invalid thread context provided");
539 goto end;
540 }
541
ab0ee2ca
JG
542 health_code_update();
543
544 ret = init_thread_state(handle, &state);
545 if (ret) {
546 goto end;
547 }
548
ab0ee2ca
JG
549 while (true) {
550 int fd_count, i;
551
552 health_poll_entry();
553 DBG("[notification-thread] Entering poll wait");
554 ret = lttng_poll_wait(&state.events, -1);
555 DBG("[notification-thread] Poll wait returned (%i)", ret);
556 health_poll_exit();
557 if (ret < 0) {
558 /*
559 * Restart interrupted system call.
560 */
561 if (errno == EINTR) {
562 continue;
563 }
564 ERR("[notification-thread] Error encountered during lttng_poll_wait (%i)", ret);
565 goto error;
566 }
567
568 fd_count = ret;
569 for (i = 0; i < fd_count; i++) {
570 int fd = LTTNG_POLL_GETFD(&state.events, i);
571 uint32_t revents = LTTNG_POLL_GETEV(&state.events, i);
572
573 DBG("[notification-thread] Handling fd (%i) activity (%u)", fd, revents);
574
575 if (fd == state.notification_channel_socket) {
576 if (revents & LPOLLIN) {
577 ret = handle_notification_thread_client_connect(
578 &state);
579 if (ret < 0) {
580 goto error;
581 }
582 } else if (revents &
583 (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
584 ERR("[notification-thread] Notification socket poll error");
585 goto error;
586 } else {
587 ERR("[notification-thread] Unexpected poll events %u for notification socket %i", revents, fd);
588 goto error;
589 }
814b4934 590 } else if (fd == lttng_pipe_get_readfd(handle->cmd_queue.event_pipe)) {
ab0ee2ca
JG
591 ret = handle_notification_thread_command(handle,
592 &state);
593 if (ret < 0) {
594 DBG("[notification-thread] Error encountered while servicing command queue");
595 goto error;
596 } else if (ret > 0) {
597 goto exit;
598 }
599 } else if (fd == handle->channel_monitoring_pipes.ust32_consumer ||
600 fd == handle->channel_monitoring_pipes.ust64_consumer ||
601 fd == handle->channel_monitoring_pipes.kernel_consumer) {
602 ret = handle_channel_monitoring_pipe(fd,
603 revents, handle, &state);
604 if (ret) {
605 goto error;
606 }
607 } else {
608 /* Activity on a client's socket. */
609 if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
610 /*
611 * It doesn't matter if a command was
612 * pending on the client socket at this
613 * point since it now has no way to
614 * receive the notifications to which
615 * it was subscribing or unsubscribing.
616 */
617 ret = handle_notification_thread_client_disconnect(
618 fd, &state);
619 if (ret) {
620 goto error;
621 }
622 } else {
623 if (revents & LPOLLIN) {
624 ret = handle_notification_thread_client_in(
625 &state, fd);
626 if (ret) {
627 goto error;
628 }
629 }
630
631 if (revents & LPOLLOUT) {
632 ret = handle_notification_thread_client_out(
633 &state, fd);
634 if (ret) {
635 goto error;
636 }
637 }
638 }
639 }
640 }
641 }
642exit:
643error:
644 fini_thread_state(&state);
f620cc28 645end:
ab0ee2ca
JG
646 rcu_thread_offline();
647 rcu_unregister_thread();
f620cc28 648 health_unregister(health_sessiond);
ab0ee2ca
JG
649 return NULL;
650}
c8a9de5a
JG
651
652static
653bool shutdown_notification_thread(void *thread_data)
654{
655 struct notification_thread_handle *handle = thread_data;
656
657 notification_thread_command_quit(handle);
658 return true;
659}
660
4a91420c
JG
661struct lttng_thread *launch_notification_thread(
662 struct notification_thread_handle *handle)
c8a9de5a
JG
663{
664 struct lttng_thread *thread;
665
666 thread = lttng_thread_create("Notification",
667 thread_notification,
668 shutdown_notification_thread,
669 NULL,
670 handle);
671 if (!thread) {
672 goto error;
673 }
674
675 /*
676 * Wait for the thread to be marked as "ready" before returning
677 * as other subsystems depend on the notification subsystem
678 * (e.g. rotation thread).
679 */
680 wait_until_thread_is_ready(handle);
4a91420c 681 return thread;
c8a9de5a 682error:
4a91420c 683 return NULL;
c8a9de5a 684}
This page took 0.099299 seconds and 4 git commands to generate.