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