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