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