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