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