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