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