e1c74e358ebff36c9426cba3c2cc82a1ece4b3c8
[lttng-tools.git] / src / bin / lttng-relayd / main.cpp
1 /*
2 * Copyright (C) 2012 Julien Desfossez <jdesfossez@efficios.com>
3 * Copyright (C) 2012 David Goulet <dgoulet@efficios.com>
4 * Copyright (C) 2013 Jérémie Galarneau <jeremie.galarneau@efficios.com>
5 * Copyright (C) 2015 Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
6 *
7 * SPDX-License-Identifier: GPL-2.0-only
8 *
9 */
10
11 #define _LGPL_SOURCE
12 #include <getopt.h>
13 #include <grp.h>
14 #include <limits.h>
15 #include <pthread.h>
16 #include <signal.h>
17 #include <stdio.h>
18 #include <stdlib.h>
19 #include <string.h>
20 #include <sys/mman.h>
21 #include <sys/mount.h>
22 #include <sys/resource.h>
23 #include <sys/socket.h>
24 #include <sys/stat.h>
25 #include <sys/types.h>
26 #include <sys/wait.h>
27 #include <sys/resource.h>
28 #include <inttypes.h>
29 #include <urcu/futex.h>
30 #include <urcu/uatomic.h>
31 #include <urcu/rculist.h>
32 #include <unistd.h>
33 #include <fcntl.h>
34 #include <strings.h>
35 #include <ctype.h>
36 #include <algorithm>
37
38 #include <lttng/lttng.h>
39 #include <common/common.hpp>
40 #include <common/compat/poll.hpp>
41 #include <common/compat/socket.hpp>
42 #include <common/compat/endian.hpp>
43 #include <common/compat/getenv.hpp>
44 #include <common/defaults.hpp>
45 #include <common/daemonize.hpp>
46 #include <common/futex.hpp>
47 #include <common/sessiond-comm/sessiond-comm.hpp>
48 #include <common/sessiond-comm/inet.hpp>
49 #include <common/sessiond-comm/relayd.hpp>
50 #include <common/uri.hpp>
51 #include <common/utils.hpp>
52 #include <common/path.hpp>
53 #include <common/align.hpp>
54 #include <common/ini-config/ini-config.hpp>
55 #include <common/dynamic-buffer.hpp>
56 #include <common/buffer-view.hpp>
57 #include <common/string-utils/format.hpp>
58 #include <common/fd-tracker/fd-tracker.hpp>
59 #include <common/fd-tracker/utils.hpp>
60
61 #include "backward-compatibility-group-by.hpp"
62 #include "cmd.hpp"
63 #include "connection.hpp"
64 #include "ctf-trace.hpp"
65 #include "health-relayd.hpp"
66 #include "index.hpp"
67 #include "live.hpp"
68 #include "lttng-relayd.hpp"
69 #include "session.hpp"
70 #include "sessiond-trace-chunks.hpp"
71 #include "stream.hpp"
72 #include "tcp_keep_alive.hpp"
73 #include "testpoint.hpp"
74 #include "tracefile-array.hpp"
75 #include "utils.hpp"
76 #include "version.hpp"
77 #include "viewer-stream.hpp"
78
79 static const char *help_msg =
80 #ifdef LTTNG_EMBED_HELP
81 #include <lttng-relayd.8.h>
82 #else
83 NULL
84 #endif
85 ;
86
87 enum relay_connection_status {
88 RELAY_CONNECTION_STATUS_OK,
89 /* An error occurred while processing an event on the connection. */
90 RELAY_CONNECTION_STATUS_ERROR,
91 /* Connection closed/shutdown cleanly. */
92 RELAY_CONNECTION_STATUS_CLOSED,
93 };
94
95 /* command line options */
96 char *opt_output_path, *opt_working_directory;
97 static int opt_daemon, opt_background, opt_print_version, opt_allow_clear = 1;
98 enum relay_group_output_by opt_group_output_by = RELAYD_GROUP_OUTPUT_BY_UNKNOWN;
99
100 /* Argument variables */
101 int lttng_opt_quiet; /* not static in error.h */
102 int lttng_opt_verbose; /* not static in error.h */
103 int lttng_opt_mi; /* not static in error.h */
104
105 /*
106 * We need to wait for listener and live listener threads, as well as
107 * health check thread, before being ready to signal readiness.
108 */
109 #define NR_LTTNG_RELAY_READY 3
110 static int lttng_relay_ready = NR_LTTNG_RELAY_READY;
111
112 /* Size of receive buffer. */
113 #define RECV_DATA_BUFFER_SIZE 65536
114
115 static int recv_child_signal; /* Set to 1 when a SIGUSR1 signal is received. */
116 static pid_t child_ppid; /* Internal parent PID use with daemonize. */
117
118 static struct lttng_uri *control_uri;
119 static struct lttng_uri *data_uri;
120 static struct lttng_uri *live_uri;
121
122 const char *progname;
123
124 const char *tracing_group_name = DEFAULT_TRACING_GROUP;
125 static int tracing_group_name_override;
126
127 const char * const config_section_name = "relayd";
128
129 /*
130 * Quit pipe for all threads. This permits a single cancellation point
131 * for all threads when receiving an event on the pipe.
132 */
133 int thread_quit_pipe[2] = { -1, -1 };
134
135 /*
136 * This pipe is used to inform the worker thread that a command is queued and
137 * ready to be processed.
138 */
139 static int relay_conn_pipe[2] = { -1, -1 };
140
141 /* Shared between threads */
142 static int dispatch_thread_exit;
143
144 static pthread_t listener_thread;
145 static pthread_t dispatcher_thread;
146 static pthread_t worker_thread;
147 static pthread_t health_thread;
148
149 /*
150 * last_relay_stream_id_lock protects last_relay_stream_id increment
151 * atomicity on 32-bit architectures.
152 */
153 static pthread_mutex_t last_relay_stream_id_lock = PTHREAD_MUTEX_INITIALIZER;
154 static uint64_t last_relay_stream_id;
155
156 /*
157 * Relay command queue.
158 *
159 * The relay_thread_listener and relay_thread_dispatcher communicate with this
160 * queue.
161 */
162 static struct relay_conn_queue relay_conn_queue;
163
164 /* Cap of file desriptors to be in simultaneous use by the relay daemon. */
165 static unsigned int lttng_opt_fd_pool_size = -1;
166
167 /* Global relay stream hash table. */
168 struct lttng_ht *relay_streams_ht;
169
170 /* Global relay viewer stream hash table. */
171 struct lttng_ht *viewer_streams_ht;
172
173 /* Global relay sessions hash table. */
174 struct lttng_ht *sessions_ht;
175
176 /* Relayd health monitoring */
177 struct health_app *health_relayd;
178
179 struct sessiond_trace_chunk_registry *sessiond_trace_chunk_registry;
180
181 /* Global fd tracker. */
182 struct fd_tracker *the_fd_tracker;
183
184 static struct option long_options[] = {
185 { "control-port", 1, 0, 'C', },
186 { "data-port", 1, 0, 'D', },
187 { "live-port", 1, 0, 'L', },
188 { "daemonize", 0, 0, 'd', },
189 { "background", 0, 0, 'b', },
190 { "group", 1, 0, 'g', },
191 { "fd-pool-size", 1, 0, '\0', },
192 { "help", 0, 0, 'h', },
193 { "output", 1, 0, 'o', },
194 { "verbose", 0, 0, 'v', },
195 { "config", 1, 0, 'f' },
196 { "version", 0, 0, 'V' },
197 { "working-directory", 1, 0, 'w', },
198 { "group-output-by-session", 0, 0, 's', },
199 { "group-output-by-host", 0, 0, 'p', },
200 { "disallow-clear", 0, 0, 'x' },
201 { NULL, 0, 0, 0, },
202 };
203
204 static const char *config_ignore_options[] = { "help", "config", "version" };
205
206 static void print_version(void) {
207 fprintf(stdout, "%s\n", VERSION);
208 }
209
210 static void relayd_config_log(void)
211 {
212 DBG("LTTng-relayd " VERSION " - " VERSION_NAME "%s%s",
213 GIT_VERSION[0] == '\0' ? "" : " - " GIT_VERSION,
214 EXTRA_VERSION_NAME[0] == '\0' ? "" : " - " EXTRA_VERSION_NAME);
215 if (EXTRA_VERSION_DESCRIPTION[0] != '\0') {
216 DBG("LTTng-relayd extra version description:\n\t" EXTRA_VERSION_DESCRIPTION "\n");
217 }
218 if (EXTRA_VERSION_PATCHES[0] != '\0') {
219 DBG("LTTng-relayd extra patches:\n\t" EXTRA_VERSION_PATCHES "\n");
220 }
221 }
222
223 /*
224 * Take an option from the getopt output and set it in the right variable to be
225 * used later.
226 *
227 * Return 0 on success else a negative value.
228 */
229 static int set_option(int opt, const char *arg, const char *optname)
230 {
231 int ret;
232
233 switch (opt) {
234 case 0:
235 if (!strcmp(optname, "fd-pool-size")) {
236 unsigned long v;
237
238 errno = 0;
239 v = strtoul(arg, NULL, 0);
240 if (errno != 0 || !isdigit((unsigned char) arg[0])) {
241 ERR("Wrong value in --fd-pool-size parameter: %s", arg);
242 ret = -1;
243 goto end;
244 }
245 if (v >= UINT_MAX) {
246 ERR("File descriptor cap overflow in --fd-pool-size parameter: %s", arg);
247 ret = -1;
248 goto end;
249 }
250 lttng_opt_fd_pool_size = (unsigned int) v;
251 } else {
252 fprintf(stderr, "unknown option %s", optname);
253 if (arg) {
254 fprintf(stderr, " with arg %s\n", arg);
255 }
256 }
257 break;
258 case 'C':
259 if (lttng_is_setuid_setgid()) {
260 WARN("Getting '%s' argument from setuid/setgid binary refused for security reasons.",
261 "-C, --control-port");
262 } else {
263 ret = uri_parse(arg, &control_uri);
264 if (ret < 0) {
265 ERR("Invalid control URI specified");
266 goto end;
267 }
268 if (control_uri->port == 0) {
269 control_uri->port = DEFAULT_NETWORK_CONTROL_PORT;
270 }
271 }
272 break;
273 case 'D':
274 if (lttng_is_setuid_setgid()) {
275 WARN("Getting '%s' argument from setuid/setgid binary refused for security reasons.",
276 "-D, -data-port");
277 } else {
278 ret = uri_parse(arg, &data_uri);
279 if (ret < 0) {
280 ERR("Invalid data URI specified");
281 goto end;
282 }
283 if (data_uri->port == 0) {
284 data_uri->port = DEFAULT_NETWORK_DATA_PORT;
285 }
286 }
287 break;
288 case 'L':
289 if (lttng_is_setuid_setgid()) {
290 WARN("Getting '%s' argument from setuid/setgid binary refused for security reasons.",
291 "-L, -live-port");
292 } else {
293 ret = uri_parse(arg, &live_uri);
294 if (ret < 0) {
295 ERR("Invalid live URI specified");
296 goto end;
297 }
298 if (live_uri->port == 0) {
299 live_uri->port = DEFAULT_NETWORK_VIEWER_PORT;
300 }
301 }
302 break;
303 case 'd':
304 opt_daemon = 1;
305 break;
306 case 'b':
307 opt_background = 1;
308 break;
309 case 'g':
310 if (lttng_is_setuid_setgid()) {
311 WARN("Getting '%s' argument from setuid/setgid binary refused for security reasons.",
312 "-g, --group");
313 } else {
314 tracing_group_name = strdup(arg);
315 if (tracing_group_name == NULL) {
316 ret = -errno;
317 PERROR("strdup");
318 goto end;
319 }
320 tracing_group_name_override = 1;
321 }
322 break;
323 case 'h':
324 ret = utils_show_help(8, "lttng-relayd", help_msg);
325 if (ret) {
326 ERR("Cannot show --help for `lttng-relayd`");
327 perror("exec");
328 }
329 exit(EXIT_FAILURE);
330 case 'V':
331 opt_print_version = 1;
332 break;
333 case 'o':
334 if (lttng_is_setuid_setgid()) {
335 WARN("Getting '%s' argument from setuid/setgid binary refused for security reasons.",
336 "-o, --output");
337 } else {
338 ret = asprintf(&opt_output_path, "%s", arg);
339 if (ret < 0) {
340 ret = -errno;
341 PERROR("asprintf opt_output_path");
342 goto end;
343 }
344 }
345 break;
346 case 'w':
347 if (lttng_is_setuid_setgid()) {
348 WARN("Getting '%s' argument from setuid/setgid binary refused for security reasons.",
349 "-w, --working-directory");
350 } else {
351 ret = asprintf(&opt_working_directory, "%s", arg);
352 if (ret < 0) {
353 ret = -errno;
354 PERROR("asprintf opt_working_directory");
355 goto end;
356 }
357 }
358 break;
359
360 case 'v':
361 /* Verbose level can increase using multiple -v */
362 if (arg) {
363 lttng_opt_verbose = config_parse_value(arg);
364 } else {
365 /* Only 3 level of verbosity (-vvv). */
366 if (lttng_opt_verbose < 3) {
367 lttng_opt_verbose += 1;
368 }
369 }
370 break;
371 case 's':
372 if (opt_group_output_by != RELAYD_GROUP_OUTPUT_BY_UNKNOWN) {
373 ERR("Cannot set --group-output-by-session, another --group-output-by argument is present");
374 exit(EXIT_FAILURE);
375 }
376 opt_group_output_by = RELAYD_GROUP_OUTPUT_BY_SESSION;
377 break;
378 case 'p':
379 if (opt_group_output_by != RELAYD_GROUP_OUTPUT_BY_UNKNOWN) {
380 ERR("Cannot set --group-output-by-host, another --group-output-by argument is present");
381 exit(EXIT_FAILURE);
382 }
383 opt_group_output_by = RELAYD_GROUP_OUTPUT_BY_HOST;
384 break;
385 case 'x':
386 /* Disallow clear */
387 opt_allow_clear = 0;
388 break;
389 default:
390 /* Unknown option or other error.
391 * Error is printed by getopt, just return */
392 ret = -1;
393 goto end;
394 }
395
396 /* All good. */
397 ret = 0;
398
399 end:
400 return ret;
401 }
402
403 /*
404 * config_entry_handler_cb used to handle options read from a config file.
405 * See config_entry_handler_cb comment in common/config/session-config.h for the
406 * return value conventions.
407 */
408 static int config_entry_handler(const struct config_entry *entry,
409 void *unused __attribute__((unused)))
410 {
411 int ret = 0, i;
412
413 if (!entry || !entry->name || !entry->value) {
414 ret = -EINVAL;
415 goto end;
416 }
417
418 /* Check if the option is to be ignored */
419 for (i = 0; i < sizeof(config_ignore_options) / sizeof(char *); i++) {
420 if (!strcmp(entry->name, config_ignore_options[i])) {
421 goto end;
422 }
423 }
424
425 for (i = 0; i < (sizeof(long_options) / sizeof(struct option)) - 1; i++) {
426 /* Ignore if entry name is not fully matched. */
427 if (strcmp(entry->name, long_options[i].name)) {
428 continue;
429 }
430
431 /*
432 * If the option takes no argument on the command line,
433 * we have to check if the value is "true". We support
434 * non-zero numeric values, true, on and yes.
435 */
436 if (!long_options[i].has_arg) {
437 ret = config_parse_value(entry->value);
438 if (ret <= 0) {
439 if (ret) {
440 WARN("Invalid configuration value \"%s\" for option %s",
441 entry->value, entry->name);
442 }
443 /* False, skip boolean config option. */
444 goto end;
445 }
446 }
447
448 ret = set_option(long_options[i].val, entry->value, entry->name);
449 goto end;
450 }
451
452 WARN("Unrecognized option \"%s\" in daemon configuration file.",
453 entry->name);
454
455 end:
456 return ret;
457 }
458
459 static int parse_env_options(void)
460 {
461 int ret = 0;
462 char *value = NULL;
463
464 value = lttng_secure_getenv(DEFAULT_LTTNG_RELAYD_WORKING_DIRECTORY_ENV);
465 if (value) {
466 opt_working_directory = strdup(value);
467 if (!opt_working_directory) {
468 ERR("Failed to allocate working directory string (\"%s\")",
469 value);
470 ret = -1;
471 }
472 }
473 return ret;
474 }
475
476 static int set_fd_pool_size(void)
477 {
478 int ret = 0;
479 struct rlimit rlimit;
480
481 ret = getrlimit(RLIMIT_NOFILE, &rlimit);
482 if (ret) {
483 PERROR("Failed to get file descriptor limit");
484 ret = -1;
485 goto end;
486 }
487
488 DBG("File descriptor count limits are %" PRIu64 " (soft) and %" PRIu64 " (hard)",
489 (uint64_t) rlimit.rlim_cur,
490 (uint64_t) rlimit.rlim_max);
491 if (lttng_opt_fd_pool_size == -1) {
492 /* Use default value (soft limit - reserve). */
493 if (rlimit.rlim_cur < DEFAULT_RELAYD_MIN_FD_POOL_SIZE) {
494 ERR("The process' file number limit is too low (%" PRIu64 "). The process' file number limit must be set to at least %i.",
495 (uint64_t) rlimit.rlim_cur, DEFAULT_RELAYD_MIN_FD_POOL_SIZE);
496 ret = -1;
497 goto end;
498 }
499 lttng_opt_fd_pool_size = rlimit.rlim_cur -
500 DEFAULT_RELAYD_FD_POOL_SIZE_RESERVE;
501 goto end;
502 }
503
504 if (lttng_opt_fd_pool_size < DEFAULT_RELAYD_MIN_FD_POOL_SIZE) {
505 ERR("File descriptor pool size must be set to at least %d",
506 DEFAULT_RELAYD_MIN_FD_POOL_SIZE);
507 ret = -1;
508 goto end;
509 }
510
511 if (lttng_opt_fd_pool_size > rlimit.rlim_cur) {
512 ERR("File descriptor pool size argument (%u) exceeds the process' soft limit (%" PRIu64 ").",
513 lttng_opt_fd_pool_size, (uint64_t) rlimit.rlim_cur);
514 ret = -1;
515 goto end;
516 }
517
518 DBG("File descriptor pool size argument (%u) adjusted to %u to accommodates transient fd uses",
519 lttng_opt_fd_pool_size,
520 lttng_opt_fd_pool_size - DEFAULT_RELAYD_FD_POOL_SIZE_RESERVE);
521 lttng_opt_fd_pool_size -= DEFAULT_RELAYD_FD_POOL_SIZE_RESERVE;
522 end:
523 return ret;
524 }
525
526 static int set_options(int argc, char **argv)
527 {
528 int c, ret = 0, option_index = 0, retval = 0;
529 int orig_optopt = optopt, orig_optind = optind;
530 char *default_address, *optstring;
531 char *config_path = NULL;
532
533 optstring = utils_generate_optstring(long_options,
534 sizeof(long_options) / sizeof(struct option));
535 if (!optstring) {
536 retval = -ENOMEM;
537 goto exit;
538 }
539
540 /* Check for the --config option */
541
542 while ((c = getopt_long(argc, argv, optstring, long_options,
543 &option_index)) != -1) {
544 if (c == '?') {
545 retval = -EINVAL;
546 goto exit;
547 } else if (c != 'f') {
548 continue;
549 }
550
551 if (lttng_is_setuid_setgid()) {
552 WARN("Getting '%s' argument from setuid/setgid binary refused for security reasons.",
553 "-f, --config");
554 } else {
555 free(config_path);
556 config_path = utils_expand_path(optarg);
557 if (!config_path) {
558 ERR("Failed to resolve path: %s", optarg);
559 }
560 }
561 }
562
563 ret = config_get_section_entries(config_path, config_section_name,
564 config_entry_handler, NULL);
565 if (ret) {
566 if (ret > 0) {
567 ERR("Invalid configuration option at line %i", ret);
568 }
569 retval = -1;
570 goto exit;
571 }
572
573 /* Reset getopt's global state */
574 optopt = orig_optopt;
575 optind = orig_optind;
576 while (1) {
577 c = getopt_long(argc, argv, optstring, long_options, &option_index);
578 if (c == -1) {
579 break;
580 }
581
582 ret = set_option(c, optarg, long_options[option_index].name);
583 if (ret < 0) {
584 retval = -1;
585 goto exit;
586 }
587 }
588
589 /* assign default values */
590 if (control_uri == NULL) {
591 ret = asprintf(&default_address,
592 "tcp://" DEFAULT_NETWORK_CONTROL_BIND_ADDRESS ":%d",
593 DEFAULT_NETWORK_CONTROL_PORT);
594 if (ret < 0) {
595 PERROR("asprintf default data address");
596 retval = -1;
597 goto exit;
598 }
599
600 ret = uri_parse(default_address, &control_uri);
601 free(default_address);
602 if (ret < 0) {
603 ERR("Invalid control URI specified");
604 retval = -1;
605 goto exit;
606 }
607 }
608 if (data_uri == NULL) {
609 ret = asprintf(&default_address,
610 "tcp://" DEFAULT_NETWORK_DATA_BIND_ADDRESS ":%d",
611 DEFAULT_NETWORK_DATA_PORT);
612 if (ret < 0) {
613 PERROR("asprintf default data address");
614 retval = -1;
615 goto exit;
616 }
617
618 ret = uri_parse(default_address, &data_uri);
619 free(default_address);
620 if (ret < 0) {
621 ERR("Invalid data URI specified");
622 retval = -1;
623 goto exit;
624 }
625 }
626 if (live_uri == NULL) {
627 ret = asprintf(&default_address,
628 "tcp://" DEFAULT_NETWORK_VIEWER_BIND_ADDRESS ":%d",
629 DEFAULT_NETWORK_VIEWER_PORT);
630 if (ret < 0) {
631 PERROR("asprintf default viewer control address");
632 retval = -1;
633 goto exit;
634 }
635
636 ret = uri_parse(default_address, &live_uri);
637 free(default_address);
638 if (ret < 0) {
639 ERR("Invalid viewer control URI specified");
640 retval = -1;
641 goto exit;
642 }
643 }
644 ret = set_fd_pool_size();
645 if (ret) {
646 retval = -1;
647 goto exit;
648 }
649
650 if (opt_group_output_by == RELAYD_GROUP_OUTPUT_BY_UNKNOWN) {
651 opt_group_output_by = RELAYD_GROUP_OUTPUT_BY_HOST;
652 }
653 if (opt_allow_clear) {
654 /* Check if env variable exists. */
655 const char *value = lttng_secure_getenv(DEFAULT_LTTNG_RELAYD_DISALLOW_CLEAR_ENV);
656 if (value) {
657 ret = config_parse_value(value);
658 if (ret < 0) {
659 ERR("Invalid value for %s specified", DEFAULT_LTTNG_RELAYD_DISALLOW_CLEAR_ENV);
660 retval = -1;
661 goto exit;
662 }
663 opt_allow_clear = !ret;
664 }
665 }
666
667 exit:
668 free(config_path);
669 free(optstring);
670 return retval;
671 }
672
673 static void print_global_objects(void)
674 {
675 print_viewer_streams();
676 print_relay_streams();
677 print_sessions();
678 }
679
680 static int noop_close(void *data __attribute__((unused)),
681 int *fds __attribute__((unused)))
682 {
683 return 0;
684 }
685
686 static void untrack_stdio(void)
687 {
688 int fds[] = { fileno(stdout), fileno(stderr) };
689
690 /*
691 * noop_close is used since we don't really want to close
692 * the stdio output fds; we merely want to stop tracking them.
693 */
694 (void) fd_tracker_close_unsuspendable_fd(the_fd_tracker,
695 fds, 2, noop_close, NULL);
696 }
697
698 /*
699 * Cleanup the daemon
700 */
701 static void relayd_cleanup(void)
702 {
703 print_global_objects();
704
705 DBG("Cleaning up");
706
707 if (viewer_streams_ht)
708 lttng_ht_destroy(viewer_streams_ht);
709 if (relay_streams_ht)
710 lttng_ht_destroy(relay_streams_ht);
711 if (sessions_ht)
712 lttng_ht_destroy(sessions_ht);
713
714 free(opt_output_path);
715 free(opt_working_directory);
716
717 if (health_relayd) {
718 health_app_destroy(health_relayd);
719 }
720 /* Close thread quit pipes */
721 if (health_quit_pipe[0] != -1) {
722 (void) fd_tracker_util_pipe_close(
723 the_fd_tracker, health_quit_pipe);
724 }
725 if (thread_quit_pipe[0] != -1) {
726 (void) fd_tracker_util_pipe_close(
727 the_fd_tracker, thread_quit_pipe);
728 }
729 if (sessiond_trace_chunk_registry) {
730 sessiond_trace_chunk_registry_destroy(
731 sessiond_trace_chunk_registry);
732 }
733 if (the_fd_tracker) {
734 untrack_stdio();
735 /*
736 * fd_tracker_destroy() will log the contents of the fd-tracker
737 * if a leak is detected.
738 */
739 fd_tracker_destroy(the_fd_tracker);
740 }
741
742 uri_free(control_uri);
743 uri_free(data_uri);
744 /* Live URI is freed in the live thread. */
745
746 if (tracing_group_name_override) {
747 free((void *) tracing_group_name);
748 }
749 }
750
751 /*
752 * Write to writable pipe used to notify a thread.
753 */
754 static int notify_thread_pipe(int wpipe)
755 {
756 ssize_t ret;
757
758 ret = lttng_write(wpipe, "!", 1);
759 if (ret < 1) {
760 PERROR("write poll pipe");
761 goto end;
762 }
763 ret = 0;
764 end:
765 return ret;
766 }
767
768 static int notify_health_quit_pipe(int *pipe)
769 {
770 ssize_t ret;
771
772 ret = lttng_write(pipe[1], "4", 1);
773 if (ret < 1) {
774 PERROR("write relay health quit");
775 goto end;
776 }
777 ret = 0;
778 end:
779 return ret;
780 }
781
782 /*
783 * Stop all relayd and relayd-live threads.
784 */
785 int lttng_relay_stop_threads(void)
786 {
787 int retval = 0;
788
789 /* Stopping all threads */
790 DBG("Terminating all threads");
791 if (notify_thread_pipe(thread_quit_pipe[1])) {
792 ERR("write error on thread quit pipe");
793 retval = -1;
794 }
795
796 if (notify_health_quit_pipe(health_quit_pipe)) {
797 ERR("write error on health quit pipe");
798 }
799
800 /* Dispatch thread */
801 CMM_STORE_SHARED(dispatch_thread_exit, 1);
802 futex_nto1_wake(&relay_conn_queue.futex);
803
804 if (relayd_live_stop()) {
805 ERR("Error stopping live threads");
806 retval = -1;
807 }
808 return retval;
809 }
810
811 /*
812 * Signal handler for the daemon
813 *
814 * Simply stop all worker threads, leaving main() return gracefully after
815 * joining all threads and calling cleanup().
816 */
817 static void sighandler(int sig)
818 {
819 switch (sig) {
820 case SIGINT:
821 DBG("SIGINT caught");
822 if (lttng_relay_stop_threads()) {
823 ERR("Error stopping threads");
824 }
825 break;
826 case SIGTERM:
827 DBG("SIGTERM caught");
828 if (lttng_relay_stop_threads()) {
829 ERR("Error stopping threads");
830 }
831 break;
832 case SIGUSR1:
833 CMM_STORE_SHARED(recv_child_signal, 1);
834 break;
835 default:
836 break;
837 }
838 }
839
840 /*
841 * Setup signal handler for :
842 * SIGINT, SIGTERM, SIGPIPE
843 */
844 static int set_signal_handler(void)
845 {
846 int ret = 0;
847 struct sigaction sa;
848 sigset_t sigset;
849
850 if ((ret = sigemptyset(&sigset)) < 0) {
851 PERROR("sigemptyset");
852 return ret;
853 }
854
855 sa.sa_mask = sigset;
856 sa.sa_flags = 0;
857
858 sa.sa_handler = sighandler;
859 if ((ret = sigaction(SIGTERM, &sa, NULL)) < 0) {
860 PERROR("sigaction");
861 return ret;
862 }
863
864 if ((ret = sigaction(SIGINT, &sa, NULL)) < 0) {
865 PERROR("sigaction");
866 return ret;
867 }
868
869 if ((ret = sigaction(SIGUSR1, &sa, NULL)) < 0) {
870 PERROR("sigaction");
871 return ret;
872 }
873
874 sa.sa_handler = SIG_IGN;
875 if ((ret = sigaction(SIGPIPE, &sa, NULL)) < 0) {
876 PERROR("sigaction");
877 return ret;
878 }
879
880 DBG("Signal handler set for SIGTERM, SIGUSR1, SIGPIPE and SIGINT");
881
882 return ret;
883 }
884
885 void lttng_relay_notify_ready(void)
886 {
887 /* Notify the parent of the fork() process that we are ready. */
888 if (opt_daemon || opt_background) {
889 if (uatomic_sub_return(&lttng_relay_ready, 1) == 0) {
890 kill(child_ppid, SIGUSR1);
891 }
892 }
893 }
894
895 /*
896 * Init thread quit pipe.
897 *
898 * Return -1 on error or 0 if all pipes are created.
899 */
900 static int init_thread_quit_pipe(void)
901 {
902 return fd_tracker_util_pipe_open_cloexec(
903 the_fd_tracker, "Quit pipe", thread_quit_pipe);
904 }
905
906 /*
907 * Init health quit pipe.
908 *
909 * Return -1 on error or 0 if all pipes are created.
910 */
911 static int init_health_quit_pipe(void)
912 {
913 return fd_tracker_util_pipe_open_cloexec(the_fd_tracker,
914 "Health quit pipe", health_quit_pipe);
915 }
916
917 /*
918 * Create a poll set with O_CLOEXEC and add the thread quit pipe to the set.
919 */
920 static int create_named_thread_poll_set(struct lttng_poll_event *events,
921 int size, const char *name)
922 {
923 int ret;
924
925 if (events == NULL || size == 0) {
926 ret = -1;
927 goto error;
928 }
929
930 ret = fd_tracker_util_poll_create(the_fd_tracker,
931 name, events, 1, LTTNG_CLOEXEC);
932 if (ret) {
933 PERROR("Failed to create \"%s\" poll file descriptor", name);
934 goto error;
935 }
936
937 /* Add quit pipe */
938 ret = lttng_poll_add(events, thread_quit_pipe[0], LPOLLIN | LPOLLERR);
939 if (ret < 0) {
940 goto error;
941 }
942
943 return 0;
944
945 error:
946 return ret;
947 }
948
949 /*
950 * Check if the thread quit pipe was triggered.
951 *
952 * Return 1 if it was triggered else 0;
953 */
954 static int check_thread_quit_pipe(int fd, uint32_t events)
955 {
956 if (fd == thread_quit_pipe[0] && (events & LPOLLIN)) {
957 return 1;
958 }
959
960 return 0;
961 }
962
963 static int create_sock(void *data, int *out_fd)
964 {
965 int ret;
966 struct lttcomm_sock *sock = (lttcomm_sock *) data;
967
968 ret = lttcomm_create_sock(sock);
969 if (ret < 0) {
970 goto end;
971 }
972
973 *out_fd = sock->fd;
974 end:
975 return ret;
976 }
977
978 static int close_sock(void *data, int *in_fd __attribute__((unused)))
979 {
980 struct lttcomm_sock *sock = (lttcomm_sock *) data;
981
982 return sock->ops->close(sock);
983 }
984
985 static int accept_sock(void *data, int *out_fd)
986 {
987 int ret = 0;
988 /* Socks is an array of in_sock, out_sock. */
989 struct lttcomm_sock **socks = (lttcomm_sock **) data;
990 struct lttcomm_sock *in_sock = socks[0];
991
992 socks[1] = in_sock->ops->accept(in_sock);
993 if (!socks[1]) {
994 ret = -1;
995 goto end;
996 }
997 *out_fd = socks[1]->fd;
998 end:
999 return ret;
1000 }
1001
1002 /*
1003 * Create and init socket from uri.
1004 */
1005 static struct lttcomm_sock *relay_socket_create(struct lttng_uri *uri,
1006 const char *name)
1007 {
1008 int ret, sock_fd;
1009 struct lttcomm_sock *sock = NULL;
1010 char uri_str[PATH_MAX];
1011 char *formated_name = NULL;
1012
1013 sock = lttcomm_alloc_sock_from_uri(uri);
1014 if (sock == NULL) {
1015 ERR("Allocating socket");
1016 goto error;
1017 }
1018
1019 /*
1020 * Don't fail to create the socket if the name can't be built as it is
1021 * only used for debugging purposes.
1022 */
1023 ret = uri_to_str_url(uri, uri_str, sizeof(uri_str));
1024 uri_str[sizeof(uri_str) - 1] = '\0';
1025 if (ret >= 0) {
1026 ret = asprintf(&formated_name, "%s socket @ %s", name,
1027 uri_str);
1028 if (ret < 0) {
1029 formated_name = NULL;
1030 }
1031 }
1032
1033 ret = fd_tracker_open_unsuspendable_fd(the_fd_tracker, &sock_fd,
1034 (const char **) (formated_name ? &formated_name : NULL),
1035 1, create_sock, sock);
1036 if (ret) {
1037 PERROR("Failed to open \"%s\" relay socket",
1038 formated_name ?: "Unknown");
1039 goto error;
1040 }
1041 DBG("Listening on %s socket %d", name, sock->fd);
1042
1043 ret = sock->ops->bind(sock);
1044 if (ret < 0) {
1045 PERROR("Failed to bind socket");
1046 goto error;
1047 }
1048
1049 ret = sock->ops->listen(sock, -1);
1050 if (ret < 0) {
1051 goto error;
1052
1053 }
1054
1055 free(formated_name);
1056 return sock;
1057
1058 error:
1059 if (sock) {
1060 lttcomm_destroy_sock(sock);
1061 }
1062 free(formated_name);
1063 return NULL;
1064 }
1065
1066 static
1067 struct lttcomm_sock *accept_relayd_sock(struct lttcomm_sock *listening_sock,
1068 const char *name)
1069 {
1070 int out_fd, ret;
1071 struct lttcomm_sock *socks[2] = { listening_sock, NULL };
1072 struct lttcomm_sock *new_sock = NULL;
1073
1074 ret = fd_tracker_open_unsuspendable_fd(
1075 the_fd_tracker, &out_fd,
1076 (const char **) &name,
1077 1, accept_sock, &socks);
1078 if (ret) {
1079 goto end;
1080 }
1081 new_sock = socks[1];
1082 DBG("%s accepted, socket %d", name, new_sock->fd);
1083 end:
1084 return new_sock;
1085 }
1086
1087 /*
1088 * This thread manages the listening for new connections on the network
1089 */
1090 static void *relay_thread_listener(void *data __attribute__((unused)))
1091 {
1092 int i, ret, pollfd, err = -1;
1093 uint32_t revents, nb_fd;
1094 struct lttng_poll_event events;
1095 struct lttcomm_sock *control_sock, *data_sock;
1096
1097 DBG("[thread] Relay listener started");
1098
1099 rcu_register_thread();
1100 health_register(health_relayd, HEALTH_RELAYD_TYPE_LISTENER);
1101
1102 health_code_update();
1103
1104 control_sock = relay_socket_create(control_uri, "Control listener");
1105 if (!control_sock) {
1106 goto error_sock_control;
1107 }
1108
1109 data_sock = relay_socket_create(data_uri, "Data listener");
1110 if (!data_sock) {
1111 goto error_sock_relay;
1112 }
1113
1114 /*
1115 * Pass 3 as size here for the thread quit pipe, control and
1116 * data socket.
1117 */
1118 ret = create_named_thread_poll_set(&events, 3, "Listener thread epoll");
1119 if (ret < 0) {
1120 goto error_create_poll;
1121 }
1122
1123 /* Add the control socket */
1124 ret = lttng_poll_add(&events, control_sock->fd, LPOLLIN | LPOLLRDHUP);
1125 if (ret < 0) {
1126 goto error_poll_add;
1127 }
1128
1129 /* Add the data socket */
1130 ret = lttng_poll_add(&events, data_sock->fd, LPOLLIN | LPOLLRDHUP);
1131 if (ret < 0) {
1132 goto error_poll_add;
1133 }
1134
1135 lttng_relay_notify_ready();
1136
1137 if (testpoint(relayd_thread_listener)) {
1138 goto error_testpoint;
1139 }
1140
1141 while (1) {
1142 health_code_update();
1143
1144 DBG("Listener accepting connections");
1145
1146 restart:
1147 health_poll_entry();
1148 ret = lttng_poll_wait(&events, -1);
1149 health_poll_exit();
1150 if (ret < 0) {
1151 /*
1152 * Restart interrupted system call.
1153 */
1154 if (errno == EINTR) {
1155 goto restart;
1156 }
1157 goto error;
1158 }
1159
1160 nb_fd = ret;
1161
1162 DBG("Relay new connection received");
1163 for (i = 0; i < nb_fd; i++) {
1164 health_code_update();
1165
1166 /* Fetch once the poll data */
1167 revents = LTTNG_POLL_GETEV(&events, i);
1168 pollfd = LTTNG_POLL_GETFD(&events, i);
1169
1170 /* Thread quit pipe has been closed. Killing thread. */
1171 ret = check_thread_quit_pipe(pollfd, revents);
1172 if (ret) {
1173 err = 0;
1174 goto exit;
1175 }
1176
1177 if (revents & LPOLLIN) {
1178 /*
1179 * A new connection is requested, therefore a
1180 * sessiond/consumerd connection is allocated in
1181 * this thread, enqueued to a global queue and
1182 * dequeued (and freed) in the worker thread.
1183 */
1184 int val = 1;
1185 struct relay_connection *new_conn;
1186 struct lttcomm_sock *newsock = NULL;
1187 enum connection_type type;
1188
1189 if (pollfd == data_sock->fd) {
1190 type = RELAY_DATA;
1191 newsock = accept_relayd_sock(data_sock,
1192 "Data socket to relayd");
1193 } else {
1194 LTTNG_ASSERT(pollfd == control_sock->fd);
1195 type = RELAY_CONTROL;
1196 newsock = accept_relayd_sock(control_sock,
1197 "Control socket to relayd");
1198 }
1199 if (!newsock) {
1200 PERROR("accepting sock");
1201 goto error;
1202 }
1203
1204 ret = setsockopt(newsock->fd, SOL_SOCKET, SO_REUSEADDR, &val,
1205 sizeof(val));
1206 if (ret < 0) {
1207 PERROR("setsockopt inet");
1208 lttcomm_destroy_sock(newsock);
1209 goto error;
1210 }
1211
1212 ret = socket_apply_keep_alive_config(newsock->fd);
1213 if (ret < 0) {
1214 ERR("Failed to apply TCP keep-alive configuration on socket (%i)",
1215 newsock->fd);
1216 lttcomm_destroy_sock(newsock);
1217 goto error;
1218 }
1219
1220 new_conn = connection_create(newsock, type);
1221 if (!new_conn) {
1222 lttcomm_destroy_sock(newsock);
1223 goto error;
1224 }
1225
1226 /* Enqueue request for the dispatcher thread. */
1227 cds_wfcq_head_ptr_t head;
1228 head.h = &relay_conn_queue.head;
1229 cds_wfcq_enqueue(head, &relay_conn_queue.tail,
1230 &new_conn->qnode);
1231
1232 /*
1233 * Wake the dispatch queue futex.
1234 * Implicit memory barrier with the
1235 * exchange in cds_wfcq_enqueue.
1236 */
1237 futex_nto1_wake(&relay_conn_queue.futex);
1238 } else if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
1239 ERR("socket poll error");
1240 goto error;
1241 } else {
1242 ERR("Unexpected poll events %u for sock %d", revents, pollfd);
1243 goto error;
1244 }
1245 }
1246 }
1247
1248 exit:
1249 error:
1250 error_poll_add:
1251 error_testpoint:
1252 (void) fd_tracker_util_poll_clean(the_fd_tracker, &events);
1253 error_create_poll:
1254 if (data_sock->fd >= 0) {
1255 int data_sock_fd = data_sock->fd;
1256
1257 ret = fd_tracker_close_unsuspendable_fd(the_fd_tracker,
1258 &data_sock_fd, 1, close_sock,
1259 data_sock);
1260 if (ret) {
1261 PERROR("Failed to close the data listener socket file descriptor");
1262 }
1263 data_sock->fd = -1;
1264 }
1265 lttcomm_destroy_sock(data_sock);
1266 error_sock_relay:
1267 if (control_sock->fd >= 0) {
1268 int control_sock_fd = control_sock->fd;
1269
1270 ret = fd_tracker_close_unsuspendable_fd(the_fd_tracker,
1271 &control_sock_fd, 1, close_sock,
1272 control_sock);
1273 if (ret) {
1274 PERROR("Failed to close the control listener socket file descriptor");
1275 }
1276 control_sock->fd = -1;
1277 }
1278 lttcomm_destroy_sock(control_sock);
1279 error_sock_control:
1280 if (err) {
1281 health_error();
1282 ERR("Health error occurred in %s", __func__);
1283 }
1284 health_unregister(health_relayd);
1285 rcu_unregister_thread();
1286 DBG("Relay listener thread cleanup complete");
1287 lttng_relay_stop_threads();
1288 return NULL;
1289 }
1290
1291 /*
1292 * This thread manages the dispatching of the requests to worker threads
1293 */
1294 static void *relay_thread_dispatcher(void *data __attribute__((unused)))
1295 {
1296 int err = -1;
1297 ssize_t ret;
1298 struct cds_wfcq_node *node;
1299 struct relay_connection *new_conn = NULL;
1300
1301 DBG("[thread] Relay dispatcher started");
1302
1303 health_register(health_relayd, HEALTH_RELAYD_TYPE_DISPATCHER);
1304
1305 if (testpoint(relayd_thread_dispatcher)) {
1306 goto error_testpoint;
1307 }
1308
1309 health_code_update();
1310
1311 for (;;) {
1312 health_code_update();
1313
1314 /* Atomically prepare the queue futex */
1315 futex_nto1_prepare(&relay_conn_queue.futex);
1316
1317 if (CMM_LOAD_SHARED(dispatch_thread_exit)) {
1318 break;
1319 }
1320
1321 do {
1322 health_code_update();
1323
1324 /* Dequeue commands */
1325 node = cds_wfcq_dequeue_blocking(&relay_conn_queue.head,
1326 &relay_conn_queue.tail);
1327 if (node == NULL) {
1328 DBG("Woken up but nothing in the relay command queue");
1329 /* Continue thread execution */
1330 break;
1331 }
1332 new_conn = caa_container_of(node, struct relay_connection, qnode);
1333
1334 DBG("Dispatching request waiting on sock %d", new_conn->sock->fd);
1335
1336 /*
1337 * Inform worker thread of the new request. This
1338 * call is blocking so we can be assured that
1339 * the data will be read at some point in time
1340 * or wait to the end of the world :)
1341 */
1342 ret = lttng_write(relay_conn_pipe[1], &new_conn, sizeof(new_conn));
1343 if (ret < 0) {
1344 PERROR("write connection pipe");
1345 connection_put(new_conn);
1346 goto error;
1347 }
1348 } while (node != NULL);
1349
1350 /* Futex wait on queue. Blocking call on futex() */
1351 health_poll_entry();
1352 futex_nto1_wait(&relay_conn_queue.futex);
1353 health_poll_exit();
1354 }
1355
1356 /* Normal exit, no error */
1357 err = 0;
1358
1359 error:
1360 error_testpoint:
1361 if (err) {
1362 health_error();
1363 ERR("Health error occurred in %s", __func__);
1364 }
1365 health_unregister(health_relayd);
1366 DBG("Dispatch thread dying");
1367 lttng_relay_stop_threads();
1368 return NULL;
1369 }
1370
1371 static bool session_streams_have_index(const struct relay_session *session)
1372 {
1373 return session->minor >= 4 && !session->snapshot;
1374 }
1375
1376 /*
1377 * Handle the RELAYD_CREATE_SESSION command.
1378 *
1379 * On success, send back the session id or else return a negative value.
1380 */
1381 static int relay_create_session(
1382 const struct lttcomm_relayd_hdr *recv_hdr __attribute__((unused)),
1383 struct relay_connection *conn,
1384 const struct lttng_buffer_view *payload)
1385 {
1386 int ret = 0;
1387 ssize_t send_ret;
1388 struct relay_session *session = NULL;
1389 struct lttcomm_relayd_create_session_reply_2_11 reply = {};
1390 char session_name[LTTNG_NAME_MAX] = {};
1391 char hostname[LTTNG_HOST_NAME_MAX] = {};
1392 uint32_t live_timer = 0;
1393 bool snapshot = false;
1394 bool session_name_contains_creation_timestamp = false;
1395 /* Left nil for peers < 2.11. */
1396 char base_path[LTTNG_PATH_MAX] = {};
1397 lttng_uuid sessiond_uuid = {};
1398 LTTNG_OPTIONAL(uint64_t) id_sessiond = {};
1399 LTTNG_OPTIONAL(uint64_t) current_chunk_id = {};
1400 LTTNG_OPTIONAL(time_t) creation_time = {};
1401 struct lttng_dynamic_buffer reply_payload;
1402
1403 lttng_dynamic_buffer_init(&reply_payload);
1404
1405 if (conn->minor < 4) {
1406 /* From 2.1 to 2.3 */
1407 ret = 0;
1408 } else if (conn->minor >= 4 && conn->minor < 11) {
1409 /* From 2.4 to 2.10 */
1410 ret = cmd_create_session_2_4(payload, session_name,
1411 hostname, &live_timer, &snapshot);
1412 } else {
1413 bool has_current_chunk;
1414 uint64_t current_chunk_id_value;
1415 time_t creation_time_value;
1416 uint64_t id_sessiond_value;
1417
1418 /* From 2.11 to ... */
1419 ret = cmd_create_session_2_11(payload, session_name, hostname,
1420 base_path, &live_timer, &snapshot, &id_sessiond_value,
1421 sessiond_uuid, &has_current_chunk,
1422 &current_chunk_id_value, &creation_time_value,
1423 &session_name_contains_creation_timestamp);
1424 if (lttng_uuid_is_nil(sessiond_uuid)) {
1425 /* The nil UUID is reserved for pre-2.11 clients. */
1426 ERR("Illegal nil UUID announced by peer in create session command");
1427 ret = -1;
1428 goto send_reply;
1429 }
1430 LTTNG_OPTIONAL_SET(&id_sessiond, id_sessiond_value);
1431 LTTNG_OPTIONAL_SET(&creation_time, creation_time_value);
1432 if (has_current_chunk) {
1433 LTTNG_OPTIONAL_SET(&current_chunk_id,
1434 current_chunk_id_value);
1435 }
1436 }
1437
1438 if (ret < 0) {
1439 goto send_reply;
1440 }
1441
1442 session = session_create(session_name, hostname, base_path, live_timer,
1443 snapshot, sessiond_uuid,
1444 id_sessiond.is_set ? &id_sessiond.value : NULL,
1445 current_chunk_id.is_set ? &current_chunk_id.value : NULL,
1446 creation_time.is_set ? &creation_time.value : NULL,
1447 conn->major, conn->minor,
1448 session_name_contains_creation_timestamp);
1449 if (!session) {
1450 ret = -1;
1451 goto send_reply;
1452 }
1453 LTTNG_ASSERT(!conn->session);
1454 conn->session = session;
1455 DBG("Created session %" PRIu64, session->id);
1456
1457 reply.generic.session_id = htobe64(session->id);
1458
1459 send_reply:
1460 if (ret < 0) {
1461 reply.generic.ret_code = htobe32(LTTNG_ERR_FATAL);
1462 } else {
1463 reply.generic.ret_code = htobe32(LTTNG_OK);
1464 }
1465
1466 if (conn->minor < 11) {
1467 /* From 2.1 to 2.10 */
1468 ret = lttng_dynamic_buffer_append(&reply_payload,
1469 &reply.generic, sizeof(reply.generic));
1470 if (ret) {
1471 ERR("Failed to append \"create session\" command reply header to payload buffer");
1472 ret = -1;
1473 goto end;
1474 }
1475 } else {
1476 const uint32_t output_path_length =
1477 session ? strlen(session->output_path) + 1 : 0;
1478
1479 reply.output_path_length = htobe32(output_path_length);
1480 ret = lttng_dynamic_buffer_append(
1481 &reply_payload, &reply, sizeof(reply));
1482 if (ret) {
1483 ERR("Failed to append \"create session\" command reply header to payload buffer");
1484 goto end;
1485 }
1486
1487 if (output_path_length) {
1488 ret = lttng_dynamic_buffer_append(&reply_payload,
1489 session->output_path,
1490 output_path_length);
1491 if (ret) {
1492 ERR("Failed to append \"create session\" command reply path to payload buffer");
1493 goto end;
1494 }
1495 }
1496 }
1497
1498 send_ret = conn->sock->ops->sendmsg(conn->sock, reply_payload.data,
1499 reply_payload.size, 0);
1500 if (send_ret < (ssize_t) reply_payload.size) {
1501 ERR("Failed to send \"create session\" command reply of %zu bytes (ret = %zd)",
1502 reply_payload.size, send_ret);
1503 ret = -1;
1504 }
1505 end:
1506 if (ret < 0 && session) {
1507 session_put(session);
1508 }
1509 lttng_dynamic_buffer_reset(&reply_payload);
1510 return ret;
1511 }
1512
1513 /*
1514 * When we have received all the streams and the metadata for a channel,
1515 * we make them visible to the viewer threads.
1516 */
1517 static void publish_connection_local_streams(struct relay_connection *conn)
1518 {
1519 struct relay_stream *stream;
1520 struct relay_session *session = conn->session;
1521
1522 /*
1523 * We publish all streams belonging to a session atomically wrt
1524 * session lock.
1525 */
1526 pthread_mutex_lock(&session->lock);
1527 rcu_read_lock();
1528 cds_list_for_each_entry_rcu(stream, &session->recv_list,
1529 recv_node) {
1530 stream_publish(stream);
1531 }
1532 rcu_read_unlock();
1533
1534 /*
1535 * Inform the viewer that there are new streams in the session.
1536 */
1537 if (session->viewer_attached) {
1538 uatomic_set(&session->new_streams, 1);
1539 }
1540 pthread_mutex_unlock(&session->lock);
1541 }
1542
1543 static int conform_channel_path(char *channel_path)
1544 {
1545 int ret = 0;
1546
1547 if (strstr("../", channel_path)) {
1548 ERR("Refusing channel path as it walks up the path hierarchy: \"%s\"",
1549 channel_path);
1550 ret = -1;
1551 goto end;
1552 }
1553
1554 if (*channel_path == '/') {
1555 const size_t len = strlen(channel_path);
1556
1557 /*
1558 * Channel paths from peers prior to 2.11 are expressed as an
1559 * absolute path that is, in reality, relative to the relay
1560 * daemon's output directory. Remove the leading slash so it
1561 * is correctly interpreted as a relative path later on.
1562 *
1563 * len (and not len - 1) is used to copy the trailing NULL.
1564 */
1565 bcopy(channel_path + 1, channel_path, len);
1566 }
1567 end:
1568 return ret;
1569 }
1570
1571 /*
1572 * relay_add_stream: allocate a new stream for a session
1573 */
1574 static int relay_add_stream(
1575 const struct lttcomm_relayd_hdr *recv_hdr __attribute__((unused)),
1576 struct relay_connection *conn,
1577 const struct lttng_buffer_view *payload)
1578 {
1579 int ret;
1580 ssize_t send_ret;
1581 struct relay_session *session = conn->session;
1582 struct relay_stream *stream = NULL;
1583 struct lttcomm_relayd_status_stream reply;
1584 struct ctf_trace *trace = NULL;
1585 uint64_t stream_handle = -1ULL;
1586 char *path_name = NULL, *channel_name = NULL;
1587 uint64_t tracefile_size = 0, tracefile_count = 0;
1588 LTTNG_OPTIONAL(uint64_t) stream_chunk_id = {};
1589
1590 if (!session || !conn->version_check_done) {
1591 ERR("Trying to add a stream before version check");
1592 ret = -1;
1593 goto end_no_session;
1594 }
1595
1596 if (session->minor == 1) {
1597 /* For 2.1 */
1598 ret = cmd_recv_stream_2_1(payload, &path_name,
1599 &channel_name);
1600 } else if (session->minor > 1 && session->minor < 11) {
1601 /* From 2.2 to 2.10 */
1602 ret = cmd_recv_stream_2_2(payload, &path_name,
1603 &channel_name, &tracefile_size, &tracefile_count);
1604 } else {
1605 /* From 2.11 to ... */
1606 ret = cmd_recv_stream_2_11(payload, &path_name,
1607 &channel_name, &tracefile_size, &tracefile_count,
1608 &stream_chunk_id.value);
1609 stream_chunk_id.is_set = true;
1610 }
1611
1612 if (ret < 0) {
1613 goto send_reply;
1614 }
1615
1616 if (conform_channel_path(path_name)) {
1617 goto send_reply;
1618 }
1619
1620 /*
1621 * Backward compatibility for --group-output-by-session.
1622 * Prior to lttng 2.11, the complete path is passed by the stream.
1623 * Starting at 2.11, lttng-relayd uses chunk. When dealing with producer
1624 * >=2.11 the chunk is responsible for the output path. When dealing
1625 * with producer < 2.11 the chunk output_path is the root output path
1626 * and the stream carries the complete path (path_name).
1627 * To support --group-output-by-session with older producer (<2.11), we
1628 * need to craft the path based on the stream path.
1629 */
1630 if (opt_group_output_by == RELAYD_GROUP_OUTPUT_BY_SESSION) {
1631 if (conn->minor < 4) {
1632 /*
1633 * From 2.1 to 2.3, the session_name is not passed on
1634 * the RELAYD_CREATE_SESSION command. The session name
1635 * is necessary to detect the presence of a base_path
1636 * inside the stream path. Without it we cannot perform
1637 * a valid group-output-by-session transformation.
1638 */
1639 WARN("Unable to perform a --group-by-session transformation for session %" PRIu64
1640 " for stream with path \"%s\" as it is produced by a peer using a protocol older than v2.4",
1641 session->id, path_name);
1642 } else if (conn->minor >= 4 && conn->minor < 11) {
1643 char *group_by_session_path_name;
1644
1645 LTTNG_ASSERT(session->session_name[0] != '\0');
1646
1647 group_by_session_path_name =
1648 backward_compat_group_by_session(
1649 path_name,
1650 session->session_name,
1651 session->creation_time.value);
1652 if (!group_by_session_path_name) {
1653 ERR("Failed to apply group by session to stream of session %" PRIu64,
1654 session->id);
1655 goto send_reply;
1656 }
1657
1658 DBG("Transformed session path from \"%s\" to \"%s\" to honor per-session name grouping",
1659 path_name, group_by_session_path_name);
1660
1661 free(path_name);
1662 path_name = group_by_session_path_name;
1663 }
1664 }
1665
1666 trace = ctf_trace_get_by_path_or_create(session, path_name);
1667 if (!trace) {
1668 goto send_reply;
1669 }
1670
1671 /* This stream here has one reference on the trace. */
1672 pthread_mutex_lock(&last_relay_stream_id_lock);
1673 stream_handle = ++last_relay_stream_id;
1674 pthread_mutex_unlock(&last_relay_stream_id_lock);
1675
1676 /* We pass ownership of path_name and channel_name. */
1677 stream = stream_create(trace, stream_handle, path_name,
1678 channel_name, tracefile_size, tracefile_count);
1679 path_name = NULL;
1680 channel_name = NULL;
1681
1682 /*
1683 * Streams are the owners of their trace. Reference to trace is
1684 * kept within stream_create().
1685 */
1686 ctf_trace_put(trace);
1687
1688 send_reply:
1689 memset(&reply, 0, sizeof(reply));
1690 reply.handle = htobe64(stream_handle);
1691 if (!stream) {
1692 reply.ret_code = htobe32(LTTNG_ERR_UNK);
1693 } else {
1694 reply.ret_code = htobe32(LTTNG_OK);
1695 }
1696
1697 send_ret = conn->sock->ops->sendmsg(conn->sock, &reply,
1698 sizeof(struct lttcomm_relayd_status_stream), 0);
1699 if (send_ret < (ssize_t) sizeof(reply)) {
1700 ERR("Failed to send \"add stream\" command reply (ret = %zd)",
1701 send_ret);
1702 ret = -1;
1703 }
1704
1705 end_no_session:
1706 free(path_name);
1707 free(channel_name);
1708 return ret;
1709 }
1710
1711 /*
1712 * relay_close_stream: close a specific stream
1713 */
1714 static int relay_close_stream(
1715 const struct lttcomm_relayd_hdr *recv_hdr __attribute__((unused)),
1716 struct relay_connection *conn,
1717 const struct lttng_buffer_view *payload)
1718 {
1719 int ret;
1720 ssize_t send_ret;
1721 struct relay_session *session = conn->session;
1722 struct lttcomm_relayd_close_stream stream_info;
1723 struct lttcomm_relayd_generic_reply reply;
1724 struct relay_stream *stream;
1725
1726 DBG("Close stream received");
1727
1728 if (!session || !conn->version_check_done) {
1729 ERR("Trying to close a stream before version check");
1730 ret = -1;
1731 goto end_no_session;
1732 }
1733
1734 if (payload->size < sizeof(stream_info)) {
1735 ERR("Unexpected payload size in \"relay_close_stream\": expected >= %zu bytes, got %zu bytes",
1736 sizeof(stream_info), payload->size);
1737 ret = -1;
1738 goto end_no_session;
1739 }
1740 memcpy(&stream_info, payload->data, sizeof(stream_info));
1741 stream_info.stream_id = be64toh(stream_info.stream_id);
1742 stream_info.last_net_seq_num = be64toh(stream_info.last_net_seq_num);
1743
1744 stream = stream_get_by_id(stream_info.stream_id);
1745 if (!stream) {
1746 ret = -1;
1747 goto end;
1748 }
1749
1750 /*
1751 * Set last_net_seq_num before the close flag. Required by data
1752 * pending check.
1753 */
1754 pthread_mutex_lock(&stream->lock);
1755 stream->last_net_seq_num = stream_info.last_net_seq_num;
1756 pthread_mutex_unlock(&stream->lock);
1757
1758 /*
1759 * This is one of the conditions which may trigger a stream close
1760 * with the others being:
1761 * 1) A close command is received for a stream
1762 * 2) The control connection owning the stream is closed
1763 * 3) We have received all of the stream's data _after_ a close
1764 * request.
1765 */
1766 try_stream_close(stream);
1767 stream_put(stream);
1768 ret = 0;
1769
1770 end:
1771 memset(&reply, 0, sizeof(reply));
1772 if (ret < 0) {
1773 reply.ret_code = htobe32(LTTNG_ERR_UNK);
1774 } else {
1775 reply.ret_code = htobe32(LTTNG_OK);
1776 }
1777 send_ret = conn->sock->ops->sendmsg(conn->sock, &reply,
1778 sizeof(struct lttcomm_relayd_generic_reply), 0);
1779 if (send_ret < (ssize_t) sizeof(reply)) {
1780 ERR("Failed to send \"close stream\" command reply (ret = %zd)",
1781 send_ret);
1782 ret = -1;
1783 }
1784
1785 end_no_session:
1786 return ret;
1787 }
1788
1789 /*
1790 * relay_reset_metadata: reset a metadata stream
1791 */
1792 static
1793 int relay_reset_metadata(
1794 const struct lttcomm_relayd_hdr *recv_hdr __attribute__((unused)),
1795 struct relay_connection *conn,
1796 const struct lttng_buffer_view *payload)
1797 {
1798 int ret;
1799 ssize_t send_ret;
1800 struct relay_session *session = conn->session;
1801 struct lttcomm_relayd_reset_metadata stream_info;
1802 struct lttcomm_relayd_generic_reply reply;
1803 struct relay_stream *stream;
1804
1805 DBG("Reset metadata received");
1806
1807 if (!session || !conn->version_check_done) {
1808 ERR("Trying to reset a metadata stream before version check");
1809 ret = -1;
1810 goto end_no_session;
1811 }
1812
1813 if (payload->size < sizeof(stream_info)) {
1814 ERR("Unexpected payload size in \"relay_reset_metadata\": expected >= %zu bytes, got %zu bytes",
1815 sizeof(stream_info), payload->size);
1816 ret = -1;
1817 goto end_no_session;
1818 }
1819 memcpy(&stream_info, payload->data, sizeof(stream_info));
1820 stream_info.stream_id = be64toh(stream_info.stream_id);
1821 stream_info.version = be64toh(stream_info.version);
1822
1823 DBG("Update metadata to version %" PRIu64, stream_info.version);
1824
1825 /* Unsupported for live sessions for now. */
1826 if (session->live_timer != 0) {
1827 ret = -1;
1828 goto end;
1829 }
1830
1831 stream = stream_get_by_id(stream_info.stream_id);
1832 if (!stream) {
1833 ret = -1;
1834 goto end;
1835 }
1836 pthread_mutex_lock(&stream->lock);
1837 if (!stream->is_metadata) {
1838 ret = -1;
1839 goto end_unlock;
1840 }
1841
1842 ret = stream_reset_file(stream);
1843 if (ret < 0) {
1844 ERR("Failed to reset metadata stream %" PRIu64
1845 ": stream_path = %s, channel = %s",
1846 stream->stream_handle, stream->path_name,
1847 stream->channel_name);
1848 goto end_unlock;
1849 }
1850 end_unlock:
1851 pthread_mutex_unlock(&stream->lock);
1852 stream_put(stream);
1853
1854 end:
1855 memset(&reply, 0, sizeof(reply));
1856 if (ret < 0) {
1857 reply.ret_code = htobe32(LTTNG_ERR_UNK);
1858 } else {
1859 reply.ret_code = htobe32(LTTNG_OK);
1860 }
1861 send_ret = conn->sock->ops->sendmsg(conn->sock, &reply,
1862 sizeof(struct lttcomm_relayd_generic_reply), 0);
1863 if (send_ret < (ssize_t) sizeof(reply)) {
1864 ERR("Failed to send \"reset metadata\" command reply (ret = %zd)",
1865 send_ret);
1866 ret = -1;
1867 }
1868
1869 end_no_session:
1870 return ret;
1871 }
1872
1873 /*
1874 * relay_unknown_command: send -1 if received unknown command
1875 */
1876 static void relay_unknown_command(struct relay_connection *conn)
1877 {
1878 struct lttcomm_relayd_generic_reply reply;
1879 ssize_t send_ret;
1880
1881 memset(&reply, 0, sizeof(reply));
1882 reply.ret_code = htobe32(LTTNG_ERR_UNK);
1883 send_ret = conn->sock->ops->sendmsg(conn->sock, &reply, sizeof(reply), 0);
1884 if (send_ret < sizeof(reply)) {
1885 ERR("Failed to send \"unknown command\" command reply (ret = %zd)", send_ret);
1886 }
1887 }
1888
1889 /*
1890 * relay_start: send an acknowledgment to the client to tell if we are
1891 * ready to receive data. We are ready if a session is established.
1892 */
1893 static int relay_start(
1894 const struct lttcomm_relayd_hdr *recv_hdr __attribute__((unused)),
1895 struct relay_connection *conn,
1896 const struct lttng_buffer_view *payload __attribute__((unused)))
1897 {
1898 int ret = 0;
1899 ssize_t send_ret;
1900 struct lttcomm_relayd_generic_reply reply;
1901 struct relay_session *session = conn->session;
1902
1903 if (!session) {
1904 DBG("Trying to start the streaming without a session established");
1905 ret = htobe32(LTTNG_ERR_UNK);
1906 }
1907
1908 memset(&reply, 0, sizeof(reply));
1909 reply.ret_code = htobe32(LTTNG_OK);
1910 send_ret = conn->sock->ops->sendmsg(conn->sock, &reply,
1911 sizeof(reply), 0);
1912 if (send_ret < (ssize_t) sizeof(reply)) {
1913 ERR("Failed to send \"relay_start\" command reply (ret = %zd)",
1914 send_ret);
1915 ret = -1;
1916 }
1917
1918 return ret;
1919 }
1920
1921 /*
1922 * relay_recv_metadata: receive the metadata for the session.
1923 */
1924 static int relay_recv_metadata(const struct lttcomm_relayd_hdr *recv_hdr,
1925 struct relay_connection *conn,
1926 const struct lttng_buffer_view *payload)
1927 {
1928 int ret = 0;
1929 struct relay_session *session = conn->session;
1930 struct lttcomm_relayd_metadata_payload metadata_payload_header;
1931 struct relay_stream *metadata_stream;
1932 uint64_t metadata_payload_size;
1933 struct lttng_buffer_view packet_view;
1934
1935 if (!session) {
1936 ERR("Metadata sent before version check");
1937 ret = -1;
1938 goto end;
1939 }
1940
1941 if (recv_hdr->data_size < sizeof(struct lttcomm_relayd_metadata_payload)) {
1942 ERR("Incorrect data size");
1943 ret = -1;
1944 goto end;
1945 }
1946 metadata_payload_size = recv_hdr->data_size -
1947 sizeof(struct lttcomm_relayd_metadata_payload);
1948
1949 memcpy(&metadata_payload_header, payload->data,
1950 sizeof(metadata_payload_header));
1951 metadata_payload_header.stream_id = be64toh(
1952 metadata_payload_header.stream_id);
1953 metadata_payload_header.padding_size = be32toh(
1954 metadata_payload_header.padding_size);
1955
1956 metadata_stream = stream_get_by_id(metadata_payload_header.stream_id);
1957 if (!metadata_stream) {
1958 ret = -1;
1959 goto end;
1960 }
1961
1962 packet_view = lttng_buffer_view_from_view(payload,
1963 sizeof(metadata_payload_header), metadata_payload_size);
1964 if (!lttng_buffer_view_is_valid(&packet_view)) {
1965 ERR("Invalid metadata packet length announced by header");
1966 ret = -1;
1967 goto end_put;
1968 }
1969
1970 pthread_mutex_lock(&metadata_stream->lock);
1971 ret = stream_write(metadata_stream, &packet_view,
1972 metadata_payload_header.padding_size);
1973 pthread_mutex_unlock(&metadata_stream->lock);
1974 if (ret){
1975 ret = -1;
1976 goto end_put;
1977 }
1978 end_put:
1979 stream_put(metadata_stream);
1980 end:
1981 return ret;
1982 }
1983
1984 /*
1985 * relay_send_version: send relayd version number
1986 */
1987 static int relay_send_version(
1988 const struct lttcomm_relayd_hdr *recv_hdr __attribute__((unused)),
1989 struct relay_connection *conn,
1990 const struct lttng_buffer_view *payload)
1991 {
1992 int ret;
1993 ssize_t send_ret;
1994 struct lttcomm_relayd_version reply, msg;
1995 bool compatible = true;
1996
1997 conn->version_check_done = true;
1998
1999 /* Get version from the other side. */
2000 if (payload->size < sizeof(msg)) {
2001 ERR("Unexpected payload size in \"relay_send_version\": expected >= %zu bytes, got %zu bytes",
2002 sizeof(msg), payload->size);
2003 ret = -1;
2004 goto end;
2005 }
2006
2007 memcpy(&msg, payload->data, sizeof(msg));
2008 msg.major = be32toh(msg.major);
2009 msg.minor = be32toh(msg.minor);
2010
2011 memset(&reply, 0, sizeof(reply));
2012 reply.major = RELAYD_VERSION_COMM_MAJOR;
2013 reply.minor = RELAYD_VERSION_COMM_MINOR;
2014
2015 /* Major versions must be the same */
2016 if (reply.major != msg.major) {
2017 DBG("Incompatible major versions (%u vs %u), deleting session",
2018 reply.major, msg.major);
2019 compatible = false;
2020 }
2021
2022 conn->major = reply.major;
2023 /* We adapt to the lowest compatible version */
2024 if (reply.minor <= msg.minor) {
2025 conn->minor = reply.minor;
2026 } else {
2027 conn->minor = msg.minor;
2028 }
2029
2030 reply.major = htobe32(reply.major);
2031 reply.minor = htobe32(reply.minor);
2032 send_ret = conn->sock->ops->sendmsg(conn->sock, &reply,
2033 sizeof(reply), 0);
2034 if (send_ret < (ssize_t) sizeof(reply)) {
2035 ERR("Failed to send \"send version\" command reply (ret = %zd)",
2036 send_ret);
2037 ret = -1;
2038 goto end;
2039 } else {
2040 ret = 0;
2041 }
2042
2043 if (!compatible) {
2044 ret = -1;
2045 goto end;
2046 }
2047
2048 DBG("Version check done using protocol %u.%u", conn->major,
2049 conn->minor);
2050
2051 end:
2052 return ret;
2053 }
2054
2055 /*
2056 * Check for data pending for a given stream id from the session daemon.
2057 */
2058 static int relay_data_pending(
2059 const struct lttcomm_relayd_hdr *recv_hdr __attribute__((unused)),
2060 struct relay_connection *conn,
2061 const struct lttng_buffer_view *payload)
2062 {
2063 struct relay_session *session = conn->session;
2064 struct lttcomm_relayd_data_pending msg;
2065 struct lttcomm_relayd_generic_reply reply;
2066 struct relay_stream *stream;
2067 ssize_t send_ret;
2068 int ret;
2069 uint64_t stream_seq;
2070
2071 DBG("Data pending command received");
2072
2073 if (!session || !conn->version_check_done) {
2074 ERR("Trying to check for data before version check");
2075 ret = -1;
2076 goto end_no_session;
2077 }
2078
2079 if (payload->size < sizeof(msg)) {
2080 ERR("Unexpected payload size in \"relay_data_pending\": expected >= %zu bytes, got %zu bytes",
2081 sizeof(msg), payload->size);
2082 ret = -1;
2083 goto end_no_session;
2084 }
2085 memcpy(&msg, payload->data, sizeof(msg));
2086 msg.stream_id = be64toh(msg.stream_id);
2087 msg.last_net_seq_num = be64toh(msg.last_net_seq_num);
2088
2089 stream = stream_get_by_id(msg.stream_id);
2090 if (stream == NULL) {
2091 ret = -1;
2092 goto end;
2093 }
2094
2095 pthread_mutex_lock(&stream->lock);
2096
2097 if (session_streams_have_index(session)) {
2098 /*
2099 * Ensure that both the index and stream data have been
2100 * flushed up to the requested point.
2101 */
2102 stream_seq = std::min(stream->prev_data_seq, stream->prev_index_seq);
2103 } else {
2104 stream_seq = stream->prev_data_seq;
2105 }
2106 DBG("Data pending for stream id %" PRIu64 ": prev_data_seq %" PRIu64
2107 ", prev_index_seq %" PRIu64
2108 ", and last_seq %" PRIu64, msg.stream_id,
2109 stream->prev_data_seq, stream->prev_index_seq,
2110 msg.last_net_seq_num);
2111
2112 /* Avoid wrapping issue */
2113 if (((int64_t) (stream_seq - msg.last_net_seq_num)) >= 0) {
2114 /* Data has in fact been written and is NOT pending */
2115 ret = 0;
2116 } else {
2117 /* Data still being streamed thus pending */
2118 ret = 1;
2119 }
2120
2121 stream->data_pending_check_done = true;
2122 pthread_mutex_unlock(&stream->lock);
2123
2124 stream_put(stream);
2125 end:
2126
2127 memset(&reply, 0, sizeof(reply));
2128 reply.ret_code = htobe32(ret);
2129 send_ret = conn->sock->ops->sendmsg(conn->sock, &reply, sizeof(reply), 0);
2130 if (send_ret < (ssize_t) sizeof(reply)) {
2131 ERR("Failed to send \"data pending\" command reply (ret = %zd)",
2132 send_ret);
2133 ret = -1;
2134 }
2135
2136 end_no_session:
2137 return ret;
2138 }
2139
2140 /*
2141 * Wait for the control socket to reach a quiescent state.
2142 *
2143 * Note that for now, when receiving this command from the session
2144 * daemon, this means that every subsequent commands or data received on
2145 * the control socket has been handled. So, this is why we simply return
2146 * OK here.
2147 */
2148 static int relay_quiescent_control(
2149 const struct lttcomm_relayd_hdr *recv_hdr __attribute__((unused)),
2150 struct relay_connection *conn,
2151 const struct lttng_buffer_view *payload)
2152 {
2153 int ret;
2154 ssize_t send_ret;
2155 struct relay_stream *stream;
2156 struct lttcomm_relayd_quiescent_control msg;
2157 struct lttcomm_relayd_generic_reply reply;
2158
2159 DBG("Checking quiescent state on control socket");
2160
2161 if (!conn->session || !conn->version_check_done) {
2162 ERR("Trying to check for data before version check");
2163 ret = -1;
2164 goto end_no_session;
2165 }
2166
2167 if (payload->size < sizeof(msg)) {
2168 ERR("Unexpected payload size in \"relay_quiescent_control\": expected >= %zu bytes, got %zu bytes",
2169 sizeof(msg), payload->size);
2170 ret = -1;
2171 goto end_no_session;
2172 }
2173 memcpy(&msg, payload->data, sizeof(msg));
2174 msg.stream_id = be64toh(msg.stream_id);
2175
2176 stream = stream_get_by_id(msg.stream_id);
2177 if (!stream) {
2178 goto reply;
2179 }
2180 pthread_mutex_lock(&stream->lock);
2181 stream->data_pending_check_done = true;
2182 pthread_mutex_unlock(&stream->lock);
2183
2184 DBG("Relay quiescent control pending flag set to %" PRIu64, msg.stream_id);
2185 stream_put(stream);
2186 reply:
2187 memset(&reply, 0, sizeof(reply));
2188 reply.ret_code = htobe32(LTTNG_OK);
2189 send_ret = conn->sock->ops->sendmsg(conn->sock, &reply, sizeof(reply), 0);
2190 if (send_ret < (ssize_t) sizeof(reply)) {
2191 ERR("Failed to send \"quiescent control\" command reply (ret = %zd)",
2192 send_ret);
2193 ret = -1;
2194 } else {
2195 ret = 0;
2196 }
2197
2198 end_no_session:
2199 return ret;
2200 }
2201
2202 /*
2203 * Initialize a data pending command. This means that a consumer is about
2204 * to ask for data pending for each stream it holds. Simply iterate over
2205 * all streams of a session and set the data_pending_check_done flag.
2206 *
2207 * This command returns to the client a LTTNG_OK code.
2208 */
2209 static int relay_begin_data_pending(const struct lttcomm_relayd_hdr *recv_hdr,
2210 struct relay_connection *conn,
2211 const struct lttng_buffer_view *payload)
2212 {
2213 int ret;
2214 ssize_t send_ret;
2215 struct lttng_ht_iter iter;
2216 struct lttcomm_relayd_begin_data_pending msg;
2217 struct lttcomm_relayd_generic_reply reply;
2218 struct relay_stream *stream;
2219
2220 LTTNG_ASSERT(recv_hdr);
2221 LTTNG_ASSERT(conn);
2222
2223 DBG("Init streams for data pending");
2224
2225 if (!conn->session || !conn->version_check_done) {
2226 ERR("Trying to check for data before version check");
2227 ret = -1;
2228 goto end_no_session;
2229 }
2230
2231 if (payload->size < sizeof(msg)) {
2232 ERR("Unexpected payload size in \"relay_begin_data_pending\": expected >= %zu bytes, got %zu bytes",
2233 sizeof(msg), payload->size);
2234 ret = -1;
2235 goto end_no_session;
2236 }
2237 memcpy(&msg, payload->data, sizeof(msg));
2238 msg.session_id = be64toh(msg.session_id);
2239
2240 /*
2241 * Iterate over all streams to set the begin data pending flag.
2242 * For now, the streams are indexed by stream handle so we have
2243 * to iterate over all streams to find the one associated with
2244 * the right session_id.
2245 */
2246 rcu_read_lock();
2247 cds_lfht_for_each_entry(relay_streams_ht->ht, &iter.iter, stream,
2248 node.node) {
2249 if (!stream_get(stream)) {
2250 continue;
2251 }
2252 if (stream->trace->session->id == msg.session_id) {
2253 pthread_mutex_lock(&stream->lock);
2254 stream->data_pending_check_done = false;
2255 pthread_mutex_unlock(&stream->lock);
2256 DBG("Set begin data pending flag to stream %" PRIu64,
2257 stream->stream_handle);
2258 }
2259 stream_put(stream);
2260 }
2261 rcu_read_unlock();
2262
2263 memset(&reply, 0, sizeof(reply));
2264 /* All good, send back reply. */
2265 reply.ret_code = htobe32(LTTNG_OK);
2266
2267 send_ret = conn->sock->ops->sendmsg(conn->sock, &reply, sizeof(reply), 0);
2268 if (send_ret < (ssize_t) sizeof(reply)) {
2269 ERR("Failed to send \"begin data pending\" command reply (ret = %zd)",
2270 send_ret);
2271 ret = -1;
2272 } else {
2273 ret = 0;
2274 }
2275
2276 end_no_session:
2277 return ret;
2278 }
2279
2280 /*
2281 * End data pending command. This will check, for a given session id, if
2282 * each stream associated with it has its data_pending_check_done flag
2283 * set. If not, this means that the client lost track of the stream but
2284 * the data is still being streamed on our side. In this case, we inform
2285 * the client that data is in flight.
2286 *
2287 * Return to the client if there is data in flight or not with a ret_code.
2288 */
2289 static int relay_end_data_pending(
2290 const struct lttcomm_relayd_hdr *recv_hdr __attribute__((unused)),
2291 struct relay_connection *conn,
2292 const struct lttng_buffer_view *payload)
2293 {
2294 int ret;
2295 ssize_t send_ret;
2296 struct lttng_ht_iter iter;
2297 struct lttcomm_relayd_end_data_pending msg;
2298 struct lttcomm_relayd_generic_reply reply;
2299 struct relay_stream *stream;
2300 uint32_t is_data_inflight = 0;
2301
2302 DBG("End data pending command");
2303
2304 if (!conn->session || !conn->version_check_done) {
2305 ERR("Trying to check for data before version check");
2306 ret = -1;
2307 goto end_no_session;
2308 }
2309
2310 if (payload->size < sizeof(msg)) {
2311 ERR("Unexpected payload size in \"relay_end_data_pending\": expected >= %zu bytes, got %zu bytes",
2312 sizeof(msg), payload->size);
2313 ret = -1;
2314 goto end_no_session;
2315 }
2316 memcpy(&msg, payload->data, sizeof(msg));
2317 msg.session_id = be64toh(msg.session_id);
2318
2319 /*
2320 * Iterate over all streams to see if the begin data pending
2321 * flag is set.
2322 */
2323 rcu_read_lock();
2324 cds_lfht_for_each_entry(relay_streams_ht->ht, &iter.iter, stream,
2325 node.node) {
2326 if (!stream_get(stream)) {
2327 continue;
2328 }
2329 if (stream->trace->session->id != msg.session_id) {
2330 stream_put(stream);
2331 continue;
2332 }
2333 pthread_mutex_lock(&stream->lock);
2334 if (!stream->data_pending_check_done) {
2335 uint64_t stream_seq;
2336
2337 if (session_streams_have_index(conn->session)) {
2338 /*
2339 * Ensure that both the index and stream data have been
2340 * flushed up to the requested point.
2341 */
2342 stream_seq = std::min(stream->prev_data_seq, stream->prev_index_seq);
2343 } else {
2344 stream_seq = stream->prev_data_seq;
2345 }
2346 if (!stream->closed || !(((int64_t) (stream_seq - stream->last_net_seq_num)) >= 0)) {
2347 is_data_inflight = 1;
2348 DBG("Data is still in flight for stream %" PRIu64,
2349 stream->stream_handle);
2350 pthread_mutex_unlock(&stream->lock);
2351 stream_put(stream);
2352 break;
2353 }
2354 }
2355 pthread_mutex_unlock(&stream->lock);
2356 stream_put(stream);
2357 }
2358 rcu_read_unlock();
2359
2360 memset(&reply, 0, sizeof(reply));
2361 /* All good, send back reply. */
2362 reply.ret_code = htobe32(is_data_inflight);
2363
2364 send_ret = conn->sock->ops->sendmsg(conn->sock, &reply, sizeof(reply), 0);
2365 if (send_ret < (ssize_t) sizeof(reply)) {
2366 ERR("Failed to send \"end data pending\" command reply (ret = %zd)",
2367 send_ret);
2368 ret = -1;
2369 } else {
2370 ret = 0;
2371 }
2372
2373 end_no_session:
2374 return ret;
2375 }
2376
2377 /*
2378 * Receive an index for a specific stream.
2379 *
2380 * Return 0 on success else a negative value.
2381 */
2382 static int relay_recv_index(
2383 const struct lttcomm_relayd_hdr *recv_hdr __attribute__((unused)),
2384 struct relay_connection *conn,
2385 const struct lttng_buffer_view *payload)
2386 {
2387 int ret;
2388 ssize_t send_ret;
2389 struct relay_session *session = conn->session;
2390 struct lttcomm_relayd_index index_info;
2391 struct lttcomm_relayd_generic_reply reply;
2392 struct relay_stream *stream;
2393 size_t msg_len;
2394
2395 LTTNG_ASSERT(conn);
2396
2397 DBG("Relay receiving index");
2398
2399 if (!session || !conn->version_check_done) {
2400 ERR("Trying to close a stream before version check");
2401 ret = -1;
2402 goto end_no_session;
2403 }
2404
2405 msg_len = lttcomm_relayd_index_len(
2406 lttng_to_index_major(conn->major, conn->minor),
2407 lttng_to_index_minor(conn->major, conn->minor));
2408 if (payload->size < msg_len) {
2409 ERR("Unexpected payload size in \"relay_recv_index\": expected >= %zu bytes, got %zu bytes",
2410 msg_len, payload->size);
2411 ret = -1;
2412 goto end_no_session;
2413 }
2414 memcpy(&index_info, payload->data, msg_len);
2415 index_info.relay_stream_id = be64toh(index_info.relay_stream_id);
2416 index_info.net_seq_num = be64toh(index_info.net_seq_num);
2417 index_info.packet_size = be64toh(index_info.packet_size);
2418 index_info.content_size = be64toh(index_info.content_size);
2419 index_info.timestamp_begin = be64toh(index_info.timestamp_begin);
2420 index_info.timestamp_end = be64toh(index_info.timestamp_end);
2421 index_info.events_discarded = be64toh(index_info.events_discarded);
2422 index_info.stream_id = be64toh(index_info.stream_id);
2423
2424 if (conn->minor >= 8) {
2425 index_info.stream_instance_id =
2426 be64toh(index_info.stream_instance_id);
2427 index_info.packet_seq_num = be64toh(index_info.packet_seq_num);
2428 } else {
2429 index_info.stream_instance_id = -1ULL;
2430 index_info.packet_seq_num = -1ULL;
2431 }
2432
2433 stream = stream_get_by_id(index_info.relay_stream_id);
2434 if (!stream) {
2435 ERR("stream_get_by_id not found");
2436 ret = -1;
2437 goto end;
2438 }
2439
2440 pthread_mutex_lock(&stream->lock);
2441 ret = stream_add_index(stream, &index_info);
2442 pthread_mutex_unlock(&stream->lock);
2443 if (ret) {
2444 goto end_stream_put;
2445 }
2446
2447 end_stream_put:
2448 stream_put(stream);
2449 end:
2450 memset(&reply, 0, sizeof(reply));
2451 if (ret < 0) {
2452 reply.ret_code = htobe32(LTTNG_ERR_UNK);
2453 } else {
2454 reply.ret_code = htobe32(LTTNG_OK);
2455 }
2456 send_ret = conn->sock->ops->sendmsg(conn->sock, &reply, sizeof(reply), 0);
2457 if (send_ret < (ssize_t) sizeof(reply)) {
2458 ERR("Failed to send \"recv index\" command reply (ret = %zd)", send_ret);
2459 ret = -1;
2460 }
2461
2462 end_no_session:
2463 return ret;
2464 }
2465
2466 /*
2467 * Receive the streams_sent message.
2468 *
2469 * Return 0 on success else a negative value.
2470 */
2471 static int relay_streams_sent(
2472 const struct lttcomm_relayd_hdr *recv_hdr __attribute__((unused)),
2473 struct relay_connection *conn,
2474 const struct lttng_buffer_view *payload __attribute__((unused)))
2475 {
2476 int ret;
2477 ssize_t send_ret;
2478 struct lttcomm_relayd_generic_reply reply;
2479
2480 LTTNG_ASSERT(conn);
2481
2482 DBG("Relay receiving streams_sent");
2483
2484 if (!conn->session || !conn->version_check_done) {
2485 ERR("Trying to close a stream before version check");
2486 ret = -1;
2487 goto end_no_session;
2488 }
2489
2490 /*
2491 * Publish every pending stream in the connection recv list which are
2492 * now ready to be used by the viewer.
2493 */
2494 publish_connection_local_streams(conn);
2495
2496 memset(&reply, 0, sizeof(reply));
2497 reply.ret_code = htobe32(LTTNG_OK);
2498 send_ret = conn->sock->ops->sendmsg(conn->sock, &reply, sizeof(reply), 0);
2499 if (send_ret < (ssize_t) sizeof(reply)) {
2500 ERR("Failed to send \"streams sent\" command reply (ret = %zd)",
2501 send_ret);
2502 ret = -1;
2503 } else {
2504 /* Success. */
2505 ret = 0;
2506 }
2507
2508 end_no_session:
2509 return ret;
2510 }
2511
2512 static ssize_t relay_unpack_rotate_streams_header(
2513 const struct lttng_buffer_view *payload,
2514 struct lttcomm_relayd_rotate_streams *_rotate_streams)
2515 {
2516 struct lttcomm_relayd_rotate_streams rotate_streams;
2517 /*
2518 * Set to the smallest version (packed) of `lttcomm_relayd_rotate_streams`.
2519 * This is the smallest version of this structure, but it can be larger;
2520 * this variable is updated once the proper size of the structure is known.
2521 *
2522 * See comment at the declaration of this structure for more information.
2523 */
2524 ssize_t header_len = sizeof(struct lttcomm_relayd_rotate_streams_packed);
2525 size_t expected_payload_size_no_padding,
2526 expected_payload_size_3_bytes_padding,
2527 expected_payload_size_7_bytes_padding;
2528
2529 if (payload->size < header_len) {
2530 ERR("Unexpected payload size in \"relay_rotate_session_stream\": expected >= %zu bytes, got %zu bytes",
2531 header_len, payload->size);
2532 goto error;
2533 }
2534
2535 /*
2536 * Some versions incorrectly omitted the LTTNG_PACKED annotation on the
2537 * `new_chunk_id` optional field of struct lttcomm_relayd_rotate_streams.
2538 *
2539 * We start by "unpacking" `stream_count` to figure out the padding length
2540 * emited by our peer.
2541 */
2542 memcpy(&rotate_streams.stream_count, payload->data,
2543 sizeof(rotate_streams.stream_count));
2544 rotate_streams = (typeof(rotate_streams)) {
2545 .stream_count = be32toh(rotate_streams.stream_count),
2546 .new_chunk_id = LTTNG_OPTIONAL_INIT_UNSET,
2547 };
2548
2549 /*
2550 * Payload size expected given the possible padding lengths in
2551 * `struct lttcomm_relayd_rotate_streams`.
2552 */
2553 expected_payload_size_no_padding = (rotate_streams.stream_count *
2554 sizeof(*rotate_streams.rotation_positions)) +
2555 sizeof(lttcomm_relayd_rotate_streams_packed);
2556 expected_payload_size_3_bytes_padding = (rotate_streams.stream_count *
2557 sizeof(*rotate_streams.rotation_positions)) +
2558 sizeof(lttcomm_relayd_rotate_streams_3_bytes_padding);
2559 expected_payload_size_7_bytes_padding = (rotate_streams.stream_count *
2560 sizeof(*rotate_streams.rotation_positions)) +
2561 sizeof(lttcomm_relayd_rotate_streams_7_bytes_padding);
2562
2563 if (payload->size == expected_payload_size_no_padding) {
2564 struct lttcomm_relayd_rotate_streams_packed packed_rotate_streams;
2565
2566 /*
2567 * This handles cases where someone might build with
2568 * -fpack-struct or any other toolchain that wouldn't produce
2569 * padding to align `value`.
2570 */
2571 DBG("Received `struct lttcomm_relayd_rotate_streams` with no padding");
2572
2573 header_len = sizeof(packed_rotate_streams);
2574 memcpy(&packed_rotate_streams, payload->data, header_len);
2575
2576 /* Unpack the packed structure to the natively-packed version. */
2577 *_rotate_streams = (typeof(*_rotate_streams)) {
2578 .stream_count = be32toh(packed_rotate_streams.stream_count),
2579 .new_chunk_id = (typeof(_rotate_streams->new_chunk_id)) {
2580 .is_set = !!packed_rotate_streams.new_chunk_id.is_set,
2581 .value = be64toh(packed_rotate_streams.new_chunk_id.value),
2582 }
2583 };
2584 } else if (payload->size == expected_payload_size_3_bytes_padding) {
2585 struct lttcomm_relayd_rotate_streams_3_bytes_padding padded_rotate_streams;
2586
2587 DBG("Received `struct lttcomm_relayd_rotate_streams` with 3 bytes of padding (4-byte aligned peer)");
2588
2589 header_len = sizeof(padded_rotate_streams);
2590 memcpy(&padded_rotate_streams, payload->data, header_len);
2591
2592 /* Unpack the 3-byte padded structure to the natively-packed version. */
2593 *_rotate_streams = (typeof(*_rotate_streams)) {
2594 .stream_count = be32toh(padded_rotate_streams.stream_count),
2595 .new_chunk_id = (typeof(_rotate_streams->new_chunk_id)) {
2596 .is_set = !!padded_rotate_streams.new_chunk_id.is_set,
2597 .value = be64toh(padded_rotate_streams.new_chunk_id.value),
2598 }
2599 };
2600 } else if (payload->size == expected_payload_size_7_bytes_padding) {
2601 struct lttcomm_relayd_rotate_streams_7_bytes_padding padded_rotate_streams;
2602
2603 DBG("Received `struct lttcomm_relayd_rotate_streams` with 7 bytes of padding (8-byte aligned peer)");
2604
2605 header_len = sizeof(padded_rotate_streams);
2606 memcpy(&padded_rotate_streams, payload->data, header_len);
2607
2608 /* Unpack the 7-byte padded structure to the natively-packed version. */
2609 *_rotate_streams = (typeof(*_rotate_streams)) {
2610 .stream_count = be32toh(padded_rotate_streams.stream_count),
2611 .new_chunk_id = (typeof(_rotate_streams->new_chunk_id)) {
2612 .is_set = !!padded_rotate_streams.new_chunk_id.is_set,
2613 .value = be64toh(padded_rotate_streams.new_chunk_id.value),
2614 }
2615 };
2616
2617 header_len = sizeof(padded_rotate_streams);
2618 } else {
2619 ERR("Unexpected payload size in \"relay_rotate_session_stream\": expected %zu, %zu or %zu bytes, got %zu bytes",
2620 expected_payload_size_no_padding,
2621 expected_payload_size_3_bytes_padding,
2622 expected_payload_size_7_bytes_padding,
2623 payload->size);
2624 goto error;
2625 }
2626
2627 return header_len;
2628 error:
2629 return -1;
2630 }
2631
2632 /*
2633 * relay_rotate_session_stream: rotate a stream to a new tracefile for the
2634 * session rotation feature (not the tracefile rotation feature).
2635 */
2636 static int relay_rotate_session_streams(
2637 const struct lttcomm_relayd_hdr *recv_hdr __attribute__((unused)),
2638 struct relay_connection *conn,
2639 const struct lttng_buffer_view *payload)
2640 {
2641 int ret = 0;
2642 uint32_t i;
2643 ssize_t send_ret;
2644 enum lttng_error_code reply_code = LTTNG_ERR_UNK;
2645 struct relay_session *session = conn->session;
2646 struct lttcomm_relayd_rotate_streams rotate_streams;
2647 struct lttcomm_relayd_generic_reply reply = {};
2648 struct relay_stream *stream = NULL;
2649 struct lttng_trace_chunk *next_trace_chunk = NULL;
2650 struct lttng_buffer_view stream_positions;
2651 char chunk_id_buf[MAX_INT_DEC_LEN(uint64_t)];
2652 const char *chunk_id_str = "none";
2653 ssize_t header_len;
2654
2655 if (!session || !conn->version_check_done) {
2656 ERR("Trying to rotate a stream before version check");
2657 ret = -1;
2658 goto end_no_reply;
2659 }
2660
2661 if (session->major == 2 && session->minor < 11) {
2662 ERR("Unsupported feature before 2.11");
2663 ret = -1;
2664 goto end_no_reply;
2665 }
2666
2667 header_len = relay_unpack_rotate_streams_header(payload, &rotate_streams);
2668 if (header_len < 0) {
2669 ret = -1;
2670 goto end_no_reply;
2671 }
2672
2673 if (rotate_streams.new_chunk_id.is_set) {
2674 /*
2675 * Retrieve the trace chunk the stream must transition to. As
2676 * per the protocol, this chunk should have been created
2677 * before this command is received.
2678 */
2679 next_trace_chunk = sessiond_trace_chunk_registry_get_chunk(
2680 sessiond_trace_chunk_registry,
2681 session->sessiond_uuid,
2682 conn->session->id_sessiond.is_set ?
2683 conn->session->id_sessiond.value :
2684 conn->session->id,
2685 rotate_streams.new_chunk_id.value);
2686 if (!next_trace_chunk) {
2687 char uuid_str[LTTNG_UUID_STR_LEN];
2688
2689 lttng_uuid_to_str(session->sessiond_uuid, uuid_str);
2690 ERR("Unknown next trace chunk in ROTATE_STREAMS command: sessiond_uuid = {%s}, session_id = %" PRIu64
2691 ", trace_chunk_id = %" PRIu64,
2692 uuid_str, session->id,
2693 rotate_streams.new_chunk_id.value);
2694 reply_code = LTTNG_ERR_INVALID_PROTOCOL;
2695 ret = -1;
2696 goto end;
2697 }
2698
2699 ret = snprintf(chunk_id_buf, sizeof(chunk_id_buf), "%" PRIu64,
2700 rotate_streams.new_chunk_id.value);
2701 if (ret < 0 || ret >= sizeof(chunk_id_buf)) {
2702 chunk_id_str = "formatting error";
2703 } else {
2704 chunk_id_str = chunk_id_buf;
2705 }
2706 }
2707
2708 DBG("Rotate %" PRIu32 " streams of session \"%s\" to chunk \"%s\"",
2709 rotate_streams.stream_count, session->session_name,
2710 chunk_id_str);
2711
2712 stream_positions = lttng_buffer_view_from_view(payload,
2713 header_len, -1);
2714 if (!stream_positions.data ||
2715 stream_positions.size <
2716 (rotate_streams.stream_count *
2717 sizeof(struct lttcomm_relayd_stream_rotation_position))) {
2718 reply_code = LTTNG_ERR_INVALID_PROTOCOL;
2719 ret = -1;
2720 goto end;
2721 }
2722
2723 for (i = 0; i < rotate_streams.stream_count; i++) {
2724 struct lttcomm_relayd_stream_rotation_position *position_comm =
2725 &((typeof(position_comm)) stream_positions.data)[i];
2726 const struct lttcomm_relayd_stream_rotation_position pos = {
2727 .stream_id = be64toh(position_comm->stream_id),
2728 .rotate_at_seq_num = be64toh(
2729 position_comm->rotate_at_seq_num),
2730 };
2731
2732 stream = stream_get_by_id(pos.stream_id);
2733 if (!stream) {
2734 reply_code = LTTNG_ERR_INVALID;
2735 ret = -1;
2736 goto end;
2737 }
2738
2739 pthread_mutex_lock(&stream->lock);
2740 ret = stream_set_pending_rotation(stream, next_trace_chunk,
2741 pos.rotate_at_seq_num);
2742 pthread_mutex_unlock(&stream->lock);
2743 if (ret) {
2744 reply_code = LTTNG_ERR_FILE_CREATION_ERROR;
2745 goto end;
2746 }
2747
2748 stream_put(stream);
2749 stream = NULL;
2750 }
2751
2752 reply_code = LTTNG_OK;
2753 ret = 0;
2754 end:
2755 if (stream) {
2756 stream_put(stream);
2757 }
2758
2759 reply.ret_code = htobe32((uint32_t) reply_code);
2760 send_ret = conn->sock->ops->sendmsg(conn->sock, &reply,
2761 sizeof(struct lttcomm_relayd_generic_reply), 0);
2762 if (send_ret < (ssize_t) sizeof(reply)) {
2763 ERR("Failed to send \"rotate session stream\" command reply (ret = %zd)",
2764 send_ret);
2765 ret = -1;
2766 }
2767 end_no_reply:
2768 lttng_trace_chunk_put(next_trace_chunk);
2769 return ret;
2770 }
2771
2772 /*
2773 * relay_create_trace_chunk: create a new trace chunk
2774 */
2775 static int relay_create_trace_chunk(
2776 const struct lttcomm_relayd_hdr *recv_hdr __attribute__((unused)),
2777 struct relay_connection *conn,
2778 const struct lttng_buffer_view *payload)
2779 {
2780 int ret = 0;
2781 ssize_t send_ret;
2782 struct relay_session *session = conn->session;
2783 struct lttcomm_relayd_create_trace_chunk *msg;
2784 struct lttcomm_relayd_generic_reply reply = {};
2785 struct lttng_buffer_view header_view;
2786 struct lttng_trace_chunk *chunk = NULL, *published_chunk = NULL;
2787 enum lttng_error_code reply_code = LTTNG_OK;
2788 enum lttng_trace_chunk_status chunk_status;
2789 const char *new_path;
2790
2791 if (!session || !conn->version_check_done) {
2792 ERR("Trying to create a trace chunk before version check");
2793 ret = -1;
2794 goto end_no_reply;
2795 }
2796
2797 if (session->major == 2 && session->minor < 11) {
2798 ERR("Chunk creation command is unsupported before 2.11");
2799 ret = -1;
2800 goto end_no_reply;
2801 }
2802
2803 header_view = lttng_buffer_view_from_view(payload, 0, sizeof(*msg));
2804 if (!lttng_buffer_view_is_valid(&header_view)) {
2805 ERR("Failed to receive payload of chunk creation command");
2806 ret = -1;
2807 goto end_no_reply;
2808 }
2809
2810 /* Convert to host endianness. */
2811 msg = (typeof(msg)) header_view.data;
2812 msg->chunk_id = be64toh(msg->chunk_id);
2813 msg->creation_timestamp = be64toh(msg->creation_timestamp);
2814 msg->override_name_length = be32toh(msg->override_name_length);
2815
2816 pthread_mutex_lock(&conn->session->lock);
2817 session->ongoing_rotation = true;
2818 if (session->current_trace_chunk &&
2819 !lttng_trace_chunk_get_name_overridden(session->current_trace_chunk)) {
2820 chunk_status = lttng_trace_chunk_rename_path(session->current_trace_chunk,
2821 DEFAULT_CHUNK_TMP_OLD_DIRECTORY);
2822 if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
2823 ERR("Failed to rename old chunk");
2824 ret = -1;
2825 reply_code = LTTNG_ERR_UNK;
2826 goto end;
2827 }
2828 }
2829 if (!session->current_trace_chunk) {
2830 if (!session->has_rotated) {
2831 new_path = "";
2832 } else {
2833 new_path = NULL;
2834 }
2835 } else {
2836 new_path = DEFAULT_CHUNK_TMP_NEW_DIRECTORY;
2837 }
2838 chunk = lttng_trace_chunk_create(
2839 msg->chunk_id, msg->creation_timestamp, new_path);
2840 if (!chunk) {
2841 ERR("Failed to create trace chunk in trace chunk creation command");
2842 ret = -1;
2843 reply_code = LTTNG_ERR_NOMEM;
2844 goto end;
2845 }
2846 lttng_trace_chunk_set_fd_tracker(chunk, the_fd_tracker);
2847
2848 if (msg->override_name_length) {
2849 const char *name;
2850 const struct lttng_buffer_view chunk_name_view =
2851 lttng_buffer_view_from_view(payload,
2852 sizeof(*msg),
2853 msg->override_name_length);
2854
2855 if (!lttng_buffer_view_is_valid(&chunk_name_view)) {
2856 ERR("Invalid payload of chunk creation command (protocol error): buffer too short for expected name length");
2857 ret = -1;
2858 reply_code = LTTNG_ERR_INVALID;
2859 goto end;
2860 }
2861
2862 name = chunk_name_view.data;
2863 if (name[msg->override_name_length - 1]) {
2864 ERR("Invalid payload of chunk creation command (protocol error): name is not null-terminated");
2865 ret = -1;
2866 reply_code = LTTNG_ERR_INVALID;
2867 goto end;
2868 }
2869
2870 chunk_status = lttng_trace_chunk_override_name(
2871 chunk, chunk_name_view.data);
2872 switch (chunk_status) {
2873 case LTTNG_TRACE_CHUNK_STATUS_OK:
2874 break;
2875 case LTTNG_TRACE_CHUNK_STATUS_INVALID_ARGUMENT:
2876 ERR("Failed to set the name of new trace chunk in trace chunk creation command (invalid name)");
2877 reply_code = LTTNG_ERR_INVALID;
2878 ret = -1;
2879 goto end;
2880 default:
2881 ERR("Failed to set the name of new trace chunk in trace chunk creation command (unknown error)");
2882 reply_code = LTTNG_ERR_UNK;
2883 ret = -1;
2884 goto end;
2885 }
2886 }
2887
2888 chunk_status = lttng_trace_chunk_set_credentials_current_user(chunk);
2889 if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
2890 reply_code = LTTNG_ERR_UNK;
2891 ret = -1;
2892 goto end;
2893 }
2894
2895 LTTNG_ASSERT(conn->session->output_directory);
2896 chunk_status = lttng_trace_chunk_set_as_owner(chunk,
2897 conn->session->output_directory);
2898 if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
2899 reply_code = LTTNG_ERR_UNK;
2900 ret = -1;
2901 goto end;
2902 }
2903
2904 published_chunk = sessiond_trace_chunk_registry_publish_chunk(
2905 sessiond_trace_chunk_registry,
2906 conn->session->sessiond_uuid,
2907 conn->session->id_sessiond.is_set ?
2908 conn->session->id_sessiond.value :
2909 conn->session->id,
2910 chunk);
2911 if (!published_chunk) {
2912 char uuid_str[LTTNG_UUID_STR_LEN];
2913
2914 lttng_uuid_to_str(conn->session->sessiond_uuid, uuid_str);
2915 ERR("Failed to publish chunk: sessiond_uuid = %s, session_id = %" PRIu64 ", chunk_id = %" PRIu64,
2916 uuid_str,
2917 conn->session->id,
2918 msg->chunk_id);
2919 ret = -1;
2920 reply_code = LTTNG_ERR_NOMEM;
2921 goto end;
2922 }
2923
2924 if (conn->session->pending_closure_trace_chunk) {
2925 /*
2926 * Invalid; this means a second create_trace_chunk command was
2927 * received before a close_trace_chunk.
2928 */
2929 ERR("Invalid trace chunk close command received; a trace chunk is already waiting for a trace chunk close command");
2930 reply_code = LTTNG_ERR_INVALID_PROTOCOL;
2931 ret = -1;
2932 goto end;
2933 }
2934 conn->session->pending_closure_trace_chunk =
2935 conn->session->current_trace_chunk;
2936 conn->session->current_trace_chunk = published_chunk;
2937 published_chunk = NULL;
2938 if (!conn->session->pending_closure_trace_chunk) {
2939 session->ongoing_rotation = false;
2940 }
2941 end:
2942 pthread_mutex_unlock(&conn->session->lock);
2943 reply.ret_code = htobe32((uint32_t) reply_code);
2944 send_ret = conn->sock->ops->sendmsg(conn->sock,
2945 &reply,
2946 sizeof(struct lttcomm_relayd_generic_reply),
2947 0);
2948 if (send_ret < (ssize_t) sizeof(reply)) {
2949 ERR("Failed to send \"create trace chunk\" command reply (ret = %zd)",
2950 send_ret);
2951 ret = -1;
2952 }
2953 end_no_reply:
2954 lttng_trace_chunk_put(chunk);
2955 lttng_trace_chunk_put(published_chunk);
2956 return ret;
2957 }
2958
2959 /*
2960 * relay_close_trace_chunk: close a trace chunk
2961 */
2962 static int relay_close_trace_chunk(
2963 const struct lttcomm_relayd_hdr *recv_hdr __attribute__((unused)),
2964 struct relay_connection *conn,
2965 const struct lttng_buffer_view *payload)
2966 {
2967 int ret = 0, buf_ret;
2968 ssize_t send_ret;
2969 struct relay_session *session = conn->session;
2970 struct lttcomm_relayd_close_trace_chunk *msg;
2971 struct lttcomm_relayd_close_trace_chunk_reply reply = {};
2972 struct lttng_buffer_view header_view;
2973 struct lttng_trace_chunk *chunk = NULL;
2974 enum lttng_error_code reply_code = LTTNG_OK;
2975 enum lttng_trace_chunk_status chunk_status;
2976 uint64_t chunk_id;
2977 LTTNG_OPTIONAL(enum lttng_trace_chunk_command_type) close_command = {};
2978 time_t close_timestamp;
2979 char closed_trace_chunk_path[LTTNG_PATH_MAX];
2980 size_t path_length = 0;
2981 const char *chunk_name = NULL;
2982 struct lttng_dynamic_buffer reply_payload;
2983 const char *new_path;
2984
2985 lttng_dynamic_buffer_init(&reply_payload);
2986
2987 if (!session || !conn->version_check_done) {
2988 ERR("Trying to close a trace chunk before version check");
2989 ret = -1;
2990 goto end_no_reply;
2991 }
2992
2993 if (session->major == 2 && session->minor < 11) {
2994 ERR("Chunk close command is unsupported before 2.11");
2995 ret = -1;
2996 goto end_no_reply;
2997 }
2998
2999 header_view = lttng_buffer_view_from_view(payload, 0, sizeof(*msg));
3000 if (!lttng_buffer_view_is_valid(&header_view)) {
3001 ERR("Failed to receive payload of chunk close command");
3002 ret = -1;
3003 goto end_no_reply;
3004 }
3005
3006 /* Convert to host endianness. */
3007 msg = (typeof(msg)) header_view.data;
3008 chunk_id = be64toh(msg->chunk_id);
3009 close_timestamp = (time_t) be64toh(msg->close_timestamp);
3010 close_command.value = (lttng_trace_chunk_command_type) be32toh(msg->close_command.value);
3011 close_command.is_set = msg->close_command.is_set;
3012
3013 chunk = sessiond_trace_chunk_registry_get_chunk(
3014 sessiond_trace_chunk_registry,
3015 conn->session->sessiond_uuid,
3016 conn->session->id_sessiond.is_set ?
3017 conn->session->id_sessiond.value :
3018 conn->session->id,
3019 chunk_id);
3020 if (!chunk) {
3021 char uuid_str[LTTNG_UUID_STR_LEN];
3022
3023 lttng_uuid_to_str(conn->session->sessiond_uuid, uuid_str);
3024 ERR("Failed to find chunk to close: sessiond_uuid = %s, session_id = %" PRIu64 ", chunk_id = %" PRIu64,
3025 uuid_str,
3026 conn->session->id,
3027 msg->chunk_id);
3028 ret = -1;
3029 reply_code = LTTNG_ERR_NOMEM;
3030 goto end;
3031 }
3032
3033 pthread_mutex_lock(&session->lock);
3034 if (close_command.is_set &&
3035 close_command.value == LTTNG_TRACE_CHUNK_COMMAND_TYPE_DELETE) {
3036 /*
3037 * Clear command. It is a protocol error to ask for a
3038 * clear on a relay which does not allow it. Querying
3039 * the configuration allows figuring out whether
3040 * clearing is allowed before doing the clear.
3041 */
3042 if (!opt_allow_clear) {
3043 ret = -1;
3044 reply_code = LTTNG_ERR_INVALID_PROTOCOL;
3045 goto end_unlock_session;
3046 }
3047 }
3048 if (session->pending_closure_trace_chunk &&
3049 session->pending_closure_trace_chunk != chunk) {
3050 ERR("Trace chunk close command for session \"%s\" does not target the trace chunk pending closure",
3051 session->session_name);
3052 reply_code = LTTNG_ERR_INVALID_PROTOCOL;
3053 ret = -1;
3054 goto end_unlock_session;
3055 }
3056
3057 if (session->current_trace_chunk && session->current_trace_chunk != chunk &&
3058 !lttng_trace_chunk_get_name_overridden(session->current_trace_chunk)) {
3059 if (close_command.is_set &&
3060 close_command.value == LTTNG_TRACE_CHUNK_COMMAND_TYPE_DELETE &&
3061 !session->has_rotated) {
3062 /* New chunk stays in session output directory. */
3063 new_path = "";
3064 } else {
3065 /* Use chunk name for new chunk. */
3066 new_path = NULL;
3067 }
3068 /* Rename new chunk path. */
3069 chunk_status = lttng_trace_chunk_rename_path(session->current_trace_chunk,
3070 new_path);
3071 if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
3072 ret = -1;
3073 goto end_unlock_session;
3074 }
3075 session->ongoing_rotation = false;
3076 }
3077 if ((!close_command.is_set ||
3078 close_command.value == LTTNG_TRACE_CHUNK_COMMAND_TYPE_NO_OPERATION) &&
3079 !lttng_trace_chunk_get_name_overridden(chunk)) {
3080 const char *old_path;
3081
3082 if (!session->has_rotated) {
3083 old_path = "";
3084 } else {
3085 old_path = NULL;
3086 }
3087 /* We need to move back the .tmp_old_chunk to its rightful place. */
3088 chunk_status = lttng_trace_chunk_rename_path(chunk, old_path);
3089 if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
3090 ret = -1;
3091 goto end_unlock_session;
3092 }
3093 }
3094 chunk_status = lttng_trace_chunk_set_close_timestamp(
3095 chunk, close_timestamp);
3096 if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
3097 ERR("Failed to set trace chunk close timestamp");
3098 ret = -1;
3099 reply_code = LTTNG_ERR_UNK;
3100 goto end_unlock_session;
3101 }
3102
3103 if (close_command.is_set) {
3104 chunk_status = lttng_trace_chunk_set_close_command(
3105 chunk, close_command.value);
3106 if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
3107 ret = -1;
3108 reply_code = LTTNG_ERR_INVALID;
3109 goto end_unlock_session;
3110 }
3111 }
3112 chunk_status = lttng_trace_chunk_get_name(chunk, &chunk_name, NULL);
3113 if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
3114 ERR("Failed to get chunk name");
3115 ret = -1;
3116 reply_code = LTTNG_ERR_UNK;
3117 goto end_unlock_session;
3118 }
3119 if (!session->has_rotated && !session->snapshot) {
3120 ret = lttng_strncpy(closed_trace_chunk_path,
3121 session->output_path,
3122 sizeof(closed_trace_chunk_path));
3123 if (ret) {
3124 ERR("Failed to send trace chunk path: path length of %zu bytes exceeds the maximal allowed length of %zu bytes",
3125 strlen(session->output_path),
3126 sizeof(closed_trace_chunk_path));
3127 reply_code = LTTNG_ERR_NOMEM;
3128 ret = -1;
3129 goto end_unlock_session;
3130 }
3131 } else {
3132 if (session->snapshot) {
3133 ret = snprintf(closed_trace_chunk_path,
3134 sizeof(closed_trace_chunk_path),
3135 "%s/%s", session->output_path,
3136 chunk_name);
3137 } else {
3138 ret = snprintf(closed_trace_chunk_path,
3139 sizeof(closed_trace_chunk_path),
3140 "%s/" DEFAULT_ARCHIVED_TRACE_CHUNKS_DIRECTORY
3141 "/%s",
3142 session->output_path, chunk_name);
3143 }
3144 if (ret < 0 || ret == sizeof(closed_trace_chunk_path)) {
3145 ERR("Failed to format closed trace chunk resulting path");
3146 reply_code = ret < 0 ? LTTNG_ERR_UNK : LTTNG_ERR_NOMEM;
3147 ret = -1;
3148 goto end_unlock_session;
3149 }
3150 }
3151 if (close_command.is_set &&
3152 close_command.value == LTTNG_TRACE_CHUNK_COMMAND_TYPE_MOVE_TO_COMPLETED) {
3153 session->has_rotated = true;
3154 }
3155 DBG("Reply chunk path on close: %s", closed_trace_chunk_path);
3156 path_length = strlen(closed_trace_chunk_path) + 1;
3157 if (path_length > UINT32_MAX) {
3158 ERR("Closed trace chunk path exceeds the maximal length allowed by the protocol");
3159 ret = -1;
3160 reply_code = LTTNG_ERR_INVALID_PROTOCOL;
3161 goto end_unlock_session;
3162 }
3163
3164 if (session->current_trace_chunk == chunk) {
3165 /*
3166 * After a trace chunk close command, no new streams
3167 * referencing the chunk may be created. Hence, on the
3168 * event that no new trace chunk have been created for
3169 * the session, the reference to the current trace chunk
3170 * is released in order to allow it to be reclaimed when
3171 * the last stream releases its reference to it.
3172 */
3173 lttng_trace_chunk_put(session->current_trace_chunk);
3174 session->current_trace_chunk = NULL;
3175 }
3176 lttng_trace_chunk_put(session->pending_closure_trace_chunk);
3177 session->pending_closure_trace_chunk = NULL;
3178 end_unlock_session:
3179 pthread_mutex_unlock(&session->lock);
3180
3181 end:
3182 reply.generic.ret_code = htobe32((uint32_t) reply_code);
3183 reply.path_length = htobe32((uint32_t) path_length);
3184 buf_ret = lttng_dynamic_buffer_append(
3185 &reply_payload, &reply, sizeof(reply));
3186 if (buf_ret) {
3187 ERR("Failed to append \"close trace chunk\" command reply header to payload buffer");
3188 goto end_no_reply;
3189 }
3190
3191 if (reply_code == LTTNG_OK) {
3192 buf_ret = lttng_dynamic_buffer_append(&reply_payload,
3193 closed_trace_chunk_path, path_length);
3194 if (buf_ret) {
3195 ERR("Failed to append \"close trace chunk\" command reply path to payload buffer");
3196 goto end_no_reply;
3197 }
3198 }
3199
3200 send_ret = conn->sock->ops->sendmsg(conn->sock,
3201 reply_payload.data,
3202 reply_payload.size,
3203 0);
3204 if (send_ret < reply_payload.size) {
3205 ERR("Failed to send \"close trace chunk\" command reply of %zu bytes (ret = %zd)",
3206 reply_payload.size, send_ret);
3207 ret = -1;
3208 goto end_no_reply;
3209 }
3210 end_no_reply:
3211 lttng_trace_chunk_put(chunk);
3212 lttng_dynamic_buffer_reset(&reply_payload);
3213 return ret;
3214 }
3215
3216 /*
3217 * relay_trace_chunk_exists: check if a trace chunk exists
3218 */
3219 static int relay_trace_chunk_exists(
3220 const struct lttcomm_relayd_hdr *recv_hdr __attribute__((unused)),
3221 struct relay_connection *conn,
3222 const struct lttng_buffer_view *payload)
3223 {
3224 int ret = 0;
3225 ssize_t send_ret;
3226 struct relay_session *session = conn->session;
3227 struct lttcomm_relayd_trace_chunk_exists *msg;
3228 struct lttcomm_relayd_trace_chunk_exists_reply reply = {};
3229 struct lttng_buffer_view header_view;
3230 uint64_t chunk_id;
3231 bool chunk_exists;
3232
3233 if (!session || !conn->version_check_done) {
3234 ERR("Trying to check for the presence of a trace chunk before version check");
3235 ret = -1;
3236 goto end_no_reply;
3237 }
3238
3239 if (session->major == 2 && session->minor < 11) {
3240 ERR("Chunk exists command is unsupported before 2.11");
3241 ret = -1;
3242 goto end_no_reply;
3243 }
3244
3245 header_view = lttng_buffer_view_from_view(payload, 0, sizeof(*msg));
3246 if (!lttng_buffer_view_is_valid(&header_view)) {
3247 ERR("Failed to receive payload of chunk exists command");
3248 ret = -1;
3249 goto end_no_reply;
3250 }
3251
3252 /* Convert to host endianness. */
3253 msg = (typeof(msg)) header_view.data;
3254 chunk_id = be64toh(msg->chunk_id);
3255
3256 ret = sessiond_trace_chunk_registry_chunk_exists(
3257 sessiond_trace_chunk_registry,
3258 conn->session->sessiond_uuid,
3259 conn->session->id,
3260 chunk_id, &chunk_exists);
3261 /*
3262 * If ret is not 0, send the reply and report the error to the caller.
3263 * It is a protocol (or internal) error and the session/connection
3264 * should be torn down.
3265 */
3266 reply.generic.ret_code = htobe32((uint32_t) (ret == 0 ? LTTNG_OK : LTTNG_ERR_INVALID_PROTOCOL));
3267 reply.trace_chunk_exists = ret == 0 ? chunk_exists : 0;
3268
3269 send_ret = conn->sock->ops->sendmsg(
3270 conn->sock, &reply, sizeof(reply), 0);
3271 if (send_ret < (ssize_t) sizeof(reply)) {
3272 ERR("Failed to send \"create trace chunk\" command reply (ret = %zd)",
3273 send_ret);
3274 ret = -1;
3275 }
3276 end_no_reply:
3277 return ret;
3278 }
3279
3280 /*
3281 * relay_get_configuration: query whether feature is available
3282 */
3283 static int relay_get_configuration(
3284 const struct lttcomm_relayd_hdr *recv_hdr __attribute__((unused)),
3285 struct relay_connection *conn,
3286 const struct lttng_buffer_view *payload)
3287 {
3288 int ret = 0;
3289 ssize_t send_ret;
3290 struct lttcomm_relayd_get_configuration *msg;
3291 struct lttcomm_relayd_get_configuration_reply reply = {};
3292 struct lttng_buffer_view header_view;
3293 uint64_t query_flags = 0;
3294 uint64_t result_flags = 0;
3295
3296 header_view = lttng_buffer_view_from_view(payload, 0, sizeof(*msg));
3297 if (!lttng_buffer_view_is_valid(&header_view)) {
3298 ERR("Failed to receive payload of chunk close command");
3299 ret = -1;
3300 goto end_no_reply;
3301 }
3302
3303 /* Convert to host endianness. */
3304 msg = (typeof(msg)) header_view.data;
3305 query_flags = be64toh(msg->query_flags);
3306
3307 if (query_flags) {
3308 ret = LTTNG_ERR_INVALID_PROTOCOL;
3309 goto reply;
3310 }
3311 if (opt_allow_clear) {
3312 result_flags |= LTTCOMM_RELAYD_CONFIGURATION_FLAG_CLEAR_ALLOWED;
3313 }
3314 ret = 0;
3315 reply:
3316 reply.generic.ret_code = htobe32((uint32_t) (ret == 0 ? LTTNG_OK : LTTNG_ERR_INVALID_PROTOCOL));
3317 reply.relayd_configuration_flags = htobe64(result_flags);
3318
3319 send_ret = conn->sock->ops->sendmsg(
3320 conn->sock, &reply, sizeof(reply), 0);
3321 if (send_ret < (ssize_t) sizeof(reply)) {
3322 ERR("Failed to send \"get configuration\" command reply (ret = %zd)",
3323 send_ret);
3324 ret = -1;
3325 }
3326 end_no_reply:
3327 return ret;
3328 }
3329
3330 static int relay_process_control_command(struct relay_connection *conn,
3331 const struct lttcomm_relayd_hdr *header,
3332 const struct lttng_buffer_view *payload)
3333 {
3334 int ret = 0;
3335
3336 DBG3("Processing \"%s\" command for socket %i",
3337 lttcomm_relayd_command_str((lttcomm_relayd_command) header->cmd),
3338 conn->sock->fd);
3339 switch (header->cmd) {
3340 case RELAYD_CREATE_SESSION:
3341 ret = relay_create_session(header, conn, payload);
3342 break;
3343 case RELAYD_ADD_STREAM:
3344 ret = relay_add_stream(header, conn, payload);
3345 break;
3346 case RELAYD_START_DATA:
3347 ret = relay_start(header, conn, payload);
3348 break;
3349 case RELAYD_SEND_METADATA:
3350 ret = relay_recv_metadata(header, conn, payload);
3351 break;
3352 case RELAYD_VERSION:
3353 ret = relay_send_version(header, conn, payload);
3354 break;
3355 case RELAYD_CLOSE_STREAM:
3356 ret = relay_close_stream(header, conn, payload);
3357 break;
3358 case RELAYD_DATA_PENDING:
3359 ret = relay_data_pending(header, conn, payload);
3360 break;
3361 case RELAYD_QUIESCENT_CONTROL:
3362 ret = relay_quiescent_control(header, conn, payload);
3363 break;
3364 case RELAYD_BEGIN_DATA_PENDING:
3365 ret = relay_begin_data_pending(header, conn, payload);
3366 break;
3367 case RELAYD_END_DATA_PENDING:
3368 ret = relay_end_data_pending(header, conn, payload);
3369 break;
3370 case RELAYD_SEND_INDEX:
3371 ret = relay_recv_index(header, conn, payload);
3372 break;
3373 case RELAYD_STREAMS_SENT:
3374 ret = relay_streams_sent(header, conn, payload);
3375 break;
3376 case RELAYD_RESET_METADATA:
3377 ret = relay_reset_metadata(header, conn, payload);
3378 break;
3379 case RELAYD_ROTATE_STREAMS:
3380 ret = relay_rotate_session_streams(header, conn, payload);
3381 break;
3382 case RELAYD_CREATE_TRACE_CHUNK:
3383 ret = relay_create_trace_chunk(header, conn, payload);
3384 break;
3385 case RELAYD_CLOSE_TRACE_CHUNK:
3386 ret = relay_close_trace_chunk(header, conn, payload);
3387 break;
3388 case RELAYD_TRACE_CHUNK_EXISTS:
3389 ret = relay_trace_chunk_exists(header, conn, payload);
3390 break;
3391 case RELAYD_GET_CONFIGURATION:
3392 ret = relay_get_configuration(header, conn, payload);
3393 break;
3394 case RELAYD_UPDATE_SYNC_INFO:
3395 default:
3396 ERR("Received unknown command (%u)", header->cmd);
3397 relay_unknown_command(conn);
3398 ret = -1;
3399 goto end;
3400 }
3401
3402 end:
3403 return ret;
3404 }
3405
3406 static enum relay_connection_status relay_process_control_receive_payload(
3407 struct relay_connection *conn)
3408 {
3409 int ret = 0;
3410 enum relay_connection_status status = RELAY_CONNECTION_STATUS_OK;
3411 struct lttng_dynamic_buffer *reception_buffer =
3412 &conn->protocol.ctrl.reception_buffer;
3413 struct ctrl_connection_state_receive_payload *state =
3414 &conn->protocol.ctrl.state.receive_payload;
3415 struct lttng_buffer_view payload_view;
3416
3417 if (state->left_to_receive == 0) {
3418 /* Short-circuit for payload-less commands. */
3419 goto reception_complete;
3420 }
3421
3422 ret = conn->sock->ops->recvmsg(conn->sock,
3423 reception_buffer->data + state->received,
3424 state->left_to_receive, MSG_DONTWAIT);
3425 if (ret < 0) {
3426 DIAGNOSTIC_PUSH
3427 DIAGNOSTIC_IGNORE_LOGICAL_OP
3428 if (errno != EAGAIN && errno != EWOULDBLOCK) {
3429 DIAGNOSTIC_POP
3430 PERROR("Unable to receive command payload on sock %d",
3431 conn->sock->fd);
3432 status = RELAY_CONNECTION_STATUS_ERROR;
3433 }
3434 goto end;
3435 } else if (ret == 0) {
3436 DBG("Socket %d performed an orderly shutdown (received EOF)", conn->sock->fd);
3437 status = RELAY_CONNECTION_STATUS_CLOSED;
3438 goto end;
3439 }
3440
3441 LTTNG_ASSERT(ret > 0);
3442 LTTNG_ASSERT(ret <= state->left_to_receive);
3443
3444 state->left_to_receive -= ret;
3445 state->received += ret;
3446
3447 if (state->left_to_receive > 0) {
3448 /*
3449 * Can't transition to the protocol's next state, wait to
3450 * receive the rest of the header.
3451 */
3452 DBG3("Partial reception of control connection protocol payload (received %" PRIu64 " bytes, %" PRIu64 " bytes left to receive, fd = %i)",
3453 state->received, state->left_to_receive,
3454 conn->sock->fd);
3455 goto end;
3456 }
3457
3458 reception_complete:
3459 DBG("Done receiving control command payload: fd = %i, payload size = %" PRIu64 " bytes",
3460 conn->sock->fd, state->received);
3461 /*
3462 * The payload required to process the command has been received.
3463 * A view to the reception buffer is forwarded to the various
3464 * commands and the state of the control is reset on success.
3465 *
3466 * Commands are responsible for sending their reply to the peer.
3467 */
3468 payload_view = lttng_buffer_view_from_dynamic_buffer(reception_buffer,
3469 0, -1);
3470 ret = relay_process_control_command(conn,
3471 &state->header, &payload_view);
3472 if (ret < 0) {
3473 status = RELAY_CONNECTION_STATUS_ERROR;
3474 goto end;
3475 }
3476
3477 ret = connection_reset_protocol_state(conn);
3478 if (ret) {
3479 status = RELAY_CONNECTION_STATUS_ERROR;
3480 }
3481 end:
3482 return status;
3483 }
3484
3485 static enum relay_connection_status relay_process_control_receive_header(
3486 struct relay_connection *conn)
3487 {
3488 int ret = 0;
3489 enum relay_connection_status status = RELAY_CONNECTION_STATUS_OK;
3490 struct lttcomm_relayd_hdr header;
3491 struct lttng_dynamic_buffer *reception_buffer =
3492 &conn->protocol.ctrl.reception_buffer;
3493 struct ctrl_connection_state_receive_header *state =
3494 &conn->protocol.ctrl.state.receive_header;
3495
3496 LTTNG_ASSERT(state->left_to_receive != 0);
3497
3498 ret = conn->sock->ops->recvmsg(conn->sock,
3499 reception_buffer->data + state->received,
3500 state->left_to_receive, MSG_DONTWAIT);
3501 if (ret < 0) {
3502 DIAGNOSTIC_PUSH
3503 DIAGNOSTIC_IGNORE_LOGICAL_OP
3504 if (errno != EAGAIN && errno != EWOULDBLOCK) {
3505 DIAGNOSTIC_POP
3506 PERROR("Unable to receive control command header on sock %d",
3507 conn->sock->fd);
3508 status = RELAY_CONNECTION_STATUS_ERROR;
3509 }
3510 goto end;
3511 } else if (ret == 0) {
3512 DBG("Socket %d performed an orderly shutdown (received EOF)", conn->sock->fd);
3513 status = RELAY_CONNECTION_STATUS_CLOSED;
3514 goto end;
3515 }
3516
3517 LTTNG_ASSERT(ret > 0);
3518 LTTNG_ASSERT(ret <= state->left_to_receive);
3519
3520 state->left_to_receive -= ret;
3521 state->received += ret;
3522
3523 if (state->left_to_receive > 0) {
3524 /*
3525 * Can't transition to the protocol's next state, wait to
3526 * receive the rest of the header.
3527 */
3528 DBG3("Partial reception of control connection protocol header (received %" PRIu64 " bytes, %" PRIu64 " bytes left to receive, fd = %i)",
3529 state->received, state->left_to_receive,
3530 conn->sock->fd);
3531 goto end;
3532 }
3533
3534 /* Transition to next state: receiving the command's payload. */
3535 conn->protocol.ctrl.state_id =
3536 CTRL_CONNECTION_STATE_RECEIVE_PAYLOAD;
3537 memcpy(&header, reception_buffer->data, sizeof(header));
3538 header.circuit_id = be64toh(header.circuit_id);
3539 header.data_size = be64toh(header.data_size);
3540 header.cmd = be32toh(header.cmd);
3541 header.cmd_version = be32toh(header.cmd_version);
3542 memcpy(&conn->protocol.ctrl.state.receive_payload.header,
3543 &header, sizeof(header));
3544
3545 DBG("Done receiving control command header: fd = %i, cmd = %s, cmd_version = %" PRIu32 ", payload size = %" PRIu64 " bytes",
3546 conn->sock->fd, lttcomm_relayd_command_str((enum lttcomm_relayd_command) header.cmd),
3547 header.cmd_version, header.data_size);
3548
3549 if (header.data_size > DEFAULT_NETWORK_RELAYD_CTRL_MAX_PAYLOAD_SIZE) {
3550 ERR("Command header indicates a payload (%" PRIu64 " bytes) that exceeds the maximal payload size allowed on a control connection.",
3551 header.data_size);
3552 status = RELAY_CONNECTION_STATUS_ERROR;
3553 goto end;
3554 }
3555
3556 conn->protocol.ctrl.state.receive_payload.left_to_receive =
3557 header.data_size;
3558 conn->protocol.ctrl.state.receive_payload.received = 0;
3559 ret = lttng_dynamic_buffer_set_size(reception_buffer,
3560 header.data_size);
3561 if (ret) {
3562 status = RELAY_CONNECTION_STATUS_ERROR;
3563 goto end;
3564 }
3565
3566 if (header.data_size == 0) {
3567 /*
3568 * Manually invoke the next state as the poll loop
3569 * will not wake-up to allow us to proceed further.
3570 */
3571 status = relay_process_control_receive_payload(conn);
3572 }
3573 end:
3574 return status;
3575 }
3576
3577 /*
3578 * Process the commands received on the control socket
3579 */
3580 static enum relay_connection_status relay_process_control(
3581 struct relay_connection *conn)
3582 {
3583 enum relay_connection_status status;
3584
3585 switch (conn->protocol.ctrl.state_id) {
3586 case CTRL_CONNECTION_STATE_RECEIVE_HEADER:
3587 status = relay_process_control_receive_header(conn);
3588 break;
3589 case CTRL_CONNECTION_STATE_RECEIVE_PAYLOAD:
3590 status = relay_process_control_receive_payload(conn);
3591 break;
3592 default:
3593 ERR("Unknown control connection protocol state encountered.");
3594 abort();
3595 }
3596
3597 return status;
3598 }
3599
3600 static enum relay_connection_status relay_process_data_receive_header(
3601 struct relay_connection *conn)
3602 {
3603 int ret;
3604 enum relay_connection_status status = RELAY_CONNECTION_STATUS_OK;
3605 struct data_connection_state_receive_header *state =
3606 &conn->protocol.data.state.receive_header;
3607 struct lttcomm_relayd_data_hdr header;
3608 struct relay_stream *stream;
3609
3610 LTTNG_ASSERT(state->left_to_receive != 0);
3611
3612 ret = conn->sock->ops->recvmsg(conn->sock,
3613 state->header_reception_buffer + state->received,
3614 state->left_to_receive, MSG_DONTWAIT);
3615 if (ret < 0) {
3616 DIAGNOSTIC_PUSH
3617 DIAGNOSTIC_IGNORE_LOGICAL_OP
3618 if (errno != EAGAIN && errno != EWOULDBLOCK) {
3619 DIAGNOSTIC_POP
3620 PERROR("Unable to receive data header on sock %d", conn->sock->fd);
3621 status = RELAY_CONNECTION_STATUS_ERROR;
3622 }
3623 goto end;
3624 } else if (ret == 0) {
3625 /* Orderly shutdown. Not necessary to print an error. */
3626 DBG("Socket %d performed an orderly shutdown (received EOF)", conn->sock->fd);
3627 status = RELAY_CONNECTION_STATUS_CLOSED;
3628 goto end;
3629 }
3630
3631 LTTNG_ASSERT(ret > 0);
3632 LTTNG_ASSERT(ret <= state->left_to_receive);
3633
3634 state->left_to_receive -= ret;
3635 state->received += ret;
3636
3637 if (state->left_to_receive > 0) {
3638 /*
3639 * Can't transition to the protocol's next state, wait to
3640 * receive the rest of the header.
3641 */
3642 DBG3("Partial reception of data connection header (received %" PRIu64 " bytes, %" PRIu64 " bytes left to receive, fd = %i)",
3643 state->received, state->left_to_receive,
3644 conn->sock->fd);
3645 goto end;
3646 }
3647
3648 /* Transition to next state: receiving the payload. */
3649 conn->protocol.data.state_id = DATA_CONNECTION_STATE_RECEIVE_PAYLOAD;
3650
3651 memcpy(&header, state->header_reception_buffer, sizeof(header));
3652 header.circuit_id = be64toh(header.circuit_id);
3653 header.stream_id = be64toh(header.stream_id);
3654 header.data_size = be32toh(header.data_size);
3655 header.net_seq_num = be64toh(header.net_seq_num);
3656 header.padding_size = be32toh(header.padding_size);
3657 memcpy(&conn->protocol.data.state.receive_payload.header, &header, sizeof(header));
3658
3659 conn->protocol.data.state.receive_payload.left_to_receive =
3660 header.data_size;
3661 conn->protocol.data.state.receive_payload.received = 0;
3662 conn->protocol.data.state.receive_payload.rotate_index = false;
3663
3664 DBG("Received data connection header on fd %i: circuit_id = %" PRIu64 ", stream_id = %" PRIu64 ", data_size = %" PRIu32 ", net_seq_num = %" PRIu64 ", padding_size = %" PRIu32,
3665 conn->sock->fd, header.circuit_id,
3666 header.stream_id, header.data_size,
3667 header.net_seq_num, header.padding_size);
3668
3669 stream = stream_get_by_id(header.stream_id);
3670 if (!stream) {
3671 DBG("relay_process_data_receive_payload: Cannot find stream %" PRIu64,
3672 header.stream_id);
3673 /* Protocol error. */
3674 status = RELAY_CONNECTION_STATUS_ERROR;
3675 goto end;
3676 }
3677
3678 pthread_mutex_lock(&stream->lock);
3679 /* Prepare stream for the reception of a new packet. */
3680 ret = stream_init_packet(stream, header.data_size,
3681 &conn->protocol.data.state.receive_payload.rotate_index);
3682 pthread_mutex_unlock(&stream->lock);
3683 if (ret) {
3684 ERR("Failed to rotate stream output file");
3685 status = RELAY_CONNECTION_STATUS_ERROR;
3686 goto end_stream_unlock;
3687 }
3688
3689 end_stream_unlock:
3690 stream_put(stream);
3691 end:
3692 return status;
3693 }
3694
3695 static enum relay_connection_status relay_process_data_receive_payload(
3696 struct relay_connection *conn)
3697 {
3698 int ret;
3699 enum relay_connection_status status = RELAY_CONNECTION_STATUS_OK;
3700 struct relay_stream *stream;
3701 struct data_connection_state_receive_payload *state =
3702 &conn->protocol.data.state.receive_payload;
3703 const size_t chunk_size = RECV_DATA_BUFFER_SIZE;
3704 char data_buffer[chunk_size];
3705 bool partial_recv = false;
3706 bool new_stream = false, close_requested = false, index_flushed = false;
3707 uint64_t left_to_receive = state->left_to_receive;
3708 struct relay_session *session;
3709
3710 DBG3("Receiving data for stream id %" PRIu64 " seqnum %" PRIu64 ", %" PRIu64" bytes received, %" PRIu64 " bytes left to receive",
3711 state->header.stream_id, state->header.net_seq_num,
3712 state->received, left_to_receive);
3713
3714 stream = stream_get_by_id(state->header.stream_id);
3715 if (!stream) {
3716 /* Protocol error. */
3717 ERR("relay_process_data_receive_payload: cannot find stream %" PRIu64,
3718 state->header.stream_id);
3719 status = RELAY_CONNECTION_STATUS_ERROR;
3720 goto end;
3721 }
3722
3723 pthread_mutex_lock(&stream->lock);
3724 session = stream->trace->session;
3725 if (!conn->session) {
3726 ret = connection_set_session(conn, session);
3727 if (ret) {
3728 status = RELAY_CONNECTION_STATUS_ERROR;
3729 goto end_stream_unlock;
3730 }
3731 }
3732
3733 /*
3734 * The size of the "chunk" received on any iteration is bounded by:
3735 * - the data left to receive,
3736 * - the data immediately available on the socket,
3737 * - the on-stack data buffer
3738 */
3739 while (left_to_receive > 0 && !partial_recv) {
3740 size_t recv_size = std::min<uint64_t>(left_to_receive, chunk_size);
3741 struct lttng_buffer_view packet_chunk;
3742
3743 ret = conn->sock->ops->recvmsg(conn->sock, data_buffer,
3744 recv_size, MSG_DONTWAIT);
3745 if (ret < 0) {
3746 DIAGNOSTIC_PUSH
3747 DIAGNOSTIC_IGNORE_LOGICAL_OP
3748 if (errno != EAGAIN && errno != EWOULDBLOCK) {
3749 DIAGNOSTIC_POP
3750 PERROR("Socket %d error", conn->sock->fd);
3751 status = RELAY_CONNECTION_STATUS_ERROR;
3752 }
3753 goto end_stream_unlock;
3754 } else if (ret == 0) {
3755 /* No more data ready to be consumed on socket. */
3756 DBG3("No more data ready for consumption on data socket of stream id %" PRIu64,
3757 state->header.stream_id);
3758 status = RELAY_CONNECTION_STATUS_CLOSED;
3759 break;
3760 } else if (ret < (int) recv_size) {
3761 /*
3762 * All the data available on the socket has been
3763 * consumed.
3764 */
3765 partial_recv = true;
3766 recv_size = ret;
3767 }
3768
3769 packet_chunk = lttng_buffer_view_init(data_buffer,
3770 0, recv_size);
3771 LTTNG_ASSERT(packet_chunk.data);
3772
3773 ret = stream_write(stream, &packet_chunk, 0);
3774 if (ret) {
3775 ERR("Relay error writing data to file");
3776 status = RELAY_CONNECTION_STATUS_ERROR;
3777 goto end_stream_unlock;
3778 }
3779
3780 left_to_receive -= recv_size;
3781 state->received += recv_size;
3782 state->left_to_receive = left_to_receive;
3783 }
3784
3785 if (state->left_to_receive > 0) {
3786 /*
3787 * Did not receive all the data expected, wait for more data to
3788 * become available on the socket.
3789 */
3790 DBG3("Partial receive on data connection of stream id %" PRIu64 ", %" PRIu64 " bytes received, %" PRIu64 " bytes left to receive",
3791 state->header.stream_id, state->received,
3792 state->left_to_receive);
3793 goto end_stream_unlock;
3794 }
3795
3796 ret = stream_write(stream, NULL, state->header.padding_size);
3797 if (ret) {
3798 status = RELAY_CONNECTION_STATUS_ERROR;
3799 goto end_stream_unlock;
3800 }
3801
3802 if (session_streams_have_index(session)) {
3803 ret = stream_update_index(stream, state->header.net_seq_num,
3804 state->rotate_index, &index_flushed,
3805 state->header.data_size + state->header.padding_size);
3806 if (ret < 0) {
3807 ERR("Failed to update index: stream %" PRIu64 " net_seq_num %" PRIu64 " ret %d",
3808 stream->stream_handle,
3809 state->header.net_seq_num, ret);
3810 status = RELAY_CONNECTION_STATUS_ERROR;
3811 goto end_stream_unlock;
3812 }
3813 }
3814
3815 if (stream->prev_data_seq == -1ULL) {
3816 new_stream = true;
3817 }
3818
3819 ret = stream_complete_packet(stream, state->header.data_size +
3820 state->header.padding_size, state->header.net_seq_num,
3821 index_flushed);
3822 if (ret) {
3823 status = RELAY_CONNECTION_STATUS_ERROR;
3824 goto end_stream_unlock;
3825 }
3826
3827 /*
3828 * Resetting the protocol state (to RECEIVE_HEADER) will trash the
3829 * contents of *state which are aliased (union) to the same location as
3830 * the new state. Don't use it beyond this point.
3831 */
3832 connection_reset_protocol_state(conn);
3833 state = NULL;
3834
3835 end_stream_unlock:
3836 close_requested = stream->close_requested;
3837 pthread_mutex_unlock(&stream->lock);
3838 if (close_requested && left_to_receive == 0) {
3839 try_stream_close(stream);
3840 }
3841
3842 if (new_stream) {
3843 pthread_mutex_lock(&session->lock);
3844 uatomic_set(&session->new_streams, 1);
3845 pthread_mutex_unlock(&session->lock);
3846 }
3847
3848 stream_put(stream);
3849 end:
3850 return status;
3851 }
3852
3853 /*
3854 * relay_process_data: Process the data received on the data socket
3855 */
3856 static enum relay_connection_status relay_process_data(
3857 struct relay_connection *conn)
3858 {
3859 enum relay_connection_status status;
3860
3861 switch (conn->protocol.data.state_id) {
3862 case DATA_CONNECTION_STATE_RECEIVE_HEADER:
3863 status = relay_process_data_receive_header(conn);
3864 break;
3865 case DATA_CONNECTION_STATE_RECEIVE_PAYLOAD:
3866 status = relay_process_data_receive_payload(conn);
3867 break;
3868 default:
3869 ERR("Unexpected data connection communication state.");
3870 abort();
3871 }
3872
3873 return status;
3874 }
3875
3876 static void cleanup_connection_pollfd(struct lttng_poll_event *events, int pollfd)
3877 {
3878 int ret;
3879
3880 (void) lttng_poll_del(events, pollfd);
3881
3882 ret = fd_tracker_close_unsuspendable_fd(the_fd_tracker, &pollfd, 1,
3883 fd_tracker_util_close_fd, NULL);
3884 if (ret < 0) {
3885 ERR("Closing pollfd %d", pollfd);
3886 }
3887 }
3888
3889 static void relay_thread_close_connection(struct lttng_poll_event *events,
3890 int pollfd, struct relay_connection *conn)
3891 {
3892 const char *type_str;
3893
3894 switch (conn->type) {
3895 case RELAY_DATA:
3896 type_str = "Data";
3897 break;
3898 case RELAY_CONTROL:
3899 type_str = "Control";
3900 break;
3901 case RELAY_VIEWER_COMMAND:
3902 type_str = "Viewer Command";
3903 break;
3904 case RELAY_VIEWER_NOTIFICATION:
3905 type_str = "Viewer Notification";
3906 break;
3907 default:
3908 type_str = "Unknown";
3909 }
3910 cleanup_connection_pollfd(events, pollfd);
3911 connection_put(conn);
3912 DBG("%s connection closed with %d", type_str, pollfd);
3913 }
3914
3915 /*
3916 * This thread does the actual work
3917 */
3918 static void *relay_thread_worker(void *data __attribute__((unused)))
3919 {
3920 int ret, err = -1, last_seen_data_fd = -1;
3921 uint32_t nb_fd;
3922 struct lttng_poll_event events;
3923 struct lttng_ht *relay_connections_ht;
3924 struct lttng_ht_iter iter;
3925 struct relay_connection *destroy_conn = NULL;
3926
3927 DBG("[thread] Relay worker started");
3928
3929 rcu_register_thread();
3930
3931 health_register(health_relayd, HEALTH_RELAYD_TYPE_WORKER);
3932
3933 if (testpoint(relayd_thread_worker)) {
3934 goto error_testpoint;
3935 }
3936
3937 health_code_update();
3938
3939 /* table of connections indexed on socket */
3940 relay_connections_ht = lttng_ht_new(0, LTTNG_HT_TYPE_ULONG);
3941 if (!relay_connections_ht) {
3942 goto relay_connections_ht_error;
3943 }
3944
3945 ret = create_named_thread_poll_set(&events, 2, "Worker thread epoll");
3946 if (ret < 0) {
3947 goto error_poll_create;
3948 }
3949
3950 ret = lttng_poll_add(&events, relay_conn_pipe[0], LPOLLIN | LPOLLRDHUP);
3951 if (ret < 0) {
3952 goto error;
3953 }
3954
3955 restart:
3956 while (1) {
3957 int idx = -1, i, seen_control = 0, last_notdel_data_fd = -1;
3958
3959 health_code_update();
3960
3961 /* Infinite blocking call, waiting for transmission */
3962 DBG3("Relayd worker thread polling...");
3963 health_poll_entry();
3964 ret = lttng_poll_wait(&events, -1);
3965 health_poll_exit();
3966 if (ret < 0) {
3967 /*
3968 * Restart interrupted system call.
3969 */
3970 if (errno == EINTR) {
3971 goto restart;
3972 }
3973 goto error;
3974 }
3975
3976 nb_fd = ret;
3977
3978 /*
3979 * Process control. The control connection is
3980 * prioritized so we don't starve it with high
3981 * throughput tracing data on the data connection.
3982 */
3983 for (i = 0; i < nb_fd; i++) {
3984 /* Fetch once the poll data */
3985 uint32_t revents = LTTNG_POLL_GETEV(&events, i);
3986 int pollfd = LTTNG_POLL_GETFD(&events, i);
3987
3988 health_code_update();
3989
3990 /* Thread quit pipe has been closed. Killing thread. */
3991 ret = check_thread_quit_pipe(pollfd, revents);
3992 if (ret) {
3993 err = 0;
3994 goto exit;
3995 }
3996
3997 /* Inspect the relay conn pipe for new connection */
3998 if (pollfd == relay_conn_pipe[0]) {
3999 if (revents & LPOLLIN) {
4000 struct relay_connection *conn;
4001
4002 ret = lttng_read(relay_conn_pipe[0], &conn, sizeof(conn));
4003 if (ret < 0) {
4004 goto error;
4005 }
4006 ret = lttng_poll_add(&events,
4007 conn->sock->fd,
4008 LPOLLIN | LPOLLRDHUP);
4009 if (ret) {
4010 ERR("Failed to add new connection file descriptor to poll set");
4011 goto error;
4012 }
4013 connection_ht_add(relay_connections_ht, conn);
4014 DBG("Connection socket %d added", conn->sock->fd);
4015 } else if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
4016 ERR("Relay connection pipe error");
4017 goto error;
4018 } else {
4019 ERR("Unexpected poll events %u for sock %d", revents, pollfd);
4020 goto error;
4021 }
4022 } else {
4023 struct relay_connection *ctrl_conn;
4024
4025 ctrl_conn = connection_get_by_sock(relay_connections_ht, pollfd);
4026 /* If not found, there is a synchronization issue. */
4027 LTTNG_ASSERT(ctrl_conn);
4028
4029 if (ctrl_conn->type == RELAY_DATA) {
4030 if (revents & LPOLLIN) {
4031 /*
4032 * Flag the last seen data fd not deleted. It will be
4033 * used as the last seen fd if any fd gets deleted in
4034 * this first loop.
4035 */
4036 last_notdel_data_fd = pollfd;
4037 }
4038 goto put_ctrl_connection;
4039 }
4040 LTTNG_ASSERT(ctrl_conn->type == RELAY_CONTROL);
4041
4042 if (revents & LPOLLIN) {
4043 enum relay_connection_status status;
4044
4045 status = relay_process_control(ctrl_conn);
4046 if (status != RELAY_CONNECTION_STATUS_OK) {
4047 /*
4048 * On socket error flag the session as aborted to force
4049 * the cleanup of its stream otherwise it can leak
4050 * during the lifetime of the relayd.
4051 *
4052 * This prevents situations in which streams can be
4053 * left opened because an index was received, the
4054 * control connection is closed, and the data
4055 * connection is closed (uncleanly) before the packet's
4056 * data provided.
4057 *
4058 * Since the control connection encountered an error,
4059 * it is okay to be conservative and close the
4060 * session right now as we can't rely on the protocol
4061 * being respected anymore.
4062 */
4063 if (status == RELAY_CONNECTION_STATUS_ERROR) {
4064 session_abort(ctrl_conn->session);
4065 }
4066
4067 /* Clear the connection on error or close. */
4068 relay_thread_close_connection(&events,
4069 pollfd,
4070 ctrl_conn);
4071 }
4072 seen_control = 1;
4073 } else if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
4074 relay_thread_close_connection(&events,
4075 pollfd, ctrl_conn);
4076 if (last_seen_data_fd == pollfd) {
4077 last_seen_data_fd = last_notdel_data_fd;
4078 }
4079 } else {
4080 ERR("Unexpected poll events %u for control sock %d",
4081 revents, pollfd);
4082 connection_put(ctrl_conn);
4083 goto error;
4084 }
4085 put_ctrl_connection:
4086 connection_put(ctrl_conn);
4087 }
4088 }
4089
4090 /*
4091 * The last loop handled a control request, go back to poll to make
4092 * sure we prioritise the control socket.
4093 */
4094 if (seen_control) {
4095 continue;
4096 }
4097
4098 if (last_seen_data_fd >= 0) {
4099 for (i = 0; i < nb_fd; i++) {
4100 int pollfd = LTTNG_POLL_GETFD(&events, i);
4101
4102 health_code_update();
4103
4104 if (last_seen_data_fd == pollfd) {
4105 idx = i;
4106 break;
4107 }
4108 }
4109 }
4110
4111 /* Process data connection. */
4112 for (i = idx + 1; i < nb_fd; i++) {
4113 /* Fetch the poll data. */
4114 uint32_t revents = LTTNG_POLL_GETEV(&events, i);
4115 int pollfd = LTTNG_POLL_GETFD(&events, i);
4116 struct relay_connection *data_conn;
4117
4118 health_code_update();
4119
4120 if (!revents) {
4121 /* No activity for this FD (poll implementation). */
4122 continue;
4123 }
4124
4125 /* Skip the command pipe. It's handled in the first loop. */
4126 if (pollfd == relay_conn_pipe[0]) {
4127 continue;
4128 }
4129
4130 data_conn = connection_get_by_sock(relay_connections_ht, pollfd);
4131 if (!data_conn) {
4132 /* Skip it. Might be removed before. */
4133 continue;
4134 }
4135 if (data_conn->type == RELAY_CONTROL) {
4136 goto put_data_connection;
4137 }
4138 LTTNG_ASSERT(data_conn->type == RELAY_DATA);
4139
4140 if (revents & LPOLLIN) {
4141 enum relay_connection_status status;
4142
4143 status = relay_process_data(data_conn);
4144 /* Connection closed or error. */
4145 if (status != RELAY_CONNECTION_STATUS_OK) {
4146 /*
4147 * On socket error flag the session as aborted to force
4148 * the cleanup of its stream otherwise it can leak
4149 * during the lifetime of the relayd.
4150 *
4151 * This prevents situations in which streams can be
4152 * left opened because an index was received, the
4153 * control connection is closed, and the data
4154 * connection is closed (uncleanly) before the packet's
4155 * data provided.
4156 *
4157 * Since the data connection encountered an error,
4158 * it is okay to be conservative and close the
4159 * session right now as we can't rely on the protocol
4160 * being respected anymore.
4161 */
4162 if (status == RELAY_CONNECTION_STATUS_ERROR) {
4163 session_abort(data_conn->session);
4164 }
4165 relay_thread_close_connection(&events, pollfd,
4166 data_conn);
4167 /*
4168 * Every goto restart call sets the last seen fd where
4169 * here we don't really care since we gracefully
4170 * continue the loop after the connection is deleted.
4171 */
4172 } else {
4173 /* Keep last seen port. */
4174 last_seen_data_fd = pollfd;
4175 connection_put(data_conn);
4176 goto restart;
4177 }
4178 } else if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
4179 relay_thread_close_connection(&events, pollfd,
4180 data_conn);
4181 } else {
4182 ERR("Unknown poll events %u for data sock %d",
4183 revents, pollfd);
4184 }
4185 put_data_connection:
4186 connection_put(data_conn);
4187 }
4188 last_seen_data_fd = -1;
4189 }
4190
4191 /* Normal exit, no error */
4192 ret = 0;
4193
4194 exit:
4195 error:
4196 /* Cleanup remaining connection object. */
4197 rcu_read_lock();
4198 cds_lfht_for_each_entry(relay_connections_ht->ht, &iter.iter,
4199 destroy_conn,
4200 sock_n.node) {
4201 health_code_update();
4202
4203 session_abort(destroy_conn->session);
4204
4205 /*
4206 * No need to grab another ref, because we own
4207 * destroy_conn.
4208 */
4209 relay_thread_close_connection(&events, destroy_conn->sock->fd,
4210 destroy_conn);
4211 }
4212 rcu_read_unlock();
4213
4214 (void) fd_tracker_util_poll_clean(the_fd_tracker, &events);
4215 error_poll_create:
4216 lttng_ht_destroy(relay_connections_ht);
4217 relay_connections_ht_error:
4218 /* Close relay conn pipes */
4219 (void) fd_tracker_util_pipe_close(the_fd_tracker,
4220 relay_conn_pipe);
4221 if (err) {
4222 DBG("Thread exited with error");
4223 }
4224 DBG("Worker thread cleanup complete");
4225 error_testpoint:
4226 if (err) {
4227 health_error();
4228 ERR("Health error occurred in %s", __func__);
4229 }
4230 health_unregister(health_relayd);
4231 rcu_unregister_thread();
4232 lttng_relay_stop_threads();
4233 return NULL;
4234 }
4235
4236 /*
4237 * Create the relay command pipe to wake thread_manage_apps.
4238 * Closed in cleanup().
4239 */
4240 static int create_relay_conn_pipe(void)
4241 {
4242 return fd_tracker_util_pipe_open_cloexec(the_fd_tracker,
4243 "Relayd connection pipe", relay_conn_pipe);
4244 }
4245
4246 static int stdio_open(void *data __attribute__((unused)), int *fds)
4247 {
4248 fds[0] = fileno(stdout);
4249 fds[1] = fileno(stderr);
4250 return 0;
4251 }
4252
4253 static int track_stdio(void)
4254 {
4255 int fds[2];
4256 const char *names[] = { "stdout", "stderr" };
4257
4258 return fd_tracker_open_unsuspendable_fd(the_fd_tracker, fds,
4259 names, 2, stdio_open, NULL);
4260 }
4261
4262 /*
4263 * main
4264 */
4265 int main(int argc, char **argv)
4266 {
4267 bool thread_is_rcu_registered = false;
4268 int ret = 0, retval = 0;
4269 void *status;
4270 char *unlinked_file_directory_path = NULL, *output_path = NULL;
4271
4272 /* Parse environment variables */
4273 ret = parse_env_options();
4274 if (ret) {
4275 retval = -1;
4276 goto exit_options;
4277 }
4278
4279 /*
4280 * Parse arguments.
4281 * Command line arguments overwrite environment.
4282 */
4283 progname = argv[0];
4284 if (set_options(argc, argv)) {
4285 retval = -1;
4286 goto exit_options;
4287 }
4288
4289 if (set_signal_handler()) {
4290 retval = -1;
4291 goto exit_options;
4292 }
4293
4294 relayd_config_log();
4295
4296 if (opt_print_version) {
4297 print_version();
4298 retval = 0;
4299 goto exit_options;
4300 }
4301
4302 ret = fclose(stdin);
4303 if (ret) {
4304 PERROR("Failed to close stdin");
4305 goto exit_options;
4306 }
4307
4308 DBG("Clear command %s", opt_allow_clear ? "allowed" : "disallowed");
4309
4310 /* Try to create directory if -o, --output is specified. */
4311 if (opt_output_path) {
4312 if (*opt_output_path != '/') {
4313 ERR("Please specify an absolute path for -o, --output PATH");
4314 retval = -1;
4315 goto exit_options;
4316 }
4317
4318 ret = utils_mkdir_recursive(opt_output_path, S_IRWXU | S_IRWXG,
4319 -1, -1);
4320 if (ret < 0) {
4321 ERR("Unable to create %s", opt_output_path);
4322 retval = -1;
4323 goto exit_options;
4324 }
4325 }
4326
4327 /* Daemonize */
4328 if (opt_daemon || opt_background) {
4329 ret = lttng_daemonize(&child_ppid, &recv_child_signal,
4330 !opt_background);
4331 if (ret < 0) {
4332 retval = -1;
4333 goto exit_options;
4334 }
4335 }
4336
4337 if (opt_working_directory) {
4338 ret = utils_change_working_directory(opt_working_directory);
4339 if (ret) {
4340 /* All errors are already logged. */
4341 goto exit_options;
4342 }
4343 }
4344
4345 sessiond_trace_chunk_registry = sessiond_trace_chunk_registry_create();
4346 if (!sessiond_trace_chunk_registry) {
4347 ERR("Failed to initialize session daemon trace chunk registry");
4348 retval = -1;
4349 goto exit_options;
4350 }
4351
4352 /*
4353 * The RCU thread registration (and use, through the fd-tracker's
4354 * creation) is done after the daemonization to allow us to not
4355 * deal with liburcu's fork() management as the call RCU needs to
4356 * be restored.
4357 */
4358 rcu_register_thread();
4359 thread_is_rcu_registered = true;
4360
4361 output_path = create_output_path("");
4362 if (!output_path) {
4363 ERR("Failed to get output path");
4364 retval = -1;
4365 goto exit_options;
4366 }
4367 ret = asprintf(&unlinked_file_directory_path, "%s/%s", output_path,
4368 DEFAULT_UNLINKED_FILES_DIRECTORY);
4369 free(output_path);
4370 if (ret < 0) {
4371 ERR("Failed to format unlinked file directory path");
4372 retval = -1;
4373 goto exit_options;
4374 }
4375 the_fd_tracker = fd_tracker_create(
4376 unlinked_file_directory_path, lttng_opt_fd_pool_size);
4377 free(unlinked_file_directory_path);
4378 if (!the_fd_tracker) {
4379 retval = -1;
4380 goto exit_options;
4381 }
4382
4383 ret = track_stdio();
4384 if (ret) {
4385 retval = -1;
4386 goto exit_options;
4387 }
4388
4389 /* Initialize thread health monitoring */
4390 health_relayd = health_app_create(NR_HEALTH_RELAYD_TYPES);
4391 if (!health_relayd) {
4392 PERROR("health_app_create error");
4393 retval = -1;
4394 goto exit_options;
4395 }
4396
4397 /* Create thread quit pipe */
4398 if (init_thread_quit_pipe()) {
4399 retval = -1;
4400 goto exit_options;
4401 }
4402
4403 /* Setup the thread apps communication pipe. */
4404 if (create_relay_conn_pipe()) {
4405 retval = -1;
4406 goto exit_options;
4407 }
4408
4409 /* Init relay command queue. */
4410 cds_wfcq_init(&relay_conn_queue.head, &relay_conn_queue.tail);
4411
4412 /* Initialize communication library */
4413 lttcomm_init();
4414 lttcomm_inet_init();
4415
4416 /* tables of sessions indexed by session ID */
4417 sessions_ht = lttng_ht_new(0, LTTNG_HT_TYPE_U64);
4418 if (!sessions_ht) {
4419 retval = -1;
4420 goto exit_options;
4421 }
4422
4423 /* tables of streams indexed by stream ID */
4424 relay_streams_ht = lttng_ht_new(0, LTTNG_HT_TYPE_U64);
4425 if (!relay_streams_ht) {
4426 retval = -1;
4427 goto exit_options;
4428 }
4429
4430 /* tables of streams indexed by stream ID */
4431 viewer_streams_ht = lttng_ht_new(0, LTTNG_HT_TYPE_U64);
4432 if (!viewer_streams_ht) {
4433 retval = -1;
4434 goto exit_options;
4435 }
4436
4437 ret = init_health_quit_pipe();
4438 if (ret) {
4439 retval = -1;
4440 goto exit_options;
4441 }
4442
4443 /* Create thread to manage the client socket */
4444 ret = pthread_create(&health_thread, default_pthread_attr(),
4445 thread_manage_health_relayd, (void *) NULL);
4446 if (ret) {
4447 errno = ret;
4448 PERROR("pthread_create health");
4449 retval = -1;
4450 goto exit_options;
4451 }
4452
4453 /* Setup the dispatcher thread */
4454 ret = pthread_create(&dispatcher_thread, default_pthread_attr(),
4455 relay_thread_dispatcher, (void *) NULL);
4456 if (ret) {
4457 errno = ret;
4458 PERROR("pthread_create dispatcher");
4459 retval = -1;
4460 goto exit_dispatcher_thread;
4461 }
4462
4463 /* Setup the worker thread */
4464 ret = pthread_create(&worker_thread, default_pthread_attr(),
4465 relay_thread_worker, NULL);
4466 if (ret) {
4467 errno = ret;
4468 PERROR("pthread_create worker");
4469 retval = -1;
4470 goto exit_worker_thread;
4471 }
4472
4473 /* Setup the listener thread */
4474 ret = pthread_create(&listener_thread, default_pthread_attr(),
4475 relay_thread_listener, (void *) NULL);
4476 if (ret) {
4477 errno = ret;
4478 PERROR("pthread_create listener");
4479 retval = -1;
4480 goto exit_listener_thread;
4481 }
4482
4483 ret = relayd_live_create(live_uri);
4484 if (ret) {
4485 ERR("Starting live viewer threads");
4486 retval = -1;
4487 goto exit_live;
4488 }
4489
4490 /*
4491 * This is where we start awaiting program completion (e.g. through
4492 * signal that asks threads to teardown).
4493 */
4494
4495 ret = relayd_live_join();
4496 if (ret) {
4497 retval = -1;
4498 }
4499 exit_live:
4500
4501 ret = pthread_join(listener_thread, &status);
4502 if (ret) {
4503 errno = ret;
4504 PERROR("pthread_join listener_thread");
4505 retval = -1;
4506 }
4507
4508 exit_listener_thread:
4509 ret = pthread_join(worker_thread, &status);
4510 if (ret) {
4511 errno = ret;
4512 PERROR("pthread_join worker_thread");
4513 retval = -1;
4514 }
4515
4516 exit_worker_thread:
4517 ret = pthread_join(dispatcher_thread, &status);
4518 if (ret) {
4519 errno = ret;
4520 PERROR("pthread_join dispatcher_thread");
4521 retval = -1;
4522 }
4523 exit_dispatcher_thread:
4524
4525 ret = pthread_join(health_thread, &status);
4526 if (ret) {
4527 errno = ret;
4528 PERROR("pthread_join health_thread");
4529 retval = -1;
4530 }
4531 exit_options:
4532 /*
4533 * Wait for all pending call_rcu work to complete before tearing
4534 * down data structures. call_rcu worker may be trying to
4535 * perform lookups in those structures.
4536 */
4537 rcu_barrier();
4538 relayd_cleanup();
4539
4540 /* Ensure all prior call_rcu are done. */
4541 rcu_barrier();
4542
4543 if (thread_is_rcu_registered) {
4544 rcu_unregister_thread();
4545 }
4546
4547 if (!retval) {
4548 exit(EXIT_SUCCESS);
4549 } else {
4550 exit(EXIT_FAILURE);
4551 }
4552 }
This page took 0.130337 seconds and 4 git commands to generate.