sessiond: Replace uses of session_trylock_list by a dedicated assert macro
[lttng-tools.git] / src / bin / lttng-sessiond / notification-thread.cpp
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.hpp>
11 #include <lttng/notification/notification-internal.hpp>
12 #include <lttng/condition/condition-internal.hpp>
13 #include <lttng/condition/buffer-usage-internal.hpp>
14 #include <common/error.hpp>
15 #include <common/config/session-config.hpp>
16 #include <common/defaults.hpp>
17 #include <common/utils.hpp>
18 #include <common/align.hpp>
19 #include <common/time.hpp>
20 #include <sys/stat.h>
21 #include <sys/eventfd.h>
22 #include <time.h>
23 #include <signal.h>
24
25 #include "notification-thread.hpp"
26 #include "notification-thread-events.hpp"
27 #include "notification-thread-commands.hpp"
28 #include "lttng-sessiond.hpp"
29 #include "health-sessiond.hpp"
30 #include "thread.hpp"
31 #include "testpoint.hpp"
32
33 #include "kernel.hpp"
34 #include <common/kernel-ctl/kernel-ctl.hpp>
35
36 #include <urcu.h>
37 #include <urcu/list.h>
38 #include <urcu/rculfhash.h>
39
40 /*
41 * Flag used to temporarily pause data consumption from testpoints.
42 *
43 * This variable is dlsym-ed from a test, so needs to be exported.
44 */
45 LTTNG_EXPORT int notifier_consumption_paused;
46
47 /*
48 * Destroy the thread data previously created by the init function.
49 */
50 void notification_thread_handle_destroy(
51 struct notification_thread_handle *handle)
52 {
53 int ret;
54
55 if (!handle) {
56 goto end;
57 }
58
59 LTTNG_ASSERT(cds_list_empty(&handle->cmd_queue.list));
60 pthread_mutex_destroy(&handle->cmd_queue.lock);
61 sem_destroy(&handle->ready);
62
63 if (handle->cmd_queue.event_fd >= 0) {
64 ret = close(handle->cmd_queue.event_fd);
65 if (ret < 0) {
66 PERROR("Failed to close notification command queue event fd");
67 }
68 }
69 if (handle->channel_monitoring_pipes.ust32_consumer >= 0) {
70 ret = close(handle->channel_monitoring_pipes.ust32_consumer);
71 if (ret) {
72 PERROR("close 32-bit consumer channel monitoring pipe");
73 }
74 }
75 if (handle->channel_monitoring_pipes.ust64_consumer >= 0) {
76 ret = close(handle->channel_monitoring_pipes.ust64_consumer);
77 if (ret) {
78 PERROR("close 64-bit consumer channel monitoring pipe");
79 }
80 }
81 if (handle->channel_monitoring_pipes.kernel_consumer >= 0) {
82 ret = close(handle->channel_monitoring_pipes.kernel_consumer);
83 if (ret) {
84 PERROR("close kernel consumer channel monitoring pipe");
85 }
86 }
87
88 end:
89 free(handle);
90 }
91
92 struct notification_thread_handle *notification_thread_handle_create(
93 struct lttng_pipe *ust32_channel_monitor_pipe,
94 struct lttng_pipe *ust64_channel_monitor_pipe,
95 struct lttng_pipe *kernel_channel_monitor_pipe)
96 {
97 int ret;
98 struct notification_thread_handle *handle;
99 int event_fd = -1;
100
101 handle = zmalloc<notification_thread_handle>();
102 if (!handle) {
103 goto end;
104 }
105
106 sem_init(&handle->ready, 0, 0);
107
108 event_fd = eventfd(0, EFD_CLOEXEC | EFD_SEMAPHORE);
109 if (event_fd < 0) {
110 PERROR("event_fd creation");
111 goto error;
112 }
113
114 handle->cmd_queue.event_fd = event_fd;
115
116 CDS_INIT_LIST_HEAD(&handle->cmd_queue.list);
117 ret = pthread_mutex_init(&handle->cmd_queue.lock, NULL);
118 if (ret) {
119 goto error;
120 }
121
122 if (ust32_channel_monitor_pipe) {
123 handle->channel_monitoring_pipes.ust32_consumer =
124 lttng_pipe_release_readfd(
125 ust32_channel_monitor_pipe);
126 if (handle->channel_monitoring_pipes.ust32_consumer < 0) {
127 goto error;
128 }
129 } else {
130 handle->channel_monitoring_pipes.ust32_consumer = -1;
131 }
132 if (ust64_channel_monitor_pipe) {
133 handle->channel_monitoring_pipes.ust64_consumer =
134 lttng_pipe_release_readfd(
135 ust64_channel_monitor_pipe);
136 if (handle->channel_monitoring_pipes.ust64_consumer < 0) {
137 goto error;
138 }
139 } else {
140 handle->channel_monitoring_pipes.ust64_consumer = -1;
141 }
142 if (kernel_channel_monitor_pipe) {
143 handle->channel_monitoring_pipes.kernel_consumer =
144 lttng_pipe_release_readfd(
145 kernel_channel_monitor_pipe);
146 if (handle->channel_monitoring_pipes.kernel_consumer < 0) {
147 goto error;
148 }
149 } else {
150 handle->channel_monitoring_pipes.kernel_consumer = -1;
151 }
152
153 end:
154 return handle;
155 error:
156 notification_thread_handle_destroy(handle);
157 return NULL;
158 }
159
160 static
161 char *get_notification_channel_sock_path(void)
162 {
163 int ret;
164 bool is_root = !getuid();
165 char *sock_path;
166
167 sock_path = calloc<char>(LTTNG_PATH_MAX);
168 if (!sock_path) {
169 goto error;
170 }
171
172 if (is_root) {
173 ret = snprintf(sock_path, LTTNG_PATH_MAX,
174 DEFAULT_GLOBAL_NOTIFICATION_CHANNEL_UNIX_SOCK);
175 if (ret < 0) {
176 goto error;
177 }
178 } else {
179 const char *home_path = utils_get_home_dir();
180
181 if (!home_path) {
182 ERR("Can't get HOME directory for socket creation");
183 goto error;
184 }
185
186 ret = snprintf(sock_path, LTTNG_PATH_MAX,
187 DEFAULT_HOME_NOTIFICATION_CHANNEL_UNIX_SOCK,
188 home_path);
189 if (ret < 0) {
190 goto error;
191 }
192 }
193
194 return sock_path;
195 error:
196 free(sock_path);
197 return NULL;
198 }
199
200 static
201 void notification_channel_socket_destroy(int fd)
202 {
203 int ret;
204 char *sock_path = get_notification_channel_sock_path();
205
206 DBG("Destroying notification channel socket");
207
208 if (sock_path) {
209 ret = unlink(sock_path);
210 free(sock_path);
211 if (ret < 0) {
212 PERROR("unlink notification channel socket");
213 }
214 }
215
216 ret = close(fd);
217 if (ret) {
218 PERROR("close notification channel socket");
219 }
220 }
221
222 static
223 int notification_channel_socket_create(void)
224 {
225 int fd = -1, ret;
226 char *sock_path = get_notification_channel_sock_path();
227
228 DBG("Creating notification channel UNIX socket at %s",
229 sock_path);
230
231 ret = lttcomm_create_unix_sock(sock_path);
232 if (ret < 0) {
233 ERR("Failed to create notification socket");
234 goto error;
235 }
236 fd = ret;
237
238 ret = chmod(sock_path, S_IRUSR | S_IWUSR | S_IRGRP | S_IWGRP);
239 if (ret < 0) {
240 ERR("Set file permissions failed: %s", sock_path);
241 PERROR("chmod notification channel socket");
242 goto error;
243 }
244
245 if (getuid() == 0) {
246 gid_t gid;
247
248 ret = utils_get_group_id(the_config.tracing_group_name.value,
249 true, &gid);
250 if (ret) {
251 /* Default to root group. */
252 gid = 0;
253 }
254
255 ret = chown(sock_path, 0, gid);
256 if (ret) {
257 ERR("Failed to set the notification channel socket's group");
258 ret = -1;
259 goto error;
260 }
261 }
262
263 DBG("Notification channel UNIX socket created (fd = %i)",
264 fd);
265 free(sock_path);
266 return fd;
267 error:
268 if (fd >= 0 && close(fd) < 0) {
269 PERROR("close notification channel socket");
270 }
271 free(sock_path);
272 return ret;
273 }
274
275 static
276 int init_poll_set(struct lttng_poll_event *poll_set,
277 struct notification_thread_handle *handle,
278 int notification_channel_socket)
279 {
280 int ret;
281
282 /*
283 * Create pollset with size 5:
284 * - notification channel socket (listen for new connections),
285 * - command queue event fd (internal sessiond commands),
286 * - consumerd (32-bit user space) channel monitor pipe,
287 * - consumerd (64-bit user space) channel monitor pipe,
288 * - consumerd (kernel) channel monitor pipe.
289 */
290 ret = lttng_poll_create(poll_set, 5, LTTNG_CLOEXEC);
291 if (ret < 0) {
292 goto end;
293 }
294
295 ret = lttng_poll_add(poll_set, notification_channel_socket,
296 LPOLLIN | LPOLLERR | LPOLLHUP | LPOLLRDHUP);
297 if (ret < 0) {
298 ERR("Failed to add notification channel socket to pollset");
299 goto error;
300 }
301 ret = lttng_poll_add(poll_set, handle->cmd_queue.event_fd,
302 LPOLLIN | LPOLLERR);
303 if (ret < 0) {
304 ERR("Failed to add notification command queue event fd to pollset");
305 goto error;
306 }
307 ret = lttng_poll_add(poll_set,
308 handle->channel_monitoring_pipes.ust32_consumer,
309 LPOLLIN | LPOLLERR);
310 if (ret < 0) {
311 ERR("Failed to add ust-32 channel monitoring pipe fd to pollset");
312 goto error;
313 }
314 ret = lttng_poll_add(poll_set,
315 handle->channel_monitoring_pipes.ust64_consumer,
316 LPOLLIN | LPOLLERR);
317 if (ret < 0) {
318 ERR("Failed to add ust-64 channel monitoring pipe fd to pollset");
319 goto error;
320 }
321 if (handle->channel_monitoring_pipes.kernel_consumer < 0) {
322 goto end;
323 }
324 ret = lttng_poll_add(poll_set,
325 handle->channel_monitoring_pipes.kernel_consumer,
326 LPOLLIN | LPOLLERR);
327 if (ret < 0) {
328 ERR("Failed to add kernel channel monitoring pipe fd to pollset");
329 goto error;
330 }
331 end:
332 return ret;
333 error:
334 lttng_poll_clean(poll_set);
335 return ret;
336 }
337
338 static
339 void fini_thread_state(struct notification_thread_state *state)
340 {
341 int ret;
342
343 if (state->client_socket_ht) {
344 ret = handle_notification_thread_client_disconnect_all(state);
345 LTTNG_ASSERT(!ret);
346 ret = cds_lfht_destroy(state->client_socket_ht, NULL);
347 LTTNG_ASSERT(!ret);
348 }
349 if (state->client_id_ht) {
350 ret = cds_lfht_destroy(state->client_id_ht, NULL);
351 LTTNG_ASSERT(!ret);
352 }
353 if (state->triggers_ht) {
354 ret = handle_notification_thread_trigger_unregister_all(state);
355 LTTNG_ASSERT(!ret);
356 ret = cds_lfht_destroy(state->triggers_ht, NULL);
357 LTTNG_ASSERT(!ret);
358 }
359 if (state->channel_triggers_ht) {
360 ret = cds_lfht_destroy(state->channel_triggers_ht, NULL);
361 LTTNG_ASSERT(!ret);
362 }
363 if (state->channel_state_ht) {
364 ret = cds_lfht_destroy(state->channel_state_ht, NULL);
365 LTTNG_ASSERT(!ret);
366 }
367 if (state->notification_trigger_clients_ht) {
368 ret = cds_lfht_destroy(state->notification_trigger_clients_ht,
369 NULL);
370 LTTNG_ASSERT(!ret);
371 }
372 if (state->channels_ht) {
373 ret = cds_lfht_destroy(state->channels_ht, NULL);
374 LTTNG_ASSERT(!ret);
375 }
376 if (state->sessions_ht) {
377 ret = cds_lfht_destroy(state->sessions_ht, NULL);
378 LTTNG_ASSERT(!ret);
379 }
380 if (state->triggers_by_name_uid_ht) {
381 ret = cds_lfht_destroy(state->triggers_by_name_uid_ht, NULL);
382 LTTNG_ASSERT(!ret);
383 }
384 if (state->trigger_tokens_ht) {
385 ret = cds_lfht_destroy(state->trigger_tokens_ht, NULL);
386 LTTNG_ASSERT(!ret);
387 }
388 /*
389 * Must be destroyed after all channels have been destroyed.
390 * See comment in struct lttng_session_trigger_list.
391 */
392 if (state->session_triggers_ht) {
393 ret = cds_lfht_destroy(state->session_triggers_ht, NULL);
394 LTTNG_ASSERT(!ret);
395 }
396 if (state->notification_channel_socket >= 0) {
397 notification_channel_socket_destroy(
398 state->notification_channel_socket);
399 }
400
401 LTTNG_ASSERT(cds_list_empty(&state->tracer_event_sources_list));
402
403 if (state->executor) {
404 action_executor_destroy(state->executor);
405 }
406 lttng_poll_clean(&state->events);
407 }
408
409 static
410 void mark_thread_as_ready(struct notification_thread_handle *handle)
411 {
412 DBG("Marking notification thread as ready");
413 sem_post(&handle->ready);
414 }
415
416 static
417 void wait_until_thread_is_ready(struct notification_thread_handle *handle)
418 {
419 DBG("Waiting for notification thread to be ready");
420 sem_wait(&handle->ready);
421 DBG("Notification thread is ready");
422 }
423
424 static
425 int init_thread_state(struct notification_thread_handle *handle,
426 struct notification_thread_state *state)
427 {
428 int ret;
429
430 memset(state, 0, sizeof(*state));
431 state->notification_channel_socket = -1;
432 state->trigger_id.next_tracer_token = 1;
433 lttng_poll_init(&state->events);
434
435 ret = notification_channel_socket_create();
436 if (ret < 0) {
437 goto end;
438 }
439 state->notification_channel_socket = ret;
440
441 ret = init_poll_set(&state->events, handle,
442 state->notification_channel_socket);
443 if (ret) {
444 goto end;
445 }
446
447 DBG("Listening on notification channel socket");
448 ret = lttcomm_listen_unix_sock(state->notification_channel_socket);
449 if (ret < 0) {
450 ERR("Listen failed on notification channel socket");
451 goto error;
452 }
453
454 state->client_socket_ht = cds_lfht_new(DEFAULT_HT_SIZE, 1, 0,
455 CDS_LFHT_AUTO_RESIZE | CDS_LFHT_ACCOUNTING, NULL);
456 if (!state->client_socket_ht) {
457 goto error;
458 }
459
460 state->client_id_ht = cds_lfht_new(DEFAULT_HT_SIZE, 1, 0,
461 CDS_LFHT_AUTO_RESIZE | CDS_LFHT_ACCOUNTING, NULL);
462 if (!state->client_id_ht) {
463 goto error;
464 }
465
466 state->channel_triggers_ht = cds_lfht_new(DEFAULT_HT_SIZE, 1, 0,
467 CDS_LFHT_AUTO_RESIZE | CDS_LFHT_ACCOUNTING, NULL);
468 if (!state->channel_triggers_ht) {
469 goto error;
470 }
471
472 state->session_triggers_ht = cds_lfht_new(DEFAULT_HT_SIZE, 1, 0,
473 CDS_LFHT_AUTO_RESIZE | CDS_LFHT_ACCOUNTING, NULL);
474 if (!state->session_triggers_ht) {
475 goto error;
476 }
477
478 state->channel_state_ht = cds_lfht_new(DEFAULT_HT_SIZE, 1, 0,
479 CDS_LFHT_AUTO_RESIZE | CDS_LFHT_ACCOUNTING, NULL);
480 if (!state->channel_state_ht) {
481 goto error;
482 }
483
484 state->notification_trigger_clients_ht = cds_lfht_new(DEFAULT_HT_SIZE,
485 1, 0, CDS_LFHT_AUTO_RESIZE | CDS_LFHT_ACCOUNTING, NULL);
486 if (!state->notification_trigger_clients_ht) {
487 goto error;
488 }
489
490 state->channels_ht = cds_lfht_new(DEFAULT_HT_SIZE,
491 1, 0, CDS_LFHT_AUTO_RESIZE | CDS_LFHT_ACCOUNTING, NULL);
492 if (!state->channels_ht) {
493 goto error;
494 }
495 state->sessions_ht = cds_lfht_new(DEFAULT_HT_SIZE,
496 1, 0, CDS_LFHT_AUTO_RESIZE | CDS_LFHT_ACCOUNTING, NULL);
497 if (!state->sessions_ht) {
498 goto error;
499 }
500 state->triggers_ht = cds_lfht_new(DEFAULT_HT_SIZE,
501 1, 0, CDS_LFHT_AUTO_RESIZE | CDS_LFHT_ACCOUNTING, NULL);
502 if (!state->triggers_ht) {
503 goto error;
504 }
505 state->triggers_by_name_uid_ht = cds_lfht_new(DEFAULT_HT_SIZE,
506 1, 0, CDS_LFHT_AUTO_RESIZE | CDS_LFHT_ACCOUNTING, NULL);
507 if (!state->triggers_by_name_uid_ht) {
508 goto error;
509 }
510
511 state->trigger_tokens_ht = cds_lfht_new(DEFAULT_HT_SIZE,
512 1, 0, CDS_LFHT_AUTO_RESIZE | CDS_LFHT_ACCOUNTING, NULL);
513 if (!state->trigger_tokens_ht) {
514 goto error;
515 }
516
517 CDS_INIT_LIST_HEAD(&state->tracer_event_sources_list);
518
519 state->executor = action_executor_create(handle);
520 if (!state->executor) {
521 goto error;
522 }
523
524 state->restart_poll = false;
525
526 mark_thread_as_ready(handle);
527 end:
528 return 0;
529 error:
530 fini_thread_state(state);
531 return -1;
532 }
533
534 static
535 int handle_channel_monitoring_pipe(int fd, uint32_t revents,
536 struct notification_thread_handle *handle,
537 struct notification_thread_state *state)
538 {
539 int ret = 0;
540 enum lttng_domain_type domain;
541
542 if (fd == handle->channel_monitoring_pipes.ust32_consumer ||
543 fd == handle->channel_monitoring_pipes.ust64_consumer) {
544 domain = LTTNG_DOMAIN_UST;
545 } else if (fd == handle->channel_monitoring_pipes.kernel_consumer) {
546 domain = LTTNG_DOMAIN_KERNEL;
547 } else {
548 abort();
549 }
550
551 if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
552 ret = lttng_poll_del(&state->events, fd);
553 if (ret) {
554 ERR("Failed to remove consumer monitoring pipe from poll set");
555 }
556 goto end;
557 }
558
559 ret = handle_notification_thread_channel_sample(
560 state, fd, domain);
561 if (ret) {
562 ERR("Consumer sample handling error occurred");
563 ret = -1;
564 goto end;
565 }
566 end:
567 return ret;
568 }
569
570 static int handle_event_notification_pipe(int event_source_fd,
571 enum lttng_domain_type domain,
572 uint32_t revents,
573 struct notification_thread_state *state)
574 {
575 int ret = 0;
576
577 if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
578 ret = handle_notification_thread_tracer_event_source_died(
579 state, event_source_fd);
580 if (ret) {
581 ERR("Failed to remove event notification pipe from poll set: fd = %d",
582 event_source_fd);
583 }
584 goto end;
585 }
586
587 if (testpoint(sessiond_handle_notifier_event_pipe)) {
588 ret = 0;
589 goto end;
590 }
591
592 if (caa_unlikely(notifier_consumption_paused)) {
593 DBG("Event notifier notification consumption paused, sleeping...");
594 sleep(1);
595 goto end;
596 }
597
598 ret = handle_notification_thread_event_notification(
599 state, event_source_fd, domain);
600 if (ret) {
601 ERR("Event notification handling error occurred for fd: %d",
602 event_source_fd);
603 ret = -1;
604 goto end;
605 }
606
607 end:
608 return ret;
609 }
610
611 /*
612 * Return the event source domain type via parameter.
613 */
614 static bool fd_is_event_notification_source(const struct notification_thread_state *state,
615 int fd,
616 enum lttng_domain_type *domain)
617 {
618 struct notification_event_tracer_event_source_element *source_element;
619
620 LTTNG_ASSERT(domain);
621
622 cds_list_for_each_entry(source_element,
623 &state->tracer_event_sources_list, node) {
624 if (source_element->fd != fd) {
625 continue;
626 }
627
628 *domain = source_element->domain;
629 return true;
630 }
631
632 return false;
633 }
634
635 /*
636 * This thread services notification channel clients and commands received
637 * from various lttng-sessiond components over a command queue.
638 */
639 static
640 void *thread_notification(void *data)
641 {
642 int ret;
643 struct notification_thread_handle *handle = (notification_thread_handle *) data;
644 struct notification_thread_state state;
645 enum lttng_domain_type domain;
646
647 DBG("Started notification thread");
648
649 health_register(the_health_sessiond, HEALTH_SESSIOND_TYPE_NOTIFICATION);
650 rcu_register_thread();
651 rcu_thread_online();
652
653 if (!handle) {
654 ERR("Invalid thread context provided");
655 goto end;
656 }
657
658 health_code_update();
659
660 ret = init_thread_state(handle, &state);
661 if (ret) {
662 goto end;
663 }
664
665 if (testpoint(sessiond_thread_notification)) {
666 goto end;
667 }
668
669 while (true) {
670 int fd_count, i;
671
672 health_poll_entry();
673 DBG("Entering poll wait");
674 ret = lttng_poll_wait(&state.events, -1);
675 DBG("Poll wait returned (%i)", ret);
676 health_poll_exit();
677 if (ret < 0) {
678 /*
679 * Restart interrupted system call.
680 */
681 if (errno == EINTR) {
682 continue;
683 }
684 ERR("Error encountered during lttng_poll_wait (%i)", ret);
685 goto error;
686 }
687
688 /*
689 * Reset restart_poll flag so that calls below might turn it
690 * on.
691 */
692 state.restart_poll = false;
693
694 fd_count = ret;
695 for (i = 0; i < fd_count; i++) {
696 int fd = LTTNG_POLL_GETFD(&state.events, i);
697 uint32_t revents = LTTNG_POLL_GETEV(&state.events, i);
698
699 DBG("Handling fd (%i) activity (%u)", fd, revents);
700
701 if (fd == state.notification_channel_socket) {
702 if (revents & LPOLLIN) {
703 ret = handle_notification_thread_client_connect(
704 &state);
705 if (ret < 0) {
706 goto error;
707 }
708 } else if (revents &
709 (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
710 ERR("Notification socket poll error");
711 goto error;
712 } else {
713 ERR("Unexpected poll events %u for notification socket %i", revents, fd);
714 goto error;
715 }
716 } else if (fd == handle->cmd_queue.event_fd) {
717 ret = handle_notification_thread_command(handle,
718 &state);
719 if (ret < 0) {
720 DBG("Error encountered while servicing command queue");
721 goto error;
722 } else if (ret > 0) {
723 goto exit;
724 }
725 } else if (fd == handle->channel_monitoring_pipes.ust32_consumer ||
726 fd == handle->channel_monitoring_pipes.ust64_consumer ||
727 fd == handle->channel_monitoring_pipes.kernel_consumer) {
728 ret = handle_channel_monitoring_pipe(fd,
729 revents, handle, &state);
730 if (ret) {
731 goto error;
732 }
733 } else if (fd_is_event_notification_source(&state, fd, &domain)) {
734 ret = handle_event_notification_pipe(fd, domain, revents, &state);
735 if (ret) {
736 goto error;
737 }
738 } else {
739 /* Activity on a client's socket. */
740 if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
741 /*
742 * It doesn't matter if a command was
743 * pending on the client socket at this
744 * point since it now has no way to
745 * receive the notifications to which
746 * it was subscribing or unsubscribing.
747 */
748 ret = handle_notification_thread_client_disconnect(
749 fd, &state);
750 if (ret) {
751 goto error;
752 }
753 } else {
754 if (revents & LPOLLIN) {
755 ret = handle_notification_thread_client_in(
756 &state, fd);
757 if (ret) {
758 goto error;
759 }
760 }
761
762 if (revents & LPOLLOUT) {
763 ret = handle_notification_thread_client_out(
764 &state, fd);
765 if (ret) {
766 goto error;
767 }
768 }
769 }
770 }
771
772 /*
773 * Calls above might have changed the state of the
774 * FDs in `state.events`. Call _poll_wait() again to
775 * ensure we have a consistent state.
776 */
777 if (state.restart_poll) {
778 break;
779 }
780 }
781 }
782 exit:
783 error:
784 fini_thread_state(&state);
785 end:
786 rcu_thread_offline();
787 rcu_unregister_thread();
788 health_unregister(the_health_sessiond);
789 return NULL;
790 }
791
792 static
793 bool shutdown_notification_thread(void *thread_data)
794 {
795 struct notification_thread_handle *handle = (notification_thread_handle *) thread_data;
796
797 notification_thread_command_quit(handle);
798 return true;
799 }
800
801 struct lttng_thread *launch_notification_thread(
802 struct notification_thread_handle *handle)
803 {
804 struct lttng_thread *thread;
805
806 thread = lttng_thread_create("Notification",
807 thread_notification,
808 shutdown_notification_thread,
809 NULL,
810 handle);
811 if (!thread) {
812 goto error;
813 }
814
815 /*
816 * Wait for the thread to be marked as "ready" before returning
817 * as other subsystems depend on the notification subsystem
818 * (e.g. rotation thread).
819 */
820 wait_until_thread_is_ready(handle);
821 return thread;
822 error:
823 return NULL;
824 }
This page took 0.048645 seconds and 4 git commands to generate.