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