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