Separate session info from channel info in notification thread
[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 {
90 int ret;
91 struct notification_thread_handle *handle;
92 struct lttng_pipe *event_pipe = NULL;
93
94 handle = zmalloc(sizeof(*handle));
95 if (!handle) {
96 goto end;
97 }
98
99 event_pipe = lttng_pipe_open(FD_CLOEXEC);
100 if (!event_pipe) {
101 ERR("event_pipe creation");
102 goto error;
103 }
104
105 handle->cmd_queue.event_pipe = event_pipe;
106 event_pipe = NULL;
107
108 CDS_INIT_LIST_HEAD(&handle->cmd_queue.list);
109 ret = pthread_mutex_init(&handle->cmd_queue.lock, NULL);
110 if (ret) {
111 goto error;
112 }
113
114 if (ust32_channel_monitor_pipe) {
115 handle->channel_monitoring_pipes.ust32_consumer =
116 lttng_pipe_release_readfd(
117 ust32_channel_monitor_pipe);
118 if (handle->channel_monitoring_pipes.ust32_consumer < 0) {
119 goto error;
120 }
121 } else {
122 handle->channel_monitoring_pipes.ust32_consumer = -1;
123 }
124 if (ust64_channel_monitor_pipe) {
125 handle->channel_monitoring_pipes.ust64_consumer =
126 lttng_pipe_release_readfd(
127 ust64_channel_monitor_pipe);
128 if (handle->channel_monitoring_pipes.ust64_consumer < 0) {
129 goto error;
130 }
131 } else {
132 handle->channel_monitoring_pipes.ust64_consumer = -1;
133 }
134 if (kernel_channel_monitor_pipe) {
135 handle->channel_monitoring_pipes.kernel_consumer =
136 lttng_pipe_release_readfd(
137 kernel_channel_monitor_pipe);
138 if (handle->channel_monitoring_pipes.kernel_consumer < 0) {
139 goto error;
140 }
141 } else {
142 handle->channel_monitoring_pipes.kernel_consumer = -1;
143 }
144 end:
145 return handle;
146 error:
147 lttng_pipe_destroy(event_pipe);
148 notification_thread_handle_destroy(handle);
149 return NULL;
150 }
151
152 static
153 char *get_notification_channel_sock_path(void)
154 {
155 int ret;
156 bool is_root = !getuid();
157 char *sock_path;
158
159 sock_path = zmalloc(LTTNG_PATH_MAX);
160 if (!sock_path) {
161 goto error;
162 }
163
164 if (is_root) {
165 ret = snprintf(sock_path, LTTNG_PATH_MAX,
166 DEFAULT_GLOBAL_NOTIFICATION_CHANNEL_UNIX_SOCK);
167 if (ret < 0) {
168 goto error;
169 }
170 } else {
171 char *home_path = utils_get_home_dir();
172
173 if (!home_path) {
174 ERR("Can't get HOME directory for socket creation");
175 goto error;
176 }
177
178 ret = snprintf(sock_path, LTTNG_PATH_MAX,
179 DEFAULT_HOME_NOTIFICATION_CHANNEL_UNIX_SOCK,
180 home_path);
181 if (ret < 0) {
182 goto error;
183 }
184 }
185
186 return sock_path;
187 error:
188 free(sock_path);
189 return NULL;
190 }
191
192 static
193 void notification_channel_socket_destroy(int fd)
194 {
195 int ret;
196 char *sock_path = get_notification_channel_sock_path();
197
198 DBG("[notification-thread] Destroying notification channel socket");
199
200 if (sock_path) {
201 ret = unlink(sock_path);
202 free(sock_path);
203 if (ret < 0) {
204 PERROR("unlink notification channel socket");
205 }
206 }
207
208 ret = close(fd);
209 if (ret) {
210 PERROR("close notification channel socket");
211 }
212 }
213
214 static
215 int notification_channel_socket_create(void)
216 {
217 int fd = -1, ret;
218 char *sock_path = get_notification_channel_sock_path();
219
220 DBG("[notification-thread] Creating notification channel UNIX socket at %s",
221 sock_path);
222
223 ret = lttcomm_create_unix_sock(sock_path);
224 if (ret < 0) {
225 ERR("[notification-thread] Failed to create notification socket");
226 goto error;
227 }
228 fd = ret;
229
230 ret = chmod(sock_path, S_IRUSR | S_IWUSR | S_IRGRP | S_IWGRP);
231 if (ret < 0) {
232 ERR("Set file permissions failed: %s", sock_path);
233 PERROR("chmod notification channel socket");
234 goto error;
235 }
236
237 if (getuid() == 0) {
238 ret = chown(sock_path, 0,
239 utils_get_group_id(config.tracing_group_name.value));
240 if (ret) {
241 ERR("Failed to set the notification channel socket's group");
242 ret = -1;
243 goto error;
244 }
245 }
246
247 DBG("[notification-thread] Notification channel UNIX socket created (fd = %i)",
248 fd);
249 free(sock_path);
250 return fd;
251 error:
252 if (fd >= 0 && close(fd) < 0) {
253 PERROR("close notification channel socket");
254 }
255 free(sock_path);
256 return ret;
257 }
258
259 static
260 int init_poll_set(struct lttng_poll_event *poll_set,
261 struct notification_thread_handle *handle,
262 int notification_channel_socket)
263 {
264 int ret;
265
266 /*
267 * Create pollset with size 5:
268 * - notification channel socket (listen for new connections),
269 * - command queue event fd (internal sessiond commands),
270 * - consumerd (32-bit user space) channel monitor pipe,
271 * - consumerd (64-bit user space) channel monitor pipe,
272 * - consumerd (kernel) channel monitor pipe.
273 */
274 ret = lttng_poll_create(poll_set, 5, LTTNG_CLOEXEC);
275 if (ret < 0) {
276 goto end;
277 }
278
279 ret = lttng_poll_add(poll_set, notification_channel_socket,
280 LPOLLIN | LPOLLERR | LPOLLHUP | LPOLLRDHUP);
281 if (ret < 0) {
282 ERR("[notification-thread] Failed to add notification channel socket to pollset");
283 goto error;
284 }
285 ret = lttng_poll_add(poll_set, lttng_pipe_get_readfd(handle->cmd_queue.event_pipe),
286 LPOLLIN | LPOLLERR);
287 if (ret < 0) {
288 ERR("[notification-thread] Failed to add notification command queue event fd to pollset");
289 goto error;
290 }
291 ret = lttng_poll_add(poll_set,
292 handle->channel_monitoring_pipes.ust32_consumer,
293 LPOLLIN | LPOLLERR);
294 if (ret < 0) {
295 ERR("[notification-thread] Failed to add ust-32 channel monitoring pipe fd to pollset");
296 goto error;
297 }
298 ret = lttng_poll_add(poll_set,
299 handle->channel_monitoring_pipes.ust64_consumer,
300 LPOLLIN | LPOLLERR);
301 if (ret < 0) {
302 ERR("[notification-thread] Failed to add ust-64 channel monitoring pipe fd to pollset");
303 goto error;
304 }
305 if (handle->channel_monitoring_pipes.kernel_consumer < 0) {
306 goto end;
307 }
308 ret = lttng_poll_add(poll_set,
309 handle->channel_monitoring_pipes.kernel_consumer,
310 LPOLLIN | LPOLLERR);
311 if (ret < 0) {
312 ERR("[notification-thread] Failed to add kernel channel monitoring pipe fd to pollset");
313 goto error;
314 }
315 end:
316 return ret;
317 error:
318 lttng_poll_clean(poll_set);
319 return ret;
320 }
321
322 static
323 void fini_thread_state(struct notification_thread_state *state)
324 {
325 int ret;
326
327 if (state->client_socket_ht) {
328 ret = handle_notification_thread_client_disconnect_all(state);
329 assert(!ret);
330 ret = cds_lfht_destroy(state->client_socket_ht, NULL);
331 assert(!ret);
332 }
333 if (state->triggers_ht) {
334 ret = handle_notification_thread_trigger_unregister_all(state);
335 assert(!ret);
336 ret = cds_lfht_destroy(state->triggers_ht, NULL);
337 assert(!ret);
338 }
339 if (state->channel_triggers_ht) {
340 ret = cds_lfht_destroy(state->channel_triggers_ht, NULL);
341 assert(!ret);
342 }
343 if (state->channel_state_ht) {
344 ret = cds_lfht_destroy(state->channel_state_ht, NULL);
345 assert(!ret);
346 }
347 if (state->notification_trigger_clients_ht) {
348 ret = cds_lfht_destroy(state->notification_trigger_clients_ht,
349 NULL);
350 assert(!ret);
351 }
352 if (state->channels_ht) {
353 ret = cds_lfht_destroy(state->channels_ht, NULL);
354 assert(!ret);
355 }
356 if (state->sessions_ht) {
357 ret = cds_lfht_destroy(state->sessions_ht, NULL);
358 assert(!ret);
359 }
360 if (state->notification_channel_socket >= 0) {
361 notification_channel_socket_destroy(
362 state->notification_channel_socket);
363 }
364 lttng_poll_clean(&state->events);
365 }
366
367 static
368 int init_thread_state(struct notification_thread_handle *handle,
369 struct notification_thread_state *state)
370 {
371 int ret;
372
373 memset(state, 0, sizeof(*state));
374 state->notification_channel_socket = -1;
375 lttng_poll_init(&state->events);
376
377 ret = notification_channel_socket_create();
378 if (ret < 0) {
379 goto end;
380 }
381 state->notification_channel_socket = ret;
382
383 ret = init_poll_set(&state->events, handle,
384 state->notification_channel_socket);
385 if (ret) {
386 goto end;
387 }
388
389 DBG("[notification-thread] Listening on notification channel socket");
390 ret = lttcomm_listen_unix_sock(state->notification_channel_socket);
391 if (ret < 0) {
392 ERR("[notification-thread] Listen failed on notification channel socket");
393 goto error;
394 }
395
396 state->client_socket_ht = cds_lfht_new(DEFAULT_HT_SIZE, 1, 0,
397 CDS_LFHT_AUTO_RESIZE | CDS_LFHT_ACCOUNTING, NULL);
398 if (!state->client_socket_ht) {
399 goto error;
400 }
401
402 state->channel_triggers_ht = cds_lfht_new(DEFAULT_HT_SIZE, 1, 0,
403 CDS_LFHT_AUTO_RESIZE | CDS_LFHT_ACCOUNTING, NULL);
404 if (!state->channel_triggers_ht) {
405 goto error;
406 }
407
408 state->channel_state_ht = cds_lfht_new(DEFAULT_HT_SIZE, 1, 0,
409 CDS_LFHT_AUTO_RESIZE | CDS_LFHT_ACCOUNTING, NULL);
410 if (!state->channel_state_ht) {
411 goto error;
412 }
413
414 state->notification_trigger_clients_ht = cds_lfht_new(DEFAULT_HT_SIZE,
415 1, 0, CDS_LFHT_AUTO_RESIZE | CDS_LFHT_ACCOUNTING, NULL);
416 if (!state->notification_trigger_clients_ht) {
417 goto error;
418 }
419
420 state->channels_ht = cds_lfht_new(DEFAULT_HT_SIZE,
421 1, 0, CDS_LFHT_AUTO_RESIZE | CDS_LFHT_ACCOUNTING, NULL);
422 if (!state->channels_ht) {
423 goto error;
424 }
425 state->sessions_ht = cds_lfht_new(DEFAULT_HT_SIZE,
426 1, 0, CDS_LFHT_AUTO_RESIZE | CDS_LFHT_ACCOUNTING, NULL);
427 if (!state->sessions_ht) {
428 goto error;
429 }
430 state->triggers_ht = cds_lfht_new(DEFAULT_HT_SIZE,
431 1, 0, CDS_LFHT_AUTO_RESIZE | CDS_LFHT_ACCOUNTING, NULL);
432 if (!state->triggers_ht) {
433 goto error;
434 }
435 end:
436 return 0;
437 error:
438 fini_thread_state(state);
439 return -1;
440 }
441
442 static
443 int handle_channel_monitoring_pipe(int fd, uint32_t revents,
444 struct notification_thread_handle *handle,
445 struct notification_thread_state *state)
446 {
447 int ret = 0;
448 enum lttng_domain_type domain;
449
450 if (fd == handle->channel_monitoring_pipes.ust32_consumer ||
451 fd == handle->channel_monitoring_pipes.ust64_consumer) {
452 domain = LTTNG_DOMAIN_UST;
453 } else if (fd == handle->channel_monitoring_pipes.kernel_consumer) {
454 domain = LTTNG_DOMAIN_KERNEL;
455 } else {
456 abort();
457 }
458
459 if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
460 ret = lttng_poll_del(&state->events, fd);
461 if (ret) {
462 ERR("[notification-thread] Failed to remove consumer monitoring pipe from poll set");
463 }
464 goto end;
465 }
466
467 ret = handle_notification_thread_channel_sample(
468 state, fd, domain);
469 if (ret) {
470 ERR("[notification-thread] Consumer sample handling error occurred");
471 ret = -1;
472 goto end;
473 }
474 end:
475 return ret;
476 }
477
478 /*
479 * This thread services notification channel clients and commands received
480 * from various lttng-sessiond components over a command queue.
481 */
482 void *thread_notification(void *data)
483 {
484 int ret;
485 struct notification_thread_handle *handle = data;
486 struct notification_thread_state state;
487
488 DBG("[notification-thread] Started notification thread");
489
490 if (!handle) {
491 ERR("[notification-thread] Invalid thread context provided");
492 goto end;
493 }
494
495 rcu_register_thread();
496 rcu_thread_online();
497
498 health_register(health_sessiond, HEALTH_SESSIOND_TYPE_NOTIFICATION);
499 health_code_update();
500
501 ret = init_thread_state(handle, &state);
502 if (ret) {
503 goto end;
504 }
505
506 /* Ready to handle client connections. */
507 sessiond_notify_ready();
508
509 while (true) {
510 int fd_count, i;
511
512 health_poll_entry();
513 DBG("[notification-thread] Entering poll wait");
514 ret = lttng_poll_wait(&state.events, -1);
515 DBG("[notification-thread] Poll wait returned (%i)", ret);
516 health_poll_exit();
517 if (ret < 0) {
518 /*
519 * Restart interrupted system call.
520 */
521 if (errno == EINTR) {
522 continue;
523 }
524 ERR("[notification-thread] Error encountered during lttng_poll_wait (%i)", ret);
525 goto error;
526 }
527
528 fd_count = ret;
529 for (i = 0; i < fd_count; i++) {
530 int fd = LTTNG_POLL_GETFD(&state.events, i);
531 uint32_t revents = LTTNG_POLL_GETEV(&state.events, i);
532
533 if (!revents) {
534 continue;
535 }
536 DBG("[notification-thread] Handling fd (%i) activity (%u)", fd, revents);
537
538 if (fd == state.notification_channel_socket) {
539 if (revents & LPOLLIN) {
540 ret = handle_notification_thread_client_connect(
541 &state);
542 if (ret < 0) {
543 goto error;
544 }
545 } else if (revents &
546 (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
547 ERR("[notification-thread] Notification socket poll error");
548 goto error;
549 } else {
550 ERR("[notification-thread] Unexpected poll events %u for notification socket %i", revents, fd);
551 goto error;
552 }
553 } else if (fd == lttng_pipe_get_readfd(handle->cmd_queue.event_pipe)) {
554 ret = handle_notification_thread_command(handle,
555 &state);
556 if (ret < 0) {
557 DBG("[notification-thread] Error encountered while servicing command queue");
558 goto error;
559 } else if (ret > 0) {
560 goto exit;
561 }
562 } else if (fd == handle->channel_monitoring_pipes.ust32_consumer ||
563 fd == handle->channel_monitoring_pipes.ust64_consumer ||
564 fd == handle->channel_monitoring_pipes.kernel_consumer) {
565 ret = handle_channel_monitoring_pipe(fd,
566 revents, handle, &state);
567 if (ret) {
568 goto error;
569 }
570 } else {
571 /* Activity on a client's socket. */
572 if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
573 /*
574 * It doesn't matter if a command was
575 * pending on the client socket at this
576 * point since it now has no way to
577 * receive the notifications to which
578 * it was subscribing or unsubscribing.
579 */
580 ret = handle_notification_thread_client_disconnect(
581 fd, &state);
582 if (ret) {
583 goto error;
584 }
585 } else {
586 if (revents & LPOLLIN) {
587 ret = handle_notification_thread_client_in(
588 &state, fd);
589 if (ret) {
590 goto error;
591 }
592 }
593
594 if (revents & LPOLLOUT) {
595 ret = handle_notification_thread_client_out(
596 &state, fd);
597 if (ret) {
598 goto error;
599 }
600 }
601 }
602 }
603 }
604 }
605 exit:
606 error:
607 fini_thread_state(&state);
608 health_unregister(health_sessiond);
609 rcu_thread_offline();
610 rcu_unregister_thread();
611 end:
612 return NULL;
613 }
This page took 0.06897 seconds and 5 git commands to generate.