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