Fix: unprivilieged sessiond agent port clashes with root sessiond
[lttng-tools.git] / src / bin / lttng-relayd / main.c
CommitLineData
b8aa1682
JD
1/*
2 * Copyright (C) 2012 - Julien Desfossez <jdesfossez@efficios.com>
3 * David Goulet <dgoulet@efficios.com>
cd60b05a 4 * 2013 - Jérémie Galarneau <jeremie.galarneau@efficios.com>
7591bab1 5 * 2015 - Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
b8aa1682
JD
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
6c1c0768 21#define _LGPL_SOURCE
b8aa1682
JD
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>
173af62f 37#include <inttypes.h>
b8aa1682
JD
38#include <urcu/futex.h>
39#include <urcu/uatomic.h>
40#include <unistd.h>
41#include <fcntl.h>
b8aa1682
JD
42
43#include <lttng/lttng.h>
44#include <common/common.h>
45#include <common/compat/poll.h>
46#include <common/compat/socket.h>
f263b7fd 47#include <common/compat/endian.h>
e8fa9fb0 48#include <common/compat/getenv.h>
b8aa1682 49#include <common/defaults.h>
3fd27398 50#include <common/daemonize.h>
b8aa1682
JD
51#include <common/futex.h>
52#include <common/sessiond-comm/sessiond-comm.h>
53#include <common/sessiond-comm/inet.h>
b8aa1682
JD
54#include <common/sessiond-comm/relayd.h>
55#include <common/uri.h>
a02de639 56#include <common/utils.h>
d3ecc550 57#include <common/align.h>
f40ef1d5 58#include <common/config/session-config.h>
7591bab1 59#include <urcu/rculist.h>
b8aa1682 60
0f907de1 61#include "cmd.h"
d3e2ba59 62#include "ctf-trace.h"
1c20f0e2 63#include "index.h"
0f907de1 64#include "utils.h"
b8aa1682 65#include "lttng-relayd.h"
d3e2ba59 66#include "live.h"
55706a7d 67#include "health-relayd.h"
9b5e0863 68#include "testpoint.h"
2f8f53af 69#include "viewer-stream.h"
2a174661
DG
70#include "session.h"
71#include "stream.h"
58eb9381 72#include "connection.h"
a44ca2ca 73#include "tracefile-array.h"
f056029c 74#include "tcp_keep_alive.h"
b8aa1682 75
4fc83d94
PP
76static const char *help_msg =
77#ifdef LTTNG_EMBED_HELP
78#include <lttng-relayd.8.h>
79#else
80NULL
81#endif
82;
83
b8aa1682 84/* command line options */
0f907de1 85char *opt_output_path;
b5218ffb 86static int opt_daemon, opt_background;
3fd27398
MD
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
93static int lttng_relay_ready = NR_LTTNG_RELAY_READY;
0848dba7
MD
94
95/* Size of receive buffer. */
96#define RECV_DATA_BUFFER_SIZE 65536
d3ecc550 97#define FILE_COPY_BUFFER_SIZE 65536
0848dba7 98
3fd27398
MD
99static int recv_child_signal; /* Set to 1 when a SIGUSR1 signal is received. */
100static pid_t child_ppid; /* Internal parent PID use with daemonize. */
101
095a4ae5
MD
102static struct lttng_uri *control_uri;
103static struct lttng_uri *data_uri;
d3e2ba59 104static struct lttng_uri *live_uri;
b8aa1682
JD
105
106const char *progname;
b8aa1682 107
65931c8b 108const char *tracing_group_name = DEFAULT_TRACING_GROUP;
cd60b05a
JG
109static int tracing_group_name_override;
110
111const char * const config_section_name = "relayd";
65931c8b 112
b8aa1682
JD
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 */
0b242f62 117int thread_quit_pipe[2] = { -1, -1 };
b8aa1682
JD
118
119/*
120 * This pipe is used to inform the worker thread that a command is queued and
121 * ready to be processed.
122 */
58eb9381 123static int relay_conn_pipe[2] = { -1, -1 };
b8aa1682 124
26c9d55e 125/* Shared between threads */
b8aa1682
JD
126static int dispatch_thread_exit;
127
128static pthread_t listener_thread;
129static pthread_t dispatcher_thread;
130static pthread_t worker_thread;
65931c8b 131static pthread_t health_thread;
b8aa1682 132
7591bab1
MD
133/*
134 * last_relay_stream_id_lock protects last_relay_stream_id increment
135 * atomicity on 32-bit architectures.
136 */
137static pthread_mutex_t last_relay_stream_id_lock = PTHREAD_MUTEX_INITIALIZER;
095a4ae5 138static uint64_t last_relay_stream_id;
b8aa1682
JD
139
140/*
141 * Relay command queue.
142 *
143 * The relay_thread_listener and relay_thread_dispatcher communicate with this
144 * queue.
145 */
58eb9381 146static struct relay_conn_queue relay_conn_queue;
b8aa1682
JD
147
148/* buffer allocated at startup, used to store the trace data */
095a4ae5
MD
149static char *data_buffer;
150static unsigned int data_buffer_size;
b8aa1682 151
d3e2ba59
JD
152/* Global relay stream hash table. */
153struct lttng_ht *relay_streams_ht;
154
92c6ca54
DG
155/* Global relay viewer stream hash table. */
156struct lttng_ht *viewer_streams_ht;
157
7591bab1
MD
158/* Global relay sessions hash table. */
159struct lttng_ht *sessions_ht;
0a6518b0 160
55706a7d 161/* Relayd health monitoring */
eea7556c 162struct health_app *health_relayd;
55706a7d 163
cd60b05a
JG
164static struct option long_options[] = {
165 { "control-port", 1, 0, 'C', },
166 { "data-port", 1, 0, 'D', },
8d5c808e 167 { "live-port", 1, 0, 'L', },
cd60b05a 168 { "daemonize", 0, 0, 'd', },
b5218ffb 169 { "background", 0, 0, 'b', },
cd60b05a
JG
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' },
3a904098 175 { "version", 0, 0, 'V' },
cd60b05a
JG
176 { NULL, 0, 0, 0, },
177};
178
3a904098 179static const char *config_ignore_options[] = { "help", "config", "version" };
cd60b05a 180
cd60b05a
JG
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 */
7591bab1 187static int set_option(int opt, const char *arg, const char *optname)
b8aa1682 188{
cd60b05a
JG
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':
e8fa9fb0
MD
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 }
cd60b05a
JG
211 }
212 break;
213 case 'D':
e8fa9fb0
MD
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 }
cd60b05a
JG
226 }
227 break;
8d5c808e 228 case 'L':
e8fa9fb0
MD
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 }
8d5c808e
AM
241 }
242 break;
cd60b05a
JG
243 case 'd':
244 opt_daemon = 1;
245 break;
b5218ffb
MD
246 case 'b':
247 opt_background = 1;
248 break;
cd60b05a 249 case 'g':
e8fa9fb0
MD
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;
330a40bb 261 }
cd60b05a
JG
262 break;
263 case 'h':
4fc83d94 264 ret = utils_show_help(8, "lttng-relayd", help_msg);
655b5cc1 265 if (ret) {
4fc83d94 266 ERR("Cannot show --help for `lttng-relayd`");
655b5cc1
PP
267 perror("exec");
268 }
cd60b05a 269 exit(EXIT_FAILURE);
3a904098
AW
270 case 'V':
271 fprintf(stdout, "%s\n", VERSION);
272 exit(EXIT_SUCCESS);
cd60b05a 273 case 'o':
e8fa9fb0
MD
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 }
cd60b05a
JG
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 {
849e5b7b
DG
291 /* Only 3 level of verbosity (-vvv). */
292 if (lttng_opt_verbose < 3) {
293 lttng_opt_verbose += 1;
294 }
cd60b05a
JG
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
307end:
308 return ret;
309}
310
311/*
312 * config_entry_handler_cb used to handle options read from a config file.
f40ef1d5 313 * See config_entry_handler_cb comment in common/config/session-config.h for the
cd60b05a
JG
314 * return value conventions.
315 */
7591bab1 316static int config_entry_handler(const struct config_entry *entry, void *unused)
cd60b05a
JG
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 /*
7591bab1
MD
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.
cd60b05a
JG
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
362end:
363 return ret;
364}
365
7591bab1 366static int set_options(int argc, char **argv)
cd60b05a 367{
178a0557 368 int c, ret = 0, option_index = 0, retval = 0;
cd60b05a
JG
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) {
178a0557 376 retval = -ENOMEM;
cd60b05a
JG
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 == '?') {
178a0557 385 retval = -EINVAL;
cd60b05a
JG
386 goto exit;
387 } else if (c != 'f') {
388 continue;
389 }
390
e8fa9fb0
MD
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 }
cd60b05a
JG
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);
cd60b05a 407 }
178a0557 408 retval = -1;
cd60b05a
JG
409 goto exit;
410 }
b8aa1682 411
cd60b05a
JG
412 /* Reset getopt's global state */
413 optopt = orig_optopt;
414 optind = orig_optind;
b8aa1682 415 while (1) {
cd60b05a 416 c = getopt_long(argc, argv, optstring, long_options, &option_index);
b8aa1682
JD
417 if (c == -1) {
418 break;
419 }
420
cd60b05a
JG
421 ret = set_option(c, optarg, long_options[option_index].name);
422 if (ret < 0) {
178a0557 423 retval = -1;
b8aa1682
JD
424 goto exit;
425 }
426 }
427
428 /* assign default values */
429 if (control_uri == NULL) {
fa91dc52
MD
430 ret = asprintf(&default_address,
431 "tcp://" DEFAULT_NETWORK_CONTROL_BIND_ADDRESS ":%d",
432 DEFAULT_NETWORK_CONTROL_PORT);
b8aa1682
JD
433 if (ret < 0) {
434 PERROR("asprintf default data address");
178a0557 435 retval = -1;
b8aa1682
JD
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");
178a0557 443 retval = -1;
b8aa1682
JD
444 goto exit;
445 }
446 }
447 if (data_uri == NULL) {
fa91dc52
MD
448 ret = asprintf(&default_address,
449 "tcp://" DEFAULT_NETWORK_DATA_BIND_ADDRESS ":%d",
450 DEFAULT_NETWORK_DATA_PORT);
b8aa1682
JD
451 if (ret < 0) {
452 PERROR("asprintf default data address");
178a0557 453 retval = -1;
b8aa1682
JD
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");
178a0557 461 retval = -1;
b8aa1682
JD
462 goto exit;
463 }
464 }
d3e2ba59 465 if (live_uri == NULL) {
fa91dc52
MD
466 ret = asprintf(&default_address,
467 "tcp://" DEFAULT_NETWORK_VIEWER_BIND_ADDRESS ":%d",
468 DEFAULT_NETWORK_VIEWER_PORT);
d3e2ba59
JD
469 if (ret < 0) {
470 PERROR("asprintf default viewer control address");
178a0557 471 retval = -1;
d3e2ba59
JD
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");
178a0557 479 retval = -1;
d3e2ba59
JD
480 goto exit;
481 }
482 }
b8aa1682
JD
483
484exit:
cd60b05a 485 free(optstring);
178a0557 486 return retval;
b8aa1682
JD
487}
488
7591bab1
MD
489static 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
b8aa1682
JD
500/*
501 * Cleanup the daemon
502 */
7591bab1 503static void relayd_cleanup(void)
b8aa1682 504{
7591bab1
MD
505 print_global_objects();
506
b8aa1682
JD
507 DBG("Cleaning up");
508
178a0557
MD
509 if (viewer_streams_ht)
510 lttng_ht_destroy(viewer_streams_ht);
511 if (relay_streams_ht)
512 lttng_ht_destroy(relay_streams_ht);
7591bab1
MD
513 if (sessions_ht)
514 lttng_ht_destroy(sessions_ht);
178a0557 515
095a4ae5
MD
516 /* free the dynamically allocated opt_output_path */
517 free(opt_output_path);
518
a02de639
CB
519 /* Close thread quit pipes */
520 utils_close_pipe(thread_quit_pipe);
521
710c1f73
DG
522 uri_free(control_uri);
523 uri_free(data_uri);
8d5c808e 524 /* Live URI is freed in the live thread. */
cd60b05a
JG
525
526 if (tracing_group_name_override) {
527 free((void *) tracing_group_name);
528 }
b8aa1682
JD
529}
530
531/*
532 * Write to writable pipe used to notify a thread.
533 */
7591bab1 534static int notify_thread_pipe(int wpipe)
b8aa1682 535{
6cd525e8 536 ssize_t ret;
b8aa1682 537
6cd525e8
MD
538 ret = lttng_write(wpipe, "!", 1);
539 if (ret < 1) {
b8aa1682 540 PERROR("write poll pipe");
b4aacfdc 541 goto end;
b8aa1682 542 }
b4aacfdc
MD
543 ret = 0;
544end:
b8aa1682
JD
545 return ret;
546}
547
7591bab1 548static int notify_health_quit_pipe(int *pipe)
65931c8b 549{
6cd525e8 550 ssize_t ret;
65931c8b 551
6cd525e8
MD
552 ret = lttng_write(pipe[1], "4", 1);
553 if (ret < 1) {
65931c8b 554 PERROR("write relay health quit");
b4aacfdc 555 goto end;
65931c8b 556 }
b4aacfdc
MD
557 ret = 0;
558end:
559 return ret;
65931c8b
MD
560}
561
b8aa1682 562/*
b4aacfdc 563 * Stop all relayd and relayd-live threads.
b8aa1682 564 */
b4aacfdc 565int lttng_relay_stop_threads(void)
b8aa1682 566{
b4aacfdc 567 int retval = 0;
b8aa1682
JD
568
569 /* Stopping all threads */
570 DBG("Terminating all threads");
b4aacfdc 571 if (notify_thread_pipe(thread_quit_pipe[1])) {
b8aa1682 572 ERR("write error on thread quit pipe");
b4aacfdc 573 retval = -1;
b8aa1682
JD
574 }
575
b4aacfdc
MD
576 if (notify_health_quit_pipe(health_quit_pipe)) {
577 ERR("write error on health quit pipe");
578 }
65931c8b 579
b8aa1682 580 /* Dispatch thread */
26c9d55e 581 CMM_STORE_SHARED(dispatch_thread_exit, 1);
58eb9381 582 futex_nto1_wake(&relay_conn_queue.futex);
178a0557 583
b4aacfdc 584 if (relayd_live_stop()) {
178a0557 585 ERR("Error stopping live threads");
b4aacfdc 586 retval = -1;
178a0557 587 }
b4aacfdc 588 return retval;
b8aa1682
JD
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 */
7591bab1 597static void sighandler(int sig)
b8aa1682
JD
598{
599 switch (sig) {
b8aa1682
JD
600 case SIGINT:
601 DBG("SIGINT caught");
b4aacfdc
MD
602 if (lttng_relay_stop_threads()) {
603 ERR("Error stopping threads");
604 }
b8aa1682
JD
605 break;
606 case SIGTERM:
607 DBG("SIGTERM caught");
b4aacfdc
MD
608 if (lttng_relay_stop_threads()) {
609 ERR("Error stopping threads");
610 }
b8aa1682 611 break;
3fd27398
MD
612 case SIGUSR1:
613 CMM_STORE_SHARED(recv_child_signal, 1);
614 break;
b8aa1682
JD
615 default:
616 break;
617 }
618}
619
620/*
621 * Setup signal handler for :
622 * SIGINT, SIGTERM, SIGPIPE
623 */
7591bab1 624static int set_signal_handler(void)
b8aa1682
JD
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
b8aa1682
JD
635 sa.sa_mask = sigset;
636 sa.sa_flags = 0;
0072e5e2
MD
637
638 sa.sa_handler = sighandler;
b8aa1682
JD
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
0072e5e2 649 if ((ret = sigaction(SIGUSR1, &sa, NULL)) < 0) {
b8aa1682
JD
650 PERROR("sigaction");
651 return ret;
652 }
653
0072e5e2
MD
654 sa.sa_handler = SIG_IGN;
655 if ((ret = sigaction(SIGPIPE, &sa, NULL)) < 0) {
3fd27398
MD
656 PERROR("sigaction");
657 return ret;
658 }
659
660 DBG("Signal handler set for SIGTERM, SIGUSR1, SIGPIPE and SIGINT");
b8aa1682
JD
661
662 return ret;
663}
664
3fd27398
MD
665void 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
b8aa1682
JD
675/*
676 * Init thread quit pipe.
677 *
678 * Return -1 on error or 0 if all pipes are created.
679 */
7591bab1 680static int init_thread_quit_pipe(void)
b8aa1682 681{
a02de639 682 int ret;
b8aa1682 683
a02de639 684 ret = utils_create_pipe_cloexec(thread_quit_pipe);
b8aa1682 685
b8aa1682
JD
686 return ret;
687}
688
689/*
690 * Create a poll set with O_CLOEXEC and add the thread quit pipe to the set.
691 */
7591bab1 692static int create_thread_poll_set(struct lttng_poll_event *events, int size)
b8aa1682
JD
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 */
c7759e6a 707 ret = lttng_poll_add(events, thread_quit_pipe[0], LPOLLIN | LPOLLERR);
b8aa1682
JD
708 if (ret < 0) {
709 goto error;
710 }
711
712 return 0;
713
714error:
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 */
7591bab1 723static int check_thread_quit_pipe(int fd, uint32_t events)
b8aa1682
JD
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 */
7591bab1 735static struct lttcomm_sock *relay_socket_create(struct lttng_uri *uri)
b8aa1682
JD
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) {
2288467f 754 PERROR("Failed to bind socket");
b8aa1682
JD
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
766error:
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 */
7591bab1 776static void *relay_thread_listener(void *data)
b8aa1682 777{
095a4ae5 778 int i, ret, pollfd, err = -1;
b8aa1682
JD
779 uint32_t revents, nb_fd;
780 struct lttng_poll_event events;
781 struct lttcomm_sock *control_sock, *data_sock;
782
b8aa1682
JD
783 DBG("[thread] Relay listener started");
784
55706a7d
MD
785 health_register(health_relayd, HEALTH_RELAYD_TYPE_LISTENER);
786
f385ae0a
MD
787 health_code_update();
788
7591bab1 789 control_sock = relay_socket_create(control_uri);
b8aa1682 790 if (!control_sock) {
095a4ae5 791 goto error_sock_control;
b8aa1682
JD
792 }
793
7591bab1 794 data_sock = relay_socket_create(data_uri);
b8aa1682 795 if (!data_sock) {
095a4ae5 796 goto error_sock_relay;
b8aa1682
JD
797 }
798
799 /*
7591bab1
MD
800 * Pass 3 as size here for the thread quit pipe, control and
801 * data socket.
b8aa1682
JD
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
3fd27398
MD
820 lttng_relay_notify_ready();
821
9b5e0863
MD
822 if (testpoint(relayd_thread_listener)) {
823 goto error_testpoint;
824 }
825
b8aa1682 826 while (1) {
f385ae0a
MD
827 health_code_update();
828
b8aa1682
JD
829 DBG("Listener accepting connections");
830
b8aa1682 831restart:
f385ae0a 832 health_poll_entry();
b8aa1682 833 ret = lttng_poll_wait(&events, -1);
f385ae0a 834 health_poll_exit();
b8aa1682
JD
835 if (ret < 0) {
836 /*
837 * Restart interrupted system call.
838 */
839 if (errno == EINTR) {
840 goto restart;
841 }
842 goto error;
843 }
844
0d9c5d77
DG
845 nb_fd = ret;
846
b8aa1682
JD
847 DBG("Relay new connection received");
848 for (i = 0; i < nb_fd; i++) {
f385ae0a
MD
849 health_code_update();
850
b8aa1682
JD
851 /* Fetch once the poll data */
852 revents = LTTNG_POLL_GETEV(&events, i);
853 pollfd = LTTNG_POLL_GETFD(&events, i);
854
fd20dac9 855 if (!revents) {
7591bab1
MD
856 /*
857 * No activity for this FD (poll
858 * implementation).
859 */
fd20dac9
MD
860 continue;
861 }
862
b8aa1682
JD
863 /* Thread quit pipe has been closed. Killing thread. */
864 ret = check_thread_quit_pipe(pollfd, revents);
865 if (ret) {
095a4ae5
MD
866 err = 0;
867 goto exit;
b8aa1682
JD
868 }
869
03e43155 870 if (revents & LPOLLIN) {
4b7f17b2 871 /*
7591bab1
MD
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.
4b7f17b2 876 */
58eb9381
DG
877 int val = 1;
878 struct relay_connection *new_conn;
4b7f17b2 879 struct lttcomm_sock *newsock;
7591bab1 880 enum connection_type type;
b8aa1682
JD
881
882 if (pollfd == data_sock->fd) {
7591bab1 883 type = RELAY_DATA;
b8aa1682 884 newsock = data_sock->ops->accept(data_sock);
58eb9381
DG
885 DBG("Relay data connection accepted, socket %d",
886 newsock->fd);
4b7f17b2
MD
887 } else {
888 assert(pollfd == control_sock->fd);
7591bab1 889 type = RELAY_CONTROL;
b8aa1682 890 newsock = control_sock->ops->accept(control_sock);
58eb9381
DG
891 DBG("Relay control connection accepted, socket %d",
892 newsock->fd);
b8aa1682 893 }
58eb9381
DG
894 if (!newsock) {
895 PERROR("accepting sock");
58eb9381
DG
896 goto error;
897 }
898
899 ret = setsockopt(newsock->fd, SOL_SOCKET, SO_REUSEADDR, &val,
900 sizeof(val));
b8aa1682
JD
901 if (ret < 0) {
902 PERROR("setsockopt inet");
4b7f17b2 903 lttcomm_destroy_sock(newsock);
b8aa1682
JD
904 goto error;
905 }
f056029c
JR
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
7591bab1
MD
915 new_conn = connection_create(newsock, type);
916 if (!new_conn) {
917 lttcomm_destroy_sock(newsock);
918 goto error;
919 }
58eb9381
DG
920
921 /* Enqueue request for the dispatcher thread. */
8bdee6e2
SM
922 cds_wfcq_enqueue(&relay_conn_queue.head, &relay_conn_queue.tail,
923 &new_conn->qnode);
b8aa1682
JD
924
925 /*
7591bab1
MD
926 * Wake the dispatch queue futex.
927 * Implicit memory barrier with the
928 * exchange in cds_wfcq_enqueue.
b8aa1682 929 */
58eb9381 930 futex_nto1_wake(&relay_conn_queue.futex);
03e43155
MD
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;
b8aa1682
JD
937 }
938 }
939 }
940
095a4ae5 941exit:
b8aa1682
JD
942error:
943error_poll_add:
9b5e0863 944error_testpoint:
b8aa1682
JD
945 lttng_poll_clean(&events);
946error_create_poll:
095a4ae5
MD
947 if (data_sock->fd >= 0) {
948 ret = data_sock->ops->close(data_sock);
b8aa1682
JD
949 if (ret) {
950 PERROR("close");
951 }
b8aa1682 952 }
095a4ae5
MD
953 lttcomm_destroy_sock(data_sock);
954error_sock_relay:
955 if (control_sock->fd >= 0) {
956 ret = control_sock->ops->close(control_sock);
b8aa1682
JD
957 if (ret) {
958 PERROR("close");
959 }
b8aa1682 960 }
095a4ae5
MD
961 lttcomm_destroy_sock(control_sock);
962error_sock_control:
963 if (err) {
f385ae0a
MD
964 health_error();
965 ERR("Health error occurred in %s", __func__);
095a4ae5 966 }
55706a7d 967 health_unregister(health_relayd);
b8aa1682 968 DBG("Relay listener thread cleanup complete");
b4aacfdc 969 lttng_relay_stop_threads();
b8aa1682
JD
970 return NULL;
971}
972
973/*
974 * This thread manages the dispatching of the requests to worker threads
975 */
7591bab1 976static void *relay_thread_dispatcher(void *data)
b8aa1682 977{
6cd525e8
MD
978 int err = -1;
979 ssize_t ret;
8bdee6e2 980 struct cds_wfcq_node *node;
58eb9381 981 struct relay_connection *new_conn = NULL;
b8aa1682
JD
982
983 DBG("[thread] Relay dispatcher started");
984
55706a7d
MD
985 health_register(health_relayd, HEALTH_RELAYD_TYPE_DISPATCHER);
986
9b5e0863
MD
987 if (testpoint(relayd_thread_dispatcher)) {
988 goto error_testpoint;
989 }
990
f385ae0a
MD
991 health_code_update();
992
0ed3b1a8 993 for (;;) {
f385ae0a
MD
994 health_code_update();
995
b8aa1682 996 /* Atomically prepare the queue futex */
58eb9381 997 futex_nto1_prepare(&relay_conn_queue.futex);
b8aa1682 998
0ed3b1a8
MD
999 if (CMM_LOAD_SHARED(dispatch_thread_exit)) {
1000 break;
1001 }
1002
b8aa1682 1003 do {
f385ae0a
MD
1004 health_code_update();
1005
b8aa1682 1006 /* Dequeue commands */
8bdee6e2
SM
1007 node = cds_wfcq_dequeue_blocking(&relay_conn_queue.head,
1008 &relay_conn_queue.tail);
b8aa1682
JD
1009 if (node == NULL) {
1010 DBG("Woken up but nothing in the relay command queue");
1011 /* Continue thread execution */
1012 break;
1013 }
58eb9381 1014 new_conn = caa_container_of(node, struct relay_connection, qnode);
b8aa1682 1015
58eb9381 1016 DBG("Dispatching request waiting on sock %d", new_conn->sock->fd);
b8aa1682
JD
1017
1018 /*
7591bab1
MD
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 :)
b8aa1682 1023 */
58eb9381
DG
1024 ret = lttng_write(relay_conn_pipe[1], &new_conn, sizeof(new_conn));
1025 if (ret < 0) {
1026 PERROR("write connection pipe");
7591bab1 1027 connection_put(new_conn);
b8aa1682
JD
1028 goto error;
1029 }
1030 } while (node != NULL);
1031
1032 /* Futex wait on queue. Blocking call on futex() */
f385ae0a 1033 health_poll_entry();
58eb9381 1034 futex_nto1_wait(&relay_conn_queue.futex);
f385ae0a 1035 health_poll_exit();
b8aa1682
JD
1036 }
1037
f385ae0a
MD
1038 /* Normal exit, no error */
1039 err = 0;
1040
b8aa1682 1041error:
9b5e0863 1042error_testpoint:
f385ae0a
MD
1043 if (err) {
1044 health_error();
1045 ERR("Health error occurred in %s", __func__);
1046 }
55706a7d 1047 health_unregister(health_relayd);
b8aa1682 1048 DBG("Dispatch thread dying");
b4aacfdc 1049 lttng_relay_stop_threads();
b8aa1682
JD
1050 return NULL;
1051}
1052
b8aa1682 1053/*
7591bab1 1054 * Set index data from the control port to a given index object.
b8aa1682 1055 */
7591bab1 1056static int set_index_control_data(struct relay_index *index,
234cd636
JD
1057 struct lttcomm_relayd_index *data,
1058 struct relay_connection *conn)
1c20f0e2 1059{
7591bab1 1060 struct ctf_packet_index index_data;
1c20f0e2
JD
1061
1062 /*
7591bab1
MD
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.
1c20f0e2 1067 */
7591bab1
MD
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;
234cd636
JD
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
7591bab1 1080 return relay_index_set_data(index, &index_data);
1c20f0e2
JD
1081}
1082
c5b6f4f0
DG
1083/*
1084 * Handle the RELAYD_CREATE_SESSION command.
1085 *
1086 * On success, send back the session id or else return a negative value.
1087 */
7591bab1 1088static int relay_create_session(struct lttcomm_relayd_hdr *recv_hdr,
58eb9381 1089 struct relay_connection *conn)
c5b6f4f0
DG
1090{
1091 int ret = 0, send_ret;
1092 struct relay_session *session;
1093 struct lttcomm_relayd_status_session reply;
36d2e35d 1094 char session_name[LTTNG_NAME_MAX];
9cf3eb20 1095 char hostname[LTTNG_HOST_NAME_MAX];
7591bab1
MD
1096 uint32_t live_timer = 0;
1097 bool snapshot = false;
c5b6f4f0 1098
36d2e35d 1099 memset(session_name, 0, LTTNG_NAME_MAX);
9cf3eb20 1100 memset(hostname, 0, LTTNG_HOST_NAME_MAX);
c5b6f4f0
DG
1101
1102 memset(&reply, 0, sizeof(reply));
1103
58eb9381 1104 switch (conn->minor) {
2a174661
DG
1105 case 1:
1106 case 2:
1107 case 3:
1108 break;
1109 case 4: /* LTTng sessiond 2.4 */
1110 default:
7591bab1
MD
1111 ret = cmd_create_session_2_4(conn, session_name,
1112 hostname, &live_timer, &snapshot);
1113 }
1114 if (ret < 0) {
1115 goto send_reply;
d3e2ba59
JD
1116 }
1117
7591bab1
MD
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;
c5b6f4f0
DG
1126 DBG("Created session %" PRIu64, session->id);
1127
7591bab1
MD
1128 reply.session_id = htobe64(session->id);
1129
1130send_reply:
c5b6f4f0
DG
1131 if (ret < 0) {
1132 reply.ret_code = htobe32(LTTNG_ERR_FATAL);
1133 } else {
1134 reply.ret_code = htobe32(LTTNG_OK);
1135 }
1136
58eb9381 1137 send_ret = conn->sock->ops->sendmsg(conn->sock, &reply, sizeof(reply), 0);
c5b6f4f0
DG
1138 if (send_ret < 0) {
1139 ERR("Relayd sending session id");
4169f5ad 1140 ret = send_ret;
c5b6f4f0
DG
1141 }
1142
1143 return ret;
1144}
1145
a4baae1b
JD
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 */
7591bab1 1150static void publish_connection_local_streams(struct relay_connection *conn)
a4baae1b 1151{
7591bab1
MD
1152 struct relay_stream *stream;
1153 struct relay_session *session = conn->session;
a4baae1b 1154
7591bab1
MD
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);
a4baae1b 1164 }
7591bab1 1165 rcu_read_unlock();
a4baae1b 1166
7591bab1
MD
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);
a4baae1b
JD
1174}
1175
b8aa1682
JD
1176/*
1177 * relay_add_stream: allocate a new stream for a session
1178 */
7591bab1 1179static int relay_add_stream(struct lttcomm_relayd_hdr *recv_hdr,
58eb9381 1180 struct relay_connection *conn)
b8aa1682 1181{
7591bab1
MD
1182 int ret;
1183 ssize_t send_ret;
58eb9381 1184 struct relay_session *session = conn->session;
b8aa1682
JD
1185 struct relay_stream *stream = NULL;
1186 struct lttcomm_relayd_status_stream reply;
4030a636 1187 struct ctf_trace *trace = NULL;
7591bab1
MD
1188 uint64_t stream_handle = -1ULL;
1189 char *path_name = NULL, *channel_name = NULL;
1190 uint64_t tracefile_size = 0, tracefile_count = 0;
b8aa1682 1191
58eb9381 1192 if (!session || conn->version_check_done == 0) {
b8aa1682
JD
1193 ERR("Trying to add a stream before version check");
1194 ret = -1;
1195 goto end_no_session;
1196 }
1197
7591bab1
MD
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);
0f907de1 1202 break;
7591bab1 1203 case 2: /* LTTng sessiond 2.2. Allocates path_name and channel_name. */
0f907de1 1204 default:
7591bab1
MD
1205 ret = cmd_recv_stream_2_2(conn, &path_name,
1206 &channel_name, &tracefile_size, &tracefile_count);
0f907de1
JD
1207 break;
1208 }
1209 if (ret < 0) {
7591bab1 1210 goto send_reply;
b8aa1682
JD
1211 }
1212
7591bab1 1213 trace = ctf_trace_get_by_path_or_create(session, path_name);
2a174661 1214 if (!trace) {
7591bab1 1215 goto send_reply;
2a174661 1216 }
7591bab1 1217 /* This stream here has one reference on the trace. */
2a174661 1218
7591bab1
MD
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);
d3e2ba59 1222
7591bab1
MD
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;
a4baae1b 1228
2a174661 1229 /*
7591bab1
MD
1230 * Streams are the owners of their trace. Reference to trace is
1231 * kept within stream_create().
2a174661 1232 */
7591bab1 1233 ctf_trace_put(trace);
d3e2ba59 1234
7591bab1 1235send_reply:
53efb85a 1236 memset(&reply, 0, sizeof(reply));
7591bab1
MD
1237 reply.handle = htobe64(stream_handle);
1238 if (!stream) {
f73fabfd 1239 reply.ret_code = htobe32(LTTNG_ERR_UNK);
b8aa1682 1240 } else {
f73fabfd 1241 reply.ret_code = htobe32(LTTNG_OK);
b8aa1682 1242 }
5af40280 1243
58eb9381 1244 send_ret = conn->sock->ops->sendmsg(conn->sock, &reply,
b8aa1682
JD
1245 sizeof(struct lttcomm_relayd_status_stream), 0);
1246 if (send_ret < 0) {
1247 ERR("Relay sending stream id");
7591bab1 1248 ret = (int) send_ret;
b8aa1682
JD
1249 }
1250
1251end_no_session:
7591bab1
MD
1252 free(path_name);
1253 free(channel_name);
0f907de1 1254 return ret;
b8aa1682
JD
1255}
1256
173af62f
DG
1257/*
1258 * relay_close_stream: close a specific stream
1259 */
7591bab1 1260static int relay_close_stream(struct lttcomm_relayd_hdr *recv_hdr,
58eb9381 1261 struct relay_connection *conn)
173af62f 1262{
94d49140 1263 int ret, send_ret;
58eb9381 1264 struct relay_session *session = conn->session;
173af62f
DG
1265 struct lttcomm_relayd_close_stream stream_info;
1266 struct lttcomm_relayd_generic_reply reply;
1267 struct relay_stream *stream;
173af62f
DG
1268
1269 DBG("Close stream received");
1270
58eb9381 1271 if (!session || conn->version_check_done == 0) {
173af62f
DG
1272 ERR("Trying to close a stream before version check");
1273 ret = -1;
1274 goto end_no_session;
1275 }
1276
58eb9381 1277 ret = conn->sock->ops->recvmsg(conn->sock, &stream_info,
7c5aef62 1278 sizeof(struct lttcomm_relayd_close_stream), 0);
173af62f 1279 if (ret < sizeof(struct lttcomm_relayd_close_stream)) {
a6cd2b97
DG
1280 if (ret == 0) {
1281 /* Orderly shutdown. Not necessary to print an error. */
58eb9381 1282 DBG("Socket %d did an orderly shutdown", conn->sock->fd);
a6cd2b97
DG
1283 } else {
1284 ERR("Relay didn't receive valid add_stream struct size : %d", ret);
1285 }
173af62f
DG
1286 ret = -1;
1287 goto end_no_session;
1288 }
1289
7591bab1 1290 stream = stream_get_by_id(be64toh(stream_info.stream_id));
173af62f
DG
1291 if (!stream) {
1292 ret = -1;
7591bab1 1293 goto end;
173af62f 1294 }
77f7bd85
MD
1295
1296 /*
1297 * Set last_net_seq_num before the close flag. Required by data
1298 * pending check.
1299 */
7591bab1 1300 pthread_mutex_lock(&stream->lock);
8e2583a4 1301 stream->last_net_seq_num = be64toh(stream_info.last_net_seq_num);
77f7bd85
MD
1302 pthread_mutex_unlock(&stream->lock);
1303
bda7c7b9
JG
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);
7591bab1
MD
1313 if (stream->is_metadata) {
1314 struct relay_viewer_stream *vstream;
173af62f 1315
7591bab1
MD
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 }
7591bab1 1331 stream_put(stream);
173af62f 1332
7591bab1 1333end:
53efb85a 1334 memset(&reply, 0, sizeof(reply));
173af62f 1335 if (ret < 0) {
f73fabfd 1336 reply.ret_code = htobe32(LTTNG_ERR_UNK);
173af62f 1337 } else {
f73fabfd 1338 reply.ret_code = htobe32(LTTNG_OK);
173af62f 1339 }
58eb9381 1340 send_ret = conn->sock->ops->sendmsg(conn->sock, &reply,
173af62f
DG
1341 sizeof(struct lttcomm_relayd_generic_reply), 0);
1342 if (send_ret < 0) {
1343 ERR("Relay sending stream id");
4169f5ad 1344 ret = send_ret;
173af62f
DG
1345 }
1346
1347end_no_session:
1348 return ret;
1349}
1350
93ec662e
JD
1351/*
1352 * relay_reset_metadata: reset a metadata stream
1353 */
1354static
1355int 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
1413end_unlock:
1414 pthread_mutex_unlock(&stream->lock);
1415 stream_put(stream);
1416
1417end:
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
1431end_no_session:
1432 return ret;
1433}
1434
b8aa1682
JD
1435/*
1436 * relay_unknown_command: send -1 if received unknown command
1437 */
7591bab1 1438static void relay_unknown_command(struct relay_connection *conn)
b8aa1682
JD
1439{
1440 struct lttcomm_relayd_generic_reply reply;
1441 int ret;
1442
53efb85a 1443 memset(&reply, 0, sizeof(reply));
f73fabfd 1444 reply.ret_code = htobe32(LTTNG_ERR_UNK);
58eb9381 1445 ret = conn->sock->ops->sendmsg(conn->sock, &reply,
b8aa1682
JD
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 */
7591bab1 1456static int relay_start(struct lttcomm_relayd_hdr *recv_hdr,
58eb9381 1457 struct relay_connection *conn)
b8aa1682 1458{
f73fabfd 1459 int ret = htobe32(LTTNG_OK);
b8aa1682 1460 struct lttcomm_relayd_generic_reply reply;
58eb9381 1461 struct relay_session *session = conn->session;
b8aa1682
JD
1462
1463 if (!session) {
1464 DBG("Trying to start the streaming without a session established");
f73fabfd 1465 ret = htobe32(LTTNG_ERR_UNK);
b8aa1682
JD
1466 }
1467
53efb85a 1468 memset(&reply, 0, sizeof(reply));
b8aa1682 1469 reply.ret_code = ret;
58eb9381 1470 ret = conn->sock->ops->sendmsg(conn->sock, &reply,
b8aa1682
JD
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
1d4dfdef
DG
1479/*
1480 * Append padding to the file pointed by the file descriptor fd.
1481 */
1482static int write_padding_to_file(int fd, uint32_t size)
1483{
6cd525e8 1484 ssize_t ret = 0;
1d4dfdef
DG
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
6cd525e8
MD
1498 ret = lttng_write(fd, zeros, size);
1499 if (ret < size) {
1d4dfdef
DG
1500 PERROR("write padding to file");
1501 }
1502
e986c7a1
DG
1503 free(zeros);
1504
1d4dfdef
DG
1505end:
1506 return ret;
1507}
1508
d3ecc550
JD
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 */
1514static
1515int 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
1540end:
1541 return ret;
1542}
1543
1544static
1545int 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
1573end:
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 */
1588static
1589int rotate_truncate_stream(struct relay_stream *stream)
1590{
1591 int ret, new_fd;
a5df8828 1592 off_t lseek_ret;
d3ecc550
JD
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 */
a5df8828 1619 lseek_ret = lseek(stream->stream_fd->fd,
d3ecc550 1620 stream->pos_after_last_complete_data_index, SEEK_SET);
a5df8828 1621 if (lseek_ret < 0) {
d3ecc550 1622 PERROR("seek truncate stream");
a5df8828 1623 ret = -1;
d3ecc550
JD
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
1708end:
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 */
1718static
1719int 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
3765c8cc
JG
1728 if (stream->prev_seq < stream->rotate_at_seq_num ||
1729 stream->prev_seq == -1ULL) {
d3ecc550
JD
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
1749end:
1750 return ret;
1751}
1752
b8aa1682 1753/*
7591bab1 1754 * relay_recv_metadata: receive the metadata for the session.
b8aa1682 1755 */
7591bab1 1756static int relay_recv_metadata(struct lttcomm_relayd_hdr *recv_hdr,
58eb9381 1757 struct relay_connection *conn)
b8aa1682 1758{
32d1569c 1759 int ret = 0;
6cd525e8 1760 ssize_t size_ret;
58eb9381 1761 struct relay_session *session = conn->session;
b8aa1682
JD
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
f6416125
MD
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
b8aa1682 1780 if (data_buffer_size < data_size) {
d7b3776f 1781 /* In case the realloc fails, we can free the memory */
c617c0c6
MD
1782 char *tmp_data_ptr;
1783
1784 tmp_data_ptr = realloc(data_buffer, data_size);
1785 if (!tmp_data_ptr) {
b8aa1682 1786 ERR("Allocating data buffer");
c617c0c6 1787 free(data_buffer);
b8aa1682
JD
1788 ret = -1;
1789 goto end;
1790 }
c617c0c6 1791 data_buffer = tmp_data_ptr;
b8aa1682
JD
1792 data_buffer_size = data_size;
1793 }
1794 memset(data_buffer, 0, data_size);
77c7c900 1795 DBG2("Relay receiving metadata, waiting for %" PRIu64 " bytes", data_size);
32d1569c
JG
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) {
a6cd2b97 1799 /* Orderly shutdown. Not necessary to print an error. */
58eb9381 1800 DBG("Socket %d did an orderly shutdown", conn->sock->fd);
a6cd2b97
DG
1801 } else {
1802 ERR("Relay didn't receive the whole metadata");
1803 }
b8aa1682 1804 ret = -1;
b8aa1682
JD
1805 goto end;
1806 }
1807 metadata_struct = (struct lttcomm_relayd_metadata_payload *) data_buffer;
9d1bbf21 1808
7591bab1 1809 metadata_stream = stream_get_by_id(be64toh(metadata_struct->stream_id));
b8aa1682
JD
1810 if (!metadata_stream) {
1811 ret = -1;
7591bab1 1812 goto end;
b8aa1682
JD
1813 }
1814
7591bab1
MD
1815 pthread_mutex_lock(&metadata_stream->lock);
1816
1817 size_ret = lttng_write(metadata_stream->stream_fd->fd, metadata_struct->payload,
6cd525e8
MD
1818 payload_size);
1819 if (size_ret < payload_size) {
b8aa1682
JD
1820 ERR("Relay error writing metadata on file");
1821 ret = -1;
7591bab1 1822 goto end_put;
b8aa1682 1823 }
1d4dfdef 1824
32d1569c 1825 size_ret = write_padding_to_file(metadata_stream->stream_fd->fd,
1d4dfdef 1826 be32toh(metadata_struct->padding_size));
32d1569c 1827 if (size_ret < 0) {
7591bab1 1828 goto end_put;
1d4dfdef 1829 }
2a174661 1830
7591bab1 1831 metadata_stream->metadata_received +=
d3e2ba59 1832 payload_size + be32toh(metadata_struct->padding_size);
7591bab1
MD
1833 DBG2("Relay metadata written. Updated metadata_received %" PRIu64,
1834 metadata_stream->metadata_received);
1d4dfdef 1835
d3ecc550
JD
1836 ret = try_rotate_stream(metadata_stream);
1837 if (ret < 0) {
1838 goto end_put;
1839 }
1840
7591bab1
MD
1841end_put:
1842 pthread_mutex_unlock(&metadata_stream->lock);
1843 stream_put(metadata_stream);
b8aa1682
JD
1844end:
1845 return ret;
1846}
1847
1848/*
1849 * relay_send_version: send relayd version number
1850 */
7591bab1 1851static int relay_send_version(struct lttcomm_relayd_hdr *recv_hdr,
58eb9381 1852 struct relay_connection *conn)
b8aa1682 1853{
7f51dcba 1854 int ret;
092b6259 1855 struct lttcomm_relayd_version reply, msg;
87cb6359 1856 bool compatible = true;
b8aa1682 1857
58eb9381 1858 conn->version_check_done = 1;
b8aa1682 1859
092b6259 1860 /* Get version from the other side. */
58eb9381 1861 ret = conn->sock->ops->recvmsg(conn->sock, &msg, sizeof(msg), 0);
092b6259 1862 if (ret < 0 || ret != sizeof(msg)) {
a6cd2b97
DG
1863 if (ret == 0) {
1864 /* Orderly shutdown. Not necessary to print an error. */
58eb9381 1865 DBG("Socket %d did an orderly shutdown", conn->sock->fd);
a6cd2b97
DG
1866 } else {
1867 ERR("Relay failed to receive the version values.");
1868 }
092b6259 1869 ret = -1;
092b6259
DG
1870 goto end;
1871 }
1872
53efb85a 1873 memset(&reply, 0, sizeof(reply));
d83a952c
MD
1874 reply.major = RELAYD_VERSION_COMM_MAJOR;
1875 reply.minor = RELAYD_VERSION_COMM_MINOR;
d4519fa3
JD
1876
1877 /* Major versions must be the same */
1878 if (reply.major != be32toh(msg.major)) {
6151a90f
JD
1879 DBG("Incompatible major versions (%u vs %u), deleting session",
1880 reply.major, be32toh(msg.major));
87cb6359 1881 compatible = false;
d4519fa3
JD
1882 }
1883
58eb9381 1884 conn->major = reply.major;
0f907de1
JD
1885 /* We adapt to the lowest compatible version */
1886 if (reply.minor <= be32toh(msg.minor)) {
58eb9381 1887 conn->minor = reply.minor;
0f907de1 1888 } else {
58eb9381 1889 conn->minor = be32toh(msg.minor);
0f907de1
JD
1890 }
1891
6151a90f
JD
1892 reply.major = htobe32(reply.major);
1893 reply.minor = htobe32(reply.minor);
58eb9381 1894 ret = conn->sock->ops->sendmsg(conn->sock, &reply,
6151a90f
JD
1895 sizeof(struct lttcomm_relayd_version), 0);
1896 if (ret < 0) {
1897 ERR("Relay sending version");
1898 }
1899
87cb6359
JD
1900 if (!compatible) {
1901 ret = -1;
1902 goto end;
1903 }
1904
58eb9381
DG
1905 DBG("Version check done using protocol %u.%u", conn->major,
1906 conn->minor);
b8aa1682
JD
1907
1908end:
1909 return ret;
1910}
1911
c8f59ee5 1912/*
6d805429 1913 * Check for data pending for a given stream id from the session daemon.
c8f59ee5 1914 */
7591bab1 1915static int relay_data_pending(struct lttcomm_relayd_hdr *recv_hdr,
58eb9381 1916 struct relay_connection *conn)
c8f59ee5 1917{
58eb9381 1918 struct relay_session *session = conn->session;
6d805429 1919 struct lttcomm_relayd_data_pending msg;
c8f59ee5
DG
1920 struct lttcomm_relayd_generic_reply reply;
1921 struct relay_stream *stream;
1922 int ret;
c8f59ee5
DG
1923 uint64_t last_net_seq_num, stream_id;
1924
6d805429 1925 DBG("Data pending command received");
c8f59ee5 1926
58eb9381 1927 if (!session || conn->version_check_done == 0) {
c8f59ee5
DG
1928 ERR("Trying to check for data before version check");
1929 ret = -1;
1930 goto end_no_session;
1931 }
1932
58eb9381 1933 ret = conn->sock->ops->recvmsg(conn->sock, &msg, sizeof(msg), 0);
c8f59ee5 1934 if (ret < sizeof(msg)) {
a6cd2b97
DG
1935 if (ret == 0) {
1936 /* Orderly shutdown. Not necessary to print an error. */
58eb9381 1937 DBG("Socket %d did an orderly shutdown", conn->sock->fd);
a6cd2b97
DG
1938 } else {
1939 ERR("Relay didn't receive valid data_pending struct size : %d",
1940 ret);
1941 }
c8f59ee5
DG
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
7591bab1 1949 stream = stream_get_by_id(stream_id);
de91f48a 1950 if (stream == NULL) {
c8f59ee5 1951 ret = -1;
7591bab1 1952 goto end;
c8f59ee5
DG
1953 }
1954
7591bab1
MD
1955 pthread_mutex_lock(&stream->lock);
1956
6d805429 1957 DBG("Data pending for stream id %" PRIu64 " prev_seq %" PRIu64
c8f59ee5
DG
1958 " and last_seq %" PRIu64, stream_id, stream->prev_seq,
1959 last_net_seq_num);
1960
33832e64 1961 /* Avoid wrapping issue */
39df6d9f 1962 if (((int64_t) (stream->prev_seq - last_net_seq_num)) >= 0) {
6d805429 1963 /* Data has in fact been written and is NOT pending */
c8f59ee5 1964 ret = 0;
6d805429
DG
1965 } else {
1966 /* Data still being streamed thus pending */
1967 ret = 1;
c8f59ee5
DG
1968 }
1969
7591bab1
MD
1970 stream->data_pending_check_done = true;
1971 pthread_mutex_unlock(&stream->lock);
f7079f67 1972
7591bab1
MD
1973 stream_put(stream);
1974end:
c8f59ee5 1975
53efb85a 1976 memset(&reply, 0, sizeof(reply));
c8f59ee5 1977 reply.ret_code = htobe32(ret);
58eb9381 1978 ret = conn->sock->ops->sendmsg(conn->sock, &reply, sizeof(reply), 0);
c8f59ee5 1979 if (ret < 0) {
6d805429 1980 ERR("Relay data pending ret code failed");
c8f59ee5
DG
1981 }
1982
1983end_no_session:
1984 return ret;
1985}
1986
1987/*
1988 * Wait for the control socket to reach a quiescent state.
1989 *
7591bab1
MD
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.
c8f59ee5 1994 */
7591bab1 1995static int relay_quiescent_control(struct lttcomm_relayd_hdr *recv_hdr,
58eb9381 1996 struct relay_connection *conn)
c8f59ee5
DG
1997{
1998 int ret;
ad7051c0
DG
1999 uint64_t stream_id;
2000 struct relay_stream *stream;
ad7051c0 2001 struct lttcomm_relayd_quiescent_control msg;
c8f59ee5
DG
2002 struct lttcomm_relayd_generic_reply reply;
2003
2004 DBG("Checking quiescent state on control socket");
2005
58eb9381 2006 if (!conn->session || conn->version_check_done == 0) {
ad7051c0
DG
2007 ERR("Trying to check for data before version check");
2008 ret = -1;
2009 goto end_no_session;
2010 }
2011
58eb9381 2012 ret = conn->sock->ops->recvmsg(conn->sock, &msg, sizeof(msg), 0);
ad7051c0 2013 if (ret < sizeof(msg)) {
a6cd2b97
DG
2014 if (ret == 0) {
2015 /* Orderly shutdown. Not necessary to print an error. */
58eb9381 2016 DBG("Socket %d did an orderly shutdown", conn->sock->fd);
a6cd2b97
DG
2017 } else {
2018 ERR("Relay didn't receive valid begin data_pending struct size: %d",
2019 ret);
2020 }
ad7051c0
DG
2021 ret = -1;
2022 goto end_no_session;
2023 }
2024
2025 stream_id = be64toh(msg.stream_id);
7591bab1
MD
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);
2035reply:
53efb85a 2036 memset(&reply, 0, sizeof(reply));
c8f59ee5 2037 reply.ret_code = htobe32(LTTNG_OK);
58eb9381 2038 ret = conn->sock->ops->sendmsg(conn->sock, &reply, sizeof(reply), 0);
c8f59ee5 2039 if (ret < 0) {
6d805429 2040 ERR("Relay data quiescent control ret code failed");
c8f59ee5
DG
2041 }
2042
ad7051c0 2043end_no_session:
c8f59ee5
DG
2044 return ret;
2045}
2046
f7079f67 2047/*
7591bab1
MD
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.
f7079f67
DG
2051 *
2052 * This command returns to the client a LTTNG_OK code.
2053 */
7591bab1 2054static int relay_begin_data_pending(struct lttcomm_relayd_hdr *recv_hdr,
58eb9381 2055 struct relay_connection *conn)
f7079f67
DG
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);
58eb9381 2065 assert(conn);
f7079f67
DG
2066
2067 DBG("Init streams for data pending");
2068
58eb9381 2069 if (!conn->session || conn->version_check_done == 0) {
f7079f67
DG
2070 ERR("Trying to check for data before version check");
2071 ret = -1;
2072 goto end_no_session;
2073 }
2074
58eb9381 2075 ret = conn->sock->ops->recvmsg(conn->sock, &msg, sizeof(msg), 0);
f7079f67 2076 if (ret < sizeof(msg)) {
a6cd2b97
DG
2077 if (ret == 0) {
2078 /* Orderly shutdown. Not necessary to print an error. */
58eb9381 2079 DBG("Socket %d did an orderly shutdown", conn->sock->fd);
a6cd2b97
DG
2080 } else {
2081 ERR("Relay didn't receive valid begin data_pending struct size: %d",
2082 ret);
2083 }
f7079f67
DG
2084 ret = -1;
2085 goto end_no_session;
2086 }
2087
2088 session_id = be64toh(msg.session_id);
2089
2090 /*
7591bab1
MD
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.
f7079f67
DG
2095 */
2096 rcu_read_lock();
d3e2ba59 2097 cds_lfht_for_each_entry(relay_streams_ht->ht, &iter.iter, stream,
2a174661 2098 node.node) {
7591bab1
MD
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);
f7079f67
DG
2106 DBG("Set begin data pending flag to stream %" PRIu64,
2107 stream->stream_handle);
2108 }
7591bab1 2109 stream_put(stream);
f7079f67
DG
2110 }
2111 rcu_read_unlock();
2112
53efb85a 2113 memset(&reply, 0, sizeof(reply));
f7079f67
DG
2114 /* All good, send back reply. */
2115 reply.ret_code = htobe32(LTTNG_OK);
2116
58eb9381 2117 ret = conn->sock->ops->sendmsg(conn->sock, &reply, sizeof(reply), 0);
f7079f67
DG
2118 if (ret < 0) {
2119 ERR("Relay begin data pending send reply failed");
2120 }
2121
2122end_no_session:
2123 return ret;
2124}
2125
2126/*
7591bab1
MD
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.
f7079f67
DG
2132 *
2133 * Return to the client if there is data in flight or not with a ret_code.
2134 */
7591bab1 2135static int relay_end_data_pending(struct lttcomm_relayd_hdr *recv_hdr,
58eb9381 2136 struct relay_connection *conn)
f7079f67
DG
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
f7079f67
DG
2146 DBG("End data pending command");
2147
58eb9381 2148 if (!conn->session || conn->version_check_done == 0) {
f7079f67
DG
2149 ERR("Trying to check for data before version check");
2150 ret = -1;
2151 goto end_no_session;
2152 }
2153
58eb9381 2154 ret = conn->sock->ops->recvmsg(conn->sock, &msg, sizeof(msg), 0);
f7079f67 2155 if (ret < sizeof(msg)) {
a6cd2b97
DG
2156 if (ret == 0) {
2157 /* Orderly shutdown. Not necessary to print an error. */
58eb9381 2158 DBG("Socket %d did an orderly shutdown", conn->sock->fd);
a6cd2b97
DG
2159 } else {
2160 ERR("Relay didn't receive valid end data_pending struct size: %d",
2161 ret);
2162 }
f7079f67
DG
2163 ret = -1;
2164 goto end_no_session;
2165 }
2166
2167 session_id = be64toh(msg.session_id);
2168
7591bab1
MD
2169 /*
2170 * Iterate over all streams to see if the begin data pending
2171 * flag is set.
2172 */
f7079f67 2173 rcu_read_lock();
d3e2ba59 2174 cds_lfht_for_each_entry(relay_streams_ht->ht, &iter.iter, stream,
2a174661 2175 node.node) {
7591bab1
MD
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 }
f7079f67 2193 }
7591bab1
MD
2194 pthread_mutex_unlock(&stream->lock);
2195 stream_put(stream);
f7079f67
DG
2196 }
2197 rcu_read_unlock();
2198
53efb85a 2199 memset(&reply, 0, sizeof(reply));
f7079f67
DG
2200 /* All good, send back reply. */
2201 reply.ret_code = htobe32(is_data_inflight);
2202
58eb9381 2203 ret = conn->sock->ops->sendmsg(conn->sock, &reply, sizeof(reply), 0);
f7079f67
DG
2204 if (ret < 0) {
2205 ERR("Relay end data pending send reply failed");
2206 }
2207
2208end_no_session:
2209 return ret;
2210}
2211
1c20f0e2
JD
2212/*
2213 * Receive an index for a specific stream.
2214 *
2215 * Return 0 on success else a negative value.
2216 */
7591bab1 2217static int relay_recv_index(struct lttcomm_relayd_hdr *recv_hdr,
58eb9381 2218 struct relay_connection *conn)
1c20f0e2 2219{
7591bab1 2220 int ret, send_ret;
58eb9381 2221 struct relay_session *session = conn->session;
1c20f0e2 2222 struct lttcomm_relayd_index index_info;
7591bab1 2223 struct relay_index *index;
1c20f0e2
JD
2224 struct lttcomm_relayd_generic_reply reply;
2225 struct relay_stream *stream;
2226 uint64_t net_seq_num;
f8f3885c 2227 size_t msg_len;
1c20f0e2 2228
58eb9381 2229 assert(conn);
1c20f0e2
JD
2230
2231 DBG("Relay receiving index");
2232
58eb9381 2233 if (!session || conn->version_check_done == 0) {
1c20f0e2
JD
2234 ERR("Trying to close a stream before version check");
2235 ret = -1;
2236 goto end_no_session;
2237 }
2238
f8f3885c
MD
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));
58eb9381 2242 ret = conn->sock->ops->recvmsg(conn->sock, &index_info,
f8f3885c
MD
2243 msg_len, 0);
2244 if (ret < msg_len) {
1c20f0e2
JD
2245 if (ret == 0) {
2246 /* Orderly shutdown. Not necessary to print an error. */
58eb9381 2247 DBG("Socket %d did an orderly shutdown", conn->sock->fd);
1c20f0e2
JD
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
7591bab1 2257 stream = stream_get_by_id(be64toh(index_info.relay_stream_id));
1c20f0e2 2258 if (!stream) {
7591bab1 2259 ERR("stream_get_by_id not found");
1c20f0e2 2260 ret = -1;
7591bab1 2261 goto end;
1c20f0e2 2262 }
7591bab1 2263 pthread_mutex_lock(&stream->lock);
1c20f0e2 2264
d3e2ba59
JD
2265 /* Live beacon handling */
2266 if (index_info.packet_size == 0) {
7591bab1
MD
2267 DBG("Received live beacon for stream %" PRIu64,
2268 stream->stream_handle);
d3e2ba59
JD
2269
2270 /*
7591bab1
MD
2271 * Only flag a stream inactive when it has already
2272 * received data and no indexes are in flight.
d3e2ba59 2273 */
a44ca2ca 2274 if (stream->index_received_seqcount > 0
7591bab1
MD
2275 && stream->indexes_in_flight == 0) {
2276 stream->beacon_ts_end =
2277 be64toh(index_info.timestamp_end);
d3e2ba59
JD
2278 }
2279 ret = 0;
7591bab1 2280 goto end_stream_put;
d3e2ba59
JD
2281 } else {
2282 stream->beacon_ts_end = -1ULL;
2283 }
2284
528f2ffa
JD
2285 if (stream->ctf_stream_id == -1ULL) {
2286 stream->ctf_stream_id = be64toh(index_info.stream_id);
2287 }
7591bab1
MD
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;
1c20f0e2 2293 }
234cd636 2294 if (set_index_control_data(index, &index_info, conn)) {
7591bab1
MD
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) {
a44ca2ca
MD
2302 tracefile_array_commit_seq(stream->tfa);
2303 stream->index_received_seqcount++;
d3ecc550 2304 stream->pos_after_last_complete_data_index += index->total_size;
7591bab1
MD
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;
1c20f0e2
JD
2312 }
2313
7591bab1
MD
2314end_stream_put:
2315 pthread_mutex_unlock(&stream->lock);
2316 stream_put(stream);
2317
2318end:
1c20f0e2 2319
53efb85a 2320 memset(&reply, 0, sizeof(reply));
1c20f0e2
JD
2321 if (ret < 0) {
2322 reply.ret_code = htobe32(LTTNG_ERR_UNK);
2323 } else {
2324 reply.ret_code = htobe32(LTTNG_OK);
2325 }
58eb9381 2326 send_ret = conn->sock->ops->sendmsg(conn->sock, &reply, sizeof(reply), 0);
1c20f0e2
JD
2327 if (send_ret < 0) {
2328 ERR("Relay sending close index id reply");
2329 ret = send_ret;
2330 }
2331
2332end_no_session:
2333 return ret;
2334}
2335
a4baae1b
JD
2336/*
2337 * Receive the streams_sent message.
2338 *
2339 * Return 0 on success else a negative value.
2340 */
7591bab1 2341static int relay_streams_sent(struct lttcomm_relayd_hdr *recv_hdr,
58eb9381 2342 struct relay_connection *conn)
a4baae1b
JD
2343{
2344 int ret, send_ret;
2345 struct lttcomm_relayd_generic_reply reply;
2346
58eb9381 2347 assert(conn);
a4baae1b
JD
2348
2349 DBG("Relay receiving streams_sent");
2350
58eb9381 2351 if (!conn->session || conn->version_check_done == 0) {
a4baae1b
JD
2352 ERR("Trying to close a stream before version check");
2353 ret = -1;
2354 goto end_no_session;
2355 }
2356
2357 /*
7591bab1
MD
2358 * Publish every pending stream in the connection recv list which are
2359 * now ready to be used by the viewer.
4a9daf17 2360 */
7591bab1 2361 publish_connection_local_streams(conn);
4a9daf17 2362
53efb85a 2363 memset(&reply, 0, sizeof(reply));
a4baae1b 2364 reply.ret_code = htobe32(LTTNG_OK);
58eb9381 2365 send_ret = conn->sock->ops->sendmsg(conn->sock, &reply, sizeof(reply), 0);
a4baae1b
JD
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
2374end_no_session:
2375 return ret;
2376}
2377
d3ecc550
JD
2378/*
2379 * relay_rotate_stream: rotate a stream to a new tracefile for the session
2380 * rotation feature (not the tracefile rotation feature).
2381 */
2382static 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
2495end_stream_unlock:
2496 pthread_mutex_unlock(&stream->lock);
2497 stream_put(stream);
2498end:
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
2512end_no_reply:
2513 free(new_pathname);
2514 return ret;
2515}
2516
a1ae2ea5
JD
2517/*
2518 * relay_mkdir: Create a folder on the disk.
2519 */
2520static 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) {
00fb02ac 2532 ERR("Trying to create a directory before version check");
a1ae2ea5
JD
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
2607end:
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
2623end_no_session:
2624 free(path);
2625 free(path_info);
2626 return ret;
2627}
2628
00fb02ac
JD
2629static 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 */
2650static 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
2775end:
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
2789end_no_reply:
2790 free(received_paths);
2791 free(complete_old_path);
2792 free(complete_new_path);
2793 return ret;
2794}
2795
e2acc8d2
JD
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 */
2804static
2805int 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;
d88744a4 2810 struct lttcomm_relayd_rotate_pending_reply reply;
e2acc8d2
JD
2811 struct lttng_ht_iter iter;
2812 struct relay_stream *stream;
4f5fb4c3 2813 int ret = 0;
e2acc8d2
JD
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
2887send_reply:
2888 rcu_read_unlock();
2889 memset(&reply, 0, sizeof(reply));
d88744a4
JD
2890 reply.generic.ret_code = htobe32((uint32_t) LTTNG_OK);
2891 reply.is_pending = (uint8_t) !!rotate_pending;
e2acc8d2
JD
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
2899end_no_reply:
2900 return ret;
2901}
2902
b8aa1682 2903/*
d3e2ba59 2904 * Process the commands received on the control socket
b8aa1682 2905 */
7591bab1 2906static int relay_process_control(struct lttcomm_relayd_hdr *recv_hdr,
58eb9381 2907 struct relay_connection *conn)
b8aa1682
JD
2908{
2909 int ret = 0;
2910
2911 switch (be32toh(recv_hdr->cmd)) {
b8aa1682 2912 case RELAYD_CREATE_SESSION:
58eb9381 2913 ret = relay_create_session(recv_hdr, conn);
b8aa1682 2914 break;
b8aa1682 2915 case RELAYD_ADD_STREAM:
58eb9381 2916 ret = relay_add_stream(recv_hdr, conn);
b8aa1682
JD
2917 break;
2918 case RELAYD_START_DATA:
58eb9381 2919 ret = relay_start(recv_hdr, conn);
b8aa1682
JD
2920 break;
2921 case RELAYD_SEND_METADATA:
58eb9381 2922 ret = relay_recv_metadata(recv_hdr, conn);
b8aa1682
JD
2923 break;
2924 case RELAYD_VERSION:
58eb9381 2925 ret = relay_send_version(recv_hdr, conn);
b8aa1682 2926 break;
173af62f 2927 case RELAYD_CLOSE_STREAM:
58eb9381 2928 ret = relay_close_stream(recv_hdr, conn);
173af62f 2929 break;
6d805429 2930 case RELAYD_DATA_PENDING:
58eb9381 2931 ret = relay_data_pending(recv_hdr, conn);
c8f59ee5
DG
2932 break;
2933 case RELAYD_QUIESCENT_CONTROL:
58eb9381 2934 ret = relay_quiescent_control(recv_hdr, conn);
c8f59ee5 2935 break;
f7079f67 2936 case RELAYD_BEGIN_DATA_PENDING:
58eb9381 2937 ret = relay_begin_data_pending(recv_hdr, conn);
f7079f67
DG
2938 break;
2939 case RELAYD_END_DATA_PENDING:
58eb9381 2940 ret = relay_end_data_pending(recv_hdr, conn);
f7079f67 2941 break;
1c20f0e2 2942 case RELAYD_SEND_INDEX:
58eb9381 2943 ret = relay_recv_index(recv_hdr, conn);
1c20f0e2 2944 break;
a4baae1b 2945 case RELAYD_STREAMS_SENT:
58eb9381 2946 ret = relay_streams_sent(recv_hdr, conn);
a4baae1b 2947 break;
93ec662e
JD
2948 case RELAYD_RESET_METADATA:
2949 ret = relay_reset_metadata(recv_hdr, conn);
2950 break;
d3ecc550
JD
2951 case RELAYD_ROTATE_STREAM:
2952 ret = relay_rotate_session_stream(recv_hdr, conn);
2953 break;
00fb02ac
JD
2954 case RELAYD_ROTATE_RENAME:
2955 ret = relay_rotate_rename(recv_hdr, conn);
2956 break;
e2acc8d2
JD
2957 case RELAYD_ROTATE_PENDING:
2958 ret = relay_rotate_pending(recv_hdr, conn);
2959 break;
a1ae2ea5
JD
2960 case RELAYD_MKDIR:
2961 ret = relay_mkdir(recv_hdr, conn);
2962 break;
b8aa1682
JD
2963 case RELAYD_UPDATE_SYNC_INFO:
2964 default:
2965 ERR("Received unknown command (%u)", be32toh(recv_hdr->cmd));
58eb9381 2966 relay_unknown_command(conn);
b8aa1682
JD
2967 ret = -1;
2968 goto end;
2969 }
2970
2971end:
2972 return ret;
2973}
2974
7d2f7452
DG
2975/*
2976 * Handle index for a data stream.
2977 *
7591bab1 2978 * Called with the stream lock held.
7d2f7452
DG
2979 *
2980 * Return 0 on success else a negative value.
2981 */
2982static int handle_index_data(struct relay_stream *stream, uint64_t net_seq_num,
d3ecc550 2983 int rotate_index, bool *flushed, uint64_t total_size)
7d2f7452 2984{
7591bab1
MD
2985 int ret = 0;
2986 uint64_t data_offset;
2987 struct relay_index *index;
7d2f7452 2988
7d2f7452
DG
2989 /* Get data offset because we are about to update the index. */
2990 data_offset = htobe64(stream->tracefile_size_current);
2991
65d66b19
MD
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);
7591bab1 2994
7d2f7452 2995 /*
7591bab1
MD
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.
7d2f7452 2999 */
7591bab1 3000 index = relay_index_get_by_id_or_create(stream, net_seq_num);
7d2f7452 3001 if (!index) {
7591bab1
MD
3002 ret = -1;
3003 goto end;
7d2f7452
DG
3004 }
3005
f8f3885c 3006 if (rotate_index || !stream->index_file) {
d3ecc550
JD
3007 ret = create_rotate_index_file(stream);
3008 if (ret < 0) {
3009 ERR("Failed to rotate index");
7591bab1
MD
3010 /* Put self-ref for this index due to error. */
3011 relay_index_put(index);
f8f3885c 3012 index = NULL;
7591bab1 3013 goto end;
7d2f7452 3014 }
7d2f7452
DG
3015 }
3016
f8f3885c 3017 if (relay_index_set_file(index, stream->index_file, data_offset)) {
7591bab1
MD
3018 ret = -1;
3019 /* Put self-ref for this index due to error. */
3020 relay_index_put(index);
f8f3885c 3021 index = NULL;
7591bab1 3022 goto end;
7d2f7452
DG
3023 }
3024
7591bab1
MD
3025 ret = relay_index_try_flush(index);
3026 if (ret == 0) {
a44ca2ca
MD
3027 tracefile_array_commit_seq(stream->tfa);
3028 stream->index_received_seqcount++;
d3ecc550 3029 *flushed = true;
7591bab1 3030 } else if (ret > 0) {
d3ecc550 3031 index->total_size = total_size;
7591bab1
MD
3032 /* No flush. */
3033 ret = 0;
3034 } else {
3035 /* Put self-ref for this index due to error. */
3036 relay_index_put(index);
f8f3885c 3037 index = NULL;
7591bab1
MD
3038 ret = -1;
3039 }
3040end:
7d2f7452
DG
3041 return ret;
3042}
3043
b8aa1682
JD
3044/*
3045 * relay_process_data: Process the data received on the data socket
3046 */
7591bab1 3047static int relay_process_data(struct relay_connection *conn)
b8aa1682 3048{
7d2f7452 3049 int ret = 0, rotate_index = 0;
6cd525e8 3050 ssize_t size_ret;
b8aa1682
JD
3051 struct relay_stream *stream;
3052 struct lttcomm_relayd_data_hdr data_hdr;
7d2f7452 3053 uint64_t stream_id;
173af62f 3054 uint64_t net_seq_num;
b8aa1682 3055 uint32_t data_size;
2a174661 3056 struct relay_session *session;
bda7c7b9 3057 bool new_stream = false, close_requested = false;
0848dba7
MD
3058 size_t chunk_size = RECV_DATA_BUFFER_SIZE;
3059 size_t recv_off = 0;
3060 char data_buffer[chunk_size];
d3ecc550 3061 bool index_flushed = false;
b8aa1682 3062
58eb9381 3063 ret = conn->sock->ops->recvmsg(conn->sock, &data_hdr,
7c5aef62 3064 sizeof(struct lttcomm_relayd_data_hdr), 0);
b8aa1682 3065 if (ret <= 0) {
a6cd2b97
DG
3066 if (ret == 0) {
3067 /* Orderly shutdown. Not necessary to print an error. */
58eb9381 3068 DBG("Socket %d did an orderly shutdown", conn->sock->fd);
a6cd2b97 3069 } else {
58eb9381 3070 ERR("Unable to receive data header on sock %d", conn->sock->fd);
a6cd2b97 3071 }
b8aa1682
JD
3072 ret = -1;
3073 goto end;
3074 }
3075
3076 stream_id = be64toh(data_hdr.stream_id);
7591bab1 3077 stream = stream_get_by_id(stream_id);
b8aa1682 3078 if (!stream) {
65d66b19 3079 ERR("relay_process_data: Cannot find stream %" PRIu64, stream_id);
b8aa1682 3080 ret = -1;
7591bab1 3081 goto end;
b8aa1682 3082 }
7591bab1 3083 session = stream->trace->session;
b8aa1682 3084 data_size = be32toh(data_hdr.data_size);
b8aa1682 3085
173af62f
DG
3086 net_seq_num = be64toh(data_hdr.net_seq_num);
3087
77c7c900 3088 DBG3("Receiving data of size %u for stream id %" PRIu64 " seqnum %" PRIu64,
173af62f 3089 data_size, stream_id, net_seq_num);
b8aa1682 3090
7591bab1
MD
3091 pthread_mutex_lock(&stream->lock);
3092
1c20f0e2 3093 /* Check if a rotation is needed. */
0f907de1
JD
3094 if (stream->tracefile_size > 0 &&
3095 (stream->tracefile_size_current + data_size) >
3096 stream->tracefile_size) {
a44ca2ca
MD
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;
6b6b9a5a 3104
7591bab1
MD
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,
a44ca2ca 3109 &new_id, &stream->stream_fd->fd);
0f907de1 3110 if (ret < 0) {
1c20f0e2 3111 ERR("Rotating stream output file");
7591bab1
MD
3112 goto end_stream_unlock;
3113 }
7591bab1
MD
3114 /*
3115 * Reset current size because we just performed a stream
3116 * rotation.
3117 */
a6976990 3118 stream->tracefile_size_current = 0;
1c20f0e2
JD
3119 rotate_index = 1;
3120 }
3121
1c20f0e2 3122 /*
7591bab1
MD
3123 * Index are handled in protocol version 2.4 and above. Also,
3124 * snapshot and index are NOT supported.
1c20f0e2 3125 */
2a174661 3126 if (session->minor >= 4 && !session->snapshot) {
d3ecc550
JD
3127 ret = handle_index_data(stream, net_seq_num, rotate_index,
3128 &index_flushed,
3129 data_size + be32toh(data_hdr.padding_size));
1c20f0e2 3130 if (ret < 0) {
65d66b19
MD
3131 ERR("handle_index_data: fail stream %" PRIu64 " net_seq_num %" PRIu64 " ret %d",
3132 stream->stream_handle, net_seq_num, ret);
7591bab1 3133 goto end_stream_unlock;
1c20f0e2 3134 }
1c20f0e2
JD
3135 }
3136
0848dba7
MD
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);
1d4dfdef 3139
0848dba7
MD
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 }
5ab7344e 3164
7591bab1
MD
3165 ret = write_padding_to_file(stream->stream_fd->fd,
3166 be32toh(data_hdr.padding_size));
1d4dfdef 3167 if (ret < 0) {
65d66b19
MD
3168 ERR("write_padding_to_file: fail stream %" PRIu64 " net_seq_num %" PRIu64 " ret %d",
3169 stream->stream_handle, net_seq_num, ret);
7591bab1 3170 goto end_stream_unlock;
1d4dfdef 3171 }
7591bab1
MD
3172 stream->tracefile_size_current +=
3173 data_size + be32toh(data_hdr.padding_size);
c0bae11d
MD
3174 if (stream->prev_seq == -1ULL) {
3175 new_stream = true;
3176 }
d3ecc550
JD
3177 if (index_flushed) {
3178 stream->pos_after_last_complete_data_index =
3179 stream->tracefile_size_current;
3180 }
c0bae11d 3181
173af62f
DG
3182 stream->prev_seq = net_seq_num;
3183
d3ecc550
JD
3184 ret = try_rotate_stream(stream);
3185 if (ret < 0) {
3186 goto end_stream_unlock;
3187 }
3188
7591bab1 3189end_stream_unlock:
bda7c7b9 3190 close_requested = stream->close_requested;
7591bab1 3191 pthread_mutex_unlock(&stream->lock);
bda7c7b9
JG
3192 if (close_requested) {
3193 try_stream_close(stream);
3194 }
3195
c0bae11d
MD
3196 if (new_stream) {
3197 pthread_mutex_lock(&session->lock);
3198 uatomic_set(&session->new_streams, 1);
3199 pthread_mutex_unlock(&session->lock);
3200 }
7591bab1 3201 stream_put(stream);
b8aa1682
JD
3202end:
3203 return ret;
3204}
3205
7591bab1 3206static void cleanup_connection_pollfd(struct lttng_poll_event *events, int pollfd)
b8aa1682
JD
3207{
3208 int ret;
3209
58eb9381 3210 (void) lttng_poll_del(events, pollfd);
b8aa1682
JD
3211
3212 ret = close(pollfd);
3213 if (ret < 0) {
3214 ERR("Closing pollfd %d", pollfd);
3215 }
3216}
3217
7591bab1
MD
3218static void relay_thread_close_connection(struct lttng_poll_event *events,
3219 int pollfd, struct relay_connection *conn)
9d1bbf21 3220{
7591bab1 3221 const char *type_str;
2a174661 3222
7591bab1
MD
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";
9d1bbf21 3238 }
7591bab1
MD
3239 cleanup_connection_pollfd(events, pollfd);
3240 connection_put(conn);
3241 DBG("%s connection closed with %d", type_str, pollfd);
b8aa1682
JD
3242}
3243
3244/*
3245 * This thread does the actual work
3246 */
7591bab1 3247static void *relay_thread_worker(void *data)
b8aa1682 3248{
beaad64c
DG
3249 int ret, err = -1, last_seen_data_fd = -1;
3250 uint32_t nb_fd;
b8aa1682
JD
3251 struct lttng_poll_event events;
3252 struct lttng_ht *relay_connections_ht;
b8aa1682 3253 struct lttng_ht_iter iter;
b8aa1682 3254 struct lttcomm_relayd_hdr recv_hdr;
90e7d72f 3255 struct relay_connection *destroy_conn = NULL;
b8aa1682
JD
3256
3257 DBG("[thread] Relay worker started");
3258
9d1bbf21
MD
3259 rcu_register_thread();
3260
55706a7d
MD
3261 health_register(health_relayd, HEALTH_RELAYD_TYPE_WORKER);
3262
9b5e0863
MD
3263 if (testpoint(relayd_thread_worker)) {
3264 goto error_testpoint;
3265 }
3266
f385ae0a
MD
3267 health_code_update();
3268
b8aa1682
JD
3269 /* table of connections indexed on socket */
3270 relay_connections_ht = lttng_ht_new(0, LTTNG_HT_TYPE_ULONG);
095a4ae5
MD
3271 if (!relay_connections_ht) {
3272 goto relay_connections_ht_error;
3273 }
b8aa1682 3274
b8aa1682
JD
3275 ret = create_thread_poll_set(&events, 2);
3276 if (ret < 0) {
3277 goto error_poll_create;
3278 }
3279
58eb9381 3280 ret = lttng_poll_add(&events, relay_conn_pipe[0], LPOLLIN | LPOLLRDHUP);
b8aa1682
JD
3281 if (ret < 0) {
3282 goto error;
3283 }
3284
beaad64c 3285restart:
b8aa1682 3286 while (1) {
beaad64c
DG
3287 int idx = -1, i, seen_control = 0, last_notdel_data_fd = -1;
3288
f385ae0a
MD
3289 health_code_update();
3290
b8aa1682 3291 /* Infinite blocking call, waiting for transmission */
87c1611d 3292 DBG3("Relayd worker thread polling...");
f385ae0a 3293 health_poll_entry();
b8aa1682 3294 ret = lttng_poll_wait(&events, -1);
f385ae0a 3295 health_poll_exit();
b8aa1682
JD
3296 if (ret < 0) {
3297 /*
3298 * Restart interrupted system call.
3299 */
3300 if (errno == EINTR) {
3301 goto restart;
3302 }
3303 goto error;
3304 }
3305
0d9c5d77
DG
3306 nb_fd = ret;
3307
beaad64c 3308 /*
7591bab1
MD
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.
beaad64c 3312 */
b8aa1682
JD
3313 for (i = 0; i < nb_fd; i++) {
3314 /* Fetch once the poll data */
beaad64c
DG
3315 uint32_t revents = LTTNG_POLL_GETEV(&events, i);
3316 int pollfd = LTTNG_POLL_GETFD(&events, i);
b8aa1682 3317
f385ae0a
MD
3318 health_code_update();