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