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