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