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