Add a comment clarifying the ownership of triggers
[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"
40
41#include <urcu.h>
42#include <urcu/list.h>
43#include <urcu/rculfhash.h>
44
ab0ee2ca
JG
45/*
46 * Destroy the thread data previously created by the init function.
47 */
48void notification_thread_handle_destroy(
49 struct notification_thread_handle *handle)
50{
51 int ret;
ab0ee2ca
JG
52
53 if (!handle) {
54 goto end;
55 }
56
8ada111f 57 assert(cds_list_empty(&handle->cmd_queue.list));
ab0ee2ca
JG
58 pthread_mutex_destroy(&handle->cmd_queue.lock);
59
814b4934
JR
60 if (handle->cmd_queue.event_pipe) {
61 lttng_pipe_destroy(handle->cmd_queue.event_pipe);
62 }
ab0ee2ca
JG
63 if (handle->channel_monitoring_pipes.ust32_consumer >= 0) {
64 ret = close(handle->channel_monitoring_pipes.ust32_consumer);
65 if (ret) {
66 PERROR("close 32-bit consumer channel monitoring pipe");
67 }
68 }
69 if (handle->channel_monitoring_pipes.ust64_consumer >= 0) {
70 ret = close(handle->channel_monitoring_pipes.ust64_consumer);
71 if (ret) {
72 PERROR("close 64-bit consumer channel monitoring pipe");
73 }
74 }
75 if (handle->channel_monitoring_pipes.kernel_consumer >= 0) {
76 ret = close(handle->channel_monitoring_pipes.kernel_consumer);
77 if (ret) {
78 PERROR("close kernel consumer channel monitoring pipe");
79 }
80 }
81end:
82 free(handle);
83}
84
85struct notification_thread_handle *notification_thread_handle_create(
86 struct lttng_pipe *ust32_channel_monitor_pipe,
87 struct lttng_pipe *ust64_channel_monitor_pipe,
90936dcf
JD
88 struct lttng_pipe *kernel_channel_monitor_pipe,
89 sem_t *notification_thread_ready)
ab0ee2ca
JG
90{
91 int ret;
92 struct notification_thread_handle *handle;
814b4934 93 struct lttng_pipe *event_pipe = NULL;
ab0ee2ca
JG
94
95 handle = zmalloc(sizeof(*handle));
96 if (!handle) {
97 goto end;
98 }
99
18d08850 100 event_pipe = lttng_pipe_open(FD_CLOEXEC);
814b4934
JR
101 if (!event_pipe) {
102 ERR("event_pipe creation");
ab0ee2ca
JG
103 goto error;
104 }
814b4934
JR
105
106 handle->cmd_queue.event_pipe = event_pipe;
107 event_pipe = NULL;
108
ab0ee2ca
JG
109 CDS_INIT_LIST_HEAD(&handle->cmd_queue.list);
110 ret = pthread_mutex_init(&handle->cmd_queue.lock, NULL);
111 if (ret) {
112 goto error;
113 }
114
115 if (ust32_channel_monitor_pipe) {
116 handle->channel_monitoring_pipes.ust32_consumer =
117 lttng_pipe_release_readfd(
118 ust32_channel_monitor_pipe);
119 if (handle->channel_monitoring_pipes.ust32_consumer < 0) {
120 goto error;
121 }
122 } else {
123 handle->channel_monitoring_pipes.ust32_consumer = -1;
124 }
125 if (ust64_channel_monitor_pipe) {
126 handle->channel_monitoring_pipes.ust64_consumer =
127 lttng_pipe_release_readfd(
128 ust64_channel_monitor_pipe);
129 if (handle->channel_monitoring_pipes.ust64_consumer < 0) {
130 goto error;
131 }
132 } else {
133 handle->channel_monitoring_pipes.ust64_consumer = -1;
134 }
135 if (kernel_channel_monitor_pipe) {
136 handle->channel_monitoring_pipes.kernel_consumer =
137 lttng_pipe_release_readfd(
138 kernel_channel_monitor_pipe);
139 if (handle->channel_monitoring_pipes.kernel_consumer < 0) {
140 goto error;
141 }
142 } else {
143 handle->channel_monitoring_pipes.kernel_consumer = -1;
144 }
90936dcf 145 handle->notification_thread_ready = notification_thread_ready;
ab0ee2ca
JG
146end:
147 return handle;
148error:
814b4934 149 lttng_pipe_destroy(event_pipe);
ab0ee2ca
JG
150 notification_thread_handle_destroy(handle);
151 return NULL;
152}
153
154static
155char *get_notification_channel_sock_path(void)
156{
157 int ret;
158 bool is_root = !getuid();
159 char *sock_path;
160
161 sock_path = zmalloc(LTTNG_PATH_MAX);
162 if (!sock_path) {
163 goto error;
164 }
165
166 if (is_root) {
167 ret = snprintf(sock_path, LTTNG_PATH_MAX,
168 DEFAULT_GLOBAL_NOTIFICATION_CHANNEL_UNIX_SOCK);
169 if (ret < 0) {
170 goto error;
171 }
172 } else {
173 char *home_path = utils_get_home_dir();
174
175 if (!home_path) {
176 ERR("Can't get HOME directory for socket creation");
177 goto error;
178 }
179
180 ret = snprintf(sock_path, LTTNG_PATH_MAX,
181 DEFAULT_HOME_NOTIFICATION_CHANNEL_UNIX_SOCK,
182 home_path);
183 if (ret < 0) {
184 goto error;
185 }
186 }
187
188 return sock_path;
189error:
190 free(sock_path);
191 return NULL;
192}
193
194static
195void notification_channel_socket_destroy(int fd)
196{
197 int ret;
198 char *sock_path = get_notification_channel_sock_path();
199
200 DBG("[notification-thread] Destroying notification channel socket");
201
202 if (sock_path) {
203 ret = unlink(sock_path);
204 free(sock_path);
205 if (ret < 0) {
206 PERROR("unlink notification channel socket");
207 }
208 }
209
210 ret = close(fd);
211 if (ret) {
212 PERROR("close notification channel socket");
213 }
214}
215
216static
217int notification_channel_socket_create(void)
218{
219 int fd = -1, ret;
220 char *sock_path = get_notification_channel_sock_path();
221
222 DBG("[notification-thread] Creating notification channel UNIX socket at %s",
223 sock_path);
224
225 ret = lttcomm_create_unix_sock(sock_path);
226 if (ret < 0) {
227 ERR("[notification-thread] Failed to create notification socket");
228 goto error;
229 }
230 fd = ret;
231
232 ret = chmod(sock_path, S_IRUSR | S_IWUSR | S_IRGRP | S_IWGRP);
233 if (ret < 0) {
234 ERR("Set file permissions failed: %s", sock_path);
235 PERROR("chmod notification channel socket");
236 goto error;
237 }
238
239 if (getuid() == 0) {
240 ret = chown(sock_path, 0,
e6142f2e 241 utils_get_group_id(config.tracing_group_name.value));
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 }
335 if (state->triggers_ht) {
336 ret = handle_notification_thread_trigger_unregister_all(state);
337 assert(!ret);
338 ret = cds_lfht_destroy(state->triggers_ht, NULL);
339 assert(!ret);
340 }
341 if (state->channel_triggers_ht) {
342 ret = cds_lfht_destroy(state->channel_triggers_ht, NULL);
343 assert(!ret);
344 }
345 if (state->channel_state_ht) {
346 ret = cds_lfht_destroy(state->channel_state_ht, NULL);
347 assert(!ret);
348 }
349 if (state->notification_trigger_clients_ht) {
350 ret = cds_lfht_destroy(state->notification_trigger_clients_ht,
351 NULL);
352 assert(!ret);
353 }
354 if (state->channels_ht) {
8abe313a
JG
355 ret = cds_lfht_destroy(state->channels_ht, NULL);
356 assert(!ret);
357 }
358 if (state->sessions_ht) {
359 ret = cds_lfht_destroy(state->sessions_ht, NULL);
ab0ee2ca
JG
360 assert(!ret);
361 }
ab0ee2ca
JG
362 if (state->notification_channel_socket >= 0) {
363 notification_channel_socket_destroy(
364 state->notification_channel_socket);
365 }
366 lttng_poll_clean(&state->events);
367}
368
369static
370int init_thread_state(struct notification_thread_handle *handle,
371 struct notification_thread_state *state)
372{
373 int ret;
374
375 memset(state, 0, sizeof(*state));
376 state->notification_channel_socket = -1;
377 lttng_poll_init(&state->events);
378
379 ret = notification_channel_socket_create();
380 if (ret < 0) {
381 goto end;
382 }
383 state->notification_channel_socket = ret;
384
385 ret = init_poll_set(&state->events, handle,
386 state->notification_channel_socket);
387 if (ret) {
388 goto end;
389 }
390
391 DBG("[notification-thread] Listening on notification channel socket");
392 ret = lttcomm_listen_unix_sock(state->notification_channel_socket);
393 if (ret < 0) {
394 ERR("[notification-thread] Listen failed on notification channel socket");
395 goto error;
396 }
397
398 state->client_socket_ht = cds_lfht_new(DEFAULT_HT_SIZE, 1, 0,
399 CDS_LFHT_AUTO_RESIZE | CDS_LFHT_ACCOUNTING, NULL);
400 if (!state->client_socket_ht) {
401 goto error;
402 }
403
404 state->channel_triggers_ht = cds_lfht_new(DEFAULT_HT_SIZE, 1, 0,
405 CDS_LFHT_AUTO_RESIZE | CDS_LFHT_ACCOUNTING, NULL);
406 if (!state->channel_triggers_ht) {
407 goto error;
408 }
409
410 state->channel_state_ht = cds_lfht_new(DEFAULT_HT_SIZE, 1, 0,
411 CDS_LFHT_AUTO_RESIZE | CDS_LFHT_ACCOUNTING, NULL);
412 if (!state->channel_state_ht) {
413 goto error;
414 }
415
416 state->notification_trigger_clients_ht = cds_lfht_new(DEFAULT_HT_SIZE,
417 1, 0, CDS_LFHT_AUTO_RESIZE | CDS_LFHT_ACCOUNTING, NULL);
418 if (!state->notification_trigger_clients_ht) {
419 goto error;
420 }
421
422 state->channels_ht = cds_lfht_new(DEFAULT_HT_SIZE,
423 1, 0, CDS_LFHT_AUTO_RESIZE | CDS_LFHT_ACCOUNTING, NULL);
424 if (!state->channels_ht) {
425 goto error;
426 }
8abe313a
JG
427 state->sessions_ht = cds_lfht_new(DEFAULT_HT_SIZE,
428 1, 0, CDS_LFHT_AUTO_RESIZE | CDS_LFHT_ACCOUNTING, NULL);
429 if (!state->sessions_ht) {
430 goto error;
431 }
ab0ee2ca
JG
432 state->triggers_ht = cds_lfht_new(DEFAULT_HT_SIZE,
433 1, 0, CDS_LFHT_AUTO_RESIZE | CDS_LFHT_ACCOUNTING, NULL);
434 if (!state->triggers_ht) {
435 goto error;
436 }
90936dcf 437 sem_post(handle->notification_thread_ready);
ab0ee2ca
JG
438end:
439 return 0;
440error:
441 fini_thread_state(state);
442 return -1;
443}
444
445static
446int handle_channel_monitoring_pipe(int fd, uint32_t revents,
447 struct notification_thread_handle *handle,
448 struct notification_thread_state *state)
449{
450 int ret = 0;
451 enum lttng_domain_type domain;
452
453 if (fd == handle->channel_monitoring_pipes.ust32_consumer ||
454 fd == handle->channel_monitoring_pipes.ust64_consumer) {
455 domain = LTTNG_DOMAIN_UST;
456 } else if (fd == handle->channel_monitoring_pipes.kernel_consumer) {
457 domain = LTTNG_DOMAIN_KERNEL;
458 } else {
459 abort();
460 }
461
462 if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
463 ret = lttng_poll_del(&state->events, fd);
464 if (ret) {
465 ERR("[notification-thread] Failed to remove consumer monitoring pipe from poll set");
466 }
467 goto end;
468 }
469
470 ret = handle_notification_thread_channel_sample(
471 state, fd, domain);
472 if (ret) {
4149ace8 473 ERR("[notification-thread] Consumer sample handling error occurred");
ab0ee2ca
JG
474 ret = -1;
475 goto end;
476 }
477end:
478 return ret;
479}
480
481/*
482 * This thread services notification channel clients and commands received
483 * from various lttng-sessiond components over a command queue.
484 */
485void *thread_notification(void *data)
486{
487 int ret;
488 struct notification_thread_handle *handle = data;
489 struct notification_thread_state state;
490
491 DBG("[notification-thread] Started notification thread");
492
493 if (!handle) {
494 ERR("[notification-thread] Invalid thread context provided");
495 goto end;
496 }
497
498 rcu_register_thread();
499 rcu_thread_online();
500
501 health_register(health_sessiond, HEALTH_SESSIOND_TYPE_NOTIFICATION);
502 health_code_update();
503
504 ret = init_thread_state(handle, &state);
505 if (ret) {
506 goto end;
507 }
508
509 /* Ready to handle client connections. */
510 sessiond_notify_ready();
511
512 while (true) {
513 int fd_count, i;
514
515 health_poll_entry();
516 DBG("[notification-thread] Entering poll wait");
517 ret = lttng_poll_wait(&state.events, -1);
518 DBG("[notification-thread] Poll wait returned (%i)", ret);
519 health_poll_exit();
520 if (ret < 0) {
521 /*
522 * Restart interrupted system call.
523 */
524 if (errno == EINTR) {
525 continue;
526 }
527 ERR("[notification-thread] Error encountered during lttng_poll_wait (%i)", ret);
528 goto error;
529 }
530
531 fd_count = ret;
532 for (i = 0; i < fd_count; i++) {
533 int fd = LTTNG_POLL_GETFD(&state.events, i);
534 uint32_t revents = LTTNG_POLL_GETEV(&state.events, i);
535
2d01d977
JG
536 if (!revents) {
537 continue;
538 }
ab0ee2ca
JG
539 DBG("[notification-thread] Handling fd (%i) activity (%u)", fd, revents);
540
541 if (fd == state.notification_channel_socket) {
542 if (revents & LPOLLIN) {
543 ret = handle_notification_thread_client_connect(
544 &state);
545 if (ret < 0) {
546 goto error;
547 }
548 } else if (revents &
549 (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
550 ERR("[notification-thread] Notification socket poll error");
551 goto error;
552 } else {
553 ERR("[notification-thread] Unexpected poll events %u for notification socket %i", revents, fd);
554 goto error;
555 }
814b4934 556 } else if (fd == lttng_pipe_get_readfd(handle->cmd_queue.event_pipe)) {
ab0ee2ca
JG
557 ret = handle_notification_thread_command(handle,
558 &state);
559 if (ret < 0) {
560 DBG("[notification-thread] Error encountered while servicing command queue");
561 goto error;
562 } else if (ret > 0) {
563 goto exit;
564 }
565 } else if (fd == handle->channel_monitoring_pipes.ust32_consumer ||
566 fd == handle->channel_monitoring_pipes.ust64_consumer ||
567 fd == handle->channel_monitoring_pipes.kernel_consumer) {
568 ret = handle_channel_monitoring_pipe(fd,
569 revents, handle, &state);
570 if (ret) {
571 goto error;
572 }
573 } else {
574 /* Activity on a client's socket. */
575 if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
576 /*
577 * It doesn't matter if a command was
578 * pending on the client socket at this
579 * point since it now has no way to
580 * receive the notifications to which
581 * it was subscribing or unsubscribing.
582 */
583 ret = handle_notification_thread_client_disconnect(
584 fd, &state);
585 if (ret) {
586 goto error;
587 }
588 } else {
589 if (revents & LPOLLIN) {
590 ret = handle_notification_thread_client_in(
591 &state, fd);
592 if (ret) {
593 goto error;
594 }
595 }
596
597 if (revents & LPOLLOUT) {
598 ret = handle_notification_thread_client_out(
599 &state, fd);
600 if (ret) {
601 goto error;
602 }
603 }
604 }
605 }
606 }
607 }
608exit:
609error:
610 fini_thread_state(&state);
611 health_unregister(health_sessiond);
612 rcu_thread_offline();
613 rcu_unregister_thread();
614end:
615 return NULL;
616}
This page took 0.083896 seconds and 5 git commands to generate.