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