Load session daemon options from configuration file
[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>
4 *
5 * This program is free software; you can redistribute it and/or modify
6 * it under the terms of the GNU General Public License, version 2 only,
7 * as published by the Free Software Foundation.
8 *
9 * This program is distributed in the hope that it will be useful, but WITHOUT
10 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
11 * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
12 * more details.
13 *
14 * You should have received a copy of the GNU General Public License along
15 * with this program; if not, write to the Free Software Foundation, Inc.,
16 * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
17 */
18
19#define _GNU_SOURCE
20#include <getopt.h>
21#include <grp.h>
22#include <limits.h>
23#include <pthread.h>
24#include <signal.h>
25#include <stdio.h>
26#include <stdlib.h>
27#include <string.h>
28#include <sys/mman.h>
29#include <sys/mount.h>
30#include <sys/resource.h>
31#include <sys/socket.h>
32#include <sys/stat.h>
33#include <sys/types.h>
34#include <sys/wait.h>
173af62f 35#include <inttypes.h>
b8aa1682
JD
36#include <urcu/futex.h>
37#include <urcu/uatomic.h>
38#include <unistd.h>
39#include <fcntl.h>
40#include <config.h>
41
42#include <lttng/lttng.h>
43#include <common/common.h>
44#include <common/compat/poll.h>
45#include <common/compat/socket.h>
46#include <common/defaults.h>
47#include <common/futex.h>
48#include <common/sessiond-comm/sessiond-comm.h>
49#include <common/sessiond-comm/inet.h>
b8aa1682
JD
50#include <common/sessiond-comm/relayd.h>
51#include <common/uri.h>
a02de639 52#include <common/utils.h>
b8aa1682 53
0f907de1 54#include "cmd.h"
d3e2ba59 55#include "ctf-trace.h"
1c20f0e2 56#include "index.h"
0f907de1 57#include "utils.h"
b8aa1682 58#include "lttng-relayd.h"
d3e2ba59 59#include "live.h"
55706a7d 60#include "health-relayd.h"
b8aa1682
JD
61
62/* command line options */
0f907de1 63char *opt_output_path;
b8aa1682 64static int opt_daemon;
095a4ae5
MD
65static struct lttng_uri *control_uri;
66static struct lttng_uri *data_uri;
d3e2ba59 67static struct lttng_uri *live_uri;
b8aa1682
JD
68
69const char *progname;
b8aa1682 70
65931c8b
MD
71const char *tracing_group_name = DEFAULT_TRACING_GROUP;
72
b8aa1682
JD
73/*
74 * Quit pipe for all threads. This permits a single cancellation point
75 * for all threads when receiving an event on the pipe.
76 */
77static int thread_quit_pipe[2] = { -1, -1 };
78
79/*
80 * This pipe is used to inform the worker thread that a command is queued and
81 * ready to be processed.
82 */
83static int relay_cmd_pipe[2] = { -1, -1 };
84
26c9d55e 85/* Shared between threads */
b8aa1682
JD
86static int dispatch_thread_exit;
87
88static pthread_t listener_thread;
89static pthread_t dispatcher_thread;
90static pthread_t worker_thread;
65931c8b 91static pthread_t health_thread;
b8aa1682 92
095a4ae5
MD
93static uint64_t last_relay_stream_id;
94static uint64_t last_relay_session_id;
b8aa1682
JD
95
96/*
97 * Relay command queue.
98 *
99 * The relay_thread_listener and relay_thread_dispatcher communicate with this
100 * queue.
101 */
102static struct relay_cmd_queue relay_cmd_queue;
103
104/* buffer allocated at startup, used to store the trace data */
095a4ae5
MD
105static char *data_buffer;
106static unsigned int data_buffer_size;
b8aa1682 107
1c20f0e2
JD
108/* We need those values for the file/dir creation. */
109static uid_t relayd_uid;
110static gid_t relayd_gid;
111
d3e2ba59
JD
112/* Global relay stream hash table. */
113struct lttng_ht *relay_streams_ht;
114
92c6ca54
DG
115/* Global relay viewer stream hash table. */
116struct lttng_ht *viewer_streams_ht;
117
0a6518b0
DG
118/* Global hash table that stores relay index object. */
119struct lttng_ht *indexes_ht;
120
55706a7d 121/* Relayd health monitoring */
eea7556c 122struct health_app *health_relayd;
55706a7d 123
b8aa1682
JD
124/*
125 * usage function on stderr
126 */
127static
128void usage(void)
129{
130 fprintf(stderr, "Usage: %s OPTIONS\n\nOptions:\n", progname);
994fa64f
DG
131 fprintf(stderr, " -h, --help Display this usage.\n");
132 fprintf(stderr, " -d, --daemonize Start as a daemon.\n");
133 fprintf(stderr, " -C, --control-port URL Control port listening.\n");
134 fprintf(stderr, " -D, --data-port URL Data port listening.\n");
135 fprintf(stderr, " -o, --output PATH Output path for traces. Must use an absolute path.\n");
136 fprintf(stderr, " -v, --verbose Verbose mode. Activate DBG() macro.\n");
65931c8b 137 fprintf(stderr, " -g, --group NAME Specify the tracing group name. (default: tracing)\n");
b8aa1682
JD
138}
139
140static
141int parse_args(int argc, char **argv)
142{
143 int c;
144 int ret = 0;
145 char *default_address;
146
147 static struct option long_options[] = {
e3678fd8
MD
148 { "control-port", 1, 0, 'C', },
149 { "data-port", 1, 0, 'D', },
150 { "daemonize", 0, 0, 'd', },
65931c8b 151 { "group", 1, 0, 'g', },
e3678fd8
MD
152 { "help", 0, 0, 'h', },
153 { "output", 1, 0, 'o', },
154 { "verbose", 0, 0, 'v', },
095a4ae5 155 { NULL, 0, 0, 0, },
b8aa1682
JD
156 };
157
158 while (1) {
159 int option_index = 0;
65931c8b 160 c = getopt_long(argc, argv, "dhv" "C:D:o:g:",
b8aa1682
JD
161 long_options, &option_index);
162 if (c == -1) {
163 break;
164 }
165
166 switch (c) {
167 case 0:
168 fprintf(stderr, "option %s", long_options[option_index].name);
169 if (optarg) {
170 fprintf(stderr, " with arg %s\n", optarg);
171 }
172 break;
173 case 'C':
174 ret = uri_parse(optarg, &control_uri);
175 if (ret < 0) {
176 ERR("Invalid control URI specified");
177 goto exit;
178 }
179 if (control_uri->port == 0) {
180 control_uri->port = DEFAULT_NETWORK_CONTROL_PORT;
181 }
182 break;
183 case 'D':
184 ret = uri_parse(optarg, &data_uri);
185 if (ret < 0) {
186 ERR("Invalid data URI specified");
187 goto exit;
188 }
189 if (data_uri->port == 0) {
190 data_uri->port = DEFAULT_NETWORK_DATA_PORT;
191 }
192 break;
193 case 'd':
194 opt_daemon = 1;
195 break;
65931c8b
MD
196 case 'g':
197 tracing_group_name = optarg;
198 break;
b8aa1682
JD
199 case 'h':
200 usage();
201 exit(EXIT_FAILURE);
202 case 'o':
203 ret = asprintf(&opt_output_path, "%s", optarg);
204 if (ret < 0) {
205 PERROR("asprintf opt_output_path");
206 goto exit;
207 }
208 break;
209 case 'v':
210 /* Verbose level can increase using multiple -v */
211 lttng_opt_verbose += 1;
212 break;
213 default:
214 /* Unknown option or other error.
215 * Error is printed by getopt, just return */
216 ret = -1;
217 goto exit;
218 }
219 }
220
221 /* assign default values */
222 if (control_uri == NULL) {
223 ret = asprintf(&default_address, "tcp://0.0.0.0:%d",
224 DEFAULT_NETWORK_CONTROL_PORT);
225 if (ret < 0) {
226 PERROR("asprintf default data address");
227 goto exit;
228 }
229
230 ret = uri_parse(default_address, &control_uri);
231 free(default_address);
232 if (ret < 0) {
233 ERR("Invalid control URI specified");
234 goto exit;
235 }
236 }
237 if (data_uri == NULL) {
238 ret = asprintf(&default_address, "tcp://0.0.0.0:%d",
239 DEFAULT_NETWORK_DATA_PORT);
240 if (ret < 0) {
241 PERROR("asprintf default data address");
242 goto exit;
243 }
244
245 ret = uri_parse(default_address, &data_uri);
246 free(default_address);
247 if (ret < 0) {
248 ERR("Invalid data URI specified");
249 goto exit;
250 }
251 }
d3e2ba59
JD
252 if (live_uri == NULL) {
253 ret = asprintf(&default_address, "tcp://0.0.0.0:%d",
254 DEFAULT_NETWORK_VIEWER_PORT);
255 if (ret < 0) {
256 PERROR("asprintf default viewer control address");
257 goto exit;
258 }
259
260 ret = uri_parse(default_address, &live_uri);
261 free(default_address);
262 if (ret < 0) {
263 ERR("Invalid viewer control URI specified");
264 goto exit;
265 }
266 }
b8aa1682
JD
267
268exit:
269 return ret;
270}
271
272/*
273 * Cleanup the daemon
274 */
275static
276void cleanup(void)
277{
b8aa1682
JD
278 DBG("Cleaning up");
279
095a4ae5
MD
280 /* free the dynamically allocated opt_output_path */
281 free(opt_output_path);
282
a02de639
CB
283 /* Close thread quit pipes */
284 utils_close_pipe(thread_quit_pipe);
285
710c1f73
DG
286 uri_free(control_uri);
287 uri_free(data_uri);
b8aa1682
JD
288}
289
290/*
291 * Write to writable pipe used to notify a thread.
292 */
293static
294int notify_thread_pipe(int wpipe)
295{
6cd525e8 296 ssize_t ret;
b8aa1682 297
6cd525e8
MD
298 ret = lttng_write(wpipe, "!", 1);
299 if (ret < 1) {
b8aa1682
JD
300 PERROR("write poll pipe");
301 }
302
303 return ret;
304}
305
65931c8b
MD
306static void notify_health_quit_pipe(int *pipe)
307{
6cd525e8 308 ssize_t ret;
65931c8b 309
6cd525e8
MD
310 ret = lttng_write(pipe[1], "4", 1);
311 if (ret < 1) {
65931c8b
MD
312 PERROR("write relay health quit");
313 }
314}
315
b8aa1682
JD
316/*
317 * Stop all threads by closing the thread quit pipe.
318 */
319static
320void stop_threads(void)
321{
322 int ret;
323
324 /* Stopping all threads */
325 DBG("Terminating all threads");
326 ret = notify_thread_pipe(thread_quit_pipe[1]);
327 if (ret < 0) {
328 ERR("write error on thread quit pipe");
329 }
330
65931c8b
MD
331 notify_health_quit_pipe(health_quit_pipe);
332
b8aa1682 333 /* Dispatch thread */
26c9d55e 334 CMM_STORE_SHARED(dispatch_thread_exit, 1);
b8aa1682
JD
335 futex_nto1_wake(&relay_cmd_queue.futex);
336}
337
338/*
339 * Signal handler for the daemon
340 *
341 * Simply stop all worker threads, leaving main() return gracefully after
342 * joining all threads and calling cleanup().
343 */
344static
345void sighandler(int sig)
346{
347 switch (sig) {
348 case SIGPIPE:
349 DBG("SIGPIPE caught");
350 return;
351 case SIGINT:
352 DBG("SIGINT caught");
353 stop_threads();
354 break;
355 case SIGTERM:
356 DBG("SIGTERM caught");
357 stop_threads();
358 break;
359 default:
360 break;
361 }
362}
363
364/*
365 * Setup signal handler for :
366 * SIGINT, SIGTERM, SIGPIPE
367 */
368static
369int set_signal_handler(void)
370{
371 int ret = 0;
372 struct sigaction sa;
373 sigset_t sigset;
374
375 if ((ret = sigemptyset(&sigset)) < 0) {
376 PERROR("sigemptyset");
377 return ret;
378 }
379
380 sa.sa_handler = sighandler;
381 sa.sa_mask = sigset;
382 sa.sa_flags = 0;
383 if ((ret = sigaction(SIGTERM, &sa, NULL)) < 0) {
384 PERROR("sigaction");
385 return ret;
386 }
387
388 if ((ret = sigaction(SIGINT, &sa, NULL)) < 0) {
389 PERROR("sigaction");
390 return ret;
391 }
392
393 if ((ret = sigaction(SIGPIPE, &sa, NULL)) < 0) {
394 PERROR("sigaction");
395 return ret;
396 }
397
398 DBG("Signal handler set for SIGTERM, SIGPIPE and SIGINT");
399
400 return ret;
401}
402
403/*
404 * Init thread quit pipe.
405 *
406 * Return -1 on error or 0 if all pipes are created.
407 */
408static
409int init_thread_quit_pipe(void)
410{
a02de639 411 int ret;
b8aa1682 412
a02de639 413 ret = utils_create_pipe_cloexec(thread_quit_pipe);
b8aa1682 414
b8aa1682
JD
415 return ret;
416}
417
418/*
419 * Create a poll set with O_CLOEXEC and add the thread quit pipe to the set.
420 */
421static
422int create_thread_poll_set(struct lttng_poll_event *events, int size)
423{
424 int ret;
425
426 if (events == NULL || size == 0) {
427 ret = -1;
428 goto error;
429 }
430
431 ret = lttng_poll_create(events, size, LTTNG_CLOEXEC);
432 if (ret < 0) {
433 goto error;
434 }
435
436 /* Add quit pipe */
437 ret = lttng_poll_add(events, thread_quit_pipe[0], LPOLLIN);
438 if (ret < 0) {
439 goto error;
440 }
441
442 return 0;
443
444error:
445 return ret;
446}
447
448/*
449 * Check if the thread quit pipe was triggered.
450 *
451 * Return 1 if it was triggered else 0;
452 */
453static
454int check_thread_quit_pipe(int fd, uint32_t events)
455{
456 if (fd == thread_quit_pipe[0] && (events & LPOLLIN)) {
457 return 1;
458 }
459
460 return 0;
461}
462
463/*
464 * Create and init socket from uri.
465 */
466static
467struct lttcomm_sock *relay_init_sock(struct lttng_uri *uri)
468{
469 int ret;
470 struct lttcomm_sock *sock = NULL;
471
472 sock = lttcomm_alloc_sock_from_uri(uri);
473 if (sock == NULL) {
474 ERR("Allocating socket");
475 goto error;
476 }
477
478 ret = lttcomm_create_sock(sock);
479 if (ret < 0) {
480 goto error;
481 }
482 DBG("Listening on sock %d", sock->fd);
483
484 ret = sock->ops->bind(sock);
485 if (ret < 0) {
486 goto error;
487 }
488
489 ret = sock->ops->listen(sock, -1);
490 if (ret < 0) {
491 goto error;
492
493 }
494
495 return sock;
496
497error:
498 if (sock) {
499 lttcomm_destroy_sock(sock);
500 }
501 return NULL;
502}
503
173af62f
DG
504/*
505 * Return nonzero if stream needs to be closed.
506 */
507static
508int close_stream_check(struct relay_stream *stream)
509{
510
511 if (stream->close_flag && stream->prev_seq == stream->last_net_seq_num) {
f7079f67
DG
512 /*
513 * We are about to close the stream so set the data pending flag to 1
514 * which will make the end data pending command skip the stream which
515 * is now closed and ready. Note that after proceeding to a file close,
516 * the written file is ready for reading.
517 */
518 stream->data_pending_check_done = 1;
173af62f
DG
519 return 1;
520 }
521 return 0;
522}
523
b8aa1682
JD
524/*
525 * This thread manages the listening for new connections on the network
526 */
527static
528void *relay_thread_listener(void *data)
529{
095a4ae5 530 int i, ret, pollfd, err = -1;
b8aa1682
JD
531 int val = 1;
532 uint32_t revents, nb_fd;
533 struct lttng_poll_event events;
534 struct lttcomm_sock *control_sock, *data_sock;
535
b8aa1682
JD
536 DBG("[thread] Relay listener started");
537
55706a7d
MD
538 health_register(health_relayd, HEALTH_RELAYD_TYPE_LISTENER);
539
f385ae0a
MD
540 health_code_update();
541
b8aa1682
JD
542 control_sock = relay_init_sock(control_uri);
543 if (!control_sock) {
095a4ae5 544 goto error_sock_control;
b8aa1682
JD
545 }
546
547 data_sock = relay_init_sock(data_uri);
548 if (!data_sock) {
095a4ae5 549 goto error_sock_relay;
b8aa1682
JD
550 }
551
552 /*
553 * Pass 3 as size here for the thread quit pipe, control and data socket.
554 */
555 ret = create_thread_poll_set(&events, 3);
556 if (ret < 0) {
557 goto error_create_poll;
558 }
559
560 /* Add the control socket */
561 ret = lttng_poll_add(&events, control_sock->fd, LPOLLIN | LPOLLRDHUP);
562 if (ret < 0) {
563 goto error_poll_add;
564 }
565
566 /* Add the data socket */
567 ret = lttng_poll_add(&events, data_sock->fd, LPOLLIN | LPOLLRDHUP);
568 if (ret < 0) {
569 goto error_poll_add;
570 }
571
572 while (1) {
f385ae0a
MD
573 health_code_update();
574
b8aa1682
JD
575 DBG("Listener accepting connections");
576
b8aa1682 577restart:
f385ae0a 578 health_poll_entry();
b8aa1682 579 ret = lttng_poll_wait(&events, -1);
f385ae0a 580 health_poll_exit();
b8aa1682
JD
581 if (ret < 0) {
582 /*
583 * Restart interrupted system call.
584 */
585 if (errno == EINTR) {
586 goto restart;
587 }
588 goto error;
589 }
590
0d9c5d77
DG
591 nb_fd = ret;
592
b8aa1682
JD
593 DBG("Relay new connection received");
594 for (i = 0; i < nb_fd; i++) {
f385ae0a
MD
595 health_code_update();
596
b8aa1682
JD
597 /* Fetch once the poll data */
598 revents = LTTNG_POLL_GETEV(&events, i);
599 pollfd = LTTNG_POLL_GETFD(&events, i);
600
601 /* Thread quit pipe has been closed. Killing thread. */
602 ret = check_thread_quit_pipe(pollfd, revents);
603 if (ret) {
095a4ae5
MD
604 err = 0;
605 goto exit;
b8aa1682
JD
606 }
607
608 if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
609 ERR("socket poll error");
610 goto error;
611 } else if (revents & LPOLLIN) {
4b7f17b2
MD
612 /*
613 * Get allocated in this thread,
614 * enqueued to a global queue, dequeued
615 * and freed in the worker thread.
616 */
617 struct relay_command *relay_cmd;
618 struct lttcomm_sock *newsock;
b8aa1682
JD
619
620 relay_cmd = zmalloc(sizeof(struct relay_command));
621 if (relay_cmd == NULL) {
622 PERROR("relay command zmalloc");
623 goto error;
624 }
625
626 if (pollfd == data_sock->fd) {
627 newsock = data_sock->ops->accept(data_sock);
4b7f17b2 628 if (!newsock) {
b8aa1682 629 PERROR("accepting data sock");
4b7f17b2 630 free(relay_cmd);
b8aa1682
JD
631 goto error;
632 }
633 relay_cmd->type = RELAY_DATA;
634 DBG("Relay data connection accepted, socket %d", newsock->fd);
4b7f17b2
MD
635 } else {
636 assert(pollfd == control_sock->fd);
b8aa1682 637 newsock = control_sock->ops->accept(control_sock);
4b7f17b2 638 if (!newsock) {
b8aa1682 639 PERROR("accepting control sock");
4b7f17b2 640 free(relay_cmd);
b8aa1682
JD
641 goto error;
642 }
643 relay_cmd->type = RELAY_CONTROL;
644 DBG("Relay control connection accepted, socket %d", newsock->fd);
645 }
646 ret = setsockopt(newsock->fd, SOL_SOCKET, SO_REUSEADDR,
647 &val, sizeof(int));
648 if (ret < 0) {
649 PERROR("setsockopt inet");
4b7f17b2
MD
650 lttcomm_destroy_sock(newsock);
651 free(relay_cmd);
b8aa1682
JD
652 goto error;
653 }
654 relay_cmd->sock = newsock;
655 /*
656 * Lock free enqueue the request.
657 */
658 cds_wfq_enqueue(&relay_cmd_queue.queue, &relay_cmd->node);
659
660 /*
661 * Wake the dispatch queue futex. Implicit memory
662 * barrier with the exchange in cds_wfq_enqueue.
663 */
664 futex_nto1_wake(&relay_cmd_queue.futex);
665 }
666 }
667 }
668
095a4ae5 669exit:
b8aa1682
JD
670error:
671error_poll_add:
672 lttng_poll_clean(&events);
673error_create_poll:
095a4ae5
MD
674 if (data_sock->fd >= 0) {
675 ret = data_sock->ops->close(data_sock);
b8aa1682
JD
676 if (ret) {
677 PERROR("close");
678 }
b8aa1682 679 }
095a4ae5
MD
680 lttcomm_destroy_sock(data_sock);
681error_sock_relay:
682 if (control_sock->fd >= 0) {
683 ret = control_sock->ops->close(control_sock);
b8aa1682
JD
684 if (ret) {
685 PERROR("close");
686 }
b8aa1682 687 }
095a4ae5
MD
688 lttcomm_destroy_sock(control_sock);
689error_sock_control:
690 if (err) {
f385ae0a
MD
691 health_error();
692 ERR("Health error occurred in %s", __func__);
095a4ae5 693 }
55706a7d 694 health_unregister(health_relayd);
b8aa1682
JD
695 DBG("Relay listener thread cleanup complete");
696 stop_threads();
b8aa1682
JD
697 return NULL;
698}
699
700/*
701 * This thread manages the dispatching of the requests to worker threads
702 */
703static
704void *relay_thread_dispatcher(void *data)
705{
6cd525e8
MD
706 int err = -1;
707 ssize_t ret;
b8aa1682
JD
708 struct cds_wfq_node *node;
709 struct relay_command *relay_cmd = NULL;
710
711 DBG("[thread] Relay dispatcher started");
712
55706a7d
MD
713 health_register(health_relayd, HEALTH_RELAYD_TYPE_DISPATCHER);
714
f385ae0a
MD
715 health_code_update();
716
26c9d55e 717 while (!CMM_LOAD_SHARED(dispatch_thread_exit)) {
f385ae0a
MD
718 health_code_update();
719
b8aa1682
JD
720 /* Atomically prepare the queue futex */
721 futex_nto1_prepare(&relay_cmd_queue.futex);
722
723 do {
f385ae0a
MD
724 health_code_update();
725
b8aa1682
JD
726 /* Dequeue commands */
727 node = cds_wfq_dequeue_blocking(&relay_cmd_queue.queue);
728 if (node == NULL) {
729 DBG("Woken up but nothing in the relay command queue");
730 /* Continue thread execution */
731 break;
732 }
733
734 relay_cmd = caa_container_of(node, struct relay_command, node);
735 DBG("Dispatching request waiting on sock %d", relay_cmd->sock->fd);
736
737 /*
738 * Inform worker thread of the new request. This
739 * call is blocking so we can be assured that the data will be read
740 * at some point in time or wait to the end of the world :)
741 */
6cd525e8
MD
742 ret = lttng_write(relay_cmd_pipe[1], relay_cmd,
743 sizeof(struct relay_command));
b8aa1682 744 free(relay_cmd);
6cd525e8 745 if (ret < sizeof(struct relay_command)) {
b8aa1682
JD
746 PERROR("write cmd pipe");
747 goto error;
748 }
749 } while (node != NULL);
750
751 /* Futex wait on queue. Blocking call on futex() */
f385ae0a 752 health_poll_entry();
b8aa1682 753 futex_nto1_wait(&relay_cmd_queue.futex);
f385ae0a 754 health_poll_exit();
b8aa1682
JD
755 }
756
f385ae0a
MD
757 /* Normal exit, no error */
758 err = 0;
759
b8aa1682 760error:
f385ae0a
MD
761 if (err) {
762 health_error();
763 ERR("Health error occurred in %s", __func__);
764 }
55706a7d 765 health_unregister(health_relayd);
b8aa1682
JD
766 DBG("Dispatch thread dying");
767 stop_threads();
768 return NULL;
769}
770
de91f48a
DG
771/*
772 * Get stream from stream id.
773 * Need to be called with RCU read-side lock held.
774 */
d3e2ba59 775struct relay_stream *relay_stream_find_by_id(uint64_t stream_id)
de91f48a
DG
776{
777 struct lttng_ht_node_ulong *node;
778 struct lttng_ht_iter iter;
779 struct relay_stream *ret;
780
d3e2ba59 781 lttng_ht_lookup(relay_streams_ht,
de91f48a
DG
782 (void *)((unsigned long) stream_id),
783 &iter);
784 node = lttng_ht_iter_get_node_ulong(&iter);
785 if (node == NULL) {
786 DBG("Relay stream %" PRIu64 " not found", stream_id);
787 ret = NULL;
788 goto end;
789 }
790
791 ret = caa_container_of(node, struct relay_stream, stream_n);
792
793end:
794 return ret;
795}
796
9d1bbf21
MD
797static
798void deferred_free_stream(struct rcu_head *head)
799{
800 struct relay_stream *stream =
801 caa_container_of(head, struct relay_stream, rcu_node);
d3e2ba59 802
0f907de1
JD
803 free(stream->path_name);
804 free(stream->channel_name);
9d1bbf21
MD
805 free(stream);
806}
807
d3e2ba59
JD
808static
809void deferred_free_session(struct rcu_head *head)
810{
811 struct relay_session *session =
812 caa_container_of(head, struct relay_session, rcu_node);
813 free(session);
814}
815
4c80d9a2
DG
816/*
817 * Close a given stream. The stream is freed using a call RCU.
818 *
819 * RCU read side lock MUST be acquired. If NO close_stream_check() was called
820 * BEFORE the stream lock MUST be acquired.
821 */
c07d8a78 822static void destroy_stream(struct relay_stream *stream)
94d49140
JD
823{
824 int delret;
825 struct relay_viewer_stream *vstream;
826 struct lttng_ht_iter iter;
827
828 assert(stream);
94d49140
JD
829
830 delret = close(stream->fd);
831 if (delret < 0) {
832 PERROR("close stream");
833 }
834
835 if (stream->index_fd >= 0) {
836 delret = close(stream->index_fd);
837 if (delret < 0) {
838 PERROR("close stream index_fd");
839 }
840 }
841
92c6ca54 842 vstream = live_find_viewer_stream_by_id(stream->stream_handle);
94d49140
JD
843 if (vstream) {
844 /*
845 * Set the last good value into the viewer stream. This is done
846 * right before the stream gets deleted from the hash table. The
847 * lookup failure on the live thread side of a stream indicates
848 * that the viewer stream index received value should be used.
849 */
cef0f7d5 850 pthread_mutex_lock(&stream->viewer_stream_rotation_lock);
94d49140 851 vstream->total_index_received = stream->total_index_received;
a020f610 852 vstream->tracefile_count_last = stream->tracefile_count_current;
cef0f7d5
JD
853 vstream->close_write_flag = 1;
854 pthread_mutex_unlock(&stream->viewer_stream_rotation_lock);
94d49140
JD
855 }
856
efc8a87f
DG
857 /* Cleanup index of that stream. */
858 relay_index_destroy_by_stream_id(stream->stream_handle);
859
94d49140
JD
860 iter.iter.node = &stream->stream_n.node;
861 delret = lttng_ht_del(relay_streams_ht, &iter);
862 assert(!delret);
863 iter.iter.node = &stream->ctf_trace_node.node;
c07d8a78 864 delret = lttng_ht_del(stream->ctf_traces_ht, &iter);
94d49140 865 assert(!delret);
157df586
JD
866
867 if (stream->ctf_trace) {
868 ctf_trace_try_destroy(stream->ctf_trace);
869 }
870
94d49140
JD
871 call_rcu(&stream->rcu_node, deferred_free_stream);
872 DBG("Closed tracefile %d from close stream", stream->fd);
873}
874
b8aa1682
JD
875/*
876 * relay_delete_session: Free all memory associated with a session and
877 * close all the FDs
878 */
879static
d3e2ba59
JD
880void relay_delete_session(struct relay_command *cmd,
881 struct lttng_ht *sessions_ht)
b8aa1682
JD
882{
883 struct lttng_ht_iter iter;
884 struct lttng_ht_node_ulong *node;
885 struct relay_stream *stream;
886 int ret;
887
095a4ae5 888 if (!cmd->session) {
b8aa1682 889 return;
095a4ae5 890 }
b8aa1682 891
77c7c900 892 DBG("Relay deleting session %" PRIu64, cmd->session->id);
5b6d8097 893
9d1bbf21 894 rcu_read_lock();
d3e2ba59 895 cds_lfht_for_each_entry(relay_streams_ht->ht, &iter.iter, node, node) {
b8aa1682 896 node = lttng_ht_iter_get_node_ulong(&iter);
4c80d9a2
DG
897 if (!node) {
898 continue;
b8aa1682 899 }
4c80d9a2
DG
900 stream = caa_container_of(node, struct relay_stream, stream_n);
901 if (stream->session == cmd->session) {
c07d8a78 902 destroy_stream(stream);
87b576ec
JD
903 cmd->session->stream_count--;
904 assert(cmd->session->stream_count >= 0);
4c80d9a2 905 }
b8aa1682 906 }
4c80d9a2
DG
907
908 /* Make this session not visible anymore. */
d3e2ba59
JD
909 iter.iter.node = &cmd->session->session_n.node;
910 ret = lttng_ht_del(sessions_ht, &iter);
911 assert(!ret);
4c80d9a2 912 call_rcu(&cmd->session->rcu_node, deferred_free_session);
9d1bbf21 913 rcu_read_unlock();
b8aa1682
JD
914}
915
1c20f0e2
JD
916/*
917 * Copy index data from the control port to a given index object.
918 */
919static void copy_index_control_data(struct relay_index *index,
920 struct lttcomm_relayd_index *data)
921{
922 assert(index);
923 assert(data);
924
925 /*
926 * The index on disk is encoded in big endian, so we don't need to convert
927 * the data received on the network. The data_offset value is NEVER
928 * modified here and is updated by the data thread.
929 */
930 index->index_data.packet_size = data->packet_size;
931 index->index_data.content_size = data->content_size;
932 index->index_data.timestamp_begin = data->timestamp_begin;
933 index->index_data.timestamp_end = data->timestamp_end;
934 index->index_data.events_discarded = data->events_discarded;
935 index->index_data.stream_id = data->stream_id;
936}
937
c5b6f4f0
DG
938/*
939 * Handle the RELAYD_CREATE_SESSION command.
940 *
941 * On success, send back the session id or else return a negative value.
942 */
943static
944int relay_create_session(struct lttcomm_relayd_hdr *recv_hdr,
d3e2ba59
JD
945 struct relay_command *cmd,
946 struct lttng_ht *sessions_ht)
c5b6f4f0
DG
947{
948 int ret = 0, send_ret;
949 struct relay_session *session;
950 struct lttcomm_relayd_status_session reply;
951
952 assert(recv_hdr);
953 assert(cmd);
954
955 memset(&reply, 0, sizeof(reply));
956
957 session = zmalloc(sizeof(struct relay_session));
958 if (session == NULL) {
959 PERROR("relay session zmalloc");
960 ret = -1;
961 goto error;
962 }
963
964 session->id = ++last_relay_session_id;
965 session->sock = cmd->sock;
7d2f7452
DG
966 session->minor = cmd->minor;
967 session->major = cmd->major;
c5b6f4f0
DG
968 cmd->session = session;
969
970 reply.session_id = htobe64(session->id);
971
d3e2ba59
JD
972 switch (cmd->minor) {
973 case 4: /* LTTng sessiond 2.4 */
974 default:
975 ret = cmd_create_session_2_4(cmd, session);
976 break;
977 }
978
979 lttng_ht_node_init_ulong(&session->session_n,
980 (unsigned long) session->id);
981 lttng_ht_add_unique_ulong(sessions_ht,
982 &session->session_n);
983
c5b6f4f0
DG
984 DBG("Created session %" PRIu64, session->id);
985
986error:
987 if (ret < 0) {
988 reply.ret_code = htobe32(LTTNG_ERR_FATAL);
989 } else {
990 reply.ret_code = htobe32(LTTNG_OK);
991 }
992
993 send_ret = cmd->sock->ops->sendmsg(cmd->sock, &reply, sizeof(reply), 0);
994 if (send_ret < 0) {
995 ERR("Relayd sending session id");
4169f5ad 996 ret = send_ret;
c5b6f4f0
DG
997 }
998
999 return ret;
1000}
1001
b8aa1682
JD
1002/*
1003 * relay_add_stream: allocate a new stream for a session
1004 */
1005static
1006int relay_add_stream(struct lttcomm_relayd_hdr *recv_hdr,
d3e2ba59 1007 struct relay_command *cmd, struct lttng_ht *sessions_ht)
b8aa1682
JD
1008{
1009 struct relay_session *session = cmd->session;
b8aa1682
JD
1010 struct relay_stream *stream = NULL;
1011 struct lttcomm_relayd_status_stream reply;
b8aa1682
JD
1012 int ret, send_ret;
1013
c5b6f4f0 1014 if (!session || cmd->version_check_done == 0) {
b8aa1682
JD
1015 ERR("Trying to add a stream before version check");
1016 ret = -1;
1017 goto end_no_session;
1018 }
1019
b8aa1682
JD
1020 stream = zmalloc(sizeof(struct relay_stream));
1021 if (stream == NULL) {
1022 PERROR("relay stream zmalloc");
1023 ret = -1;
1024 goto end_no_session;
1025 }
1026
0f907de1
JD
1027 switch (cmd->minor) {
1028 case 1: /* LTTng sessiond 2.1 */
1029 ret = cmd_recv_stream_2_1(cmd, stream);
1030 break;
1031 case 2: /* LTTng sessiond 2.2 */
1032 default:
1033 ret = cmd_recv_stream_2_2(cmd, stream);
1034 break;
1035 }
1036 if (ret < 0) {
1037 goto err_free_stream;
1038 }
1039
9d1bbf21 1040 rcu_read_lock();
b8aa1682 1041 stream->stream_handle = ++last_relay_stream_id;
173af62f 1042 stream->prev_seq = -1ULL;
b8aa1682 1043 stream->session = session;
1c20f0e2 1044 stream->index_fd = -1;
d3e2ba59
JD
1045 stream->read_index_fd = -1;
1046 stream->ctf_trace = NULL;
1047 pthread_mutex_init(&stream->lock, NULL);
b8aa1682 1048
0f907de1 1049 ret = utils_mkdir_recursive(stream->path_name, S_IRWXU | S_IRWXG);
b8aa1682 1050 if (ret < 0) {
b8aa1682
JD
1051 ERR("relay creating output directory");
1052 goto end;
1053 }
1054
be96a7d1
DG
1055 /*
1056 * No need to use run_as API here because whatever we receives, the relayd
1057 * uses its own credentials for the stream files.
1058 */
0f907de1 1059 ret = utils_create_stream_file(stream->path_name, stream->channel_name,
1c20f0e2 1060 stream->tracefile_size, 0, relayd_uid, relayd_gid, NULL);
b8aa1682 1061 if (ret < 0) {
0f907de1 1062 ERR("Create output file");
b8aa1682
JD
1063 goto end;
1064 }
b8aa1682 1065 stream->fd = ret;
0f907de1
JD
1066 if (stream->tracefile_size) {
1067 DBG("Tracefile %s/%s_0 created", stream->path_name, stream->channel_name);
1068 } else {
1069 DBG("Tracefile %s/%s created", stream->path_name, stream->channel_name);
1070 }
b8aa1682 1071
d3e2ba59
JD
1072 if (!strncmp(stream->channel_name, DEFAULT_METADATA_NAME, NAME_MAX)) {
1073 stream->metadata_flag = 1;
1074 /*
1075 * When we receive a new metadata stream, we create a new
1076 * ctf_trace and we assign this ctf_trace to all streams with
1077 * the same path.
1078 *
1079 * If later on we receive a new stream for the same ctf_trace,
1080 * we copy the information from the first hit in the HT to the
1081 * new stream.
1082 */
1083 stream->ctf_trace = ctf_trace_create();
1084 if (!stream->ctf_trace) {
1085 ret = -1;
1086 goto end;
1087 }
1088 stream->ctf_trace->refcount++;
1089 stream->ctf_trace->metadata_stream = stream;
1090 }
1091 ctf_trace_assign(cmd->ctf_traces_ht, stream);
c07d8a78 1092 stream->ctf_traces_ht = cmd->ctf_traces_ht;
d3e2ba59 1093
b8aa1682
JD
1094 lttng_ht_node_init_ulong(&stream->stream_n,
1095 (unsigned long) stream->stream_handle);
d3e2ba59 1096 lttng_ht_add_unique_ulong(relay_streams_ht,
b8aa1682
JD
1097 &stream->stream_n);
1098
d3e2ba59
JD
1099 lttng_ht_node_init_str(&stream->ctf_trace_node, stream->path_name);
1100 lttng_ht_add_str(cmd->ctf_traces_ht, &stream->ctf_trace_node);
87b576ec 1101 session->stream_count++;
d3e2ba59 1102
1c20f0e2
JD
1103 DBG("Relay new stream added %s with ID %" PRIu64, stream->channel_name,
1104 stream->stream_handle);
b8aa1682
JD
1105
1106end:
5af40280 1107 reply.handle = htobe64(stream->stream_handle);
b8aa1682
JD
1108 /* send the session id to the client or a negative return code on error */
1109 if (ret < 0) {
f73fabfd 1110 reply.ret_code = htobe32(LTTNG_ERR_UNK);
5af40280
CB
1111 /* stream was not properly added to the ht, so free it */
1112 free(stream);
b8aa1682 1113 } else {
f73fabfd 1114 reply.ret_code = htobe32(LTTNG_OK);
b8aa1682 1115 }
5af40280 1116
b8aa1682
JD
1117 send_ret = cmd->sock->ops->sendmsg(cmd->sock, &reply,
1118 sizeof(struct lttcomm_relayd_status_stream), 0);
1119 if (send_ret < 0) {
1120 ERR("Relay sending stream id");
4169f5ad 1121 ret = send_ret;
b8aa1682 1122 }
9d1bbf21 1123 rcu_read_unlock();
b8aa1682
JD
1124
1125end_no_session:
1126 return ret;
0f907de1
JD
1127
1128err_free_stream:
1129 free(stream->path_name);
1130 free(stream->channel_name);
1131 free(stream);
1132 return ret;
b8aa1682
JD
1133}
1134
173af62f
DG
1135/*
1136 * relay_close_stream: close a specific stream
1137 */
1138static
1139int relay_close_stream(struct lttcomm_relayd_hdr *recv_hdr,
92c6ca54 1140 struct relay_command *cmd)
173af62f 1141{
94d49140 1142 int ret, send_ret;
173af62f
DG
1143 struct relay_session *session = cmd->session;
1144 struct lttcomm_relayd_close_stream stream_info;
1145 struct lttcomm_relayd_generic_reply reply;
1146 struct relay_stream *stream;
173af62f
DG
1147
1148 DBG("Close stream received");
1149
c5b6f4f0 1150 if (!session || cmd->version_check_done == 0) {
173af62f
DG
1151 ERR("Trying to close a stream before version check");
1152 ret = -1;
1153 goto end_no_session;
1154 }
1155
1156 ret = cmd->sock->ops->recvmsg(cmd->sock, &stream_info,
7c5aef62 1157 sizeof(struct lttcomm_relayd_close_stream), 0);
173af62f 1158 if (ret < sizeof(struct lttcomm_relayd_close_stream)) {
a6cd2b97
DG
1159 if (ret == 0) {
1160 /* Orderly shutdown. Not necessary to print an error. */
1161 DBG("Socket %d did an orderly shutdown", cmd->sock->fd);
1162 } else {
1163 ERR("Relay didn't receive valid add_stream struct size : %d", ret);
1164 }
173af62f
DG
1165 ret = -1;
1166 goto end_no_session;
1167 }
1168
1169 rcu_read_lock();
d3e2ba59 1170 stream = relay_stream_find_by_id(be64toh(stream_info.stream_id));
173af62f
DG
1171 if (!stream) {
1172 ret = -1;
1173 goto end_unlock;
1174 }
1175
8e2583a4 1176 stream->last_net_seq_num = be64toh(stream_info.last_net_seq_num);
173af62f 1177 stream->close_flag = 1;
87b576ec
JD
1178 session->stream_count--;
1179 assert(session->stream_count >= 0);
173af62f
DG
1180
1181 if (close_stream_check(stream)) {
c07d8a78 1182 destroy_stream(stream);
173af62f
DG
1183 }
1184
1185end_unlock:
1186 rcu_read_unlock();
1187
1188 if (ret < 0) {
f73fabfd 1189 reply.ret_code = htobe32(LTTNG_ERR_UNK);
173af62f 1190 } else {
f73fabfd 1191 reply.ret_code = htobe32(LTTNG_OK);
173af62f
DG
1192 }
1193 send_ret = cmd->sock->ops->sendmsg(cmd->sock, &reply,
1194 sizeof(struct lttcomm_relayd_generic_reply), 0);
1195 if (send_ret < 0) {
1196 ERR("Relay sending stream id");
4169f5ad 1197 ret = send_ret;
173af62f
DG
1198 }
1199
1200end_no_session:
1201 return ret;
1202}
1203
b8aa1682
JD
1204/*
1205 * relay_unknown_command: send -1 if received unknown command
1206 */
1207static
1208void relay_unknown_command(struct relay_command *cmd)
1209{
1210 struct lttcomm_relayd_generic_reply reply;
1211 int ret;
1212
f73fabfd 1213 reply.ret_code = htobe32(LTTNG_ERR_UNK);
b8aa1682
JD
1214 ret = cmd->sock->ops->sendmsg(cmd->sock, &reply,
1215 sizeof(struct lttcomm_relayd_generic_reply), 0);
1216 if (ret < 0) {
1217 ERR("Relay sending unknown command");
1218 }
1219}
1220
1221/*
1222 * relay_start: send an acknowledgment to the client to tell if we are
1223 * ready to receive data. We are ready if a session is established.
1224 */
1225static
1226int relay_start(struct lttcomm_relayd_hdr *recv_hdr,
1227 struct relay_command *cmd)
1228{
f73fabfd 1229 int ret = htobe32(LTTNG_OK);
b8aa1682
JD
1230 struct lttcomm_relayd_generic_reply reply;
1231 struct relay_session *session = cmd->session;
1232
1233 if (!session) {
1234 DBG("Trying to start the streaming without a session established");
f73fabfd 1235 ret = htobe32(LTTNG_ERR_UNK);
b8aa1682
JD
1236 }
1237
1238 reply.ret_code = ret;
1239 ret = cmd->sock->ops->sendmsg(cmd->sock, &reply,
1240 sizeof(struct lttcomm_relayd_generic_reply), 0);
1241 if (ret < 0) {
1242 ERR("Relay sending start ack");
1243 }
1244
1245 return ret;
1246}
1247
1d4dfdef
DG
1248/*
1249 * Append padding to the file pointed by the file descriptor fd.
1250 */
1251static int write_padding_to_file(int fd, uint32_t size)
1252{
6cd525e8 1253 ssize_t ret = 0;
1d4dfdef
DG
1254 char *zeros;
1255
1256 if (size == 0) {
1257 goto end;
1258 }
1259
1260 zeros = zmalloc(size);
1261 if (zeros == NULL) {
1262 PERROR("zmalloc zeros for padding");
1263 ret = -1;
1264 goto end;
1265 }
1266
6cd525e8
MD
1267 ret = lttng_write(fd, zeros, size);
1268 if (ret < size) {
1d4dfdef
DG
1269 PERROR("write padding to file");
1270 }
1271
e986c7a1
DG
1272 free(zeros);
1273
1d4dfdef
DG
1274end:
1275 return ret;
1276}
1277
b8aa1682
JD
1278/*
1279 * relay_recv_metadata: receive the metada for the session.
1280 */
1281static
1282int relay_recv_metadata(struct lttcomm_relayd_hdr *recv_hdr,
d3e2ba59 1283 struct relay_command *cmd)
b8aa1682 1284{
f73fabfd 1285 int ret = htobe32(LTTNG_OK);
6cd525e8 1286 ssize_t size_ret;
b8aa1682
JD
1287 struct relay_session *session = cmd->session;
1288 struct lttcomm_relayd_metadata_payload *metadata_struct;
1289 struct relay_stream *metadata_stream;
1290 uint64_t data_size, payload_size;
1291
1292 if (!session) {
1293 ERR("Metadata sent before version check");
1294 ret = -1;
1295 goto end;
1296 }
1297
f6416125
MD
1298 data_size = payload_size = be64toh(recv_hdr->data_size);
1299 if (data_size < sizeof(struct lttcomm_relayd_metadata_payload)) {
1300 ERR("Incorrect data size");
1301 ret = -1;
1302 goto end;
1303 }
1304 payload_size -= sizeof(struct lttcomm_relayd_metadata_payload);
1305
b8aa1682 1306 if (data_buffer_size < data_size) {
d7b3776f 1307 /* In case the realloc fails, we can free the memory */
c617c0c6
MD
1308 char *tmp_data_ptr;
1309
1310 tmp_data_ptr = realloc(data_buffer, data_size);
1311 if (!tmp_data_ptr) {
b8aa1682 1312 ERR("Allocating data buffer");
c617c0c6 1313 free(data_buffer);
b8aa1682
JD
1314 ret = -1;
1315 goto end;
1316 }
c617c0c6 1317 data_buffer = tmp_data_ptr;
b8aa1682
JD
1318 data_buffer_size = data_size;
1319 }
1320 memset(data_buffer, 0, data_size);
77c7c900 1321 DBG2("Relay receiving metadata, waiting for %" PRIu64 " bytes", data_size);
7c5aef62 1322 ret = cmd->sock->ops->recvmsg(cmd->sock, data_buffer, data_size, 0);
b8aa1682 1323 if (ret < 0 || ret != data_size) {
a6cd2b97
DG
1324 if (ret == 0) {
1325 /* Orderly shutdown. Not necessary to print an error. */
1326 DBG("Socket %d did an orderly shutdown", cmd->sock->fd);
1327 } else {
1328 ERR("Relay didn't receive the whole metadata");
1329 }
b8aa1682 1330 ret = -1;
b8aa1682
JD
1331 goto end;
1332 }
1333 metadata_struct = (struct lttcomm_relayd_metadata_payload *) data_buffer;
9d1bbf21
MD
1334
1335 rcu_read_lock();
d3e2ba59
JD
1336 metadata_stream = relay_stream_find_by_id(
1337 be64toh(metadata_struct->stream_id));
b8aa1682
JD
1338 if (!metadata_stream) {
1339 ret = -1;
9d1bbf21 1340 goto end_unlock;
b8aa1682
JD
1341 }
1342
6cd525e8
MD
1343 size_ret = lttng_write(metadata_stream->fd, metadata_struct->payload,
1344 payload_size);
1345 if (size_ret < payload_size) {
b8aa1682
JD
1346 ERR("Relay error writing metadata on file");
1347 ret = -1;
9d1bbf21 1348 goto end_unlock;
b8aa1682 1349 }
1d4dfdef
DG
1350
1351 ret = write_padding_to_file(metadata_stream->fd,
1352 be32toh(metadata_struct->padding_size));
1353 if (ret < 0) {
1354 goto end_unlock;
1355 }
d3e2ba59
JD
1356 metadata_stream->ctf_trace->metadata_received +=
1357 payload_size + be32toh(metadata_struct->padding_size);
1d4dfdef 1358
b8aa1682
JD
1359 DBG2("Relay metadata written");
1360
9d1bbf21 1361end_unlock:
6e3c5836 1362 rcu_read_unlock();
b8aa1682
JD
1363end:
1364 return ret;
1365}
1366
1367/*
1368 * relay_send_version: send relayd version number
1369 */
1370static
1371int relay_send_version(struct lttcomm_relayd_hdr *recv_hdr,
d3e2ba59 1372 struct relay_command *cmd, struct lttng_ht *sessions_ht)
b8aa1682 1373{
7f51dcba 1374 int ret;
092b6259 1375 struct lttcomm_relayd_version reply, msg;
b8aa1682 1376
c5b6f4f0
DG
1377 assert(cmd);
1378
1379 cmd->version_check_done = 1;
b8aa1682 1380
092b6259 1381 /* Get version from the other side. */
7c5aef62 1382 ret = cmd->sock->ops->recvmsg(cmd->sock, &msg, sizeof(msg), 0);
092b6259 1383 if (ret < 0 || ret != sizeof(msg)) {
a6cd2b97
DG
1384 if (ret == 0) {
1385 /* Orderly shutdown. Not necessary to print an error. */
1386 DBG("Socket %d did an orderly shutdown", cmd->sock->fd);
1387 } else {
1388 ERR("Relay failed to receive the version values.");
1389 }
092b6259 1390 ret = -1;
092b6259
DG
1391 goto end;
1392 }
1393
d83a952c
MD
1394 reply.major = RELAYD_VERSION_COMM_MAJOR;
1395 reply.minor = RELAYD_VERSION_COMM_MINOR;
d4519fa3
JD
1396
1397 /* Major versions must be the same */
1398 if (reply.major != be32toh(msg.major)) {
6151a90f
JD
1399 DBG("Incompatible major versions (%u vs %u), deleting session",
1400 reply.major, be32toh(msg.major));
d3e2ba59 1401 relay_delete_session(cmd, sessions_ht);
d4519fa3
JD
1402 ret = 0;
1403 goto end;
1404 }
1405
0f907de1
JD
1406 cmd->major = reply.major;
1407 /* We adapt to the lowest compatible version */
1408 if (reply.minor <= be32toh(msg.minor)) {
1409 cmd->minor = reply.minor;
1410 } else {
1411 cmd->minor = be32toh(msg.minor);
1412 }
1413
6151a90f
JD
1414 reply.major = htobe32(reply.major);
1415 reply.minor = htobe32(reply.minor);
1416 ret = cmd->sock->ops->sendmsg(cmd->sock, &reply,
1417 sizeof(struct lttcomm_relayd_version), 0);
1418 if (ret < 0) {
1419 ERR("Relay sending version");
1420 }
1421
0f907de1
JD
1422 DBG("Version check done using protocol %u.%u", cmd->major,
1423 cmd->minor);
b8aa1682
JD
1424
1425end:
1426 return ret;
1427}
1428
c8f59ee5 1429/*
6d805429 1430 * Check for data pending for a given stream id from the session daemon.
c8f59ee5
DG
1431 */
1432static
6d805429 1433int relay_data_pending(struct lttcomm_relayd_hdr *recv_hdr,
d3e2ba59 1434 struct relay_command *cmd)
c8f59ee5
DG
1435{
1436 struct relay_session *session = cmd->session;
6d805429 1437 struct lttcomm_relayd_data_pending msg;
c8f59ee5
DG
1438 struct lttcomm_relayd_generic_reply reply;
1439 struct relay_stream *stream;
1440 int ret;
c8f59ee5
DG
1441 uint64_t last_net_seq_num, stream_id;
1442
6d805429 1443 DBG("Data pending command received");
c8f59ee5 1444
c5b6f4f0 1445 if (!session || cmd->version_check_done == 0) {
c8f59ee5
DG
1446 ERR("Trying to check for data before version check");
1447 ret = -1;
1448 goto end_no_session;
1449 }
1450
7c5aef62 1451 ret = cmd->sock->ops->recvmsg(cmd->sock, &msg, sizeof(msg), 0);
c8f59ee5 1452 if (ret < sizeof(msg)) {
a6cd2b97
DG
1453 if (ret == 0) {
1454 /* Orderly shutdown. Not necessary to print an error. */
1455 DBG("Socket %d did an orderly shutdown", cmd->sock->fd);
1456 } else {
1457 ERR("Relay didn't receive valid data_pending struct size : %d",
1458 ret);
1459 }
c8f59ee5
DG
1460 ret = -1;
1461 goto end_no_session;
1462 }
1463
1464 stream_id = be64toh(msg.stream_id);
1465 last_net_seq_num = be64toh(msg.last_net_seq_num);
1466
1467 rcu_read_lock();
d3e2ba59 1468 stream = relay_stream_find_by_id(stream_id);
de91f48a 1469 if (stream == NULL) {
c8f59ee5
DG
1470 ret = -1;
1471 goto end_unlock;
1472 }
1473
6d805429 1474 DBG("Data pending for stream id %" PRIu64 " prev_seq %" PRIu64
c8f59ee5
DG
1475 " and last_seq %" PRIu64, stream_id, stream->prev_seq,
1476 last_net_seq_num);
1477
33832e64 1478 /* Avoid wrapping issue */
39df6d9f 1479 if (((int64_t) (stream->prev_seq - last_net_seq_num)) >= 0) {
6d805429 1480 /* Data has in fact been written and is NOT pending */
c8f59ee5 1481 ret = 0;
6d805429
DG
1482 } else {
1483 /* Data still being streamed thus pending */
1484 ret = 1;
c8f59ee5
DG
1485 }
1486
f7079f67
DG
1487 /* Pending check is now done. */
1488 stream->data_pending_check_done = 1;
1489
c8f59ee5
DG
1490end_unlock:
1491 rcu_read_unlock();
1492
1493 reply.ret_code = htobe32(ret);
1494 ret = cmd->sock->ops->sendmsg(cmd->sock, &reply, sizeof(reply), 0);
1495 if (ret < 0) {
6d805429 1496 ERR("Relay data pending ret code failed");
c8f59ee5
DG
1497 }
1498
1499end_no_session:
1500 return ret;
1501}
1502
1503/*
1504 * Wait for the control socket to reach a quiescent state.
1505 *
1506 * Note that for now, when receiving this command from the session daemon, this
1507 * means that every subsequent commands or data received on the control socket
1508 * has been handled. So, this is why we simply return OK here.
1509 */
1510static
1511int relay_quiescent_control(struct lttcomm_relayd_hdr *recv_hdr,
d3e2ba59 1512 struct relay_command *cmd)
c8f59ee5
DG
1513{
1514 int ret;
ad7051c0
DG
1515 uint64_t stream_id;
1516 struct relay_stream *stream;
1517 struct lttng_ht_iter iter;
1518 struct lttcomm_relayd_quiescent_control msg;
c8f59ee5
DG
1519 struct lttcomm_relayd_generic_reply reply;
1520
1521 DBG("Checking quiescent state on control socket");
1522
ad7051c0
DG
1523 if (!cmd->session || cmd->version_check_done == 0) {
1524 ERR("Trying to check for data before version check");
1525 ret = -1;
1526 goto end_no_session;
1527 }
1528
1529 ret = cmd->sock->ops->recvmsg(cmd->sock, &msg, sizeof(msg), 0);
1530 if (ret < sizeof(msg)) {
a6cd2b97
DG
1531 if (ret == 0) {
1532 /* Orderly shutdown. Not necessary to print an error. */
1533 DBG("Socket %d did an orderly shutdown", cmd->sock->fd);
1534 } else {
1535 ERR("Relay didn't receive valid begin data_pending struct size: %d",
1536 ret);
1537 }
ad7051c0
DG
1538 ret = -1;
1539 goto end_no_session;
1540 }
1541
1542 stream_id = be64toh(msg.stream_id);
1543
1544 rcu_read_lock();
d3e2ba59
JD
1545 cds_lfht_for_each_entry(relay_streams_ht->ht, &iter.iter, stream,
1546 stream_n.node) {
ad7051c0
DG
1547 if (stream->stream_handle == stream_id) {
1548 stream->data_pending_check_done = 1;
1549 DBG("Relay quiescent control pending flag set to %" PRIu64,
1550 stream_id);
1551 break;
1552 }
1553 }
1554 rcu_read_unlock();
1555
c8f59ee5
DG
1556 reply.ret_code = htobe32(LTTNG_OK);
1557 ret = cmd->sock->ops->sendmsg(cmd->sock, &reply, sizeof(reply), 0);
1558 if (ret < 0) {
6d805429 1559 ERR("Relay data quiescent control ret code failed");
c8f59ee5
DG
1560 }
1561
ad7051c0 1562end_no_session:
c8f59ee5
DG
1563 return ret;
1564}
1565
f7079f67
DG
1566/*
1567 * Initialize a data pending command. This means that a client is about to ask
1568 * for data pending for each stream he/she holds. Simply iterate over all
1569 * streams of a session and set the data_pending_check_done flag.
1570 *
1571 * This command returns to the client a LTTNG_OK code.
1572 */
1573static
1574int relay_begin_data_pending(struct lttcomm_relayd_hdr *recv_hdr,
d3e2ba59 1575 struct relay_command *cmd)
f7079f67
DG
1576{
1577 int ret;
1578 struct lttng_ht_iter iter;
1579 struct lttcomm_relayd_begin_data_pending msg;
1580 struct lttcomm_relayd_generic_reply reply;
1581 struct relay_stream *stream;
1582 uint64_t session_id;
1583
1584 assert(recv_hdr);
1585 assert(cmd);
f7079f67
DG
1586
1587 DBG("Init streams for data pending");
1588
1589 if (!cmd->session || cmd->version_check_done == 0) {
1590 ERR("Trying to check for data before version check");
1591 ret = -1;
1592 goto end_no_session;
1593 }
1594
1595 ret = cmd->sock->ops->recvmsg(cmd->sock, &msg, sizeof(msg), 0);
1596 if (ret < sizeof(msg)) {
a6cd2b97
DG
1597 if (ret == 0) {
1598 /* Orderly shutdown. Not necessary to print an error. */
1599 DBG("Socket %d did an orderly shutdown", cmd->sock->fd);
1600 } else {
1601 ERR("Relay didn't receive valid begin data_pending struct size: %d",
1602 ret);
1603 }
f7079f67
DG
1604 ret = -1;
1605 goto end_no_session;
1606 }
1607
1608 session_id = be64toh(msg.session_id);
1609
1610 /*
1611 * Iterate over all streams to set the begin data pending flag. For now, the
1612 * streams are indexed by stream handle so we have to iterate over all
1613 * streams to find the one associated with the right session_id.
1614 */
1615 rcu_read_lock();
d3e2ba59
JD
1616 cds_lfht_for_each_entry(relay_streams_ht->ht, &iter.iter, stream,
1617 stream_n.node) {
f7079f67
DG
1618 if (stream->session->id == session_id) {
1619 stream->data_pending_check_done = 0;
1620 DBG("Set begin data pending flag to stream %" PRIu64,
1621 stream->stream_handle);
1622 }
1623 }
1624 rcu_read_unlock();
1625
1626 /* All good, send back reply. */
1627 reply.ret_code = htobe32(LTTNG_OK);
1628
1629 ret = cmd->sock->ops->sendmsg(cmd->sock, &reply, sizeof(reply), 0);
1630 if (ret < 0) {
1631 ERR("Relay begin data pending send reply failed");
1632 }
1633
1634end_no_session:
1635 return ret;
1636}
1637
1638/*
1639 * End data pending command. This will check, for a given session id, if each
1640 * stream associated with it has its data_pending_check_done flag set. If not,
1641 * this means that the client lost track of the stream but the data is still
1642 * being streamed on our side. In this case, we inform the client that data is
1643 * inflight.
1644 *
1645 * Return to the client if there is data in flight or not with a ret_code.
1646 */
1647static
1648int relay_end_data_pending(struct lttcomm_relayd_hdr *recv_hdr,
d3e2ba59 1649 struct relay_command *cmd)
f7079f67
DG
1650{
1651 int ret;
1652 struct lttng_ht_iter iter;
1653 struct lttcomm_relayd_end_data_pending msg;
1654 struct lttcomm_relayd_generic_reply reply;
1655 struct relay_stream *stream;
1656 uint64_t session_id;
1657 uint32_t is_data_inflight = 0;
1658
1659 assert(recv_hdr);
1660 assert(cmd);
f7079f67
DG
1661
1662 DBG("End data pending command");
1663
1664 if (!cmd->session || cmd->version_check_done == 0) {
1665 ERR("Trying to check for data before version check");
1666 ret = -1;
1667 goto end_no_session;
1668 }
1669
1670 ret = cmd->sock->ops->recvmsg(cmd->sock, &msg, sizeof(msg), 0);
1671 if (ret < sizeof(msg)) {
a6cd2b97
DG
1672 if (ret == 0) {
1673 /* Orderly shutdown. Not necessary to print an error. */
1674 DBG("Socket %d did an orderly shutdown", cmd->sock->fd);
1675 } else {
1676 ERR("Relay didn't receive valid end data_pending struct size: %d",
1677 ret);
1678 }
f7079f67
DG
1679 ret = -1;
1680 goto end_no_session;
1681 }
1682
1683 session_id = be64toh(msg.session_id);
1684
1685 /* Iterate over all streams to see if the begin data pending flag is set. */
1686 rcu_read_lock();
d3e2ba59
JD
1687 cds_lfht_for_each_entry(relay_streams_ht->ht, &iter.iter, stream,
1688 stream_n.node) {
f7079f67
DG
1689 if (stream->session->id == session_id &&
1690 !stream->data_pending_check_done) {
1691 is_data_inflight = 1;
1692 DBG("Data is still in flight for stream %" PRIu64,
1693 stream->stream_handle);
1694 break;
1695 }
1696 }
1697 rcu_read_unlock();
1698
1699 /* All good, send back reply. */
1700 reply.ret_code = htobe32(is_data_inflight);
1701
1702 ret = cmd->sock->ops->sendmsg(cmd->sock, &reply, sizeof(reply), 0);
1703 if (ret < 0) {
1704 ERR("Relay end data pending send reply failed");
1705 }
1706
1707end_no_session:
1708 return ret;
1709}
1710
1c20f0e2
JD
1711/*
1712 * Receive an index for a specific stream.
1713 *
1714 * Return 0 on success else a negative value.
1715 */
1716static
1717int relay_recv_index(struct lttcomm_relayd_hdr *recv_hdr,
0a6518b0 1718 struct relay_command *cmd)
1c20f0e2
JD
1719{
1720 int ret, send_ret, index_created = 0;
1721 struct relay_session *session = cmd->session;
1722 struct lttcomm_relayd_index index_info;
1723 struct relay_index *index, *wr_index = NULL;
1724 struct lttcomm_relayd_generic_reply reply;
1725 struct relay_stream *stream;
1726 uint64_t net_seq_num;
1727
1728 assert(cmd);
1c20f0e2
JD
1729
1730 DBG("Relay receiving index");
1731
1732 if (!session || cmd->version_check_done == 0) {
1733 ERR("Trying to close a stream before version check");
1734 ret = -1;
1735 goto end_no_session;
1736 }
1737
1738 ret = cmd->sock->ops->recvmsg(cmd->sock, &index_info,
1739 sizeof(index_info), 0);
1740 if (ret < sizeof(index_info)) {
1741 if (ret == 0) {
1742 /* Orderly shutdown. Not necessary to print an error. */
1743 DBG("Socket %d did an orderly shutdown", cmd->sock->fd);
1744 } else {
1745 ERR("Relay didn't receive valid index struct size : %d", ret);
1746 }
1747 ret = -1;
1748 goto end_no_session;
1749 }
1750
1751 net_seq_num = be64toh(index_info.net_seq_num);
1752
1753 rcu_read_lock();
d3e2ba59 1754 stream = relay_stream_find_by_id(be64toh(index_info.relay_stream_id));
1c20f0e2
JD
1755 if (!stream) {
1756 ret = -1;
1757 goto end_rcu_unlock;
1758 }
1759
d3e2ba59
JD
1760 /* Live beacon handling */
1761 if (index_info.packet_size == 0) {
1762 DBG("Received live beacon for stream %" PRIu64, stream->stream_handle);
1763
1764 /*
1765 * Only flag a stream inactive when it has already received data.
1766 */
1767 if (stream->total_index_received > 0) {
1768 stream->beacon_ts_end = be64toh(index_info.timestamp_end);
1769 }
1770 ret = 0;
1771 goto end_rcu_unlock;
1772 } else {
1773 stream->beacon_ts_end = -1ULL;
1774 }
1775
0a6518b0 1776 index = relay_index_find(stream->stream_handle, net_seq_num);
1c20f0e2
JD
1777 if (!index) {
1778 /* A successful creation will add the object to the HT. */
1779 index = relay_index_create(stream->stream_handle, net_seq_num);
1780 if (!index) {
1781 goto end_rcu_unlock;
1782 }
1783 index_created = 1;
1784 }
1785
1786 copy_index_control_data(index, &index_info);
1787
1788 if (index_created) {
1789 /*
1790 * Try to add the relay index object to the hash table. If an object
1791 * already exist, destroy back the index created, set the data in this
1792 * object and write it on disk.
1793 */
0a6518b0 1794 relay_index_add(index, &wr_index);
1c20f0e2
JD
1795 if (wr_index) {
1796 copy_index_control_data(wr_index, &index_info);
1797 free(index);
1798 }
1799 } else {
1800 /* The index already exists so write it on disk. */
1801 wr_index = index;
1802 }
1803
1804 /* Do we have a writable ready index to write on disk. */
1805 if (wr_index) {
1806 /* Starting at 2.4, create the index file if none available. */
1807 if (cmd->minor >= 4 && stream->index_fd < 0) {
1808 ret = index_create_file(stream->path_name, stream->channel_name,
1809 relayd_uid, relayd_gid, stream->tracefile_size,
1810 stream->tracefile_count_current);
1811 if (ret < 0) {
1812 goto end_rcu_unlock;
1813 }
1814 stream->index_fd = ret;
1815 }
1816
0a6518b0 1817 ret = relay_index_write(wr_index->fd, wr_index);
1c20f0e2
JD
1818 if (ret < 0) {
1819 goto end_rcu_unlock;
1820 }
d3e2ba59 1821 stream->total_index_received++;
1c20f0e2
JD
1822 }
1823
1824end_rcu_unlock:
1825 rcu_read_unlock();
1826
1827 if (ret < 0) {
1828 reply.ret_code = htobe32(LTTNG_ERR_UNK);
1829 } else {
1830 reply.ret_code = htobe32(LTTNG_OK);
1831 }
1832 send_ret = cmd->sock->ops->sendmsg(cmd->sock, &reply, sizeof(reply), 0);
1833 if (send_ret < 0) {
1834 ERR("Relay sending close index id reply");
1835 ret = send_ret;
1836 }
1837
1838end_no_session:
1839 return ret;
1840}
1841
b8aa1682 1842/*
d3e2ba59 1843 * Process the commands received on the control socket
b8aa1682
JD
1844 */
1845static
1846int relay_process_control(struct lttcomm_relayd_hdr *recv_hdr,
d3e2ba59 1847 struct relay_command *cmd, struct relay_local_data *ctx)
b8aa1682
JD
1848{
1849 int ret = 0;
1850
1851 switch (be32toh(recv_hdr->cmd)) {
b8aa1682 1852 case RELAYD_CREATE_SESSION:
d3e2ba59 1853 ret = relay_create_session(recv_hdr, cmd, ctx->sessions_ht);
b8aa1682 1854 break;
b8aa1682 1855 case RELAYD_ADD_STREAM:
d3e2ba59 1856 ret = relay_add_stream(recv_hdr, cmd, ctx->sessions_ht);
b8aa1682
JD
1857 break;
1858 case RELAYD_START_DATA:
1859 ret = relay_start(recv_hdr, cmd);
1860 break;
1861 case RELAYD_SEND_METADATA:
d3e2ba59 1862 ret = relay_recv_metadata(recv_hdr, cmd);
b8aa1682
JD
1863 break;
1864 case RELAYD_VERSION:
d3e2ba59 1865 ret = relay_send_version(recv_hdr, cmd, ctx->sessions_ht);
b8aa1682 1866 break;
173af62f 1867 case RELAYD_CLOSE_STREAM:
92c6ca54 1868 ret = relay_close_stream(recv_hdr, cmd);
173af62f 1869 break;
6d805429 1870 case RELAYD_DATA_PENDING:
d3e2ba59 1871 ret = relay_data_pending(recv_hdr, cmd);
c8f59ee5
DG
1872 break;
1873 case RELAYD_QUIESCENT_CONTROL:
d3e2ba59 1874 ret = relay_quiescent_control(recv_hdr, cmd);
c8f59ee5 1875 break;
f7079f67 1876 case RELAYD_BEGIN_DATA_PENDING:
d3e2ba59 1877 ret = relay_begin_data_pending(recv_hdr, cmd);
f7079f67
DG
1878 break;
1879 case RELAYD_END_DATA_PENDING:
d3e2ba59 1880 ret = relay_end_data_pending(recv_hdr, cmd);
f7079f67 1881 break;
1c20f0e2 1882 case RELAYD_SEND_INDEX:
0a6518b0 1883 ret = relay_recv_index(recv_hdr, cmd);
1c20f0e2 1884 break;
b8aa1682
JD
1885 case RELAYD_UPDATE_SYNC_INFO:
1886 default:
1887 ERR("Received unknown command (%u)", be32toh(recv_hdr->cmd));
1888 relay_unknown_command(cmd);
1889 ret = -1;
1890 goto end;
1891 }
1892
1893end:
1894 return ret;
1895}
1896
7d2f7452
DG
1897/*
1898 * Handle index for a data stream.
1899 *
1900 * RCU read side lock MUST be acquired.
1901 *
1902 * Return 0 on success else a negative value.
1903 */
1904static int handle_index_data(struct relay_stream *stream, uint64_t net_seq_num,
1905 int rotate_index)
1906{
1907 int ret = 0, index_created = 0;
1908 uint64_t stream_id, data_offset;
1909 struct relay_index *index, *wr_index = NULL;
1910
1911 assert(stream);
1912
1913 stream_id = stream->stream_handle;
1914 /* Get data offset because we are about to update the index. */
1915 data_offset = htobe64(stream->tracefile_size_current);
1916
1917 /*
1918 * Lookup for an existing index for that stream id/sequence number. If on
1919 * exists, the control thread already received the data for it thus we need
1920 * to write it on disk.
1921 */
1922 index = relay_index_find(stream_id, net_seq_num);
1923 if (!index) {
1924 /* A successful creation will add the object to the HT. */
1925 index = relay_index_create(stream_id, net_seq_num);
1926 if (!index) {
1927 ret = -1;
1928 goto error;
1929 }
1930 index_created = 1;
1931 }
1932
1933 if (rotate_index || stream->index_fd < 0) {
1934 index->to_close_fd = stream->index_fd;
1935 ret = index_create_file(stream->path_name, stream->channel_name,
1936 relayd_uid, relayd_gid, stream->tracefile_size,
1937 stream->tracefile_count_current);
1938 if (ret < 0) {
1939 /* This will close the stream's index fd if one. */
1940 relay_index_free_safe(index);
1941 goto error;
1942 }
1943 stream->index_fd = ret;
1944 }
1945 index->fd = stream->index_fd;
1946 index->index_data.offset = data_offset;
1947
1948 if (index_created) {
1949 /*
1950 * Try to add the relay index object to the hash table. If an object
1951 * already exist, destroy back the index created and set the data.
1952 */
1953 relay_index_add(index, &wr_index);
1954 if (wr_index) {
1955 /* Copy back data from the created index. */
1956 wr_index->fd = index->fd;
1957 wr_index->to_close_fd = index->to_close_fd;
1958 wr_index->index_data.offset = data_offset;
1959 free(index);
1960 }
1961 } else {
1962 /* The index already exists so write it on disk. */
1963 wr_index = index;
1964 }
1965
1966 /* Do we have a writable ready index to write on disk. */
1967 if (wr_index) {
1968 ret = relay_index_write(wr_index->fd, wr_index);
1969 if (ret < 0) {
1970 goto error;
1971 }
1972 stream->total_index_received++;
1973 }
1974
1975error:
1976 return ret;
1977}
1978
b8aa1682
JD
1979/*
1980 * relay_process_data: Process the data received on the data socket
1981 */
1982static
0a6518b0 1983int relay_process_data(struct relay_command *cmd)
b8aa1682 1984{
7d2f7452 1985 int ret = 0, rotate_index = 0;
6cd525e8 1986 ssize_t size_ret;
b8aa1682
JD
1987 struct relay_stream *stream;
1988 struct lttcomm_relayd_data_hdr data_hdr;
7d2f7452 1989 uint64_t stream_id;
173af62f 1990 uint64_t net_seq_num;
b8aa1682
JD
1991 uint32_t data_size;
1992
1993 ret = cmd->sock->ops->recvmsg(cmd->sock, &data_hdr,
7c5aef62 1994 sizeof(struct lttcomm_relayd_data_hdr), 0);
b8aa1682 1995 if (ret <= 0) {
a6cd2b97
DG
1996 if (ret == 0) {
1997 /* Orderly shutdown. Not necessary to print an error. */
1998 DBG("Socket %d did an orderly shutdown", cmd->sock->fd);
1999 } else {
2000 ERR("Unable to receive data header on sock %d", cmd->sock->fd);
2001 }
b8aa1682
JD
2002 ret = -1;
2003 goto end;
2004 }
2005
2006 stream_id = be64toh(data_hdr.stream_id);
9d1bbf21
MD
2007
2008 rcu_read_lock();
d3e2ba59 2009 stream = relay_stream_find_by_id(stream_id);
b8aa1682
JD
2010 if (!stream) {
2011 ret = -1;
1c20f0e2 2012 goto end_rcu_unlock;
b8aa1682
JD
2013 }
2014
2015 data_size = be32toh(data_hdr.data_size);
2016 if (data_buffer_size < data_size) {
c617c0c6
MD
2017 char *tmp_data_ptr;
2018
2019 tmp_data_ptr = realloc(data_buffer, data_size);
2020 if (!tmp_data_ptr) {
b8aa1682 2021 ERR("Allocating data buffer");
c617c0c6 2022 free(data_buffer);
b8aa1682 2023 ret = -1;
1c20f0e2 2024 goto end_rcu_unlock;
b8aa1682 2025 }
c617c0c6 2026 data_buffer = tmp_data_ptr;
b8aa1682
JD
2027 data_buffer_size = data_size;
2028 }
2029 memset(data_buffer, 0, data_size);
2030
173af62f
DG
2031 net_seq_num = be64toh(data_hdr.net_seq_num);
2032
77c7c900 2033 DBG3("Receiving data of size %u for stream id %" PRIu64 " seqnum %" PRIu64,
173af62f 2034 data_size, stream_id, net_seq_num);
7c5aef62 2035 ret = cmd->sock->ops->recvmsg(cmd->sock, data_buffer, data_size, 0);
b8aa1682 2036 if (ret <= 0) {
a6cd2b97
DG
2037 if (ret == 0) {
2038 /* Orderly shutdown. Not necessary to print an error. */
2039 DBG("Socket %d did an orderly shutdown", cmd->sock->fd);
2040 }
b8aa1682 2041 ret = -1;
1c20f0e2 2042 goto end_rcu_unlock;
b8aa1682
JD
2043 }
2044
1c20f0e2 2045 /* Check if a rotation is needed. */
0f907de1
JD
2046 if (stream->tracefile_size > 0 &&
2047 (stream->tracefile_size_current + data_size) >
2048 stream->tracefile_size) {
6b6b9a5a
JD
2049 struct relay_viewer_stream *vstream;
2050 uint64_t new_id;
2051
2052 new_id = (stream->tracefile_count_current + 1) %
2053 stream->tracefile_count;
2054 /*
2055 * When we wrap-around back to 0, we start overwriting old
2056 * trace data.
2057 */
2058 if (!stream->tracefile_overwrite && new_id == 0) {
2059 stream->tracefile_overwrite = 1;
2060 }
2061 pthread_mutex_lock(&stream->viewer_stream_rotation_lock);
2062 if (stream->tracefile_overwrite) {
2063 stream->oldest_tracefile_id =
2064 (stream->oldest_tracefile_id + 1) %
2065 stream->tracefile_count;
2066 }
2067 vstream = live_find_viewer_stream_by_id(stream->stream_handle);
2068 if (vstream) {
2069 /*
2070 * The viewer is reading a file about to be
2071 * overwritten. Close the FDs it is
2072 * currently using and let it handle the fault.
2073 */
2074 if (vstream->tracefile_count_current == new_id) {
cef0f7d5 2075 pthread_mutex_lock(&vstream->overwrite_lock);
6b6b9a5a 2076 vstream->abort_flag = 1;
cef0f7d5 2077 pthread_mutex_unlock(&vstream->overwrite_lock);
6b6b9a5a
JD
2078 DBG("Streaming side setting abort_flag on stream %s_%lu\n",
2079 stream->channel_name, new_id);
2080 } else if (vstream->tracefile_count_current ==
2081 stream->tracefile_count_current) {
2082 /*
2083 * The reader and writer were in the
2084 * same trace file, inform the viewer
2085 * that no new index will ever be added
2086 * to this file.
2087 */
2088 vstream->close_write_flag = 1;
2089 }
2090 }
1c20f0e2
JD
2091 ret = utils_rotate_stream_file(stream->path_name, stream->channel_name,
2092 stream->tracefile_size, stream->tracefile_count,
2093 relayd_uid, relayd_gid, stream->fd,
2094 &(stream->tracefile_count_current), &stream->fd);
cef0f7d5 2095 stream->total_index_received = 0;
6b6b9a5a 2096 pthread_mutex_unlock(&stream->viewer_stream_rotation_lock);
0f907de1 2097 if (ret < 0) {
1c20f0e2
JD
2098 ERR("Rotating stream output file");
2099 goto end_rcu_unlock;
0f907de1 2100 }
a6976990
DG
2101 /* Reset current size because we just perform a stream rotation. */
2102 stream->tracefile_size_current = 0;
1c20f0e2
JD
2103 rotate_index = 1;
2104 }
2105
1c20f0e2 2106 /*
7d2f7452
DG
2107 * Index are handled in protocol version 2.4 and above. Also, snapshot and
2108 * index are NOT supported.
1c20f0e2 2109 */
7d2f7452
DG
2110 if (stream->session->minor >= 4 && !stream->session->snapshot) {
2111 ret = handle_index_data(stream, net_seq_num, rotate_index);
1c20f0e2 2112 if (ret < 0) {
1c20f0e2
JD
2113 goto end_rcu_unlock;
2114 }
1c20f0e2
JD
2115 }
2116
7d2f7452 2117 /* Write data to stream output fd. */
6cd525e8
MD
2118 size_ret = lttng_write(stream->fd, data_buffer, data_size);
2119 if (size_ret < data_size) {
b8aa1682
JD
2120 ERR("Relay error writing data to file");
2121 ret = -1;
1c20f0e2 2122 goto end_rcu_unlock;
b8aa1682 2123 }
1d4dfdef 2124
5ab7344e
JD
2125 DBG2("Relay wrote %d bytes to tracefile for stream id %" PRIu64,
2126 ret, stream->stream_handle);
2127
1d4dfdef
DG
2128 ret = write_padding_to_file(stream->fd, be32toh(data_hdr.padding_size));
2129 if (ret < 0) {
1c20f0e2 2130 goto end_rcu_unlock;
1d4dfdef 2131 }
1c20f0e2 2132 stream->tracefile_size_current += data_size + be32toh(data_hdr.padding_size);
1d4dfdef 2133
173af62f
DG
2134 stream->prev_seq = net_seq_num;
2135
2136 /* Check if we need to close the FD */
2137 if (close_stream_check(stream)) {
c07d8a78 2138 destroy_stream(stream);
173af62f
DG
2139 }
2140
1c20f0e2 2141end_rcu_unlock:
9d1bbf21 2142 rcu_read_unlock();
b8aa1682
JD
2143end:
2144 return ret;
2145}
2146
2147static
9d1bbf21 2148void relay_cleanup_poll_connection(struct lttng_poll_event *events, int pollfd)
b8aa1682
JD
2149{
2150 int ret;
2151
b8aa1682
JD
2152 lttng_poll_del(events, pollfd);
2153
2154 ret = close(pollfd);
2155 if (ret < 0) {
2156 ERR("Closing pollfd %d", pollfd);
2157 }
2158}
2159
2160static
2161int relay_add_connection(int fd, struct lttng_poll_event *events,
2162 struct lttng_ht *relay_connections_ht)
2163{
b8aa1682 2164 struct relay_command *relay_connection;
6cd525e8 2165 ssize_t ret;
b8aa1682
JD
2166
2167 relay_connection = zmalloc(sizeof(struct relay_command));
2168 if (relay_connection == NULL) {
2169 PERROR("Relay command zmalloc");
9d1bbf21 2170 goto error;
b8aa1682 2171 }
6cd525e8
MD
2172 ret = lttng_read(fd, relay_connection, sizeof(struct relay_command));
2173 if (ret < sizeof(struct relay_command)) {
b8aa1682 2174 PERROR("read relay cmd pipe");
9d1bbf21 2175 goto error_read;
b8aa1682
JD
2176 }
2177
c07d8a78
DG
2178 /*
2179 * Only used by the control side and the reference is copied inside each
2180 * stream from that connection. Thus a destroy HT must be done after every
2181 * stream has been destroyed.
2182 */
2183 if (relay_connection->type == RELAY_CONTROL) {
2184 relay_connection->ctf_traces_ht = lttng_ht_new(0,
2185 LTTNG_HT_TYPE_STRING);
2186 if (!relay_connection->ctf_traces_ht) {
2187 goto error_read;
2188 }
d3e2ba59
JD
2189 }
2190
b8aa1682
JD
2191 lttng_ht_node_init_ulong(&relay_connection->sock_n,
2192 (unsigned long) relay_connection->sock->fd);
9d1bbf21 2193 rcu_read_lock();
b8aa1682
JD
2194 lttng_ht_add_unique_ulong(relay_connections_ht,
2195 &relay_connection->sock_n);
9d1bbf21
MD
2196 rcu_read_unlock();
2197 return lttng_poll_add(events,
b8aa1682
JD
2198 relay_connection->sock->fd,
2199 LPOLLIN | LPOLLRDHUP);
2200
9d1bbf21
MD
2201error_read:
2202 free(relay_connection);
2203error:
2204 return -1;
2205}
2206
2207static
2208void deferred_free_connection(struct rcu_head *head)
2209{
2210 struct relay_command *relay_connection =
2211 caa_container_of(head, struct relay_command, rcu_node);
5b6d8097
DG
2212
2213 lttcomm_destroy_sock(relay_connection->sock);
9d1bbf21
MD
2214 free(relay_connection);
2215}
2216
2217static
2218void relay_del_connection(struct lttng_ht *relay_connections_ht,
d3e2ba59
JD
2219 struct lttng_ht_iter *iter, struct relay_command *relay_connection,
2220 struct lttng_ht *sessions_ht)
9d1bbf21
MD
2221{
2222 int ret;
2223
2224 ret = lttng_ht_del(relay_connections_ht, iter);
2225 assert(!ret);
c07d8a78 2226
9d1bbf21 2227 if (relay_connection->type == RELAY_CONTROL) {
d3e2ba59 2228 relay_delete_session(relay_connection, sessions_ht);
c07d8a78 2229 lttng_ht_destroy(relay_connection->ctf_traces_ht);
9d1bbf21 2230 }
5b6d8097 2231
c07d8a78 2232 call_rcu(&relay_connection->rcu_node, deferred_free_connection);
b8aa1682
JD
2233}
2234
2235/*
2236 * This thread does the actual work
2237 */
2238static
2239void *relay_thread_worker(void *data)
2240{
beaad64c
DG
2241 int ret, err = -1, last_seen_data_fd = -1;
2242 uint32_t nb_fd;
b8aa1682
JD
2243 struct relay_command *relay_connection;
2244 struct lttng_poll_event events;
2245 struct lttng_ht *relay_connections_ht;
2246 struct lttng_ht_node_ulong *node;
2247 struct lttng_ht_iter iter;
b8aa1682 2248 struct lttcomm_relayd_hdr recv_hdr;
d3e2ba59
JD
2249 struct relay_local_data *relay_ctx = (struct relay_local_data *) data;
2250 struct lttng_ht *sessions_ht = relay_ctx->sessions_ht;
b8aa1682
JD
2251
2252 DBG("[thread] Relay worker started");
2253
9d1bbf21
MD
2254 rcu_register_thread();
2255
55706a7d
MD
2256 health_register(health_relayd, HEALTH_RELAYD_TYPE_WORKER);
2257
f385ae0a
MD
2258 health_code_update();
2259
b8aa1682
JD
2260 /* table of connections indexed on socket */
2261 relay_connections_ht = lttng_ht_new(0, LTTNG_HT_TYPE_ULONG);
095a4ae5
MD
2262 if (!relay_connections_ht) {
2263 goto relay_connections_ht_error;
2264 }
b8aa1682 2265
1c20f0e2
JD
2266 /* Tables of received indexes indexed by index handle and net_seq_num. */
2267 indexes_ht = lttng_ht_new(0, LTTNG_HT_TYPE_TWO_U64);
2268 if (!indexes_ht) {
2269 goto indexes_ht_error;
2270 }
2271
b8aa1682
JD
2272 ret = create_thread_poll_set(&events, 2);
2273 if (ret < 0) {
2274 goto error_poll_create;
2275 }
2276
2277 ret = lttng_poll_add(&events, relay_cmd_pipe[0], LPOLLIN | LPOLLRDHUP);
2278 if (ret < 0) {
2279 goto error;
2280 }
2281
beaad64c 2282restart:
b8aa1682 2283 while (1) {
beaad64c
DG
2284 int idx = -1, i, seen_control = 0, last_notdel_data_fd = -1;
2285
f385ae0a
MD
2286 health_code_update();
2287
b8aa1682 2288 /* Infinite blocking call, waiting for transmission */
87c1611d 2289 DBG3("Relayd worker thread polling...");
f385ae0a 2290 health_poll_entry();
b8aa1682 2291 ret = lttng_poll_wait(&events, -1);
f385ae0a 2292 health_poll_exit();
b8aa1682
JD
2293 if (ret < 0) {
2294 /*
2295 * Restart interrupted system call.
2296 */
2297 if (errno == EINTR) {
2298 goto restart;
2299 }
2300 goto error;
2301 }
2302
0d9c5d77
DG
2303 nb_fd = ret;
2304
beaad64c
DG
2305 /*
2306 * Process control. The control connection is prioritised so we don't
2307 * starve it with high throughout put tracing data on the data
2308 * connection.
2309 */
b8aa1682
JD
2310 for (i = 0; i < nb_fd; i++) {
2311 /* Fetch once the poll data */
beaad64c
DG
2312 uint32_t revents = LTTNG_POLL_GETEV(&events, i);
2313 int pollfd = LTTNG_POLL_GETFD(&events, i);
b8aa1682 2314
f385ae0a
MD
2315 health_code_update();
2316
b8aa1682
JD
2317 /* Thread quit pipe has been closed. Killing thread. */
2318 ret = check_thread_quit_pipe(pollfd, revents);
2319 if (ret) {
095a4ae5
MD
2320 err = 0;
2321 goto exit;
b8aa1682
JD
2322 }
2323
2324 /* Inspect the relay cmd pipe for new connection */
2325 if (pollfd == relay_cmd_pipe[0]) {
2326 if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
2327 ERR("Relay pipe error");
2328 goto error;
2329 } else if (revents & LPOLLIN) {
2330 DBG("Relay command received");
2331 ret = relay_add_connection(relay_cmd_pipe[0],
2332 &events, relay_connections_ht);
2333 if (ret < 0) {
2334 goto error;
2335 }
2336 }
beaad64c 2337 } else if (revents) {
9d1bbf21 2338 rcu_read_lock();
b8aa1682
JD
2339 lttng_ht_lookup(relay_connections_ht,
2340 (void *)((unsigned long) pollfd),
2341 &iter);
2342 node = lttng_ht_iter_get_node_ulong(&iter);
2343 if (node == NULL) {
2344 DBG2("Relay sock %d not found", pollfd);
9d1bbf21 2345 rcu_read_unlock();
b8aa1682
JD
2346 goto error;
2347 }
2348 relay_connection = caa_container_of(node,
2349 struct relay_command, sock_n);
2350
2351 if (revents & (LPOLLERR)) {
2352 ERR("POLL ERROR");
9d1bbf21
MD
2353 relay_cleanup_poll_connection(&events, pollfd);
2354 relay_del_connection(relay_connections_ht,
d3e2ba59 2355 &iter, relay_connection, sessions_ht);
beaad64c
DG
2356 if (last_seen_data_fd == pollfd) {
2357 last_seen_data_fd = last_notdel_data_fd;
2358 }
b8aa1682
JD
2359 } else if (revents & (LPOLLHUP | LPOLLRDHUP)) {
2360 DBG("Socket %d hung up", pollfd);
9d1bbf21
MD
2361 relay_cleanup_poll_connection(&events, pollfd);
2362 relay_del_connection(relay_connections_ht,
d3e2ba59 2363 &iter, relay_connection, sessions_ht);
beaad64c
DG
2364 if (last_seen_data_fd == pollfd) {
2365 last_seen_data_fd = last_notdel_data_fd;
2366 }
b8aa1682
JD
2367 } else if (revents & LPOLLIN) {
2368 /* control socket */
2369 if (relay_connection->type == RELAY_CONTROL) {
2370 ret = relay_connection->sock->ops->recvmsg(
2371 relay_connection->sock, &recv_hdr,
7c5aef62 2372 sizeof(struct lttcomm_relayd_hdr), 0);
b8aa1682
JD
2373 /* connection closed */
2374 if (ret <= 0) {
9d1bbf21
MD
2375 relay_cleanup_poll_connection(&events, pollfd);
2376 relay_del_connection(relay_connections_ht,
d3e2ba59 2377 &iter, relay_connection, sessions_ht);
b8aa1682
JD
2378 DBG("Control connection closed with %d", pollfd);
2379 } else {
2380 if (relay_connection->session) {
77c7c900 2381 DBG2("Relay worker receiving data for session : %" PRIu64,
b8aa1682
JD
2382 relay_connection->session->id);
2383 }
2384 ret = relay_process_control(&recv_hdr,
d3e2ba59 2385 relay_connection, relay_ctx);
b8aa1682 2386 if (ret < 0) {
beaad64c 2387 /* Clear the session on error. */
9d1bbf21
MD
2388 relay_cleanup_poll_connection(&events, pollfd);
2389 relay_del_connection(relay_connections_ht,
d3e2ba59 2390 &iter, relay_connection, sessions_ht);
b8aa1682
JD
2391 DBG("Connection closed with %d", pollfd);
2392 }
beaad64c 2393 seen_control = 1;
b8aa1682 2394 }
beaad64c
DG
2395 } else {
2396 /*
2397 * Flag the last seen data fd not deleted. It will be
2398 * used as the last seen fd if any fd gets deleted in
2399 * this first loop.
2400 */
2401 last_notdel_data_fd = pollfd;
2402 }
2403 }
2404 rcu_read_unlock();
2405 }
2406 }
2407
2408 /*
2409 * The last loop handled a control request, go back to poll to make
2410 * sure we prioritise the control socket.
2411 */
2412 if (seen_control) {
2413 continue;
2414 }
2415
2416 if (last_seen_data_fd >= 0) {
2417 for (i = 0; i < nb_fd; i++) {
2418 int pollfd = LTTNG_POLL_GETFD(&events, i);
f385ae0a
MD
2419
2420 health_code_update();
2421
beaad64c
DG
2422 if (last_seen_data_fd == pollfd) {
2423 idx = i;
2424 break;
2425 }
2426 }
2427 }
2428
2429 /* Process data connection. */
2430 for (i = idx + 1; i < nb_fd; i++) {
2431 /* Fetch the poll data. */
2432 uint32_t revents = LTTNG_POLL_GETEV(&events, i);
2433 int pollfd = LTTNG_POLL_GETFD(&events, i);
2434
f385ae0a
MD
2435 health_code_update();
2436
beaad64c
DG
2437 /* Skip the command pipe. It's handled in the first loop. */
2438 if (pollfd == relay_cmd_pipe[0]) {
2439 continue;
2440 }
2441
2442 if (revents) {
2443 rcu_read_lock();
2444 lttng_ht_lookup(relay_connections_ht,
2445 (void *)((unsigned long) pollfd),
2446 &iter);
2447 node = lttng_ht_iter_get_node_ulong(&iter);
2448 if (node == NULL) {
2449 /* Skip it. Might be removed before. */
2450 rcu_read_unlock();
2451 continue;
2452 }
2453 relay_connection = caa_container_of(node,
2454 struct relay_command, sock_n);
2455
2456 if (revents & LPOLLIN) {
2457 if (relay_connection->type != RELAY_DATA) {
2458 continue;
2459 }
2460
0a6518b0 2461 ret = relay_process_data(relay_connection);
beaad64c
DG
2462 /* connection closed */
2463 if (ret < 0) {
2464 relay_cleanup_poll_connection(&events, pollfd);
2465 relay_del_connection(relay_connections_ht,
d3e2ba59 2466 &iter, relay_connection, sessions_ht);
beaad64c
DG
2467 DBG("Data connection closed with %d", pollfd);
2468 /*
2469 * Every goto restart call sets the last seen fd where
2470 * here we don't really care since we gracefully
2471 * continue the loop after the connection is deleted.
2472 */
2473 } else {
2474 /* Keep last seen port. */
2475 last_seen_data_fd = pollfd;
2476 rcu_read_unlock();
2477 goto restart;
b8aa1682
JD
2478 }
2479 }
9d1bbf21 2480 rcu_read_unlock();
b8aa1682
JD
2481 }
2482 }
beaad64c 2483 last_seen_data_fd = -1;
b8aa1682
JD
2484 }
2485
f385ae0a
MD
2486 /* Normal exit, no error */
2487 ret = 0;
2488
095a4ae5 2489exit:
b8aa1682
JD
2490error:
2491 lttng_poll_clean(&events);
2492
2493 /* empty the hash table and free the memory */
9d1bbf21 2494 rcu_read_lock();
b8aa1682 2495 cds_lfht_for_each_entry(relay_connections_ht->ht, &iter.iter, node, node) {
f385ae0a
MD
2496 health_code_update();
2497
b8aa1682
JD
2498 node = lttng_ht_iter_get_node_ulong(&iter);
2499 if (node) {
2500 relay_connection = caa_container_of(node,
2501 struct relay_command, sock_n);
9d1bbf21 2502 relay_del_connection(relay_connections_ht,
d3e2ba59 2503 &iter, relay_connection, sessions_ht);
b8aa1682 2504 }
b8aa1682 2505 }
94d49140 2506 rcu_read_unlock();
7d2f7452
DG
2507error_poll_create:
2508 lttng_ht_destroy(indexes_ht);
1c20f0e2 2509indexes_ht_error:
b8aa1682 2510 lttng_ht_destroy(relay_connections_ht);
095a4ae5 2511relay_connections_ht_error:
6620da75
DG
2512 /* Close relay cmd pipes */
2513 utils_close_pipe(relay_cmd_pipe);
095a4ae5
MD
2514 if (err) {
2515 DBG("Thread exited with error");
2516 }
b8aa1682 2517 DBG("Worker thread cleanup complete");
095a4ae5 2518 free(data_buffer);
f385ae0a
MD
2519 if (err) {
2520 health_error();
2521 ERR("Health error occurred in %s", __func__);
2522 }
2523 health_unregister(health_relayd);
9d1bbf21 2524 rcu_unregister_thread();
f385ae0a 2525 stop_threads();
b8aa1682
JD
2526 return NULL;
2527}
2528
2529/*
2530 * Create the relay command pipe to wake thread_manage_apps.
2531 * Closed in cleanup().
2532 */
2533static int create_relay_cmd_pipe(void)
2534{
a02de639 2535 int ret;
b8aa1682 2536
a02de639 2537 ret = utils_create_pipe_cloexec(relay_cmd_pipe);
b8aa1682 2538
b8aa1682
JD
2539 return ret;
2540}
2541
2542/*
2543 * main
2544 */
2545int main(int argc, char **argv)
2546{
2547 int ret = 0;
2548 void *status;
d3e2ba59 2549 struct relay_local_data *relay_ctx;
b8aa1682
JD
2550
2551 /* Create thread quit pipe */
2552 if ((ret = init_thread_quit_pipe()) < 0) {
2553 goto error;
2554 }
2555
2556 /* Parse arguments */
2557 progname = argv[0];
c617c0c6 2558 if ((ret = parse_args(argc, argv)) < 0) {
a02de639 2559 goto exit;
b8aa1682
JD
2560 }
2561
2562 if ((ret = set_signal_handler()) < 0) {
2563 goto exit;
2564 }
2565
4d513a50
DG
2566 /* Try to create directory if -o, --output is specified. */
2567 if (opt_output_path) {
994fa64f
DG
2568 if (*opt_output_path != '/') {
2569 ERR("Please specify an absolute path for -o, --output PATH");
2570 goto exit;
2571 }
2572
4d513a50
DG
2573 ret = utils_mkdir_recursive(opt_output_path, S_IRWXU | S_IRWXG);
2574 if (ret < 0) {
2575 ERR("Unable to create %s", opt_output_path);
2576 goto exit;
2577 }
2578 }
2579
b8aa1682
JD
2580 /* Daemonize */
2581 if (opt_daemon) {
2582 ret = daemon(0, 0);
2583 if (ret < 0) {
2584 PERROR("daemon");
a02de639 2585 goto exit;
b8aa1682
JD
2586 }
2587 }
2588
1c20f0e2
JD
2589 /* We need those values for the file/dir creation. */
2590 relayd_uid = getuid();
2591 relayd_gid = getgid();
b8aa1682 2592
1c20f0e2
JD
2593 /* Check if daemon is UID = 0 */
2594 if (relayd_uid == 0) {
b8aa1682
JD
2595 if (control_uri->port < 1024 || data_uri->port < 1024) {
2596 ERR("Need to be root to use ports < 1024");
2597 ret = -1;
a02de639 2598 goto exit;
b8aa1682
JD
2599 }
2600 }
2601
2602 /* Setup the thread apps communication pipe. */
2603 if ((ret = create_relay_cmd_pipe()) < 0) {
2604 goto exit;
2605 }
2606
2607 /* Init relay command queue. */
2608 cds_wfq_init(&relay_cmd_queue.queue);
2609
2610 /* Set up max poll set size */
2611 lttng_poll_set_max_size();
2612
554831e7
MD
2613 /* Initialize communication library */
2614 lttcomm_init();
2615
d3e2ba59
JD
2616 relay_ctx = zmalloc(sizeof(struct relay_local_data));
2617 if (!relay_ctx) {
2618 PERROR("relay_ctx");
2619 goto exit;
2620 }
2621
2622 /* tables of sessions indexed by session ID */
2623 relay_ctx->sessions_ht = lttng_ht_new(0, LTTNG_HT_TYPE_ULONG);
2624 if (!relay_ctx->sessions_ht) {
2625 goto exit_relay_ctx_sessions;
2626 }
2627
2628 /* tables of streams indexed by stream ID */
2629 relay_streams_ht = lttng_ht_new(0, LTTNG_HT_TYPE_ULONG);
2630 if (!relay_streams_ht) {
2631 goto exit_relay_ctx_streams;
2632 }
2633
2634 /* tables of streams indexed by stream ID */
92c6ca54
DG
2635 viewer_streams_ht = lttng_ht_new(0, LTTNG_HT_TYPE_U64);
2636 if (!viewer_streams_ht) {
d3e2ba59
JD
2637 goto exit_relay_ctx_viewer_streams;
2638 }
2639
55706a7d
MD
2640 /* Initialize thread health monitoring */
2641 health_relayd = health_app_create(NR_HEALTH_RELAYD_TYPES);
2642 if (!health_relayd) {
2643 PERROR("health_app_create error");
2644 goto exit_health_app_create;
2645 }
2646
65931c8b
MD
2647 ret = utils_create_pipe(health_quit_pipe);
2648 if (ret < 0) {
2649 goto error_health_pipe;
2650 }
2651
2652 /* Create thread to manage the client socket */
2653 ret = pthread_create(&health_thread, NULL,
2654 thread_manage_health, (void *) NULL);
2655 if (ret != 0) {
2656 PERROR("pthread_create health");
2657 goto health_error;
2658 }
2659
b8aa1682
JD
2660 /* Setup the dispatcher thread */
2661 ret = pthread_create(&dispatcher_thread, NULL,
2662 relay_thread_dispatcher, (void *) NULL);
2663 if (ret != 0) {
2664 PERROR("pthread_create dispatcher");
2665 goto exit_dispatcher;
2666 }
2667
2668 /* Setup the worker thread */
2669 ret = pthread_create(&worker_thread, NULL,
d3e2ba59 2670 relay_thread_worker, (void *) relay_ctx);
b8aa1682
JD
2671 if (ret != 0) {
2672 PERROR("pthread_create worker");
2673 goto exit_worker;
2674 }
2675
2676 /* Setup the listener thread */
2677 ret = pthread_create(&listener_thread, NULL,
2678 relay_thread_listener, (void *) NULL);
2679 if (ret != 0) {
2680 PERROR("pthread_create listener");
2681 goto exit_listener;
2682 }
2683
42415026 2684 ret = live_start_threads(live_uri, relay_ctx, thread_quit_pipe);
d3e2ba59
JD
2685 if (ret != 0) {
2686 ERR("Starting live viewer threads");
50138f51 2687 goto exit_live;
d3e2ba59
JD
2688 }
2689
50138f51 2690exit_live:
b8aa1682
JD
2691 ret = pthread_join(listener_thread, &status);
2692 if (ret != 0) {
2693 PERROR("pthread_join");
2694 goto error; /* join error, exit without cleanup */
2695 }
2696
50138f51 2697exit_listener:
b8aa1682
JD
2698 ret = pthread_join(worker_thread, &status);
2699 if (ret != 0) {
2700 PERROR("pthread_join");
2701 goto error; /* join error, exit without cleanup */
2702 }
2703
50138f51 2704exit_worker:
b8aa1682
JD
2705 ret = pthread_join(dispatcher_thread, &status);
2706 if (ret != 0) {
2707 PERROR("pthread_join");
2708 goto error; /* join error, exit without cleanup */
2709 }
42415026 2710
50138f51 2711exit_dispatcher:
65931c8b
MD
2712 ret = pthread_join(health_thread, &status);
2713 if (ret != 0) {
2714 PERROR("pthread_join health thread");
2715 goto error; /* join error, exit without cleanup */
2716 }
2717
bd11e201
MD
2718 /*
2719 * Stop live threads only after joining other threads.
2720 */
2721 live_stop_threads();
2722
65931c8b
MD
2723health_error:
2724 utils_close_pipe(health_quit_pipe);
2725
2726error_health_pipe:
55706a7d
MD
2727 health_app_destroy(health_relayd);
2728
2729exit_health_app_create:
92c6ca54 2730 lttng_ht_destroy(viewer_streams_ht);
d3e2ba59
JD
2731
2732exit_relay_ctx_viewer_streams:
2733 lttng_ht_destroy(relay_streams_ht);
2734
2735exit_relay_ctx_streams:
2736 lttng_ht_destroy(relay_ctx->sessions_ht);
2737
2738exit_relay_ctx_sessions:
2739 free(relay_ctx);
b8aa1682
JD
2740
2741exit:
2742 cleanup();
2743 if (!ret) {
2744 exit(EXIT_SUCCESS);
2745 }
a02de639 2746
b8aa1682
JD
2747error:
2748 exit(EXIT_FAILURE);
2749}
This page took 0.168296 seconds and 5 git commands to generate.