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