4ce38e180e106986ae630a0c85483e93568405a3
[lttng-tools.git] / src / bin / lttng-sessiond / notification-thread.c
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>
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
45 /*
46 * Destroy the thread data previously created by the init function.
47 */
48 void notification_thread_handle_destroy(
49 struct notification_thread_handle *handle)
50 {
51 int ret;
52
53 if (!handle) {
54 goto end;
55 }
56
57 assert(cds_list_empty(&handle->cmd_queue.list));
58 pthread_mutex_destroy(&handle->cmd_queue.lock);
59
60 if (handle->cmd_queue.event_pipe) {
61 lttng_pipe_destroy(handle->cmd_queue.event_pipe);
62 }
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 }
81 end:
82 free(handle);
83 }
84
85 struct notification_thread_handle *notification_thread_handle_create(
86 struct lttng_pipe *ust32_channel_monitor_pipe,
87 struct lttng_pipe *ust64_channel_monitor_pipe,
88 struct lttng_pipe *kernel_channel_monitor_pipe,
89 sem_t *notification_thread_ready)
90 {
91 int ret;
92 struct notification_thread_handle *handle;
93 struct lttng_pipe *event_pipe = NULL;
94
95 handle = zmalloc(sizeof(*handle));
96 if (!handle) {
97 goto end;
98 }
99
100 event_pipe = lttng_pipe_open(FD_CLOEXEC);
101 if (!event_pipe) {
102 ERR("event_pipe creation");
103 goto error;
104 }
105
106 handle->cmd_queue.event_pipe = event_pipe;
107 event_pipe = NULL;
108
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 }
145 handle->notification_thread_ready = notification_thread_ready;
146 end:
147 return handle;
148 error:
149 lttng_pipe_destroy(event_pipe);
150 notification_thread_handle_destroy(handle);
151 return NULL;
152 }
153
154 static
155 char *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;
189 error:
190 free(sock_path);
191 return NULL;
192 }
193
194 static
195 void 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
216 static
217 int 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,
241 utils_get_group_id(config.tracing_group_name.value));
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;
253 error:
254 if (fd >= 0 && close(fd) < 0) {
255 PERROR("close notification channel socket");
256 }
257 free(sock_path);
258 return ret;
259 }
260
261 static
262 int 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 }
287 ret = lttng_poll_add(poll_set, lttng_pipe_get_readfd(handle->cmd_queue.event_pipe),
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 }
317 end:
318 return ret;
319 error:
320 lttng_poll_clean(poll_set);
321 return ret;
322 }
323
324 static
325 void 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) {
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);
360 assert(!ret);
361 }
362 /*
363 * Must be destroyed after all channels have been destroyed.
364 * See comment in struct lttng_session_trigger_list.
365 */
366 if (state->session_triggers_ht) {
367 ret = cds_lfht_destroy(state->session_triggers_ht, NULL);
368 assert(!ret);
369 }
370 if (state->notification_channel_socket >= 0) {
371 notification_channel_socket_destroy(
372 state->notification_channel_socket);
373 }
374 lttng_poll_clean(&state->events);
375 }
376
377 static
378 int init_thread_state(struct notification_thread_handle *handle,
379 struct notification_thread_state *state)
380 {
381 int ret;
382
383 memset(state, 0, sizeof(*state));
384 state->notification_channel_socket = -1;
385 lttng_poll_init(&state->events);
386
387 ret = notification_channel_socket_create();
388 if (ret < 0) {
389 goto end;
390 }
391 state->notification_channel_socket = ret;
392
393 ret = init_poll_set(&state->events, handle,
394 state->notification_channel_socket);
395 if (ret) {
396 goto end;
397 }
398
399 DBG("[notification-thread] Listening on notification channel socket");
400 ret = lttcomm_listen_unix_sock(state->notification_channel_socket);
401 if (ret < 0) {
402 ERR("[notification-thread] Listen failed on notification channel socket");
403 goto error;
404 }
405
406 state->client_socket_ht = cds_lfht_new(DEFAULT_HT_SIZE, 1, 0,
407 CDS_LFHT_AUTO_RESIZE | CDS_LFHT_ACCOUNTING, NULL);
408 if (!state->client_socket_ht) {
409 goto error;
410 }
411
412 state->channel_triggers_ht = cds_lfht_new(DEFAULT_HT_SIZE, 1, 0,
413 CDS_LFHT_AUTO_RESIZE | CDS_LFHT_ACCOUNTING, NULL);
414 if (!state->channel_triggers_ht) {
415 goto error;
416 }
417
418 state->session_triggers_ht = cds_lfht_new(DEFAULT_HT_SIZE, 1, 0,
419 CDS_LFHT_AUTO_RESIZE | CDS_LFHT_ACCOUNTING, NULL);
420 if (!state->session_triggers_ht) {
421 goto error;
422 }
423
424 state->channel_state_ht = cds_lfht_new(DEFAULT_HT_SIZE, 1, 0,
425 CDS_LFHT_AUTO_RESIZE | CDS_LFHT_ACCOUNTING, NULL);
426 if (!state->channel_state_ht) {
427 goto error;
428 }
429
430 state->notification_trigger_clients_ht = cds_lfht_new(DEFAULT_HT_SIZE,
431 1, 0, CDS_LFHT_AUTO_RESIZE | CDS_LFHT_ACCOUNTING, NULL);
432 if (!state->notification_trigger_clients_ht) {
433 goto error;
434 }
435
436 state->channels_ht = cds_lfht_new(DEFAULT_HT_SIZE,
437 1, 0, CDS_LFHT_AUTO_RESIZE | CDS_LFHT_ACCOUNTING, NULL);
438 if (!state->channels_ht) {
439 goto error;
440 }
441 state->sessions_ht = cds_lfht_new(DEFAULT_HT_SIZE,
442 1, 0, CDS_LFHT_AUTO_RESIZE | CDS_LFHT_ACCOUNTING, NULL);
443 if (!state->sessions_ht) {
444 goto error;
445 }
446 state->triggers_ht = cds_lfht_new(DEFAULT_HT_SIZE,
447 1, 0, CDS_LFHT_AUTO_RESIZE | CDS_LFHT_ACCOUNTING, NULL);
448 if (!state->triggers_ht) {
449 goto error;
450 }
451 sem_post(handle->notification_thread_ready);
452 end:
453 return 0;
454 error:
455 fini_thread_state(state);
456 return -1;
457 }
458
459 static
460 int handle_channel_monitoring_pipe(int fd, uint32_t revents,
461 struct notification_thread_handle *handle,
462 struct notification_thread_state *state)
463 {
464 int ret = 0;
465 enum lttng_domain_type domain;
466
467 if (fd == handle->channel_monitoring_pipes.ust32_consumer ||
468 fd == handle->channel_monitoring_pipes.ust64_consumer) {
469 domain = LTTNG_DOMAIN_UST;
470 } else if (fd == handle->channel_monitoring_pipes.kernel_consumer) {
471 domain = LTTNG_DOMAIN_KERNEL;
472 } else {
473 abort();
474 }
475
476 if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
477 ret = lttng_poll_del(&state->events, fd);
478 if (ret) {
479 ERR("[notification-thread] Failed to remove consumer monitoring pipe from poll set");
480 }
481 goto end;
482 }
483
484 ret = handle_notification_thread_channel_sample(
485 state, fd, domain);
486 if (ret) {
487 ERR("[notification-thread] Consumer sample handling error occurred");
488 ret = -1;
489 goto end;
490 }
491 end:
492 return ret;
493 }
494
495 /*
496 * This thread services notification channel clients and commands received
497 * from various lttng-sessiond components over a command queue.
498 */
499 void *thread_notification(void *data)
500 {
501 int ret;
502 struct notification_thread_handle *handle = data;
503 struct notification_thread_state state;
504
505 DBG("[notification-thread] Started notification thread");
506
507 if (!handle) {
508 ERR("[notification-thread] Invalid thread context provided");
509 goto end;
510 }
511
512 rcu_register_thread();
513 rcu_thread_online();
514
515 health_register(health_sessiond, HEALTH_SESSIOND_TYPE_NOTIFICATION);
516 health_code_update();
517
518 ret = init_thread_state(handle, &state);
519 if (ret) {
520 goto end;
521 }
522
523 /* Ready to handle client connections. */
524 sessiond_notify_ready();
525
526 while (true) {
527 int fd_count, i;
528
529 health_poll_entry();
530 DBG("[notification-thread] Entering poll wait");
531 ret = lttng_poll_wait(&state.events, -1);
532 DBG("[notification-thread] Poll wait returned (%i)", ret);
533 health_poll_exit();
534 if (ret < 0) {
535 /*
536 * Restart interrupted system call.
537 */
538 if (errno == EINTR) {
539 continue;
540 }
541 ERR("[notification-thread] Error encountered during lttng_poll_wait (%i)", ret);
542 goto error;
543 }
544
545 fd_count = ret;
546 for (i = 0; i < fd_count; i++) {
547 int fd = LTTNG_POLL_GETFD(&state.events, i);
548 uint32_t revents = LTTNG_POLL_GETEV(&state.events, i);
549
550 if (!revents) {
551 continue;
552 }
553 DBG("[notification-thread] Handling fd (%i) activity (%u)", fd, revents);
554
555 if (fd == state.notification_channel_socket) {
556 if (revents & LPOLLIN) {
557 ret = handle_notification_thread_client_connect(
558 &state);
559 if (ret < 0) {
560 goto error;
561 }
562 } else if (revents &
563 (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
564 ERR("[notification-thread] Notification socket poll error");
565 goto error;
566 } else {
567 ERR("[notification-thread] Unexpected poll events %u for notification socket %i", revents, fd);
568 goto error;
569 }
570 } else if (fd == lttng_pipe_get_readfd(handle->cmd_queue.event_pipe)) {
571 ret = handle_notification_thread_command(handle,
572 &state);
573 if (ret < 0) {
574 DBG("[notification-thread] Error encountered while servicing command queue");
575 goto error;
576 } else if (ret > 0) {
577 goto exit;
578 }
579 } else if (fd == handle->channel_monitoring_pipes.ust32_consumer ||
580 fd == handle->channel_monitoring_pipes.ust64_consumer ||
581 fd == handle->channel_monitoring_pipes.kernel_consumer) {
582 ret = handle_channel_monitoring_pipe(fd,
583 revents, handle, &state);
584 if (ret) {
585 goto error;
586 }
587 } else {
588 /* Activity on a client's socket. */
589 if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
590 /*
591 * It doesn't matter if a command was
592 * pending on the client socket at this
593 * point since it now has no way to
594 * receive the notifications to which
595 * it was subscribing or unsubscribing.
596 */
597 ret = handle_notification_thread_client_disconnect(
598 fd, &state);
599 if (ret) {
600 goto error;
601 }
602 } else {
603 if (revents & LPOLLIN) {
604 ret = handle_notification_thread_client_in(
605 &state, fd);
606 if (ret) {
607 goto error;
608 }
609 }
610
611 if (revents & LPOLLOUT) {
612 ret = handle_notification_thread_client_out(
613 &state, fd);
614 if (ret) {
615 goto error;
616 }
617 }
618 }
619 }
620 }
621 }
622 exit:
623 error:
624 fini_thread_state(&state);
625 health_unregister(health_sessiond);
626 rcu_thread_offline();
627 rcu_unregister_thread();
628 end:
629 return NULL;
630 }
This page took 0.068002 seconds and 5 git commands to generate.