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