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