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