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