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