Introduce LTTNG_RELAYD_WORKING_DIRECTORY environment variable
[lttng-tools.git] / src / bin / lttng-relayd / main.c
1 /*
2 * Copyright (C) 2012 - Julien Desfossez <jdesfossez@efficios.com>
3 * David Goulet <dgoulet@efficios.com>
4 * 2013 - Jérémie Galarneau <jeremie.galarneau@efficios.com>
5 * 2015 - Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
6 *
7 * This program is free software; you can redistribute it and/or modify
8 * it under the terms of the GNU General Public License, version 2 only,
9 * as published by the Free Software Foundation.
10 *
11 * This program is distributed in the hope that it will be useful, but WITHOUT
12 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
13 * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
14 * more details.
15 *
16 * You should have received a copy of the GNU General Public License along
17 * with this program; if not, write to the Free Software Foundation, Inc.,
18 * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
19 */
20
21 #define _LGPL_SOURCE
22 #include <getopt.h>
23 #include <grp.h>
24 #include <limits.h>
25 #include <pthread.h>
26 #include <signal.h>
27 #include <stdio.h>
28 #include <stdlib.h>
29 #include <string.h>
30 #include <sys/mman.h>
31 #include <sys/mount.h>
32 #include <sys/resource.h>
33 #include <sys/socket.h>
34 #include <sys/stat.h>
35 #include <sys/types.h>
36 #include <sys/wait.h>
37 #include <inttypes.h>
38 #include <urcu/futex.h>
39 #include <urcu/uatomic.h>
40 #include <urcu/rculist.h>
41 #include <unistd.h>
42 #include <fcntl.h>
43 #include <strings.h>
44
45 #include <lttng/lttng.h>
46 #include <common/common.h>
47 #include <common/compat/poll.h>
48 #include <common/compat/socket.h>
49 #include <common/compat/endian.h>
50 #include <common/compat/getenv.h>
51 #include <common/defaults.h>
52 #include <common/daemonize.h>
53 #include <common/futex.h>
54 #include <common/sessiond-comm/sessiond-comm.h>
55 #include <common/sessiond-comm/inet.h>
56 #include <common/sessiond-comm/relayd.h>
57 #include <common/uri.h>
58 #include <common/utils.h>
59 #include <common/align.h>
60 #include <common/config/session-config.h>
61 #include <common/dynamic-buffer.h>
62 #include <common/buffer-view.h>
63 #include <common/string-utils/format.h>
64
65 #include "version.h"
66 #include "cmd.h"
67 #include "ctf-trace.h"
68 #include "index.h"
69 #include "utils.h"
70 #include "lttng-relayd.h"
71 #include "live.h"
72 #include "health-relayd.h"
73 #include "testpoint.h"
74 #include "viewer-stream.h"
75 #include "session.h"
76 #include "stream.h"
77 #include "connection.h"
78 #include "tracefile-array.h"
79 #include "tcp_keep_alive.h"
80 #include "sessiond-trace-chunks.h"
81
82 static const char *help_msg =
83 #ifdef LTTNG_EMBED_HELP
84 #include <lttng-relayd.8.h>
85 #else
86 NULL
87 #endif
88 ;
89
90 enum relay_connection_status {
91 RELAY_CONNECTION_STATUS_OK,
92 /* An error occurred while processing an event on the connection. */
93 RELAY_CONNECTION_STATUS_ERROR,
94 /* Connection closed/shutdown cleanly. */
95 RELAY_CONNECTION_STATUS_CLOSED,
96 };
97
98 /* command line options */
99 char *opt_output_path, *opt_working_directory;
100 static int opt_daemon, opt_background, opt_print_version;
101
102 /*
103 * We need to wait for listener and live listener threads, as well as
104 * health check thread, before being ready to signal readiness.
105 */
106 #define NR_LTTNG_RELAY_READY 3
107 static int lttng_relay_ready = NR_LTTNG_RELAY_READY;
108
109 /* Size of receive buffer. */
110 #define RECV_DATA_BUFFER_SIZE 65536
111
112 static int recv_child_signal; /* Set to 1 when a SIGUSR1 signal is received. */
113 static pid_t child_ppid; /* Internal parent PID use with daemonize. */
114
115 static struct lttng_uri *control_uri;
116 static struct lttng_uri *data_uri;
117 static struct lttng_uri *live_uri;
118
119 const char *progname;
120
121 const char *tracing_group_name = DEFAULT_TRACING_GROUP;
122 static int tracing_group_name_override;
123
124 const char * const config_section_name = "relayd";
125
126 /*
127 * Quit pipe for all threads. This permits a single cancellation point
128 * for all threads when receiving an event on the pipe.
129 */
130 int thread_quit_pipe[2] = { -1, -1 };
131
132 /*
133 * This pipe is used to inform the worker thread that a command is queued and
134 * ready to be processed.
135 */
136 static int relay_conn_pipe[2] = { -1, -1 };
137
138 /* Shared between threads */
139 static int dispatch_thread_exit;
140
141 static pthread_t listener_thread;
142 static pthread_t dispatcher_thread;
143 static pthread_t worker_thread;
144 static pthread_t health_thread;
145
146 /*
147 * last_relay_stream_id_lock protects last_relay_stream_id increment
148 * atomicity on 32-bit architectures.
149 */
150 static pthread_mutex_t last_relay_stream_id_lock = PTHREAD_MUTEX_INITIALIZER;
151 static uint64_t last_relay_stream_id;
152
153 /*
154 * Relay command queue.
155 *
156 * The relay_thread_listener and relay_thread_dispatcher communicate with this
157 * queue.
158 */
159 static struct relay_conn_queue relay_conn_queue;
160
161 /* Global relay stream hash table. */
162 struct lttng_ht *relay_streams_ht;
163
164 /* Global relay viewer stream hash table. */
165 struct lttng_ht *viewer_streams_ht;
166
167 /* Global relay sessions hash table. */
168 struct lttng_ht *sessions_ht;
169
170 /* Relayd health monitoring */
171 struct health_app *health_relayd;
172
173 struct sessiond_trace_chunk_registry *sessiond_trace_chunk_registry;
174
175 static struct option long_options[] = {
176 { "control-port", 1, 0, 'C', },
177 { "data-port", 1, 0, 'D', },
178 { "live-port", 1, 0, 'L', },
179 { "daemonize", 0, 0, 'd', },
180 { "background", 0, 0, 'b', },
181 { "group", 1, 0, 'g', },
182 { "help", 0, 0, 'h', },
183 { "output", 1, 0, 'o', },
184 { "verbose", 0, 0, 'v', },
185 { "config", 1, 0, 'f' },
186 { "version", 0, 0, 'V' },
187 { "working-directory", 1, 0, 'w', },
188 { NULL, 0, 0, 0, },
189 };
190
191 static const char *config_ignore_options[] = { "help", "config", "version" };
192
193 static void print_version(void) {
194 fprintf(stdout, "%s\n", VERSION);
195 }
196
197 static void relayd_config_log(void)
198 {
199 DBG("LTTng-relayd " VERSION " - " VERSION_NAME "%s%s",
200 GIT_VERSION[0] == '\0' ? "" : " - " GIT_VERSION,
201 EXTRA_VERSION_NAME[0] == '\0' ? "" : " - " EXTRA_VERSION_NAME);
202 if (EXTRA_VERSION_DESCRIPTION[0] != '\0') {
203 DBG("LTTng-relayd extra version description:\n\t" EXTRA_VERSION_DESCRIPTION "\n");
204 }
205 if (EXTRA_VERSION_PATCHES[0] != '\0') {
206 DBG("LTTng-relayd extra patches:\n\t" EXTRA_VERSION_PATCHES "\n");
207 }
208 }
209
210 /*
211 * Take an option from the getopt output and set it in the right variable to be
212 * used later.
213 *
214 * Return 0 on success else a negative value.
215 */
216 static int set_option(int opt, const char *arg, const char *optname)
217 {
218 int ret;
219
220 switch (opt) {
221 case 0:
222 fprintf(stderr, "option %s", optname);
223 if (arg) {
224 fprintf(stderr, " with arg %s\n", arg);
225 }
226 break;
227 case 'C':
228 if (lttng_is_setuid_setgid()) {
229 WARN("Getting '%s' argument from setuid/setgid binary refused for security reasons.",
230 "-C, --control-port");
231 } else {
232 ret = uri_parse(arg, &control_uri);
233 if (ret < 0) {
234 ERR("Invalid control URI specified");
235 goto end;
236 }
237 if (control_uri->port == 0) {
238 control_uri->port = DEFAULT_NETWORK_CONTROL_PORT;
239 }
240 }
241 break;
242 case 'D':
243 if (lttng_is_setuid_setgid()) {
244 WARN("Getting '%s' argument from setuid/setgid binary refused for security reasons.",
245 "-D, -data-port");
246 } else {
247 ret = uri_parse(arg, &data_uri);
248 if (ret < 0) {
249 ERR("Invalid data URI specified");
250 goto end;
251 }
252 if (data_uri->port == 0) {
253 data_uri->port = DEFAULT_NETWORK_DATA_PORT;
254 }
255 }
256 break;
257 case 'L':
258 if (lttng_is_setuid_setgid()) {
259 WARN("Getting '%s' argument from setuid/setgid binary refused for security reasons.",
260 "-L, -live-port");
261 } else {
262 ret = uri_parse(arg, &live_uri);
263 if (ret < 0) {
264 ERR("Invalid live URI specified");
265 goto end;
266 }
267 if (live_uri->port == 0) {
268 live_uri->port = DEFAULT_NETWORK_VIEWER_PORT;
269 }
270 }
271 break;
272 case 'd':
273 opt_daemon = 1;
274 break;
275 case 'b':
276 opt_background = 1;
277 break;
278 case 'g':
279 if (lttng_is_setuid_setgid()) {
280 WARN("Getting '%s' argument from setuid/setgid binary refused for security reasons.",
281 "-g, --group");
282 } else {
283 tracing_group_name = strdup(arg);
284 if (tracing_group_name == NULL) {
285 ret = -errno;
286 PERROR("strdup");
287 goto end;
288 }
289 tracing_group_name_override = 1;
290 }
291 break;
292 case 'h':
293 ret = utils_show_help(8, "lttng-relayd", help_msg);
294 if (ret) {
295 ERR("Cannot show --help for `lttng-relayd`");
296 perror("exec");
297 }
298 exit(EXIT_FAILURE);
299 case 'V':
300 opt_print_version = 1;
301 break;
302 case 'o':
303 if (lttng_is_setuid_setgid()) {
304 WARN("Getting '%s' argument from setuid/setgid binary refused for security reasons.",
305 "-o, --output");
306 } else {
307 ret = asprintf(&opt_output_path, "%s", arg);
308 if (ret < 0) {
309 ret = -errno;
310 PERROR("asprintf opt_output_path");
311 goto end;
312 }
313 }
314 break;
315 case 'w':
316 if (lttng_is_setuid_setgid()) {
317 WARN("Getting '%s' argument from setuid/setgid binary refused for security reasons.",
318 "-w, --working-directory");
319 } else {
320 ret = asprintf(&opt_working_directory, "%s", arg);
321 if (ret < 0) {
322 ret = -errno;
323 PERROR("asprintf opt_working_directory");
324 goto end;
325 }
326 }
327 break;
328
329 case 'v':
330 /* Verbose level can increase using multiple -v */
331 if (arg) {
332 lttng_opt_verbose = config_parse_value(arg);
333 } else {
334 /* Only 3 level of verbosity (-vvv). */
335 if (lttng_opt_verbose < 3) {
336 lttng_opt_verbose += 1;
337 }
338 }
339 break;
340 default:
341 /* Unknown option or other error.
342 * Error is printed by getopt, just return */
343 ret = -1;
344 goto end;
345 }
346
347 /* All good. */
348 ret = 0;
349
350 end:
351 return ret;
352 }
353
354 /*
355 * config_entry_handler_cb used to handle options read from a config file.
356 * See config_entry_handler_cb comment in common/config/session-config.h for the
357 * return value conventions.
358 */
359 static int config_entry_handler(const struct config_entry *entry, void *unused)
360 {
361 int ret = 0, i;
362
363 if (!entry || !entry->name || !entry->value) {
364 ret = -EINVAL;
365 goto end;
366 }
367
368 /* Check if the option is to be ignored */
369 for (i = 0; i < sizeof(config_ignore_options) / sizeof(char *); i++) {
370 if (!strcmp(entry->name, config_ignore_options[i])) {
371 goto end;
372 }
373 }
374
375 for (i = 0; i < (sizeof(long_options) / sizeof(struct option)) - 1; i++) {
376 /* Ignore if entry name is not fully matched. */
377 if (strcmp(entry->name, long_options[i].name)) {
378 continue;
379 }
380
381 /*
382 * If the option takes no argument on the command line,
383 * we have to check if the value is "true". We support
384 * non-zero numeric values, true, on and yes.
385 */
386 if (!long_options[i].has_arg) {
387 ret = config_parse_value(entry->value);
388 if (ret <= 0) {
389 if (ret) {
390 WARN("Invalid configuration value \"%s\" for option %s",
391 entry->value, entry->name);
392 }
393 /* False, skip boolean config option. */
394 goto end;
395 }
396 }
397
398 ret = set_option(long_options[i].val, entry->value, entry->name);
399 goto end;
400 }
401
402 WARN("Unrecognized option \"%s\" in daemon configuration file.",
403 entry->name);
404
405 end:
406 return ret;
407 }
408
409 static int parse_env_options(void)
410 {
411 int ret = 0;
412 char *value = NULL;
413
414 value = lttng_secure_getenv(DEFAULT_LTTNG_RELAYD_WORKING_DIRECTORY_ENV);
415 if (value) {
416 opt_working_directory = strdup(value);
417 if (!opt_working_directory) {
418 ERR("Failed to allocate working directory string (\"%s\")",
419 value);
420 ret = -1;
421 }
422 }
423 return ret;
424 }
425
426 static int set_options(int argc, char **argv)
427 {
428 int c, ret = 0, option_index = 0, retval = 0;
429 int orig_optopt = optopt, orig_optind = optind;
430 char *default_address, *optstring;
431 const char *config_path = NULL;
432
433 optstring = utils_generate_optstring(long_options,
434 sizeof(long_options) / sizeof(struct option));
435 if (!optstring) {
436 retval = -ENOMEM;
437 goto exit;
438 }
439
440 /* Check for the --config option */
441
442 while ((c = getopt_long(argc, argv, optstring, long_options,
443 &option_index)) != -1) {
444 if (c == '?') {
445 retval = -EINVAL;
446 goto exit;
447 } else if (c != 'f') {
448 continue;
449 }
450
451 if (lttng_is_setuid_setgid()) {
452 WARN("Getting '%s' argument from setuid/setgid binary refused for security reasons.",
453 "-f, --config");
454 } else {
455 config_path = utils_expand_path(optarg);
456 if (!config_path) {
457 ERR("Failed to resolve path: %s", optarg);
458 }
459 }
460 }
461
462 ret = config_get_section_entries(config_path, config_section_name,
463 config_entry_handler, NULL);
464 if (ret) {
465 if (ret > 0) {
466 ERR("Invalid configuration option at line %i", ret);
467 }
468 retval = -1;
469 goto exit;
470 }
471
472 /* Reset getopt's global state */
473 optopt = orig_optopt;
474 optind = orig_optind;
475 while (1) {
476 c = getopt_long(argc, argv, optstring, long_options, &option_index);
477 if (c == -1) {
478 break;
479 }
480
481 ret = set_option(c, optarg, long_options[option_index].name);
482 if (ret < 0) {
483 retval = -1;
484 goto exit;
485 }
486 }
487
488 /* assign default values */
489 if (control_uri == NULL) {
490 ret = asprintf(&default_address,
491 "tcp://" DEFAULT_NETWORK_CONTROL_BIND_ADDRESS ":%d",
492 DEFAULT_NETWORK_CONTROL_PORT);
493 if (ret < 0) {
494 PERROR("asprintf default data address");
495 retval = -1;
496 goto exit;
497 }
498
499 ret = uri_parse(default_address, &control_uri);
500 free(default_address);
501 if (ret < 0) {
502 ERR("Invalid control URI specified");
503 retval = -1;
504 goto exit;
505 }
506 }
507 if (data_uri == NULL) {
508 ret = asprintf(&default_address,
509 "tcp://" DEFAULT_NETWORK_DATA_BIND_ADDRESS ":%d",
510 DEFAULT_NETWORK_DATA_PORT);
511 if (ret < 0) {
512 PERROR("asprintf default data address");
513 retval = -1;
514 goto exit;
515 }
516
517 ret = uri_parse(default_address, &data_uri);
518 free(default_address);
519 if (ret < 0) {
520 ERR("Invalid data URI specified");
521 retval = -1;
522 goto exit;
523 }
524 }
525 if (live_uri == NULL) {
526 ret = asprintf(&default_address,
527 "tcp://" DEFAULT_NETWORK_VIEWER_BIND_ADDRESS ":%d",
528 DEFAULT_NETWORK_VIEWER_PORT);
529 if (ret < 0) {
530 PERROR("asprintf default viewer control address");
531 retval = -1;
532 goto exit;
533 }
534
535 ret = uri_parse(default_address, &live_uri);
536 free(default_address);
537 if (ret < 0) {
538 ERR("Invalid viewer control URI specified");
539 retval = -1;
540 goto exit;
541 }
542 }
543
544 exit:
545 free(optstring);
546 return retval;
547 }
548
549 static void print_global_objects(void)
550 {
551 rcu_register_thread();
552
553 print_viewer_streams();
554 print_relay_streams();
555 print_sessions();
556
557 rcu_unregister_thread();
558 }
559
560 /*
561 * Cleanup the daemon
562 */
563 static void relayd_cleanup(void)
564 {
565 print_global_objects();
566
567 DBG("Cleaning up");
568
569 if (viewer_streams_ht)
570 lttng_ht_destroy(viewer_streams_ht);
571 if (relay_streams_ht)
572 lttng_ht_destroy(relay_streams_ht);
573 if (sessions_ht)
574 lttng_ht_destroy(sessions_ht);
575
576 free(opt_output_path);
577 free(opt_working_directory);
578
579 /* Close thread quit pipes */
580 utils_close_pipe(thread_quit_pipe);
581
582 uri_free(control_uri);
583 uri_free(data_uri);
584 /* Live URI is freed in the live thread. */
585
586 if (tracing_group_name_override) {
587 free((void *) tracing_group_name);
588 }
589 }
590
591 /*
592 * Write to writable pipe used to notify a thread.
593 */
594 static int notify_thread_pipe(int wpipe)
595 {
596 ssize_t ret;
597
598 ret = lttng_write(wpipe, "!", 1);
599 if (ret < 1) {
600 PERROR("write poll pipe");
601 goto end;
602 }
603 ret = 0;
604 end:
605 return ret;
606 }
607
608 static int notify_health_quit_pipe(int *pipe)
609 {
610 ssize_t ret;
611
612 ret = lttng_write(pipe[1], "4", 1);
613 if (ret < 1) {
614 PERROR("write relay health quit");
615 goto end;
616 }
617 ret = 0;
618 end:
619 return ret;
620 }
621
622 /*
623 * Stop all relayd and relayd-live threads.
624 */
625 int lttng_relay_stop_threads(void)
626 {
627 int retval = 0;
628
629 /* Stopping all threads */
630 DBG("Terminating all threads");
631 if (notify_thread_pipe(thread_quit_pipe[1])) {
632 ERR("write error on thread quit pipe");
633 retval = -1;
634 }
635
636 if (notify_health_quit_pipe(health_quit_pipe)) {
637 ERR("write error on health quit pipe");
638 }
639
640 /* Dispatch thread */
641 CMM_STORE_SHARED(dispatch_thread_exit, 1);
642 futex_nto1_wake(&relay_conn_queue.futex);
643
644 if (relayd_live_stop()) {
645 ERR("Error stopping live threads");
646 retval = -1;
647 }
648 return retval;
649 }
650
651 /*
652 * Signal handler for the daemon
653 *
654 * Simply stop all worker threads, leaving main() return gracefully after
655 * joining all threads and calling cleanup().
656 */
657 static void sighandler(int sig)
658 {
659 switch (sig) {
660 case SIGINT:
661 DBG("SIGINT caught");
662 if (lttng_relay_stop_threads()) {
663 ERR("Error stopping threads");
664 }
665 break;
666 case SIGTERM:
667 DBG("SIGTERM caught");
668 if (lttng_relay_stop_threads()) {
669 ERR("Error stopping threads");
670 }
671 break;
672 case SIGUSR1:
673 CMM_STORE_SHARED(recv_child_signal, 1);
674 break;
675 default:
676 break;
677 }
678 }
679
680 /*
681 * Setup signal handler for :
682 * SIGINT, SIGTERM, SIGPIPE
683 */
684 static int set_signal_handler(void)
685 {
686 int ret = 0;
687 struct sigaction sa;
688 sigset_t sigset;
689
690 if ((ret = sigemptyset(&sigset)) < 0) {
691 PERROR("sigemptyset");
692 return ret;
693 }
694
695 sa.sa_mask = sigset;
696 sa.sa_flags = 0;
697
698 sa.sa_handler = sighandler;
699 if ((ret = sigaction(SIGTERM, &sa, NULL)) < 0) {
700 PERROR("sigaction");
701 return ret;
702 }
703
704 if ((ret = sigaction(SIGINT, &sa, NULL)) < 0) {
705 PERROR("sigaction");
706 return ret;
707 }
708
709 if ((ret = sigaction(SIGUSR1, &sa, NULL)) < 0) {
710 PERROR("sigaction");
711 return ret;
712 }
713
714 sa.sa_handler = SIG_IGN;
715 if ((ret = sigaction(SIGPIPE, &sa, NULL)) < 0) {
716 PERROR("sigaction");
717 return ret;
718 }
719
720 DBG("Signal handler set for SIGTERM, SIGUSR1, SIGPIPE and SIGINT");
721
722 return ret;
723 }
724
725 void lttng_relay_notify_ready(void)
726 {
727 /* Notify the parent of the fork() process that we are ready. */
728 if (opt_daemon || opt_background) {
729 if (uatomic_sub_return(&lttng_relay_ready, 1) == 0) {
730 kill(child_ppid, SIGUSR1);
731 }
732 }
733 }
734
735 /*
736 * Init thread quit pipe.
737 *
738 * Return -1 on error or 0 if all pipes are created.
739 */
740 static int init_thread_quit_pipe(void)
741 {
742 int ret;
743
744 ret = utils_create_pipe_cloexec(thread_quit_pipe);
745
746 return ret;
747 }
748
749 /*
750 * Create a poll set with O_CLOEXEC and add the thread quit pipe to the set.
751 */
752 static int create_thread_poll_set(struct lttng_poll_event *events, int size)
753 {
754 int ret;
755
756 if (events == NULL || size == 0) {
757 ret = -1;
758 goto error;
759 }
760
761 ret = lttng_poll_create(events, size, LTTNG_CLOEXEC);
762 if (ret < 0) {
763 goto error;
764 }
765
766 /* Add quit pipe */
767 ret = lttng_poll_add(events, thread_quit_pipe[0], LPOLLIN | LPOLLERR);
768 if (ret < 0) {
769 goto error;
770 }
771
772 return 0;
773
774 error:
775 return ret;
776 }
777
778 /*
779 * Check if the thread quit pipe was triggered.
780 *
781 * Return 1 if it was triggered else 0;
782 */
783 static int check_thread_quit_pipe(int fd, uint32_t events)
784 {
785 if (fd == thread_quit_pipe[0] && (events & LPOLLIN)) {
786 return 1;
787 }
788
789 return 0;
790 }
791
792 /*
793 * Create and init socket from uri.
794 */
795 static struct lttcomm_sock *relay_socket_create(struct lttng_uri *uri)
796 {
797 int ret;
798 struct lttcomm_sock *sock = NULL;
799
800 sock = lttcomm_alloc_sock_from_uri(uri);
801 if (sock == NULL) {
802 ERR("Allocating socket");
803 goto error;
804 }
805
806 ret = lttcomm_create_sock(sock);
807 if (ret < 0) {
808 goto error;
809 }
810 DBG("Listening on sock %d", sock->fd);
811
812 ret = sock->ops->bind(sock);
813 if (ret < 0) {
814 PERROR("Failed to bind socket");
815 goto error;
816 }
817
818 ret = sock->ops->listen(sock, -1);
819 if (ret < 0) {
820 goto error;
821
822 }
823
824 return sock;
825
826 error:
827 if (sock) {
828 lttcomm_destroy_sock(sock);
829 }
830 return NULL;
831 }
832
833 /*
834 * This thread manages the listening for new connections on the network
835 */
836 static void *relay_thread_listener(void *data)
837 {
838 int i, ret, pollfd, err = -1;
839 uint32_t revents, nb_fd;
840 struct lttng_poll_event events;
841 struct lttcomm_sock *control_sock, *data_sock;
842
843 DBG("[thread] Relay listener started");
844
845 health_register(health_relayd, HEALTH_RELAYD_TYPE_LISTENER);
846
847 health_code_update();
848
849 control_sock = relay_socket_create(control_uri);
850 if (!control_sock) {
851 goto error_sock_control;
852 }
853
854 data_sock = relay_socket_create(data_uri);
855 if (!data_sock) {
856 goto error_sock_relay;
857 }
858
859 /*
860 * Pass 3 as size here for the thread quit pipe, control and
861 * data socket.
862 */
863 ret = create_thread_poll_set(&events, 3);
864 if (ret < 0) {
865 goto error_create_poll;
866 }
867
868 /* Add the control socket */
869 ret = lttng_poll_add(&events, control_sock->fd, LPOLLIN | LPOLLRDHUP);
870 if (ret < 0) {
871 goto error_poll_add;
872 }
873
874 /* Add the data socket */
875 ret = lttng_poll_add(&events, data_sock->fd, LPOLLIN | LPOLLRDHUP);
876 if (ret < 0) {
877 goto error_poll_add;
878 }
879
880 lttng_relay_notify_ready();
881
882 if (testpoint(relayd_thread_listener)) {
883 goto error_testpoint;
884 }
885
886 while (1) {
887 health_code_update();
888
889 DBG("Listener accepting connections");
890
891 restart:
892 health_poll_entry();
893 ret = lttng_poll_wait(&events, -1);
894 health_poll_exit();
895 if (ret < 0) {
896 /*
897 * Restart interrupted system call.
898 */
899 if (errno == EINTR) {
900 goto restart;
901 }
902 goto error;
903 }
904
905 nb_fd = ret;
906
907 DBG("Relay new connection received");
908 for (i = 0; i < nb_fd; i++) {
909 health_code_update();
910
911 /* Fetch once the poll data */
912 revents = LTTNG_POLL_GETEV(&events, i);
913 pollfd = LTTNG_POLL_GETFD(&events, i);
914
915 /* Thread quit pipe has been closed. Killing thread. */
916 ret = check_thread_quit_pipe(pollfd, revents);
917 if (ret) {
918 err = 0;
919 goto exit;
920 }
921
922 if (revents & LPOLLIN) {
923 /*
924 * A new connection is requested, therefore a
925 * sessiond/consumerd connection is allocated in
926 * this thread, enqueued to a global queue and
927 * dequeued (and freed) in the worker thread.
928 */
929 int val = 1;
930 struct relay_connection *new_conn;
931 struct lttcomm_sock *newsock;
932 enum connection_type type;
933
934 if (pollfd == data_sock->fd) {
935 type = RELAY_DATA;
936 newsock = data_sock->ops->accept(data_sock);
937 DBG("Relay data connection accepted, socket %d",
938 newsock->fd);
939 } else {
940 assert(pollfd == control_sock->fd);
941 type = RELAY_CONTROL;
942 newsock = control_sock->ops->accept(control_sock);
943 DBG("Relay control connection accepted, socket %d",
944 newsock->fd);
945 }
946 if (!newsock) {
947 PERROR("accepting sock");
948 goto error;
949 }
950
951 ret = setsockopt(newsock->fd, SOL_SOCKET, SO_REUSEADDR, &val,
952 sizeof(val));
953 if (ret < 0) {
954 PERROR("setsockopt inet");
955 lttcomm_destroy_sock(newsock);
956 goto error;
957 }
958
959 ret = socket_apply_keep_alive_config(newsock->fd);
960 if (ret < 0) {
961 ERR("Failed to apply TCP keep-alive configuration on socket (%i)",
962 newsock->fd);
963 lttcomm_destroy_sock(newsock);
964 goto error;
965 }
966
967 new_conn = connection_create(newsock, type);
968 if (!new_conn) {
969 lttcomm_destroy_sock(newsock);
970 goto error;
971 }
972
973 /* Enqueue request for the dispatcher thread. */
974 cds_wfcq_enqueue(&relay_conn_queue.head, &relay_conn_queue.tail,
975 &new_conn->qnode);
976
977 /*
978 * Wake the dispatch queue futex.
979 * Implicit memory barrier with the
980 * exchange in cds_wfcq_enqueue.
981 */
982 futex_nto1_wake(&relay_conn_queue.futex);
983 } else if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
984 ERR("socket poll error");
985 goto error;
986 } else {
987 ERR("Unexpected poll events %u for sock %d", revents, pollfd);
988 goto error;
989 }
990 }
991 }
992
993 exit:
994 error:
995 error_poll_add:
996 error_testpoint:
997 lttng_poll_clean(&events);
998 error_create_poll:
999 if (data_sock->fd >= 0) {
1000 ret = data_sock->ops->close(data_sock);
1001 if (ret) {
1002 PERROR("close");
1003 }
1004 }
1005 lttcomm_destroy_sock(data_sock);
1006 error_sock_relay:
1007 if (control_sock->fd >= 0) {
1008 ret = control_sock->ops->close(control_sock);
1009 if (ret) {
1010 PERROR("close");
1011 }
1012 }
1013 lttcomm_destroy_sock(control_sock);
1014 error_sock_control:
1015 if (err) {
1016 health_error();
1017 ERR("Health error occurred in %s", __func__);
1018 }
1019 health_unregister(health_relayd);
1020 DBG("Relay listener thread cleanup complete");
1021 lttng_relay_stop_threads();
1022 return NULL;
1023 }
1024
1025 /*
1026 * This thread manages the dispatching of the requests to worker threads
1027 */
1028 static void *relay_thread_dispatcher(void *data)
1029 {
1030 int err = -1;
1031 ssize_t ret;
1032 struct cds_wfcq_node *node;
1033 struct relay_connection *new_conn = NULL;
1034
1035 DBG("[thread] Relay dispatcher started");
1036
1037 health_register(health_relayd, HEALTH_RELAYD_TYPE_DISPATCHER);
1038
1039 if (testpoint(relayd_thread_dispatcher)) {
1040 goto error_testpoint;
1041 }
1042
1043 health_code_update();
1044
1045 for (;;) {
1046 health_code_update();
1047
1048 /* Atomically prepare the queue futex */
1049 futex_nto1_prepare(&relay_conn_queue.futex);
1050
1051 if (CMM_LOAD_SHARED(dispatch_thread_exit)) {
1052 break;
1053 }
1054
1055 do {
1056 health_code_update();
1057
1058 /* Dequeue commands */
1059 node = cds_wfcq_dequeue_blocking(&relay_conn_queue.head,
1060 &relay_conn_queue.tail);
1061 if (node == NULL) {
1062 DBG("Woken up but nothing in the relay command queue");
1063 /* Continue thread execution */
1064 break;
1065 }
1066 new_conn = caa_container_of(node, struct relay_connection, qnode);
1067
1068 DBG("Dispatching request waiting on sock %d", new_conn->sock->fd);
1069
1070 /*
1071 * Inform worker thread of the new request. This
1072 * call is blocking so we can be assured that
1073 * the data will be read at some point in time
1074 * or wait to the end of the world :)
1075 */
1076 ret = lttng_write(relay_conn_pipe[1], &new_conn, sizeof(new_conn));
1077 if (ret < 0) {
1078 PERROR("write connection pipe");
1079 connection_put(new_conn);
1080 goto error;
1081 }
1082 } while (node != NULL);
1083
1084 /* Futex wait on queue. Blocking call on futex() */
1085 health_poll_entry();
1086 futex_nto1_wait(&relay_conn_queue.futex);
1087 health_poll_exit();
1088 }
1089
1090 /* Normal exit, no error */
1091 err = 0;
1092
1093 error:
1094 error_testpoint:
1095 if (err) {
1096 health_error();
1097 ERR("Health error occurred in %s", __func__);
1098 }
1099 health_unregister(health_relayd);
1100 DBG("Dispatch thread dying");
1101 lttng_relay_stop_threads();
1102 return NULL;
1103 }
1104
1105 static bool session_streams_have_index(const struct relay_session *session)
1106 {
1107 return session->minor >= 4 && !session->snapshot;
1108 }
1109
1110 /*
1111 * Handle the RELAYD_CREATE_SESSION command.
1112 *
1113 * On success, send back the session id or else return a negative value.
1114 */
1115 static int relay_create_session(const struct lttcomm_relayd_hdr *recv_hdr,
1116 struct relay_connection *conn,
1117 const struct lttng_buffer_view *payload)
1118 {
1119 int ret = 0;
1120 ssize_t send_ret;
1121 struct relay_session *session = NULL;
1122 struct lttcomm_relayd_create_session_reply_2_11 reply = {};
1123 char session_name[LTTNG_NAME_MAX] = {};
1124 char hostname[LTTNG_HOST_NAME_MAX] = {};
1125 uint32_t live_timer = 0;
1126 bool snapshot = false;
1127 bool session_name_contains_creation_timestamp = false;
1128 /* Left nil for peers < 2.11. */
1129 char base_path[LTTNG_PATH_MAX] = {};
1130 lttng_uuid sessiond_uuid = {};
1131 LTTNG_OPTIONAL(uint64_t) id_sessiond = {};
1132 LTTNG_OPTIONAL(uint64_t) current_chunk_id = {};
1133 LTTNG_OPTIONAL(time_t) creation_time = {};
1134 struct lttng_dynamic_buffer reply_payload;
1135
1136 lttng_dynamic_buffer_init(&reply_payload);
1137
1138 if (conn->minor < 4) {
1139 /* From 2.1 to 2.3 */
1140 ret = 0;
1141 } else if (conn->minor >= 4 && conn->minor < 11) {
1142 /* From 2.4 to 2.10 */
1143 ret = cmd_create_session_2_4(payload, session_name,
1144 hostname, &live_timer, &snapshot);
1145 } else {
1146 bool has_current_chunk;
1147 uint64_t current_chunk_id_value;
1148 time_t creation_time_value;
1149 uint64_t id_sessiond_value;
1150
1151 /* From 2.11 to ... */
1152 ret = cmd_create_session_2_11(payload, session_name, hostname,
1153 base_path, &live_timer, &snapshot, &id_sessiond_value,
1154 sessiond_uuid, &has_current_chunk,
1155 &current_chunk_id_value, &creation_time_value,
1156 &session_name_contains_creation_timestamp);
1157 if (lttng_uuid_is_nil(sessiond_uuid)) {
1158 /* The nil UUID is reserved for pre-2.11 clients. */
1159 ERR("Illegal nil UUID announced by peer in create session command");
1160 ret = -1;
1161 goto send_reply;
1162 }
1163 LTTNG_OPTIONAL_SET(&id_sessiond, id_sessiond_value);
1164 LTTNG_OPTIONAL_SET(&creation_time, creation_time_value);
1165 if (has_current_chunk) {
1166 LTTNG_OPTIONAL_SET(&current_chunk_id,
1167 current_chunk_id_value);
1168 }
1169 }
1170
1171 if (ret < 0) {
1172 goto send_reply;
1173 }
1174
1175 session = session_create(session_name, hostname, base_path, live_timer,
1176 snapshot, sessiond_uuid,
1177 id_sessiond.is_set ? &id_sessiond.value : NULL,
1178 current_chunk_id.is_set ? &current_chunk_id.value : NULL,
1179 creation_time.is_set ? &creation_time.value : NULL,
1180 conn->major, conn->minor,
1181 session_name_contains_creation_timestamp);
1182 if (!session) {
1183 ret = -1;
1184 goto send_reply;
1185 }
1186 assert(!conn->session);
1187 conn->session = session;
1188 DBG("Created session %" PRIu64, session->id);
1189
1190 reply.generic.session_id = htobe64(session->id);
1191
1192 send_reply:
1193 if (ret < 0) {
1194 reply.generic.ret_code = htobe32(LTTNG_ERR_FATAL);
1195 } else {
1196 reply.generic.ret_code = htobe32(LTTNG_OK);
1197 }
1198
1199 if (conn->minor < 11) {
1200 /* From 2.1 to 2.10 */
1201 ret = lttng_dynamic_buffer_append(&reply_payload,
1202 &reply.generic, sizeof(reply.generic));
1203 if (ret) {
1204 ERR("Failed to append \"create session\" command reply header to payload buffer");
1205 ret = -1;
1206 goto end;
1207 }
1208 } else {
1209 const uint32_t output_path_length =
1210 session ? strlen(session->output_path) + 1 : 0;
1211
1212 reply.output_path_length = htobe32(output_path_length);
1213 ret = lttng_dynamic_buffer_append(
1214 &reply_payload, &reply, sizeof(reply));
1215 if (ret) {
1216 ERR("Failed to append \"create session\" command reply header to payload buffer");
1217 goto end;
1218 }
1219
1220 if (output_path_length) {
1221 ret = lttng_dynamic_buffer_append(&reply_payload,
1222 session->output_path,
1223 output_path_length);
1224 if (ret) {
1225 ERR("Failed to append \"create session\" command reply path to payload buffer");
1226 goto end;
1227 }
1228 }
1229 }
1230
1231 send_ret = conn->sock->ops->sendmsg(conn->sock, reply_payload.data,
1232 reply_payload.size, 0);
1233 if (send_ret < (ssize_t) reply_payload.size) {
1234 ERR("Failed to send \"create session\" command reply of %zu bytes (ret = %zd)",
1235 reply_payload.size, send_ret);
1236 ret = -1;
1237 }
1238 end:
1239 if (ret < 0 && session) {
1240 session_put(session);
1241 }
1242 lttng_dynamic_buffer_reset(&reply_payload);
1243 return ret;
1244 }
1245
1246 /*
1247 * When we have received all the streams and the metadata for a channel,
1248 * we make them visible to the viewer threads.
1249 */
1250 static void publish_connection_local_streams(struct relay_connection *conn)
1251 {
1252 struct relay_stream *stream;
1253 struct relay_session *session = conn->session;
1254
1255 /*
1256 * We publish all streams belonging to a session atomically wrt
1257 * session lock.
1258 */
1259 pthread_mutex_lock(&session->lock);
1260 rcu_read_lock();
1261 cds_list_for_each_entry_rcu(stream, &session->recv_list,
1262 recv_node) {
1263 stream_publish(stream);
1264 }
1265 rcu_read_unlock();
1266
1267 /*
1268 * Inform the viewer that there are new streams in the session.
1269 */
1270 if (session->viewer_attached) {
1271 uatomic_set(&session->new_streams, 1);
1272 }
1273 pthread_mutex_unlock(&session->lock);
1274 }
1275
1276 static int conform_channel_path(char *channel_path)
1277 {
1278 int ret = 0;
1279
1280 if (strstr("../", channel_path)) {
1281 ERR("Refusing channel path as it walks up the path hierarchy: \"%s\"",
1282 channel_path);
1283 ret = -1;
1284 goto end;
1285 }
1286
1287 if (*channel_path == '/') {
1288 const size_t len = strlen(channel_path);
1289
1290 /*
1291 * Channel paths from peers prior to 2.11 are expressed as an
1292 * absolute path that is, in reality, relative to the relay
1293 * daemon's output directory. Remove the leading slash so it
1294 * is correctly interpreted as a relative path later on.
1295 *
1296 * len (and not len - 1) is used to copy the trailing NULL.
1297 */
1298 bcopy(channel_path + 1, channel_path, len);
1299 }
1300 end:
1301 return ret;
1302 }
1303
1304 /*
1305 * relay_add_stream: allocate a new stream for a session
1306 */
1307 static int relay_add_stream(const struct lttcomm_relayd_hdr *recv_hdr,
1308 struct relay_connection *conn,
1309 const struct lttng_buffer_view *payload)
1310 {
1311 int ret;
1312 ssize_t send_ret;
1313 struct relay_session *session = conn->session;
1314 struct relay_stream *stream = NULL;
1315 struct lttcomm_relayd_status_stream reply;
1316 struct ctf_trace *trace = NULL;
1317 uint64_t stream_handle = -1ULL;
1318 char *path_name = NULL, *channel_name = NULL;
1319 uint64_t tracefile_size = 0, tracefile_count = 0;
1320 LTTNG_OPTIONAL(uint64_t) stream_chunk_id = {};
1321
1322 if (!session || !conn->version_check_done) {
1323 ERR("Trying to add a stream before version check");
1324 ret = -1;
1325 goto end_no_session;
1326 }
1327
1328 if (session->minor == 1) {
1329 /* For 2.1 */
1330 ret = cmd_recv_stream_2_1(payload, &path_name,
1331 &channel_name);
1332 } else if (session->minor > 1 && session->minor < 11) {
1333 /* From 2.2 to 2.10 */
1334 ret = cmd_recv_stream_2_2(payload, &path_name,
1335 &channel_name, &tracefile_size, &tracefile_count);
1336 } else {
1337 /* From 2.11 to ... */
1338 ret = cmd_recv_stream_2_11(payload, &path_name,
1339 &channel_name, &tracefile_size, &tracefile_count,
1340 &stream_chunk_id.value);
1341 stream_chunk_id.is_set = true;
1342 }
1343
1344 if (ret < 0) {
1345 goto send_reply;
1346 }
1347
1348 if (conform_channel_path(path_name)) {
1349 goto send_reply;
1350 }
1351
1352 trace = ctf_trace_get_by_path_or_create(session, path_name);
1353 if (!trace) {
1354 goto send_reply;
1355 }
1356 /* This stream here has one reference on the trace. */
1357
1358 pthread_mutex_lock(&last_relay_stream_id_lock);
1359 stream_handle = ++last_relay_stream_id;
1360 pthread_mutex_unlock(&last_relay_stream_id_lock);
1361
1362 /* We pass ownership of path_name and channel_name. */
1363 stream = stream_create(trace, stream_handle, path_name,
1364 channel_name, tracefile_size, tracefile_count);
1365 path_name = NULL;
1366 channel_name = NULL;
1367
1368 /*
1369 * Streams are the owners of their trace. Reference to trace is
1370 * kept within stream_create().
1371 */
1372 ctf_trace_put(trace);
1373
1374 send_reply:
1375 memset(&reply, 0, sizeof(reply));
1376 reply.handle = htobe64(stream_handle);
1377 if (!stream) {
1378 reply.ret_code = htobe32(LTTNG_ERR_UNK);
1379 } else {
1380 reply.ret_code = htobe32(LTTNG_OK);
1381 }
1382
1383 send_ret = conn->sock->ops->sendmsg(conn->sock, &reply,
1384 sizeof(struct lttcomm_relayd_status_stream), 0);
1385 if (send_ret < (ssize_t) sizeof(reply)) {
1386 ERR("Failed to send \"add stream\" command reply (ret = %zd)",
1387 send_ret);
1388 ret = -1;
1389 }
1390
1391 end_no_session:
1392 free(path_name);
1393 free(channel_name);
1394 return ret;
1395 }
1396
1397 /*
1398 * relay_close_stream: close a specific stream
1399 */
1400 static int relay_close_stream(const struct lttcomm_relayd_hdr *recv_hdr,
1401 struct relay_connection *conn,
1402 const struct lttng_buffer_view *payload)
1403 {
1404 int ret;
1405 ssize_t send_ret;
1406 struct relay_session *session = conn->session;
1407 struct lttcomm_relayd_close_stream stream_info;
1408 struct lttcomm_relayd_generic_reply reply;
1409 struct relay_stream *stream;
1410
1411 DBG("Close stream received");
1412
1413 if (!session || !conn->version_check_done) {
1414 ERR("Trying to close a stream before version check");
1415 ret = -1;
1416 goto end_no_session;
1417 }
1418
1419 if (payload->size < sizeof(stream_info)) {
1420 ERR("Unexpected payload size in \"relay_close_stream\": expected >= %zu bytes, got %zu bytes",
1421 sizeof(stream_info), payload->size);
1422 ret = -1;
1423 goto end_no_session;
1424 }
1425 memcpy(&stream_info, payload->data, sizeof(stream_info));
1426 stream_info.stream_id = be64toh(stream_info.stream_id);
1427 stream_info.last_net_seq_num = be64toh(stream_info.last_net_seq_num);
1428
1429 stream = stream_get_by_id(stream_info.stream_id);
1430 if (!stream) {
1431 ret = -1;
1432 goto end;
1433 }
1434
1435 /*
1436 * Set last_net_seq_num before the close flag. Required by data
1437 * pending check.
1438 */
1439 pthread_mutex_lock(&stream->lock);
1440 stream->last_net_seq_num = stream_info.last_net_seq_num;
1441 pthread_mutex_unlock(&stream->lock);
1442
1443 /*
1444 * This is one of the conditions which may trigger a stream close
1445 * with the others being:
1446 * 1) A close command is received for a stream
1447 * 2) The control connection owning the stream is closed
1448 * 3) We have received all of the stream's data _after_ a close
1449 * request.
1450 */
1451 try_stream_close(stream);
1452 if (stream->is_metadata) {
1453 struct relay_viewer_stream *vstream;
1454
1455 vstream = viewer_stream_get_by_id(stream->stream_handle);
1456 if (vstream) {
1457 if (vstream->metadata_sent == stream->metadata_received) {
1458 /*
1459 * Since all the metadata has been sent to the
1460 * viewer and that we have a request to close
1461 * its stream, we can safely teardown the
1462 * corresponding metadata viewer stream.
1463 */
1464 viewer_stream_put(vstream);
1465 }
1466 /* Put local reference. */
1467 viewer_stream_put(vstream);
1468 }
1469 }
1470 stream_put(stream);
1471 ret = 0;
1472
1473 end:
1474 memset(&reply, 0, sizeof(reply));
1475 if (ret < 0) {
1476 reply.ret_code = htobe32(LTTNG_ERR_UNK);
1477 } else {
1478 reply.ret_code = htobe32(LTTNG_OK);
1479 }
1480 send_ret = conn->sock->ops->sendmsg(conn->sock, &reply,
1481 sizeof(struct lttcomm_relayd_generic_reply), 0);
1482 if (send_ret < (ssize_t) sizeof(reply)) {
1483 ERR("Failed to send \"close stream\" command reply (ret = %zd)",
1484 send_ret);
1485 ret = -1;
1486 }
1487
1488 end_no_session:
1489 return ret;
1490 }
1491
1492 /*
1493 * relay_reset_metadata: reset a metadata stream
1494 */
1495 static
1496 int relay_reset_metadata(const struct lttcomm_relayd_hdr *recv_hdr,
1497 struct relay_connection *conn,
1498 const struct lttng_buffer_view *payload)
1499 {
1500 int ret;
1501 ssize_t send_ret;
1502 struct relay_session *session = conn->session;
1503 struct lttcomm_relayd_reset_metadata stream_info;
1504 struct lttcomm_relayd_generic_reply reply;
1505 struct relay_stream *stream;
1506
1507 DBG("Reset metadata received");
1508
1509 if (!session || !conn->version_check_done) {
1510 ERR("Trying to reset a metadata stream before version check");
1511 ret = -1;
1512 goto end_no_session;
1513 }
1514
1515 if (payload->size < sizeof(stream_info)) {
1516 ERR("Unexpected payload size in \"relay_reset_metadata\": expected >= %zu bytes, got %zu bytes",
1517 sizeof(stream_info), payload->size);
1518 ret = -1;
1519 goto end_no_session;
1520 }
1521 memcpy(&stream_info, payload->data, sizeof(stream_info));
1522 stream_info.stream_id = be64toh(stream_info.stream_id);
1523 stream_info.version = be64toh(stream_info.version);
1524
1525 DBG("Update metadata to version %" PRIu64, stream_info.version);
1526
1527 /* Unsupported for live sessions for now. */
1528 if (session->live_timer != 0) {
1529 ret = -1;
1530 goto end;
1531 }
1532
1533 stream = stream_get_by_id(stream_info.stream_id);
1534 if (!stream) {
1535 ret = -1;
1536 goto end;
1537 }
1538 pthread_mutex_lock(&stream->lock);
1539 if (!stream->is_metadata) {
1540 ret = -1;
1541 goto end_unlock;
1542 }
1543
1544 ret = stream_reset_file(stream);
1545 if (ret < 0) {
1546 ERR("Failed to reset metadata stream %" PRIu64
1547 ": stream_path = %s, channel = %s",
1548 stream->stream_handle, stream->path_name,
1549 stream->channel_name);
1550 goto end_unlock;
1551 }
1552 end_unlock:
1553 pthread_mutex_unlock(&stream->lock);
1554 stream_put(stream);
1555
1556 end:
1557 memset(&reply, 0, sizeof(reply));
1558 if (ret < 0) {
1559 reply.ret_code = htobe32(LTTNG_ERR_UNK);
1560 } else {
1561 reply.ret_code = htobe32(LTTNG_OK);
1562 }
1563 send_ret = conn->sock->ops->sendmsg(conn->sock, &reply,
1564 sizeof(struct lttcomm_relayd_generic_reply), 0);
1565 if (send_ret < (ssize_t) sizeof(reply)) {
1566 ERR("Failed to send \"reset metadata\" command reply (ret = %zd)",
1567 send_ret);
1568 ret = -1;
1569 }
1570
1571 end_no_session:
1572 return ret;
1573 }
1574
1575 /*
1576 * relay_unknown_command: send -1 if received unknown command
1577 */
1578 static void relay_unknown_command(struct relay_connection *conn)
1579 {
1580 struct lttcomm_relayd_generic_reply reply;
1581 ssize_t send_ret;
1582
1583 memset(&reply, 0, sizeof(reply));
1584 reply.ret_code = htobe32(LTTNG_ERR_UNK);
1585 send_ret = conn->sock->ops->sendmsg(conn->sock, &reply, sizeof(reply), 0);
1586 if (send_ret < sizeof(reply)) {
1587 ERR("Failed to send \"unknown command\" command reply (ret = %zd)", send_ret);
1588 }
1589 }
1590
1591 /*
1592 * relay_start: send an acknowledgment to the client to tell if we are
1593 * ready to receive data. We are ready if a session is established.
1594 */
1595 static int relay_start(const struct lttcomm_relayd_hdr *recv_hdr,
1596 struct relay_connection *conn,
1597 const struct lttng_buffer_view *payload)
1598 {
1599 int ret = 0;
1600 ssize_t send_ret;
1601 struct lttcomm_relayd_generic_reply reply;
1602 struct relay_session *session = conn->session;
1603
1604 if (!session) {
1605 DBG("Trying to start the streaming without a session established");
1606 ret = htobe32(LTTNG_ERR_UNK);
1607 }
1608
1609 memset(&reply, 0, sizeof(reply));
1610 reply.ret_code = htobe32(LTTNG_OK);
1611 send_ret = conn->sock->ops->sendmsg(conn->sock, &reply,
1612 sizeof(reply), 0);
1613 if (send_ret < (ssize_t) sizeof(reply)) {
1614 ERR("Failed to send \"relay_start\" command reply (ret = %zd)",
1615 send_ret);
1616 ret = -1;
1617 }
1618
1619 return ret;
1620 }
1621
1622 /*
1623 * relay_recv_metadata: receive the metadata for the session.
1624 */
1625 static int relay_recv_metadata(const struct lttcomm_relayd_hdr *recv_hdr,
1626 struct relay_connection *conn,
1627 const struct lttng_buffer_view *payload)
1628 {
1629 int ret = 0;
1630 struct relay_session *session = conn->session;
1631 struct lttcomm_relayd_metadata_payload metadata_payload_header;
1632 struct relay_stream *metadata_stream;
1633 uint64_t metadata_payload_size;
1634 struct lttng_buffer_view packet_view;
1635
1636 if (!session) {
1637 ERR("Metadata sent before version check");
1638 ret = -1;
1639 goto end;
1640 }
1641
1642 if (recv_hdr->data_size < sizeof(struct lttcomm_relayd_metadata_payload)) {
1643 ERR("Incorrect data size");
1644 ret = -1;
1645 goto end;
1646 }
1647 metadata_payload_size = recv_hdr->data_size -
1648 sizeof(struct lttcomm_relayd_metadata_payload);
1649
1650 memcpy(&metadata_payload_header, payload->data,
1651 sizeof(metadata_payload_header));
1652 metadata_payload_header.stream_id = be64toh(
1653 metadata_payload_header.stream_id);
1654 metadata_payload_header.padding_size = be32toh(
1655 metadata_payload_header.padding_size);
1656
1657 metadata_stream = stream_get_by_id(metadata_payload_header.stream_id);
1658 if (!metadata_stream) {
1659 ret = -1;
1660 goto end;
1661 }
1662
1663 packet_view = lttng_buffer_view_from_view(payload,
1664 sizeof(metadata_payload_header), metadata_payload_size);
1665 if (!packet_view.data) {
1666 ERR("Invalid metadata packet length announced by header");
1667 ret = -1;
1668 goto end_put;
1669 }
1670
1671 pthread_mutex_lock(&metadata_stream->lock);
1672 ret = stream_write(metadata_stream, &packet_view,
1673 metadata_payload_header.padding_size);
1674 pthread_mutex_unlock(&metadata_stream->lock);
1675 if (ret){
1676 ret = -1;
1677 goto end_put;
1678 }
1679 end_put:
1680 stream_put(metadata_stream);
1681 end:
1682 return ret;
1683 }
1684
1685 /*
1686 * relay_send_version: send relayd version number
1687 */
1688 static int relay_send_version(const struct lttcomm_relayd_hdr *recv_hdr,
1689 struct relay_connection *conn,
1690 const struct lttng_buffer_view *payload)
1691 {
1692 int ret;
1693 ssize_t send_ret;
1694 struct lttcomm_relayd_version reply, msg;
1695 bool compatible = true;
1696
1697 conn->version_check_done = true;
1698
1699 /* Get version from the other side. */
1700 if (payload->size < sizeof(msg)) {
1701 ERR("Unexpected payload size in \"relay_send_version\": expected >= %zu bytes, got %zu bytes",
1702 sizeof(msg), payload->size);
1703 ret = -1;
1704 goto end;
1705 }
1706
1707 memcpy(&msg, payload->data, sizeof(msg));
1708 msg.major = be32toh(msg.major);
1709 msg.minor = be32toh(msg.minor);
1710
1711 memset(&reply, 0, sizeof(reply));
1712 reply.major = RELAYD_VERSION_COMM_MAJOR;
1713 reply.minor = RELAYD_VERSION_COMM_MINOR;
1714
1715 /* Major versions must be the same */
1716 if (reply.major != msg.major) {
1717 DBG("Incompatible major versions (%u vs %u), deleting session",
1718 reply.major, msg.major);
1719 compatible = false;
1720 }
1721
1722 conn->major = reply.major;
1723 /* We adapt to the lowest compatible version */
1724 if (reply.minor <= msg.minor) {
1725 conn->minor = reply.minor;
1726 } else {
1727 conn->minor = msg.minor;
1728 }
1729
1730 reply.major = htobe32(reply.major);
1731 reply.minor = htobe32(reply.minor);
1732 send_ret = conn->sock->ops->sendmsg(conn->sock, &reply,
1733 sizeof(reply), 0);
1734 if (send_ret < (ssize_t) sizeof(reply)) {
1735 ERR("Failed to send \"send version\" command reply (ret = %zd)",
1736 send_ret);
1737 ret = -1;
1738 goto end;
1739 } else {
1740 ret = 0;
1741 }
1742
1743 if (!compatible) {
1744 ret = -1;
1745 goto end;
1746 }
1747
1748 DBG("Version check done using protocol %u.%u", conn->major,
1749 conn->minor);
1750
1751 end:
1752 return ret;
1753 }
1754
1755 /*
1756 * Check for data pending for a given stream id from the session daemon.
1757 */
1758 static int relay_data_pending(const struct lttcomm_relayd_hdr *recv_hdr,
1759 struct relay_connection *conn,
1760 const struct lttng_buffer_view *payload)
1761 {
1762 struct relay_session *session = conn->session;
1763 struct lttcomm_relayd_data_pending msg;
1764 struct lttcomm_relayd_generic_reply reply;
1765 struct relay_stream *stream;
1766 ssize_t send_ret;
1767 int ret;
1768 uint64_t stream_seq;
1769
1770 DBG("Data pending command received");
1771
1772 if (!session || !conn->version_check_done) {
1773 ERR("Trying to check for data before version check");
1774 ret = -1;
1775 goto end_no_session;
1776 }
1777
1778 if (payload->size < sizeof(msg)) {
1779 ERR("Unexpected payload size in \"relay_data_pending\": expected >= %zu bytes, got %zu bytes",
1780 sizeof(msg), payload->size);
1781 ret = -1;
1782 goto end_no_session;
1783 }
1784 memcpy(&msg, payload->data, sizeof(msg));
1785 msg.stream_id = be64toh(msg.stream_id);
1786 msg.last_net_seq_num = be64toh(msg.last_net_seq_num);
1787
1788 stream = stream_get_by_id(msg.stream_id);
1789 if (stream == NULL) {
1790 ret = -1;
1791 goto end;
1792 }
1793
1794 pthread_mutex_lock(&stream->lock);
1795
1796 if (session_streams_have_index(session)) {
1797 /*
1798 * Ensure that both the index and stream data have been
1799 * flushed up to the requested point.
1800 */
1801 stream_seq = min(stream->prev_data_seq, stream->prev_index_seq);
1802 } else {
1803 stream_seq = stream->prev_data_seq;
1804 }
1805 DBG("Data pending for stream id %" PRIu64 ": prev_data_seq %" PRIu64
1806 ", prev_index_seq %" PRIu64
1807 ", and last_seq %" PRIu64, msg.stream_id,
1808 stream->prev_data_seq, stream->prev_index_seq,
1809 msg.last_net_seq_num);
1810
1811 /* Avoid wrapping issue */
1812 if (((int64_t) (stream_seq - msg.last_net_seq_num)) >= 0) {
1813 /* Data has in fact been written and is NOT pending */
1814 ret = 0;
1815 } else {
1816 /* Data still being streamed thus pending */
1817 ret = 1;
1818 }
1819
1820 stream->data_pending_check_done = true;
1821 pthread_mutex_unlock(&stream->lock);
1822
1823 stream_put(stream);
1824 end:
1825
1826 memset(&reply, 0, sizeof(reply));
1827 reply.ret_code = htobe32(ret);
1828 send_ret = conn->sock->ops->sendmsg(conn->sock, &reply, sizeof(reply), 0);
1829 if (send_ret < (ssize_t) sizeof(reply)) {
1830 ERR("Failed to send \"data pending\" command reply (ret = %zd)",
1831 send_ret);
1832 ret = -1;
1833 }
1834
1835 end_no_session:
1836 return ret;
1837 }
1838
1839 /*
1840 * Wait for the control socket to reach a quiescent state.
1841 *
1842 * Note that for now, when receiving this command from the session
1843 * daemon, this means that every subsequent commands or data received on
1844 * the control socket has been handled. So, this is why we simply return
1845 * OK here.
1846 */
1847 static int relay_quiescent_control(const struct lttcomm_relayd_hdr *recv_hdr,
1848 struct relay_connection *conn,
1849 const struct lttng_buffer_view *payload)
1850 {
1851 int ret;
1852 ssize_t send_ret;
1853 struct relay_stream *stream;
1854 struct lttcomm_relayd_quiescent_control msg;
1855 struct lttcomm_relayd_generic_reply reply;
1856
1857 DBG("Checking quiescent state on control socket");
1858
1859 if (!conn->session || !conn->version_check_done) {
1860 ERR("Trying to check for data before version check");
1861 ret = -1;
1862 goto end_no_session;
1863 }
1864
1865 if (payload->size < sizeof(msg)) {
1866 ERR("Unexpected payload size in \"relay_quiescent_control\": expected >= %zu bytes, got %zu bytes",
1867 sizeof(msg), payload->size);
1868 ret = -1;
1869 goto end_no_session;
1870 }
1871 memcpy(&msg, payload->data, sizeof(msg));
1872 msg.stream_id = be64toh(msg.stream_id);
1873
1874 stream = stream_get_by_id(msg.stream_id);
1875 if (!stream) {
1876 goto reply;
1877 }
1878 pthread_mutex_lock(&stream->lock);
1879 stream->data_pending_check_done = true;
1880 pthread_mutex_unlock(&stream->lock);
1881
1882 DBG("Relay quiescent control pending flag set to %" PRIu64, msg.stream_id);
1883 stream_put(stream);
1884 reply:
1885 memset(&reply, 0, sizeof(reply));
1886 reply.ret_code = htobe32(LTTNG_OK);
1887 send_ret = conn->sock->ops->sendmsg(conn->sock, &reply, sizeof(reply), 0);
1888 if (send_ret < (ssize_t) sizeof(reply)) {
1889 ERR("Failed to send \"quiescent control\" command reply (ret = %zd)",
1890 send_ret);
1891 ret = -1;
1892 } else {
1893 ret = 0;
1894 }
1895
1896 end_no_session:
1897 return ret;
1898 }
1899
1900 /*
1901 * Initialize a data pending command. This means that a consumer is about
1902 * to ask for data pending for each stream it holds. Simply iterate over
1903 * all streams of a session and set the data_pending_check_done flag.
1904 *
1905 * This command returns to the client a LTTNG_OK code.
1906 */
1907 static int relay_begin_data_pending(const struct lttcomm_relayd_hdr *recv_hdr,
1908 struct relay_connection *conn,
1909 const struct lttng_buffer_view *payload)
1910 {
1911 int ret;
1912 ssize_t send_ret;
1913 struct lttng_ht_iter iter;
1914 struct lttcomm_relayd_begin_data_pending msg;
1915 struct lttcomm_relayd_generic_reply reply;
1916 struct relay_stream *stream;
1917
1918 assert(recv_hdr);
1919 assert(conn);
1920
1921 DBG("Init streams for data pending");
1922
1923 if (!conn->session || !conn->version_check_done) {
1924 ERR("Trying to check for data before version check");
1925 ret = -1;
1926 goto end_no_session;
1927 }
1928
1929 if (payload->size < sizeof(msg)) {
1930 ERR("Unexpected payload size in \"relay_begin_data_pending\": expected >= %zu bytes, got %zu bytes",
1931 sizeof(msg), payload->size);
1932 ret = -1;
1933 goto end_no_session;
1934 }
1935 memcpy(&msg, payload->data, sizeof(msg));
1936 msg.session_id = be64toh(msg.session_id);
1937
1938 /*
1939 * Iterate over all streams to set the begin data pending flag.
1940 * For now, the streams are indexed by stream handle so we have
1941 * to iterate over all streams to find the one associated with
1942 * the right session_id.
1943 */
1944 rcu_read_lock();
1945 cds_lfht_for_each_entry(relay_streams_ht->ht, &iter.iter, stream,
1946 node.node) {
1947 if (!stream_get(stream)) {
1948 continue;
1949 }
1950 if (stream->trace->session->id == msg.session_id) {
1951 pthread_mutex_lock(&stream->lock);
1952 stream->data_pending_check_done = false;
1953 pthread_mutex_unlock(&stream->lock);
1954 DBG("Set begin data pending flag to stream %" PRIu64,
1955 stream->stream_handle);
1956 }
1957 stream_put(stream);
1958 }
1959 rcu_read_unlock();
1960
1961 memset(&reply, 0, sizeof(reply));
1962 /* All good, send back reply. */
1963 reply.ret_code = htobe32(LTTNG_OK);
1964
1965 send_ret = conn->sock->ops->sendmsg(conn->sock, &reply, sizeof(reply), 0);
1966 if (send_ret < (ssize_t) sizeof(reply)) {
1967 ERR("Failed to send \"begin data pending\" command reply (ret = %zd)",
1968 send_ret);
1969 ret = -1;
1970 } else {
1971 ret = 0;
1972 }
1973
1974 end_no_session:
1975 return ret;
1976 }
1977
1978 /*
1979 * End data pending command. This will check, for a given session id, if
1980 * each stream associated with it has its data_pending_check_done flag
1981 * set. If not, this means that the client lost track of the stream but
1982 * the data is still being streamed on our side. In this case, we inform
1983 * the client that data is in flight.
1984 *
1985 * Return to the client if there is data in flight or not with a ret_code.
1986 */
1987 static int relay_end_data_pending(const struct lttcomm_relayd_hdr *recv_hdr,
1988 struct relay_connection *conn,
1989 const struct lttng_buffer_view *payload)
1990 {
1991 int ret;
1992 ssize_t send_ret;
1993 struct lttng_ht_iter iter;
1994 struct lttcomm_relayd_end_data_pending msg;
1995 struct lttcomm_relayd_generic_reply reply;
1996 struct relay_stream *stream;
1997 uint32_t is_data_inflight = 0;
1998
1999 DBG("End data pending command");
2000
2001 if (!conn->session || !conn->version_check_done) {
2002 ERR("Trying to check for data before version check");
2003 ret = -1;
2004 goto end_no_session;
2005 }
2006
2007 if (payload->size < sizeof(msg)) {
2008 ERR("Unexpected payload size in \"relay_end_data_pending\": expected >= %zu bytes, got %zu bytes",
2009 sizeof(msg), payload->size);
2010 ret = -1;
2011 goto end_no_session;
2012 }
2013 memcpy(&msg, payload->data, sizeof(msg));
2014 msg.session_id = be64toh(msg.session_id);
2015
2016 /*
2017 * Iterate over all streams to see if the begin data pending
2018 * flag is set.
2019 */
2020 rcu_read_lock();
2021 cds_lfht_for_each_entry(relay_streams_ht->ht, &iter.iter, stream,
2022 node.node) {
2023 if (!stream_get(stream)) {
2024 continue;
2025 }
2026 if (stream->trace->session->id != msg.session_id) {
2027 stream_put(stream);
2028 continue;
2029 }
2030 pthread_mutex_lock(&stream->lock);
2031 if (!stream->data_pending_check_done) {
2032 uint64_t stream_seq;
2033
2034 if (session_streams_have_index(conn->session)) {
2035 /*
2036 * Ensure that both the index and stream data have been
2037 * flushed up to the requested point.
2038 */
2039 stream_seq = min(stream->prev_data_seq, stream->prev_index_seq);
2040 } else {
2041 stream_seq = stream->prev_data_seq;
2042 }
2043 if (!stream->closed || !(((int64_t) (stream_seq - stream->last_net_seq_num)) >= 0)) {
2044 is_data_inflight = 1;
2045 DBG("Data is still in flight for stream %" PRIu64,
2046 stream->stream_handle);
2047 pthread_mutex_unlock(&stream->lock);
2048 stream_put(stream);
2049 break;
2050 }
2051 }
2052 pthread_mutex_unlock(&stream->lock);
2053 stream_put(stream);
2054 }
2055 rcu_read_unlock();
2056
2057 memset(&reply, 0, sizeof(reply));
2058 /* All good, send back reply. */
2059 reply.ret_code = htobe32(is_data_inflight);
2060
2061 send_ret = conn->sock->ops->sendmsg(conn->sock, &reply, sizeof(reply), 0);
2062 if (send_ret < (ssize_t) sizeof(reply)) {
2063 ERR("Failed to send \"end data pending\" command reply (ret = %zd)",
2064 send_ret);
2065 ret = -1;
2066 } else {
2067 ret = 0;
2068 }
2069
2070 end_no_session:
2071 return ret;
2072 }
2073
2074 /*
2075 * Receive an index for a specific stream.
2076 *
2077 * Return 0 on success else a negative value.
2078 */
2079 static int relay_recv_index(const struct lttcomm_relayd_hdr *recv_hdr,
2080 struct relay_connection *conn,
2081 const struct lttng_buffer_view *payload)
2082 {
2083 int ret;
2084 ssize_t send_ret;
2085 struct relay_session *session = conn->session;
2086 struct lttcomm_relayd_index index_info;
2087 struct lttcomm_relayd_generic_reply reply;
2088 struct relay_stream *stream;
2089 size_t msg_len;
2090
2091 assert(conn);
2092
2093 DBG("Relay receiving index");
2094
2095 if (!session || !conn->version_check_done) {
2096 ERR("Trying to close a stream before version check");
2097 ret = -1;
2098 goto end_no_session;
2099 }
2100
2101 msg_len = lttcomm_relayd_index_len(
2102 lttng_to_index_major(conn->major, conn->minor),
2103 lttng_to_index_minor(conn->major, conn->minor));
2104 if (payload->size < msg_len) {
2105 ERR("Unexpected payload size in \"relay_recv_index\": expected >= %zu bytes, got %zu bytes",
2106 msg_len, payload->size);
2107 ret = -1;
2108 goto end_no_session;
2109 }
2110 memcpy(&index_info, payload->data, msg_len);
2111 index_info.relay_stream_id = be64toh(index_info.relay_stream_id);
2112 index_info.net_seq_num = be64toh(index_info.net_seq_num);
2113 index_info.packet_size = be64toh(index_info.packet_size);
2114 index_info.content_size = be64toh(index_info.content_size);
2115 index_info.timestamp_begin = be64toh(index_info.timestamp_begin);
2116 index_info.timestamp_end = be64toh(index_info.timestamp_end);
2117 index_info.events_discarded = be64toh(index_info.events_discarded);
2118 index_info.stream_id = be64toh(index_info.stream_id);
2119
2120 if (conn->minor >= 8) {
2121 index_info.stream_instance_id =
2122 be64toh(index_info.stream_instance_id);
2123 index_info.packet_seq_num = be64toh(index_info.packet_seq_num);
2124 }
2125
2126 stream = stream_get_by_id(index_info.relay_stream_id);
2127 if (!stream) {
2128 ERR("stream_get_by_id not found");
2129 ret = -1;
2130 goto end;
2131 }
2132
2133 pthread_mutex_lock(&stream->lock);
2134 ret = stream_add_index(stream, &index_info);
2135 pthread_mutex_unlock(&stream->lock);
2136 if (ret) {
2137 goto end_stream_put;
2138 }
2139
2140 end_stream_put:
2141 stream_put(stream);
2142 end:
2143 memset(&reply, 0, sizeof(reply));
2144 if (ret < 0) {
2145 reply.ret_code = htobe32(LTTNG_ERR_UNK);
2146 } else {
2147 reply.ret_code = htobe32(LTTNG_OK);
2148 }
2149 send_ret = conn->sock->ops->sendmsg(conn->sock, &reply, sizeof(reply), 0);
2150 if (send_ret < (ssize_t) sizeof(reply)) {
2151 ERR("Failed to send \"recv index\" command reply (ret = %zd)", send_ret);
2152 ret = -1;
2153 }
2154
2155 end_no_session:
2156 return ret;
2157 }
2158
2159 /*
2160 * Receive the streams_sent message.
2161 *
2162 * Return 0 on success else a negative value.
2163 */
2164 static int relay_streams_sent(const struct lttcomm_relayd_hdr *recv_hdr,
2165 struct relay_connection *conn,
2166 const struct lttng_buffer_view *payload)
2167 {
2168 int ret;
2169 ssize_t send_ret;
2170 struct lttcomm_relayd_generic_reply reply;
2171
2172 assert(conn);
2173
2174 DBG("Relay receiving streams_sent");
2175
2176 if (!conn->session || !conn->version_check_done) {
2177 ERR("Trying to close a stream before version check");
2178 ret = -1;
2179 goto end_no_session;
2180 }
2181
2182 /*
2183 * Publish every pending stream in the connection recv list which are
2184 * now ready to be used by the viewer.
2185 */
2186 publish_connection_local_streams(conn);
2187
2188 memset(&reply, 0, sizeof(reply));
2189 reply.ret_code = htobe32(LTTNG_OK);
2190 send_ret = conn->sock->ops->sendmsg(conn->sock, &reply, sizeof(reply), 0);
2191 if (send_ret < (ssize_t) sizeof(reply)) {
2192 ERR("Failed to send \"streams sent\" command reply (ret = %zd)",
2193 send_ret);
2194 ret = -1;
2195 } else {
2196 /* Success. */
2197 ret = 0;
2198 }
2199
2200 end_no_session:
2201 return ret;
2202 }
2203
2204 /*
2205 * relay_rotate_session_stream: rotate a stream to a new tracefile for the
2206 * session rotation feature (not the tracefile rotation feature).
2207 */
2208 static int relay_rotate_session_streams(
2209 const struct lttcomm_relayd_hdr *recv_hdr,
2210 struct relay_connection *conn,
2211 const struct lttng_buffer_view *payload)
2212 {
2213 int ret = 0;
2214 uint32_t i;
2215 ssize_t send_ret;
2216 enum lttng_error_code reply_code = LTTNG_ERR_UNK;
2217 struct relay_session *session = conn->session;
2218 struct lttcomm_relayd_rotate_streams rotate_streams;
2219 struct lttcomm_relayd_generic_reply reply = {};
2220 struct relay_stream *stream = NULL;
2221 const size_t header_len = sizeof(struct lttcomm_relayd_rotate_streams);
2222 struct lttng_trace_chunk *next_trace_chunk = NULL;
2223 struct lttng_buffer_view stream_positions;
2224 char chunk_id_buf[MAX_INT_DEC_LEN(uint64_t)];
2225 const char *chunk_id_str = "none";
2226
2227 if (!session || !conn->version_check_done) {
2228 ERR("Trying to rotate a stream before version check");
2229 ret = -1;
2230 goto end_no_reply;
2231 }
2232
2233 if (session->major == 2 && session->minor < 11) {
2234 ERR("Unsupported feature before 2.11");
2235 ret = -1;
2236 goto end_no_reply;
2237 }
2238
2239 if (payload->size < header_len) {
2240 ERR("Unexpected payload size in \"relay_rotate_session_stream\": expected >= %zu bytes, got %zu bytes",
2241 header_len, payload->size);
2242 ret = -1;
2243 goto end_no_reply;
2244 }
2245
2246 memcpy(&rotate_streams, payload->data, header_len);
2247
2248 /* Convert header to host endianness. */
2249 rotate_streams = (typeof(rotate_streams)) {
2250 .stream_count = be32toh(rotate_streams.stream_count),
2251 .new_chunk_id = (typeof(rotate_streams.new_chunk_id)) {
2252 .is_set = !!rotate_streams.new_chunk_id.is_set,
2253 .value = be64toh(rotate_streams.new_chunk_id.value),
2254 }
2255 };
2256
2257 if (rotate_streams.new_chunk_id.is_set) {
2258 /*
2259 * Retrieve the trace chunk the stream must transition to. As
2260 * per the protocol, this chunk should have been created
2261 * before this command is received.
2262 */
2263 next_trace_chunk = sessiond_trace_chunk_registry_get_chunk(
2264 sessiond_trace_chunk_registry,
2265 session->sessiond_uuid, session->id,
2266 rotate_streams.new_chunk_id.value);
2267 if (!next_trace_chunk) {
2268 char uuid_str[UUID_STR_LEN];
2269
2270 lttng_uuid_to_str(session->sessiond_uuid, uuid_str);
2271 ERR("Unknown next trace chunk in ROTATE_STREAMS command: sessiond_uuid = {%s}, session_id = %" PRIu64
2272 ", trace_chunk_id = %" PRIu64,
2273 uuid_str, session->id,
2274 rotate_streams.new_chunk_id.value);
2275 reply_code = LTTNG_ERR_INVALID_PROTOCOL;
2276 ret = -1;
2277 goto end;
2278 }
2279
2280 ret = snprintf(chunk_id_buf, sizeof(chunk_id_buf), "%" PRIu64,
2281 rotate_streams.new_chunk_id.value);
2282 if (ret < 0 || ret >= sizeof(chunk_id_buf)) {
2283 chunk_id_str = "formatting error";
2284 } else {
2285 chunk_id_str = chunk_id_buf;
2286 }
2287 session->has_rotated = true;
2288 }
2289
2290 DBG("Rotate %" PRIu32 " streams of session \"%s\" to chunk \"%s\"",
2291 rotate_streams.stream_count, session->session_name,
2292 chunk_id_str);
2293
2294 stream_positions = lttng_buffer_view_from_view(payload,
2295 sizeof(rotate_streams), -1);
2296 if (!stream_positions.data ||
2297 stream_positions.size <
2298 (rotate_streams.stream_count *
2299 sizeof(struct lttcomm_relayd_stream_rotation_position))) {
2300 reply_code = LTTNG_ERR_INVALID_PROTOCOL;
2301 ret = -1;
2302 goto end;
2303 }
2304
2305 for (i = 0; i < rotate_streams.stream_count; i++) {
2306 struct lttcomm_relayd_stream_rotation_position *position_comm =
2307 &((typeof(position_comm)) stream_positions.data)[i];
2308 const struct lttcomm_relayd_stream_rotation_position pos = {
2309 .stream_id = be64toh(position_comm->stream_id),
2310 .rotate_at_seq_num = be64toh(
2311 position_comm->rotate_at_seq_num),
2312 };
2313
2314 stream = stream_get_by_id(pos.stream_id);
2315 if (!stream) {
2316 reply_code = LTTNG_ERR_INVALID;
2317 ret = -1;
2318 goto end;
2319 }
2320
2321 pthread_mutex_lock(&stream->lock);
2322 ret = stream_set_pending_rotation(stream, next_trace_chunk,
2323 pos.rotate_at_seq_num);
2324 pthread_mutex_unlock(&stream->lock);
2325 if (ret) {
2326 reply_code = LTTNG_ERR_FILE_CREATION_ERROR;
2327 goto end;
2328 }
2329
2330 stream_put(stream);
2331 stream = NULL;
2332 }
2333
2334 reply_code = LTTNG_OK;
2335 ret = 0;
2336 end:
2337 if (stream) {
2338 stream_put(stream);
2339 }
2340
2341 reply.ret_code = htobe32((uint32_t) reply_code);
2342 send_ret = conn->sock->ops->sendmsg(conn->sock, &reply,
2343 sizeof(struct lttcomm_relayd_generic_reply), 0);
2344 if (send_ret < (ssize_t) sizeof(reply)) {
2345 ERR("Failed to send \"rotate session stream\" command reply (ret = %zd)",
2346 send_ret);
2347 ret = -1;
2348 }
2349 end_no_reply:
2350 lttng_trace_chunk_put(next_trace_chunk);
2351 return ret;
2352 }
2353
2354
2355
2356 /*
2357 * relay_create_trace_chunk: create a new trace chunk
2358 */
2359 static int relay_create_trace_chunk(const struct lttcomm_relayd_hdr *recv_hdr,
2360 struct relay_connection *conn,
2361 const struct lttng_buffer_view *payload)
2362 {
2363 int ret = 0;
2364 ssize_t send_ret;
2365 struct relay_session *session = conn->session;
2366 struct lttcomm_relayd_create_trace_chunk *msg;
2367 struct lttcomm_relayd_generic_reply reply = {};
2368 struct lttng_buffer_view header_view;
2369 struct lttng_buffer_view chunk_name_view;
2370 struct lttng_trace_chunk *chunk = NULL, *published_chunk = NULL;
2371 enum lttng_error_code reply_code = LTTNG_OK;
2372 enum lttng_trace_chunk_status chunk_status;
2373 struct lttng_directory_handle session_output;
2374
2375 if (!session || !conn->version_check_done) {
2376 ERR("Trying to create a trace chunk before version check");
2377 ret = -1;
2378 goto end_no_reply;
2379 }
2380
2381 if (session->major == 2 && session->minor < 11) {
2382 ERR("Chunk creation command is unsupported before 2.11");
2383 ret = -1;
2384 goto end_no_reply;
2385 }
2386
2387 header_view = lttng_buffer_view_from_view(payload, 0, sizeof(*msg));
2388 if (!header_view.data) {
2389 ERR("Failed to receive payload of chunk creation command");
2390 ret = -1;
2391 goto end_no_reply;
2392 }
2393
2394 /* Convert to host endianness. */
2395 msg = (typeof(msg)) header_view.data;
2396 msg->chunk_id = be64toh(msg->chunk_id);
2397 msg->creation_timestamp = be64toh(msg->creation_timestamp);
2398 msg->override_name_length = be32toh(msg->override_name_length);
2399
2400 chunk = lttng_trace_chunk_create(
2401 msg->chunk_id, msg->creation_timestamp);
2402 if (!chunk) {
2403 ERR("Failed to create trace chunk in trace chunk creation command");
2404 ret = -1;
2405 reply_code = LTTNG_ERR_NOMEM;
2406 goto end;
2407 }
2408
2409 if (msg->override_name_length) {
2410 const char *name;
2411
2412 chunk_name_view = lttng_buffer_view_from_view(payload,
2413 sizeof(*msg),
2414 msg->override_name_length);
2415 name = chunk_name_view.data;
2416 if (!name || name[msg->override_name_length - 1]) {
2417 ERR("Failed to receive payload of chunk creation command");
2418 ret = -1;
2419 reply_code = LTTNG_ERR_INVALID;
2420 goto end;
2421 }
2422
2423 chunk_status = lttng_trace_chunk_override_name(
2424 chunk, chunk_name_view.data);
2425 switch (chunk_status) {
2426 case LTTNG_TRACE_CHUNK_STATUS_OK:
2427 break;
2428 case LTTNG_TRACE_CHUNK_STATUS_INVALID_ARGUMENT:
2429 ERR("Failed to set the name of new trace chunk in trace chunk creation command (invalid name)");
2430 reply_code = LTTNG_ERR_INVALID;
2431 ret = -1;
2432 goto end;
2433 default:
2434 ERR("Failed to set the name of new trace chunk in trace chunk creation command (unknown error)");
2435 reply_code = LTTNG_ERR_UNK;
2436 ret = -1;
2437 goto end;
2438 }
2439 }
2440
2441 chunk_status = lttng_trace_chunk_set_credentials_current_user(chunk);
2442 if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
2443 reply_code = LTTNG_ERR_UNK;
2444 ret = -1;
2445 goto end;
2446 }
2447
2448 ret = session_init_output_directory_handle(
2449 conn->session, &session_output);
2450 if (ret) {
2451 reply_code = LTTNG_ERR_CREATE_DIR_FAIL;
2452 goto end;
2453 }
2454 chunk_status = lttng_trace_chunk_set_as_owner(chunk, &session_output);
2455 lttng_directory_handle_fini(&session_output);
2456 if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
2457 reply_code = LTTNG_ERR_UNK;
2458 ret = -1;
2459 goto end;
2460 }
2461
2462 published_chunk = sessiond_trace_chunk_registry_publish_chunk(
2463 sessiond_trace_chunk_registry,
2464 conn->session->sessiond_uuid,
2465 conn->session->id,
2466 chunk);
2467 if (!published_chunk) {
2468 char uuid_str[UUID_STR_LEN];
2469
2470 lttng_uuid_to_str(conn->session->sessiond_uuid, uuid_str);
2471 ERR("Failed to publish chunk: sessiond_uuid = %s, session_id = %" PRIu64 ", chunk_id = %" PRIu64,
2472 uuid_str,
2473 conn->session->id,
2474 msg->chunk_id);
2475 ret = -1;
2476 reply_code = LTTNG_ERR_NOMEM;
2477 goto end;
2478 }
2479
2480 pthread_mutex_lock(&conn->session->lock);
2481 if (conn->session->pending_closure_trace_chunk) {
2482 /*
2483 * Invalid; this means a second create_trace_chunk command was
2484 * received before a close_trace_chunk.
2485 */
2486 ERR("Invalid trace chunk close command received; a trace chunk is already waiting for a trace chunk close command");
2487 reply_code = LTTNG_ERR_INVALID_PROTOCOL;
2488 ret = -1;
2489 goto end_unlock_session;
2490 }
2491 conn->session->pending_closure_trace_chunk =
2492 conn->session->current_trace_chunk;
2493 conn->session->current_trace_chunk = published_chunk;
2494 published_chunk = NULL;
2495 end_unlock_session:
2496 pthread_mutex_unlock(&conn->session->lock);
2497 end:
2498 reply.ret_code = htobe32((uint32_t) reply_code);
2499 send_ret = conn->sock->ops->sendmsg(conn->sock,
2500 &reply,
2501 sizeof(struct lttcomm_relayd_generic_reply),
2502 0);
2503 if (send_ret < (ssize_t) sizeof(reply)) {
2504 ERR("Failed to send \"create trace chunk\" command reply (ret = %zd)",
2505 send_ret);
2506 ret = -1;
2507 }
2508 end_no_reply:
2509 lttng_trace_chunk_put(chunk);
2510 lttng_trace_chunk_put(published_chunk);
2511 return ret;
2512 }
2513
2514 /*
2515 * relay_close_trace_chunk: close a trace chunk
2516 */
2517 static int relay_close_trace_chunk(const struct lttcomm_relayd_hdr *recv_hdr,
2518 struct relay_connection *conn,
2519 const struct lttng_buffer_view *payload)
2520 {
2521 int ret = 0, buf_ret;
2522 ssize_t send_ret;
2523 struct relay_session *session = conn->session;
2524 struct lttcomm_relayd_close_trace_chunk *msg;
2525 struct lttcomm_relayd_close_trace_chunk_reply reply = {};
2526 struct lttng_buffer_view header_view;
2527 struct lttng_trace_chunk *chunk = NULL;
2528 enum lttng_error_code reply_code = LTTNG_OK;
2529 enum lttng_trace_chunk_status chunk_status;
2530 uint64_t chunk_id;
2531 LTTNG_OPTIONAL(enum lttng_trace_chunk_command_type) close_command = {};
2532 time_t close_timestamp;
2533 char closed_trace_chunk_path[LTTNG_PATH_MAX];
2534 size_t path_length = 0;
2535 const char *chunk_name = NULL;
2536 struct lttng_dynamic_buffer reply_payload;
2537
2538 lttng_dynamic_buffer_init(&reply_payload);
2539
2540 if (!session || !conn->version_check_done) {
2541 ERR("Trying to close a trace chunk before version check");
2542 ret = -1;
2543 goto end_no_reply;
2544 }
2545
2546 if (session->major == 2 && session->minor < 11) {
2547 ERR("Chunk close command is unsupported before 2.11");
2548 ret = -1;
2549 goto end_no_reply;
2550 }
2551
2552 header_view = lttng_buffer_view_from_view(payload, 0, sizeof(*msg));
2553 if (!header_view.data) {
2554 ERR("Failed to receive payload of chunk close command");
2555 ret = -1;
2556 goto end_no_reply;
2557 }
2558
2559 /* Convert to host endianness. */
2560 msg = (typeof(msg)) header_view.data;
2561 chunk_id = be64toh(msg->chunk_id);
2562 close_timestamp = (time_t) be64toh(msg->close_timestamp);
2563 close_command = (typeof(close_command)){
2564 .value = be32toh(msg->close_command.value),
2565 .is_set = msg->close_command.is_set,
2566 };
2567
2568 chunk = sessiond_trace_chunk_registry_get_chunk(
2569 sessiond_trace_chunk_registry,
2570 conn->session->sessiond_uuid,
2571 conn->session->id,
2572 chunk_id);
2573 if (!chunk) {
2574 char uuid_str[UUID_STR_LEN];
2575
2576 lttng_uuid_to_str(conn->session->sessiond_uuid, uuid_str);
2577 ERR("Failed to find chunk to close: sessiond_uuid = %s, session_id = %" PRIu64 ", chunk_id = %" PRIu64,
2578 uuid_str,
2579 conn->session->id,
2580 msg->chunk_id);
2581 ret = -1;
2582 reply_code = LTTNG_ERR_NOMEM;
2583 goto end;
2584 }
2585
2586 pthread_mutex_lock(&session->lock);
2587 if (session->pending_closure_trace_chunk &&
2588 session->pending_closure_trace_chunk != chunk) {
2589 ERR("Trace chunk close command for session \"%s\" does not target the trace chunk pending closure",
2590 session->session_name);
2591 reply_code = LTTNG_ERR_INVALID_PROTOCOL;
2592 ret = -1;
2593 goto end_unlock_session;
2594 }
2595
2596 chunk_status = lttng_trace_chunk_set_close_timestamp(
2597 chunk, close_timestamp);
2598 if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
2599 ERR("Failed to set trace chunk close timestamp");
2600 ret = -1;
2601 reply_code = LTTNG_ERR_UNK;
2602 goto end_unlock_session;
2603 }
2604
2605 if (close_command.is_set) {
2606 chunk_status = lttng_trace_chunk_set_close_command(
2607 chunk, close_command.value);
2608 if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
2609 ret = -1;
2610 reply_code = LTTNG_ERR_INVALID;
2611 goto end_unlock_session;
2612 }
2613 }
2614 chunk_status = lttng_trace_chunk_get_name(chunk, &chunk_name, NULL);
2615 if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
2616 ERR("Failed to get chunk name");
2617 ret = -1;
2618 reply_code = LTTNG_ERR_UNK;
2619 goto end_unlock_session;
2620 }
2621 if (!session->has_rotated && !session->snapshot) {
2622 ret = lttng_strncpy(closed_trace_chunk_path,
2623 session->output_path,
2624 sizeof(closed_trace_chunk_path));
2625 if (ret) {
2626 ERR("Failed to send trace chunk path: path length of %zu bytes exceeds the maximal allowed length of %zu bytes",
2627 strlen(session->output_path),
2628 sizeof(closed_trace_chunk_path));
2629 reply_code = LTTNG_ERR_NOMEM;
2630 ret = -1;
2631 goto end_unlock_session;
2632 }
2633 } else {
2634 if (session->snapshot) {
2635 ret = snprintf(closed_trace_chunk_path,
2636 sizeof(closed_trace_chunk_path),
2637 "%s/%s", session->output_path,
2638 chunk_name);
2639 } else {
2640 ret = snprintf(closed_trace_chunk_path,
2641 sizeof(closed_trace_chunk_path),
2642 "%s/" DEFAULT_ARCHIVED_TRACE_CHUNKS_DIRECTORY
2643 "/%s",
2644 session->output_path, chunk_name);
2645 }
2646 if (ret < 0 || ret == sizeof(closed_trace_chunk_path)) {
2647 ERR("Failed to format closed trace chunk resulting path");
2648 reply_code = ret < 0 ? LTTNG_ERR_UNK : LTTNG_ERR_NOMEM;
2649 ret = -1;
2650 goto end_unlock_session;
2651 }
2652 }
2653 DBG("Reply chunk path on close: %s", closed_trace_chunk_path);
2654 path_length = strlen(closed_trace_chunk_path) + 1;
2655 if (path_length > UINT32_MAX) {
2656 ERR("Closed trace chunk path exceeds the maximal length allowed by the protocol");
2657 ret = -1;
2658 reply_code = LTTNG_ERR_INVALID_PROTOCOL;
2659 goto end_unlock_session;
2660 }
2661
2662 if (session->current_trace_chunk == chunk) {
2663 /*
2664 * After a trace chunk close command, no new streams
2665 * referencing the chunk may be created. Hence, on the
2666 * event that no new trace chunk have been created for
2667 * the session, the reference to the current trace chunk
2668 * is released in order to allow it to be reclaimed when
2669 * the last stream releases its reference to it.
2670 */
2671 lttng_trace_chunk_put(session->current_trace_chunk);
2672 session->current_trace_chunk = NULL;
2673 }
2674 lttng_trace_chunk_put(session->pending_closure_trace_chunk);
2675 session->pending_closure_trace_chunk = NULL;
2676 end_unlock_session:
2677 pthread_mutex_unlock(&session->lock);
2678
2679 end:
2680 reply.generic.ret_code = htobe32((uint32_t) reply_code);
2681 reply.path_length = htobe32((uint32_t) path_length);
2682 buf_ret = lttng_dynamic_buffer_append(
2683 &reply_payload, &reply, sizeof(reply));
2684 if (buf_ret) {
2685 ERR("Failed to append \"close trace chunk\" command reply header to payload buffer");
2686 goto end_no_reply;
2687 }
2688
2689 if (reply_code == LTTNG_OK) {
2690 buf_ret = lttng_dynamic_buffer_append(&reply_payload,
2691 closed_trace_chunk_path, path_length);
2692 if (buf_ret) {
2693 ERR("Failed to append \"close trace chunk\" command reply path to payload buffer");
2694 goto end_no_reply;
2695 }
2696 }
2697
2698 send_ret = conn->sock->ops->sendmsg(conn->sock,
2699 reply_payload.data,
2700 reply_payload.size,
2701 0);
2702 if (send_ret < reply_payload.size) {
2703 ERR("Failed to send \"close trace chunk\" command reply of %zu bytes (ret = %zd)",
2704 reply_payload.size, send_ret);
2705 ret = -1;
2706 goto end_no_reply;
2707 }
2708 end_no_reply:
2709 lttng_trace_chunk_put(chunk);
2710 lttng_dynamic_buffer_reset(&reply_payload);
2711 return ret;
2712 }
2713
2714 /*
2715 * relay_trace_chunk_exists: check if a trace chunk exists
2716 */
2717 static int relay_trace_chunk_exists(const struct lttcomm_relayd_hdr *recv_hdr,
2718 struct relay_connection *conn,
2719 const struct lttng_buffer_view *payload)
2720 {
2721 int ret = 0;
2722 ssize_t send_ret;
2723 struct relay_session *session = conn->session;
2724 struct lttcomm_relayd_trace_chunk_exists *msg;
2725 struct lttcomm_relayd_trace_chunk_exists_reply reply = {};
2726 struct lttng_buffer_view header_view;
2727 uint64_t chunk_id;
2728 bool chunk_exists;
2729
2730 if (!session || !conn->version_check_done) {
2731 ERR("Trying to close a trace chunk before version check");
2732 ret = -1;
2733 goto end_no_reply;
2734 }
2735
2736 if (session->major == 2 && session->minor < 11) {
2737 ERR("Chunk close command is unsupported before 2.11");
2738 ret = -1;
2739 goto end_no_reply;
2740 }
2741
2742 header_view = lttng_buffer_view_from_view(payload, 0, sizeof(*msg));
2743 if (!header_view.data) {
2744 ERR("Failed to receive payload of chunk close command");
2745 ret = -1;
2746 goto end_no_reply;
2747 }
2748
2749 /* Convert to host endianness. */
2750 msg = (typeof(msg)) header_view.data;
2751 chunk_id = be64toh(msg->chunk_id);
2752
2753 ret = sessiond_trace_chunk_registry_chunk_exists(
2754 sessiond_trace_chunk_registry,
2755 conn->session->sessiond_uuid,
2756 conn->session->id,
2757 chunk_id, &chunk_exists);
2758 /*
2759 * If ret is not 0, send the reply and report the error to the caller.
2760 * It is a protocol (or internal) error and the session/connection
2761 * should be torn down.
2762 */
2763 reply = (typeof(reply)){
2764 .generic.ret_code = htobe32((uint32_t)
2765 (ret == 0 ? LTTNG_OK : LTTNG_ERR_INVALID_PROTOCOL)),
2766 .trace_chunk_exists = ret == 0 ? chunk_exists : 0,
2767 };
2768 send_ret = conn->sock->ops->sendmsg(
2769 conn->sock, &reply, sizeof(reply), 0);
2770 if (send_ret < (ssize_t) sizeof(reply)) {
2771 ERR("Failed to send \"create trace chunk\" command reply (ret = %zd)",
2772 send_ret);
2773 ret = -1;
2774 }
2775 end_no_reply:
2776 return ret;
2777 }
2778
2779 #define DBG_CMD(cmd_name, conn) \
2780 DBG3("Processing \"%s\" command for socket %i", cmd_name, conn->sock->fd);
2781
2782 static int relay_process_control_command(struct relay_connection *conn,
2783 const struct lttcomm_relayd_hdr *header,
2784 const struct lttng_buffer_view *payload)
2785 {
2786 int ret = 0;
2787
2788 switch (header->cmd) {
2789 case RELAYD_CREATE_SESSION:
2790 DBG_CMD("RELAYD_CREATE_SESSION", conn);
2791 ret = relay_create_session(header, conn, payload);
2792 break;
2793 case RELAYD_ADD_STREAM:
2794 DBG_CMD("RELAYD_ADD_STREAM", conn);
2795 ret = relay_add_stream(header, conn, payload);
2796 break;
2797 case RELAYD_START_DATA:
2798 DBG_CMD("RELAYD_START_DATA", conn);
2799 ret = relay_start(header, conn, payload);
2800 break;
2801 case RELAYD_SEND_METADATA:
2802 DBG_CMD("RELAYD_SEND_METADATA", conn);
2803 ret = relay_recv_metadata(header, conn, payload);
2804 break;
2805 case RELAYD_VERSION:
2806 DBG_CMD("RELAYD_VERSION", conn);
2807 ret = relay_send_version(header, conn, payload);
2808 break;
2809 case RELAYD_CLOSE_STREAM:
2810 DBG_CMD("RELAYD_CLOSE_STREAM", conn);
2811 ret = relay_close_stream(header, conn, payload);
2812 break;
2813 case RELAYD_DATA_PENDING:
2814 DBG_CMD("RELAYD_DATA_PENDING", conn);
2815 ret = relay_data_pending(header, conn, payload);
2816 break;
2817 case RELAYD_QUIESCENT_CONTROL:
2818 DBG_CMD("RELAYD_QUIESCENT_CONTROL", conn);
2819 ret = relay_quiescent_control(header, conn, payload);
2820 break;
2821 case RELAYD_BEGIN_DATA_PENDING:
2822 DBG_CMD("RELAYD_BEGIN_DATA_PENDING", conn);
2823 ret = relay_begin_data_pending(header, conn, payload);
2824 break;
2825 case RELAYD_END_DATA_PENDING:
2826 DBG_CMD("RELAYD_END_DATA_PENDING", conn);
2827 ret = relay_end_data_pending(header, conn, payload);
2828 break;
2829 case RELAYD_SEND_INDEX:
2830 DBG_CMD("RELAYD_SEND_INDEX", conn);
2831 ret = relay_recv_index(header, conn, payload);
2832 break;
2833 case RELAYD_STREAMS_SENT:
2834 DBG_CMD("RELAYD_STREAMS_SENT", conn);
2835 ret = relay_streams_sent(header, conn, payload);
2836 break;
2837 case RELAYD_RESET_METADATA:
2838 DBG_CMD("RELAYD_RESET_METADATA", conn);
2839 ret = relay_reset_metadata(header, conn, payload);
2840 break;
2841 case RELAYD_ROTATE_STREAMS:
2842 DBG_CMD("RELAYD_ROTATE_STREAMS", conn);
2843 ret = relay_rotate_session_streams(header, conn, payload);
2844 break;
2845 case RELAYD_CREATE_TRACE_CHUNK:
2846 DBG_CMD("RELAYD_CREATE_TRACE_CHUNK", conn);
2847 ret = relay_create_trace_chunk(header, conn, payload);
2848 break;
2849 case RELAYD_CLOSE_TRACE_CHUNK:
2850 DBG_CMD("RELAYD_CLOSE_TRACE_CHUNK", conn);
2851 ret = relay_close_trace_chunk(header, conn, payload);
2852 break;
2853 case RELAYD_TRACE_CHUNK_EXISTS:
2854 DBG_CMD("RELAYD_TRACE_CHUNK_EXISTS", conn);
2855 ret = relay_trace_chunk_exists(header, conn, payload);
2856 break;
2857 case RELAYD_UPDATE_SYNC_INFO:
2858 default:
2859 ERR("Received unknown command (%u)", header->cmd);
2860 relay_unknown_command(conn);
2861 ret = -1;
2862 goto end;
2863 }
2864
2865 end:
2866 return ret;
2867 }
2868
2869 static enum relay_connection_status relay_process_control_receive_payload(
2870 struct relay_connection *conn)
2871 {
2872 int ret = 0;
2873 enum relay_connection_status status = RELAY_CONNECTION_STATUS_OK;
2874 struct lttng_dynamic_buffer *reception_buffer =
2875 &conn->protocol.ctrl.reception_buffer;
2876 struct ctrl_connection_state_receive_payload *state =
2877 &conn->protocol.ctrl.state.receive_payload;
2878 struct lttng_buffer_view payload_view;
2879
2880 if (state->left_to_receive == 0) {
2881 /* Short-circuit for payload-less commands. */
2882 goto reception_complete;
2883 }
2884
2885 ret = conn->sock->ops->recvmsg(conn->sock,
2886 reception_buffer->data + state->received,
2887 state->left_to_receive, MSG_DONTWAIT);
2888 if (ret < 0) {
2889 if (errno != EAGAIN && errno != EWOULDBLOCK) {
2890 PERROR("Unable to receive command payload on sock %d",
2891 conn->sock->fd);
2892 status = RELAY_CONNECTION_STATUS_ERROR;
2893 }
2894 goto end;
2895 } else if (ret == 0) {
2896 DBG("Socket %d performed an orderly shutdown (received EOF)", conn->sock->fd);
2897 status = RELAY_CONNECTION_STATUS_CLOSED;
2898 goto end;
2899 }
2900
2901 assert(ret > 0);
2902 assert(ret <= state->left_to_receive);
2903
2904 state->left_to_receive -= ret;
2905 state->received += ret;
2906
2907 if (state->left_to_receive > 0) {
2908 /*
2909 * Can't transition to the protocol's next state, wait to
2910 * receive the rest of the header.
2911 */
2912 DBG3("Partial reception of control connection protocol payload (received %" PRIu64 " bytes, %" PRIu64 " bytes left to receive, fd = %i)",
2913 state->received, state->left_to_receive,
2914 conn->sock->fd);
2915 goto end;
2916 }
2917
2918 reception_complete:
2919 DBG("Done receiving control command payload: fd = %i, payload size = %" PRIu64 " bytes",
2920 conn->sock->fd, state->received);
2921 /*
2922 * The payload required to process the command has been received.
2923 * A view to the reception buffer is forwarded to the various
2924 * commands and the state of the control is reset on success.
2925 *
2926 * Commands are responsible for sending their reply to the peer.
2927 */
2928 payload_view = lttng_buffer_view_from_dynamic_buffer(reception_buffer,
2929 0, -1);
2930 ret = relay_process_control_command(conn,
2931 &state->header, &payload_view);
2932 if (ret < 0) {
2933 status = RELAY_CONNECTION_STATUS_ERROR;
2934 goto end;
2935 }
2936
2937 ret = connection_reset_protocol_state(conn);
2938 if (ret) {
2939 status = RELAY_CONNECTION_STATUS_ERROR;
2940 }
2941 end:
2942 return status;
2943 }
2944
2945 static enum relay_connection_status relay_process_control_receive_header(
2946 struct relay_connection *conn)
2947 {
2948 int ret = 0;
2949 enum relay_connection_status status = RELAY_CONNECTION_STATUS_OK;
2950 struct lttcomm_relayd_hdr header;
2951 struct lttng_dynamic_buffer *reception_buffer =
2952 &conn->protocol.ctrl.reception_buffer;
2953 struct ctrl_connection_state_receive_header *state =
2954 &conn->protocol.ctrl.state.receive_header;
2955
2956 assert(state->left_to_receive != 0);
2957
2958 ret = conn->sock->ops->recvmsg(conn->sock,
2959 reception_buffer->data + state->received,
2960 state->left_to_receive, MSG_DONTWAIT);
2961 if (ret < 0) {
2962 if (errno != EAGAIN && errno != EWOULDBLOCK) {
2963 PERROR("Unable to receive control command header on sock %d",
2964 conn->sock->fd);
2965 status = RELAY_CONNECTION_STATUS_ERROR;
2966 }
2967 goto end;
2968 } else if (ret == 0) {
2969 DBG("Socket %d performed an orderly shutdown (received EOF)", conn->sock->fd);
2970 status = RELAY_CONNECTION_STATUS_CLOSED;
2971 goto end;
2972 }
2973
2974 assert(ret > 0);
2975 assert(ret <= state->left_to_receive);
2976
2977 state->left_to_receive -= ret;
2978 state->received += ret;
2979
2980 if (state->left_to_receive > 0) {
2981 /*
2982 * Can't transition to the protocol's next state, wait to
2983 * receive the rest of the header.
2984 */
2985 DBG3("Partial reception of control connection protocol header (received %" PRIu64 " bytes, %" PRIu64 " bytes left to receive, fd = %i)",
2986 state->received, state->left_to_receive,
2987 conn->sock->fd);
2988 goto end;
2989 }
2990
2991 /* Transition to next state: receiving the command's payload. */
2992 conn->protocol.ctrl.state_id =
2993 CTRL_CONNECTION_STATE_RECEIVE_PAYLOAD;
2994 memcpy(&header, reception_buffer->data, sizeof(header));
2995 header.circuit_id = be64toh(header.circuit_id);
2996 header.data_size = be64toh(header.data_size);
2997 header.cmd = be32toh(header.cmd);
2998 header.cmd_version = be32toh(header.cmd_version);
2999 memcpy(&conn->protocol.ctrl.state.receive_payload.header,
3000 &header, sizeof(header));
3001
3002 DBG("Done receiving control command header: fd = %i, cmd = %" PRIu32 ", cmd_version = %" PRIu32 ", payload size = %" PRIu64 " bytes",
3003 conn->sock->fd, header.cmd, header.cmd_version,
3004 header.data_size);
3005
3006 if (header.data_size > DEFAULT_NETWORK_RELAYD_CTRL_MAX_PAYLOAD_SIZE) {
3007 ERR("Command header indicates a payload (%" PRIu64 " bytes) that exceeds the maximal payload size allowed on a control connection.",
3008 header.data_size);
3009 status = RELAY_CONNECTION_STATUS_ERROR;
3010 goto end;
3011 }
3012
3013 conn->protocol.ctrl.state.receive_payload.left_to_receive =
3014 header.data_size;
3015 conn->protocol.ctrl.state.receive_payload.received = 0;
3016 ret = lttng_dynamic_buffer_set_size(reception_buffer,
3017 header.data_size);
3018 if (ret) {
3019 status = RELAY_CONNECTION_STATUS_ERROR;
3020 goto end;
3021 }
3022
3023 if (header.data_size == 0) {
3024 /*
3025 * Manually invoke the next state as the poll loop
3026 * will not wake-up to allow us to proceed further.
3027 */
3028 status = relay_process_control_receive_payload(conn);
3029 }
3030 end:
3031 return status;
3032 }
3033
3034 /*
3035 * Process the commands received on the control socket
3036 */
3037 static enum relay_connection_status relay_process_control(
3038 struct relay_connection *conn)
3039 {
3040 enum relay_connection_status status;
3041
3042 switch (conn->protocol.ctrl.state_id) {
3043 case CTRL_CONNECTION_STATE_RECEIVE_HEADER:
3044 status = relay_process_control_receive_header(conn);
3045 break;
3046 case CTRL_CONNECTION_STATE_RECEIVE_PAYLOAD:
3047 status = relay_process_control_receive_payload(conn);
3048 break;
3049 default:
3050 ERR("Unknown control connection protocol state encountered.");
3051 abort();
3052 }
3053
3054 return status;
3055 }
3056
3057 static enum relay_connection_status relay_process_data_receive_header(
3058 struct relay_connection *conn)
3059 {
3060 int ret;
3061 enum relay_connection_status status = RELAY_CONNECTION_STATUS_OK;
3062 struct data_connection_state_receive_header *state =
3063 &conn->protocol.data.state.receive_header;
3064 struct lttcomm_relayd_data_hdr header;
3065 struct relay_stream *stream;
3066
3067 assert(state->left_to_receive != 0);
3068
3069 ret = conn->sock->ops->recvmsg(conn->sock,
3070 state->header_reception_buffer + state->received,
3071 state->left_to_receive, MSG_DONTWAIT);
3072 if (ret < 0) {
3073 if (errno != EAGAIN && errno != EWOULDBLOCK) {
3074 PERROR("Unable to receive data header on sock %d", conn->sock->fd);
3075 status = RELAY_CONNECTION_STATUS_ERROR;
3076 }
3077 goto end;
3078 } else if (ret == 0) {
3079 /* Orderly shutdown. Not necessary to print an error. */
3080 DBG("Socket %d performed an orderly shutdown (received EOF)", conn->sock->fd);
3081 status = RELAY_CONNECTION_STATUS_CLOSED;
3082 goto end;
3083 }
3084
3085 assert(ret > 0);
3086 assert(ret <= state->left_to_receive);
3087
3088 state->left_to_receive -= ret;
3089 state->received += ret;
3090
3091 if (state->left_to_receive > 0) {
3092 /*
3093 * Can't transition to the protocol's next state, wait to
3094 * receive the rest of the header.
3095 */
3096 DBG3("Partial reception of data connection header (received %" PRIu64 " bytes, %" PRIu64 " bytes left to receive, fd = %i)",
3097 state->received, state->left_to_receive,
3098 conn->sock->fd);
3099 goto end;
3100 }
3101
3102 /* Transition to next state: receiving the payload. */
3103 conn->protocol.data.state_id = DATA_CONNECTION_STATE_RECEIVE_PAYLOAD;
3104
3105 memcpy(&header, state->header_reception_buffer, sizeof(header));
3106 header.circuit_id = be64toh(header.circuit_id);
3107 header.stream_id = be64toh(header.stream_id);
3108 header.data_size = be32toh(header.data_size);
3109 header.net_seq_num = be64toh(header.net_seq_num);
3110 header.padding_size = be32toh(header.padding_size);
3111 memcpy(&conn->protocol.data.state.receive_payload.header, &header, sizeof(header));
3112
3113 conn->protocol.data.state.receive_payload.left_to_receive =
3114 header.data_size;
3115 conn->protocol.data.state.receive_payload.received = 0;
3116 conn->protocol.data.state.receive_payload.rotate_index = false;
3117
3118 DBG("Received data connection header on fd %i: circuit_id = %" PRIu64 ", stream_id = %" PRIu64 ", data_size = %" PRIu32 ", net_seq_num = %" PRIu64 ", padding_size = %" PRIu32,
3119 conn->sock->fd, header.circuit_id,
3120 header.stream_id, header.data_size,
3121 header.net_seq_num, header.padding_size);
3122
3123 stream = stream_get_by_id(header.stream_id);
3124 if (!stream) {
3125 DBG("relay_process_data_receive_payload: Cannot find stream %" PRIu64,
3126 header.stream_id);
3127 /* Protocol error. */
3128 status = RELAY_CONNECTION_STATUS_ERROR;
3129 goto end;
3130 }
3131
3132 pthread_mutex_lock(&stream->lock);
3133 /* Prepare stream for the reception of a new packet. */
3134 ret = stream_init_packet(stream, header.data_size,
3135 &conn->protocol.data.state.receive_payload.rotate_index);
3136 pthread_mutex_unlock(&stream->lock);
3137 if (ret) {
3138 ERR("Failed to rotate stream output file");
3139 status = RELAY_CONNECTION_STATUS_ERROR;
3140 goto end_stream_unlock;
3141 }
3142
3143 end_stream_unlock:
3144 stream_put(stream);
3145 end:
3146 return status;
3147 }
3148
3149 static enum relay_connection_status relay_process_data_receive_payload(
3150 struct relay_connection *conn)
3151 {
3152 int ret;
3153 enum relay_connection_status status = RELAY_CONNECTION_STATUS_OK;
3154 struct relay_stream *stream;
3155 struct data_connection_state_receive_payload *state =
3156 &conn->protocol.data.state.receive_payload;
3157 const size_t chunk_size = RECV_DATA_BUFFER_SIZE;
3158 char data_buffer[chunk_size];
3159 bool partial_recv = false;
3160 bool new_stream = false, close_requested = false, index_flushed = false;