Fix: Off by uint64_t in the metadata transfer
[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>
35#include <urcu/futex.h>
36#include <urcu/uatomic.h>
37#include <unistd.h>
38#include <fcntl.h>
39#include <config.h>
40
41#include <lttng/lttng.h>
42#include <common/common.h>
43#include <common/compat/poll.h>
44#include <common/compat/socket.h>
45#include <common/defaults.h>
46#include <common/futex.h>
47#include <common/sessiond-comm/sessiond-comm.h>
48#include <common/sessiond-comm/inet.h>
49#include <common/hashtable/hashtable.h>
50#include <common/sessiond-comm/relayd.h>
51#include <common/uri.h>
a02de639 52#include <common/utils.h>
b8aa1682
JD
53
54#include "lttng-relayd.h"
55
56/* command line options */
57static int opt_daemon;
58static char *opt_output_path;
095a4ae5
MD
59static struct lttng_uri *control_uri;
60static struct lttng_uri *data_uri;
b8aa1682
JD
61
62const char *progname;
63static int is_root; /* Set to 1 if the daemon is running as root */
64
65/*
66 * Quit pipe for all threads. This permits a single cancellation point
67 * for all threads when receiving an event on the pipe.
68 */
69static int thread_quit_pipe[2] = { -1, -1 };
70
71/*
72 * This pipe is used to inform the worker thread that a command is queued and
73 * ready to be processed.
74 */
75static int relay_cmd_pipe[2] = { -1, -1 };
76
26c9d55e 77/* Shared between threads */
b8aa1682
JD
78static int dispatch_thread_exit;
79
80static pthread_t listener_thread;
81static pthread_t dispatcher_thread;
82static pthread_t worker_thread;
83
095a4ae5
MD
84static uint64_t last_relay_stream_id;
85static uint64_t last_relay_session_id;
b8aa1682
JD
86
87/*
88 * Relay command queue.
89 *
90 * The relay_thread_listener and relay_thread_dispatcher communicate with this
91 * queue.
92 */
93static struct relay_cmd_queue relay_cmd_queue;
94
95/* buffer allocated at startup, used to store the trace data */
095a4ae5
MD
96static char *data_buffer;
97static unsigned int data_buffer_size;
b8aa1682
JD
98
99/*
100 * usage function on stderr
101 */
102static
103void usage(void)
104{
105 fprintf(stderr, "Usage: %s OPTIONS\n\nOptions:\n", progname);
106 fprintf(stderr, " -h, --help Display this usage.\n");
107 fprintf(stderr, " -d, --daemonize Start as a daemon.\n");
108 fprintf(stderr, " -C, --control-port Control port listening (URI)\n");
109 fprintf(stderr, " -D, --data-port Data port listening (URI)\n");
25672d02 110 fprintf(stderr, " -o, --output Output path for traces (PATH)\n");
b8aa1682
JD
111 fprintf(stderr, " -v, --verbose Verbose mode. Activate DBG() macro.\n");
112}
113
114static
115int parse_args(int argc, char **argv)
116{
117 int c;
118 int ret = 0;
119 char *default_address;
120
121 static struct option long_options[] = {
e3678fd8
MD
122 { "control-port", 1, 0, 'C', },
123 { "data-port", 1, 0, 'D', },
124 { "daemonize", 0, 0, 'd', },
125 { "help", 0, 0, 'h', },
126 { "output", 1, 0, 'o', },
127 { "verbose", 0, 0, 'v', },
095a4ae5 128 { NULL, 0, 0, 0, },
b8aa1682
JD
129 };
130
131 while (1) {
132 int option_index = 0;
133 c = getopt_long(argc, argv, "dhv" "C:D:o:",
134 long_options, &option_index);
135 if (c == -1) {
136 break;
137 }
138
139 switch (c) {
140 case 0:
141 fprintf(stderr, "option %s", long_options[option_index].name);
142 if (optarg) {
143 fprintf(stderr, " with arg %s\n", optarg);
144 }
145 break;
146 case 'C':
147 ret = uri_parse(optarg, &control_uri);
148 if (ret < 0) {
149 ERR("Invalid control URI specified");
150 goto exit;
151 }
152 if (control_uri->port == 0) {
153 control_uri->port = DEFAULT_NETWORK_CONTROL_PORT;
154 }
155 break;
156 case 'D':
157 ret = uri_parse(optarg, &data_uri);
158 if (ret < 0) {
159 ERR("Invalid data URI specified");
160 goto exit;
161 }
162 if (data_uri->port == 0) {
163 data_uri->port = DEFAULT_NETWORK_DATA_PORT;
164 }
165 break;
166 case 'd':
167 opt_daemon = 1;
168 break;
169 case 'h':
170 usage();
171 exit(EXIT_FAILURE);
172 case 'o':
173 ret = asprintf(&opt_output_path, "%s", optarg);
174 if (ret < 0) {
175 PERROR("asprintf opt_output_path");
176 goto exit;
177 }
178 break;
179 case 'v':
180 /* Verbose level can increase using multiple -v */
181 lttng_opt_verbose += 1;
182 break;
183 default:
184 /* Unknown option or other error.
185 * Error is printed by getopt, just return */
186 ret = -1;
187 goto exit;
188 }
189 }
190
191 /* assign default values */
192 if (control_uri == NULL) {
193 ret = asprintf(&default_address, "tcp://0.0.0.0:%d",
194 DEFAULT_NETWORK_CONTROL_PORT);
195 if (ret < 0) {
196 PERROR("asprintf default data address");
197 goto exit;
198 }
199
200 ret = uri_parse(default_address, &control_uri);
201 free(default_address);
202 if (ret < 0) {
203 ERR("Invalid control URI specified");
204 goto exit;
205 }
206 }
207 if (data_uri == NULL) {
208 ret = asprintf(&default_address, "tcp://0.0.0.0:%d",
209 DEFAULT_NETWORK_DATA_PORT);
210 if (ret < 0) {
211 PERROR("asprintf default data address");
212 goto exit;
213 }
214
215 ret = uri_parse(default_address, &data_uri);
216 free(default_address);
217 if (ret < 0) {
218 ERR("Invalid data URI specified");
219 goto exit;
220 }
221 }
222
223exit:
224 return ret;
225}
226
227/*
228 * Cleanup the daemon
229 */
230static
231void cleanup(void)
232{
b8aa1682
JD
233 DBG("Cleaning up");
234
095a4ae5
MD
235 /* free the dynamically allocated opt_output_path */
236 free(opt_output_path);
237
a02de639
CB
238 /* Close thread quit pipes */
239 utils_close_pipe(thread_quit_pipe);
240
241 /* Close relay cmd pipes */
242 utils_close_pipe(relay_cmd_pipe);
b8aa1682
JD
243}
244
245/*
246 * Write to writable pipe used to notify a thread.
247 */
248static
249int notify_thread_pipe(int wpipe)
250{
251 int ret;
252
6f94560a
MD
253 do {
254 ret = write(wpipe, "!", 1);
255 } while (ret < 0 && errno == EINTR);
b8aa1682
JD
256 if (ret < 0) {
257 PERROR("write poll pipe");
258 }
259
260 return ret;
261}
262
263/*
264 * Stop all threads by closing the thread quit pipe.
265 */
266static
267void stop_threads(void)
268{
269 int ret;
270
271 /* Stopping all threads */
272 DBG("Terminating all threads");
273 ret = notify_thread_pipe(thread_quit_pipe[1]);
274 if (ret < 0) {
275 ERR("write error on thread quit pipe");
276 }
277
278 /* Dispatch thread */
26c9d55e 279 CMM_STORE_SHARED(dispatch_thread_exit, 1);
b8aa1682
JD
280 futex_nto1_wake(&relay_cmd_queue.futex);
281}
282
283/*
284 * Signal handler for the daemon
285 *
286 * Simply stop all worker threads, leaving main() return gracefully after
287 * joining all threads and calling cleanup().
288 */
289static
290void sighandler(int sig)
291{
292 switch (sig) {
293 case SIGPIPE:
294 DBG("SIGPIPE caught");
295 return;
296 case SIGINT:
297 DBG("SIGINT caught");
298 stop_threads();
299 break;
300 case SIGTERM:
301 DBG("SIGTERM caught");
302 stop_threads();
303 break;
304 default:
305 break;
306 }
307}
308
309/*
310 * Setup signal handler for :
311 * SIGINT, SIGTERM, SIGPIPE
312 */
313static
314int set_signal_handler(void)
315{
316 int ret = 0;
317 struct sigaction sa;
318 sigset_t sigset;
319
320 if ((ret = sigemptyset(&sigset)) < 0) {
321 PERROR("sigemptyset");
322 return ret;
323 }
324
325 sa.sa_handler = sighandler;
326 sa.sa_mask = sigset;
327 sa.sa_flags = 0;
328 if ((ret = sigaction(SIGTERM, &sa, NULL)) < 0) {
329 PERROR("sigaction");
330 return ret;
331 }
332
333 if ((ret = sigaction(SIGINT, &sa, NULL)) < 0) {
334 PERROR("sigaction");
335 return ret;
336 }
337
338 if ((ret = sigaction(SIGPIPE, &sa, NULL)) < 0) {
339 PERROR("sigaction");
340 return ret;
341 }
342
343 DBG("Signal handler set for SIGTERM, SIGPIPE and SIGINT");
344
345 return ret;
346}
347
348/*
349 * Init thread quit pipe.
350 *
351 * Return -1 on error or 0 if all pipes are created.
352 */
353static
354int init_thread_quit_pipe(void)
355{
a02de639 356 int ret;
b8aa1682 357
a02de639 358 ret = utils_create_pipe_cloexec(thread_quit_pipe);
b8aa1682 359
b8aa1682
JD
360 return ret;
361}
362
363/*
364 * Create a poll set with O_CLOEXEC and add the thread quit pipe to the set.
365 */
366static
367int create_thread_poll_set(struct lttng_poll_event *events, int size)
368{
369 int ret;
370
371 if (events == NULL || size == 0) {
372 ret = -1;
373 goto error;
374 }
375
376 ret = lttng_poll_create(events, size, LTTNG_CLOEXEC);
377 if (ret < 0) {
378 goto error;
379 }
380
381 /* Add quit pipe */
382 ret = lttng_poll_add(events, thread_quit_pipe[0], LPOLLIN);
383 if (ret < 0) {
384 goto error;
385 }
386
387 return 0;
388
389error:
390 return ret;
391}
392
393/*
394 * Check if the thread quit pipe was triggered.
395 *
396 * Return 1 if it was triggered else 0;
397 */
398static
399int check_thread_quit_pipe(int fd, uint32_t events)
400{
401 if (fd == thread_quit_pipe[0] && (events & LPOLLIN)) {
402 return 1;
403 }
404
405 return 0;
406}
407
408/*
409 * Create and init socket from uri.
410 */
411static
412struct lttcomm_sock *relay_init_sock(struct lttng_uri *uri)
413{
414 int ret;
415 struct lttcomm_sock *sock = NULL;
416
417 sock = lttcomm_alloc_sock_from_uri(uri);
418 if (sock == NULL) {
419 ERR("Allocating socket");
420 goto error;
421 }
422
423 ret = lttcomm_create_sock(sock);
424 if (ret < 0) {
425 goto error;
426 }
427 DBG("Listening on sock %d", sock->fd);
428
429 ret = sock->ops->bind(sock);
430 if (ret < 0) {
431 goto error;
432 }
433
434 ret = sock->ops->listen(sock, -1);
435 if (ret < 0) {
436 goto error;
437
438 }
439
440 return sock;
441
442error:
443 if (sock) {
444 lttcomm_destroy_sock(sock);
445 }
446 return NULL;
447}
448
449/*
450 * This thread manages the listening for new connections on the network
451 */
452static
453void *relay_thread_listener(void *data)
454{
095a4ae5 455 int i, ret, pollfd, err = -1;
b8aa1682
JD
456 int val = 1;
457 uint32_t revents, nb_fd;
458 struct lttng_poll_event events;
459 struct lttcomm_sock *control_sock, *data_sock;
460
461 /*
462 * Get allocated in this thread, enqueued to a global queue, dequeued and
463 * freed in the worker thread.
464 */
465 struct relay_command *relay_cmd = NULL;
466
467 DBG("[thread] Relay listener started");
468
469 control_sock = relay_init_sock(control_uri);
470 if (!control_sock) {
095a4ae5 471 goto error_sock_control;
b8aa1682
JD
472 }
473
474 data_sock = relay_init_sock(data_uri);
475 if (!data_sock) {
095a4ae5 476 goto error_sock_relay;
b8aa1682
JD
477 }
478
479 /*
480 * Pass 3 as size here for the thread quit pipe, control and data socket.
481 */
482 ret = create_thread_poll_set(&events, 3);
483 if (ret < 0) {
484 goto error_create_poll;
485 }
486
487 /* Add the control socket */
488 ret = lttng_poll_add(&events, control_sock->fd, LPOLLIN | LPOLLRDHUP);
489 if (ret < 0) {
490 goto error_poll_add;
491 }
492
493 /* Add the data socket */
494 ret = lttng_poll_add(&events, data_sock->fd, LPOLLIN | LPOLLRDHUP);
495 if (ret < 0) {
496 goto error_poll_add;
497 }
498
499 while (1) {
500 DBG("Listener accepting connections");
501
502 nb_fd = LTTNG_POLL_GETNB(&events);
503
504restart:
505 ret = lttng_poll_wait(&events, -1);
506 if (ret < 0) {
507 /*
508 * Restart interrupted system call.
509 */
510 if (errno == EINTR) {
511 goto restart;
512 }
513 goto error;
514 }
515
516 DBG("Relay new connection received");
517 for (i = 0; i < nb_fd; i++) {
518 /* Fetch once the poll data */
519 revents = LTTNG_POLL_GETEV(&events, i);
520 pollfd = LTTNG_POLL_GETFD(&events, i);
521
522 /* Thread quit pipe has been closed. Killing thread. */
523 ret = check_thread_quit_pipe(pollfd, revents);
524 if (ret) {
095a4ae5
MD
525 err = 0;
526 goto exit;
b8aa1682
JD
527 }
528
529 if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
530 ERR("socket poll error");
531 goto error;
532 } else if (revents & LPOLLIN) {
533 struct lttcomm_sock *newsock = NULL;
534
535 relay_cmd = zmalloc(sizeof(struct relay_command));
536 if (relay_cmd == NULL) {
537 PERROR("relay command zmalloc");
538 goto error;
539 }
540
541 if (pollfd == data_sock->fd) {
542 newsock = data_sock->ops->accept(data_sock);
543 if (newsock < 0) {
544 PERROR("accepting data sock");
545 goto error;
546 }
547 relay_cmd->type = RELAY_DATA;
548 DBG("Relay data connection accepted, socket %d", newsock->fd);
549 } else if (pollfd == control_sock->fd) {
550 newsock = control_sock->ops->accept(control_sock);
551 if (newsock < 0) {
552 PERROR("accepting control sock");
553 goto error;
554 }
555 relay_cmd->type = RELAY_CONTROL;
556 DBG("Relay control connection accepted, socket %d", newsock->fd);
557 }
558 ret = setsockopt(newsock->fd, SOL_SOCKET, SO_REUSEADDR,
559 &val, sizeof(int));
560 if (ret < 0) {
561 PERROR("setsockopt inet");
562 goto error;
563 }
564 relay_cmd->sock = newsock;
565 /*
566 * Lock free enqueue the request.
567 */
568 cds_wfq_enqueue(&relay_cmd_queue.queue, &relay_cmd->node);
569
570 /*
571 * Wake the dispatch queue futex. Implicit memory
572 * barrier with the exchange in cds_wfq_enqueue.
573 */
574 futex_nto1_wake(&relay_cmd_queue.futex);
575 }
576 }
577 }
578
095a4ae5 579exit:
b8aa1682
JD
580error:
581error_poll_add:
582 lttng_poll_clean(&events);
583error_create_poll:
095a4ae5
MD
584 if (data_sock->fd >= 0) {
585 ret = data_sock->ops->close(data_sock);
b8aa1682
JD
586 if (ret) {
587 PERROR("close");
588 }
b8aa1682 589 }
095a4ae5
MD
590 lttcomm_destroy_sock(data_sock);
591error_sock_relay:
592 if (control_sock->fd >= 0) {
593 ret = control_sock->ops->close(control_sock);
b8aa1682
JD
594 if (ret) {
595 PERROR("close");
596 }
b8aa1682 597 }
095a4ae5
MD
598 lttcomm_destroy_sock(control_sock);
599error_sock_control:
600 if (err) {
601 DBG("Thread exited with error");
602 }
b8aa1682
JD
603 DBG("Relay listener thread cleanup complete");
604 stop_threads();
b8aa1682
JD
605 return NULL;
606}
607
608/*
609 * This thread manages the dispatching of the requests to worker threads
610 */
611static
612void *relay_thread_dispatcher(void *data)
613{
614 int ret;
615 struct cds_wfq_node *node;
616 struct relay_command *relay_cmd = NULL;
617
618 DBG("[thread] Relay dispatcher started");
619
26c9d55e 620 while (!CMM_LOAD_SHARED(dispatch_thread_exit)) {
b8aa1682
JD
621 /* Atomically prepare the queue futex */
622 futex_nto1_prepare(&relay_cmd_queue.futex);
623
624 do {
625 /* Dequeue commands */
626 node = cds_wfq_dequeue_blocking(&relay_cmd_queue.queue);
627 if (node == NULL) {
628 DBG("Woken up but nothing in the relay command queue");
629 /* Continue thread execution */
630 break;
631 }
632
633 relay_cmd = caa_container_of(node, struct relay_command, node);
634 DBG("Dispatching request waiting on sock %d", relay_cmd->sock->fd);
635
636 /*
637 * Inform worker thread of the new request. This
638 * call is blocking so we can be assured that the data will be read
639 * at some point in time or wait to the end of the world :)
640 */
6f94560a
MD
641 do {
642 ret = write(relay_cmd_pipe[1], relay_cmd,
643 sizeof(struct relay_command));
644 } while (ret < 0 && errno == EINTR);
b8aa1682
JD
645 free(relay_cmd);
646 if (ret < 0) {
647 PERROR("write cmd pipe");
648 goto error;
649 }
650 } while (node != NULL);
651
652 /* Futex wait on queue. Blocking call on futex() */
653 futex_nto1_wait(&relay_cmd_queue.futex);
654 }
655
656error:
657 DBG("Dispatch thread dying");
658 stop_threads();
659 return NULL;
660}
661
662/*
663 * Return the realpath(3) of the path even if the last directory token does not
664 * exist. For example, with /tmp/test1/test2, if test2/ does not exist but the
665 * /tmp/test1 does, the real path is returned. In normal time, realpath(3)
666 * fails if the end point directory does not exist.
667 */
668static
669char *expand_full_path(const char *path)
670{
671 const char *end_path = path;
095a4ae5 672 char *next, *cut_path, *expanded_path, *respath;
b8aa1682
JD
673
674 /* Find last token delimited by '/' */
675 while ((next = strpbrk(end_path + 1, "/"))) {
676 end_path = next;
677 }
678
679 /* Cut last token from original path */
680 cut_path = strndup(path, end_path - path);
681
682 expanded_path = malloc(PATH_MAX);
683 if (expanded_path == NULL) {
095a4ae5
MD
684 respath = NULL;
685 goto end;
b8aa1682
JD
686 }
687
095a4ae5
MD
688 respath = realpath(cut_path, expanded_path);
689 if (respath == NULL) {
b8aa1682
JD
690 switch (errno) {
691 case ENOENT:
692 ERR("%s: No such file or directory", cut_path);
693 break;
694 default:
695 PERROR("realpath");
696 break;
697 }
095a4ae5
MD
698 free(expanded_path);
699 } else {
700 /* Add end part to expanded path */
701 strcat(respath, end_path);
b8aa1682 702 }
095a4ae5 703end:
b8aa1682 704 free(cut_path);
095a4ae5 705 return respath;
b8aa1682
JD
706}
707
708
709/*
710 * config_get_default_path
711 *
712 * Returns the HOME directory path. Caller MUST NOT free(3) the return pointer.
713 */
714static
715char *config_get_default_path(void)
716{
717 return getenv("HOME");
718}
719
720/*
721 * Create recursively directory using the FULL path.
722 */
723static
724int mkdir_recursive(char *path, mode_t mode)
725{
726 char *p, tmp[PATH_MAX];
727 struct stat statbuf;
728 size_t len;
729 int ret;
730
731 ret = snprintf(tmp, sizeof(tmp), "%s", path);
732 if (ret < 0) {
733 PERROR("snprintf mkdir");
734 goto error;
735 }
736
737 len = ret;
738 if (tmp[len - 1] == '/') {
739 tmp[len - 1] = 0;
740 }
741
742 for (p = tmp + 1; *p; p++) {
743 if (*p == '/') {
744 *p = 0;
745 if (tmp[strlen(tmp) - 1] == '.' &&
746 tmp[strlen(tmp) - 2] == '.' &&
747 tmp[strlen(tmp) - 3] == '/') {
748 ERR("Using '/../' is not permitted in the trace path (%s)",
749 tmp);
750 ret = -1;
751 goto error;
752 }
753 ret = stat(tmp, &statbuf);
754 if (ret < 0) {
755 ret = mkdir(tmp, mode);
756 if (ret < 0) {
095a4ae5 757 if (errno != EEXIST) {
b8aa1682
JD
758 PERROR("mkdir recursive");
759 ret = -errno;
760 goto error;
761 }
762 }
763 }
764 *p = '/';
765 }
766 }
767
768 ret = mkdir(tmp, mode);
769 if (ret < 0) {
095a4ae5 770 if (errno != EEXIST) {
b8aa1682
JD
771 PERROR("mkdir recursive last piece");
772 ret = -errno;
773 } else {
774 ret = 0;
775 }
776 }
777
778error:
779 return ret;
780}
781
b8aa1682 782static
095a4ae5 783char *create_output_path_auto(char *path_name)
b8aa1682 784{
095a4ae5 785 int ret;
b8aa1682 786 char *traces_path = NULL;
095a4ae5
MD
787 char *alloc_path = NULL;
788 char *default_path;
b8aa1682 789
095a4ae5
MD
790 default_path = config_get_default_path();
791 if (default_path == NULL) {
792 ERR("Home path not found.\n \
793 Please specify an output path using -o, --output PATH");
794 goto exit;
795 }
796 alloc_path = strdup(default_path);
797 if (alloc_path == NULL) {
798 PERROR("Path allocation");
799 goto exit;
800 }
801 ret = asprintf(&traces_path, "%s/" DEFAULT_TRACE_DIR_NAME
802 "/%s", alloc_path, path_name);
803 if (ret < 0) {
804 PERROR("asprintf trace dir name");
805 goto exit;
b8aa1682 806 }
095a4ae5 807exit:
b8aa1682 808 free(alloc_path);
095a4ae5
MD
809 return traces_path;
810}
811
812static
813char *create_output_path_noauto(char *path_name)
814{
815 int ret;
816 char *traces_path = NULL;
817 char *full_path;
b8aa1682 818
095a4ae5
MD
819 full_path = expand_full_path(opt_output_path);
820 ret = asprintf(&traces_path, "%s/%s", full_path, path_name);
821 if (ret < 0) {
822 PERROR("asprintf trace dir name");
823 goto exit;
824 }
b8aa1682 825exit:
095a4ae5 826 free(full_path);
b8aa1682
JD
827 return traces_path;
828}
829
095a4ae5
MD
830/*
831 * create_output_path: create the output trace directory
832 */
833static
834char *create_output_path(char *path_name)
835{
836 if (opt_output_path == NULL) {
837 return create_output_path_auto(path_name);
838 } else {
839 return create_output_path_noauto(path_name);
840 }
841}
842
9d1bbf21
MD
843static
844void deferred_free_stream(struct rcu_head *head)
845{
846 struct relay_stream *stream =
847 caa_container_of(head, struct relay_stream, rcu_node);
848 free(stream);
849}
850
b8aa1682
JD
851/*
852 * relay_delete_session: Free all memory associated with a session and
853 * close all the FDs
854 */
855static
856void relay_delete_session(struct relay_command *cmd, struct lttng_ht *streams_ht)
857{
858 struct lttng_ht_iter iter;
859 struct lttng_ht_node_ulong *node;
860 struct relay_stream *stream;
861 int ret;
862
095a4ae5 863 if (!cmd->session) {
b8aa1682 864 return;
095a4ae5 865 }
b8aa1682
JD
866
867 DBG("Relay deleting session %lu", cmd->session->id);
868 free(cmd->session->sock);
869
9d1bbf21 870 rcu_read_lock();
b8aa1682
JD
871 cds_lfht_for_each_entry(streams_ht->ht, &iter.iter, node, node) {
872 node = lttng_ht_iter_get_node_ulong(&iter);
873 if (node) {
874 stream = caa_container_of(node,
875 struct relay_stream, stream_n);
876 if (stream->session == cmd->session) {
877 close(stream->fd);
878 ret = lttng_ht_del(streams_ht, &iter);
879 assert(!ret);
9d1bbf21
MD
880 call_rcu(&stream->rcu_node,
881 deferred_free_stream);
b8aa1682
JD
882 }
883 }
884 }
9d1bbf21 885 rcu_read_unlock();
b8aa1682
JD
886}
887
888/*
889 * relay_add_stream: allocate a new stream for a session
890 */
891static
892int relay_add_stream(struct lttcomm_relayd_hdr *recv_hdr,
893 struct relay_command *cmd, struct lttng_ht *streams_ht)
894{
895 struct relay_session *session = cmd->session;
896 struct lttcomm_relayd_add_stream stream_info;
897 struct relay_stream *stream = NULL;
898 struct lttcomm_relayd_status_stream reply;
899 char *path = NULL, *root_path = NULL;
900 int ret, send_ret;
901
902 if (!session || session->version_check_done == 0) {
903 ERR("Trying to add a stream before version check");
904 ret = -1;
905 goto end_no_session;
906 }
907
908 /* FIXME : use data_size for something ? */
909 ret = cmd->sock->ops->recvmsg(cmd->sock, &stream_info,
910 sizeof(struct lttcomm_relayd_add_stream), MSG_WAITALL);
911 if (ret < sizeof(struct lttcomm_relayd_add_stream)) {
912 ERR("Relay didn't receive valid add_stream struct size : %d", ret);
913 ret = -1;
914 goto end_no_session;
915 }
916 stream = zmalloc(sizeof(struct relay_stream));
917 if (stream == NULL) {
918 PERROR("relay stream zmalloc");
919 ret = -1;
920 goto end_no_session;
921 }
922
9d1bbf21 923 rcu_read_lock();
b8aa1682
JD
924 stream->stream_handle = ++last_relay_stream_id;
925 stream->seq = 0;
926 stream->session = session;
927
928 root_path = create_output_path(stream_info.pathname);
929 if (!root_path) {
930 ret = -1;
931 goto end;
932 }
933 ret = mkdir_recursive(root_path, S_IRWXU | S_IRWXG);
934 if (ret < 0) {
b8aa1682
JD
935 ERR("relay creating output directory");
936 goto end;
937 }
938
939 ret = asprintf(&path, "%s/%s", root_path, stream_info.channel_name);
940 if (ret < 0) {
941 PERROR("asprintf stream path");
942 goto end;
943 }
944
945 ret = open(path, O_WRONLY|O_CREAT|O_TRUNC, S_IRWXU|S_IRWXG|S_IRWXO);
946 if (ret < 0) {
947 PERROR("Relay creating trace file");
948 goto end;
949 }
950
951 stream->fd = ret;
952 DBG("Tracefile %s created", path);
953
954 lttng_ht_node_init_ulong(&stream->stream_n,
955 (unsigned long) stream->stream_handle);
956 lttng_ht_add_unique_ulong(streams_ht,
957 &stream->stream_n);
958
959 DBG("Relay new stream added %s", stream_info.channel_name);
960
961end:
962 free(path);
963 free(root_path);
964 /* send the session id to the client or a negative return code on error */
965 if (ret < 0) {
966 reply.ret_code = htobe32(LTTCOMM_ERR);
967 } else {
968 reply.ret_code = htobe32(LTTCOMM_OK);
969 }
970 reply.handle = htobe64(stream->stream_handle);
971 send_ret = cmd->sock->ops->sendmsg(cmd->sock, &reply,
972 sizeof(struct lttcomm_relayd_status_stream), 0);
973 if (send_ret < 0) {
974 ERR("Relay sending stream id");
975 }
9d1bbf21 976 rcu_read_unlock();
b8aa1682
JD
977
978end_no_session:
979 return ret;
980}
981
982/*
983 * relay_unknown_command: send -1 if received unknown command
984 */
985static
986void relay_unknown_command(struct relay_command *cmd)
987{
988 struct lttcomm_relayd_generic_reply reply;
989 int ret;
990
991 reply.ret_code = htobe32(LTTCOMM_ERR);
992 ret = cmd->sock->ops->sendmsg(cmd->sock, &reply,
993 sizeof(struct lttcomm_relayd_generic_reply), 0);
994 if (ret < 0) {
995 ERR("Relay sending unknown command");
996 }
997}
998
999/*
1000 * relay_start: send an acknowledgment to the client to tell if we are
1001 * ready to receive data. We are ready if a session is established.
1002 */
1003static
1004int relay_start(struct lttcomm_relayd_hdr *recv_hdr,
1005 struct relay_command *cmd)
1006{
1007 int ret = htobe32(LTTCOMM_OK);
1008 struct lttcomm_relayd_generic_reply reply;
1009 struct relay_session *session = cmd->session;
1010
1011 if (!session) {
1012 DBG("Trying to start the streaming without a session established");
1013 ret = htobe32(LTTCOMM_ERR);
1014 }
1015
1016 reply.ret_code = ret;
1017 ret = cmd->sock->ops->sendmsg(cmd->sock, &reply,
1018 sizeof(struct lttcomm_relayd_generic_reply), 0);
1019 if (ret < 0) {
1020 ERR("Relay sending start ack");
1021 }
1022
1023 return ret;
1024}
1025
1026/*
1027 * Get stream from stream id.
9d1bbf21 1028 * Need to be called with RCU read-side lock held.
b8aa1682
JD
1029 */
1030static
1031struct relay_stream *relay_stream_from_stream_id(uint64_t stream_id,
1032 struct lttng_ht *streams_ht)
1033{
1034 struct lttng_ht_node_ulong *node;
1035 struct lttng_ht_iter iter;
1036 struct relay_stream *ret;
1037
1038 lttng_ht_lookup(streams_ht,
1039 (void *)((unsigned long) stream_id),
1040 &iter);
1041 node = lttng_ht_iter_get_node_ulong(&iter);
1042 if (node == NULL) {
1043 DBG("Relay stream %lu not found", stream_id);
1044 ret = NULL;
1045 goto end;
1046 }
1047
1048 ret = caa_container_of(node, struct relay_stream, stream_n);
1049
1050end:
1051 return ret;
1052}
1053
1054/*
1055 * relay_recv_metadata: receive the metada for the session.
1056 */
1057static
1058int relay_recv_metadata(struct lttcomm_relayd_hdr *recv_hdr,
1059 struct relay_command *cmd, struct lttng_ht *streams_ht)
1060{
1061 int ret = htobe32(LTTCOMM_OK);
1062 struct relay_session *session = cmd->session;
1063 struct lttcomm_relayd_metadata_payload *metadata_struct;
1064 struct relay_stream *metadata_stream;
1065 uint64_t data_size, payload_size;
1066
1067 if (!session) {
1068 ERR("Metadata sent before version check");
1069 ret = -1;
1070 goto end;
1071 }
1072
1073 data_size = be64toh(recv_hdr->data_size);
87c1611d
DG
1074 payload_size = data_size;
1075 /*
1076 * Add 8 bytes (uint64_t) to the data size which is the value of the
1077 * stream_id and the payload size.
1078 */
1079 data_size += sizeof(uint64_t);
b8aa1682
JD
1080 if (data_buffer_size < data_size) {
1081 data_buffer = realloc(data_buffer, data_size);
1082 if (!data_buffer) {
1083 ERR("Allocating data buffer");
1084 ret = -1;
1085 goto end;
1086 }
1087 data_buffer_size = data_size;
1088 }
1089 memset(data_buffer, 0, data_size);
1090 DBG2("Relay receiving metadata, waiting for %lu bytes", data_size);
87c1611d
DG
1091 ret = cmd->sock->ops->recvmsg(cmd->sock, data_buffer, data_size,
1092 MSG_WAITALL);
b8aa1682
JD
1093 if (ret < 0 || ret != data_size) {
1094 ret = -1;
1095 ERR("Relay didn't receive the whole metadata");
1096 goto end;
1097 }
1098 metadata_struct = (struct lttcomm_relayd_metadata_payload *) data_buffer;
9d1bbf21
MD
1099
1100 rcu_read_lock();
b8aa1682
JD
1101 metadata_stream = relay_stream_from_stream_id(
1102 be64toh(metadata_struct->stream_id), streams_ht);
1103 if (!metadata_stream) {
1104 ret = -1;
9d1bbf21 1105 goto end_unlock;
b8aa1682
JD
1106 }
1107
6f94560a
MD
1108 do {
1109 ret = write(metadata_stream->fd, metadata_struct->payload,
1110 payload_size);
1111 } while (ret < 0 && errno == EINTR);
87c1611d 1112 if (ret < payload_size) {
b8aa1682
JD
1113 ERR("Relay error writing metadata on file");
1114 ret = -1;
9d1bbf21 1115 goto end_unlock;
b8aa1682
JD
1116 }
1117 DBG2("Relay metadata written");
1118
9d1bbf21 1119end_unlock:
6e3c5836 1120 rcu_read_unlock();
b8aa1682
JD
1121end:
1122 return ret;
1123}
1124
1125/*
1126 * relay_send_version: send relayd version number
1127 */
1128static
1129int relay_send_version(struct lttcomm_relayd_hdr *recv_hdr,
1130 struct relay_command *cmd)
1131{
1132 int ret = htobe32(LTTCOMM_OK);
1133 struct lttcomm_relayd_version reply;
1134 struct relay_session *session = NULL;
1135
1136 if (cmd->session == NULL) {
1137 session = zmalloc(sizeof(struct relay_session));
1138 if (session == NULL) {
1139 PERROR("relay session zmalloc");
1140 ret = -1;
1141 goto end;
1142 }
1143 session->id = ++last_relay_session_id;
1144 DBG("Created session %lu", session->id);
1145 cmd->session = session;
1146 }
1147 session->version_check_done = 1;
1148
1149 sscanf(VERSION, "%u.%u", &reply.major, &reply.minor);
1150 reply.major = htobe32(reply.major);
1151 reply.minor = htobe32(reply.minor);
1152 ret = cmd->sock->ops->sendmsg(cmd->sock, &reply,
1153 sizeof(struct lttcomm_relayd_version), 0);
1154 if (ret < 0) {
1155 ERR("Relay sending version");
1156 }
1157 DBG("Version check done");
1158
1159end:
1160 return ret;
1161}
1162
1163/*
1164 * relay_process_control: Process the commands received on the control socket
1165 */
1166static
1167int relay_process_control(struct lttcomm_relayd_hdr *recv_hdr,
1168 struct relay_command *cmd, struct lttng_ht *streams_ht)
1169{
1170 int ret = 0;
1171
1172 switch (be32toh(recv_hdr->cmd)) {
1173 /*
1174 case RELAYD_CREATE_SESSION:
1175 ret = relay_create_session(recv_hdr, cmd);
1176 break;
1177 */
1178 case RELAYD_ADD_STREAM:
1179 ret = relay_add_stream(recv_hdr, cmd, streams_ht);
1180 break;
1181 case RELAYD_START_DATA:
1182 ret = relay_start(recv_hdr, cmd);
1183 break;
1184 case RELAYD_SEND_METADATA:
1185 ret = relay_recv_metadata(recv_hdr, cmd, streams_ht);
1186 break;
1187 case RELAYD_VERSION:
1188 ret = relay_send_version(recv_hdr, cmd);
1189 break;
1190 case RELAYD_UPDATE_SYNC_INFO:
1191 default:
1192 ERR("Received unknown command (%u)", be32toh(recv_hdr->cmd));
1193 relay_unknown_command(cmd);
1194 ret = -1;
1195 goto end;
1196 }
1197
1198end:
1199 return ret;
1200}
1201
1202/*
1203 * relay_process_data: Process the data received on the data socket
1204 */
1205static
1206int relay_process_data(struct relay_command *cmd, struct lttng_ht *streams_ht)
1207{
1208 int ret = 0;
1209 struct relay_stream *stream;
1210 struct lttcomm_relayd_data_hdr data_hdr;
1211 uint64_t stream_id;
1212 uint32_t data_size;
1213
1214 ret = cmd->sock->ops->recvmsg(cmd->sock, &data_hdr,
1215 sizeof(struct lttcomm_relayd_data_hdr), MSG_WAITALL);
1216 if (ret <= 0) {
1217 ERR("Connections seems to be closed");
1218 ret = -1;
1219 goto end;
1220 }
1221
1222 stream_id = be64toh(data_hdr.stream_id);
9d1bbf21
MD
1223
1224 rcu_read_lock();
b8aa1682
JD
1225 stream = relay_stream_from_stream_id(stream_id, streams_ht);
1226 if (!stream) {
1227 ret = -1;
9d1bbf21 1228 goto end_unlock;
b8aa1682
JD
1229 }
1230
1231 data_size = be32toh(data_hdr.data_size);
1232 if (data_buffer_size < data_size) {
1233 data_buffer = realloc(data_buffer, data_size);
1234 if (!data_buffer) {
1235 ERR("Allocating data buffer");
1236 ret = -1;
9d1bbf21 1237 goto end_unlock;
b8aa1682
JD
1238 }
1239 data_buffer_size = data_size;
1240 }
1241 memset(data_buffer, 0, data_size);
1242
87c1611d 1243 DBG3("Receiving data of size %u for stream id %zu", data_size, stream_id);
b8aa1682
JD
1244 ret = cmd->sock->ops->recvmsg(cmd->sock, data_buffer, data_size, MSG_WAITALL);
1245 if (ret <= 0) {
1246 ret = -1;
9d1bbf21 1247 goto end_unlock;
b8aa1682
JD
1248 }
1249
6f94560a
MD
1250 do {
1251 ret = write(stream->fd, data_buffer, data_size);
1252 } while (ret < 0 && errno == EINTR);
b8aa1682
JD
1253 if (ret < data_size) {
1254 ERR("Relay error writing data to file");
1255 ret = -1;
9d1bbf21 1256 goto end_unlock;
b8aa1682
JD
1257 }
1258 DBG2("Relay wrote %d bytes to tracefile for stream id %lu", ret, stream->stream_handle);
1259
9d1bbf21
MD
1260end_unlock:
1261 rcu_read_unlock();
b8aa1682
JD
1262end:
1263 return ret;
1264}
1265
1266static
9d1bbf21 1267void relay_cleanup_poll_connection(struct lttng_poll_event *events, int pollfd)
b8aa1682
JD
1268{
1269 int ret;
1270
b8aa1682
JD
1271 lttng_poll_del(events, pollfd);
1272
1273 ret = close(pollfd);
1274 if (ret < 0) {
1275 ERR("Closing pollfd %d", pollfd);
1276 }
1277}
1278
1279static
1280int relay_add_connection(int fd, struct lttng_poll_event *events,
1281 struct lttng_ht *relay_connections_ht)
1282{
b8aa1682 1283 struct relay_command *relay_connection;
9d1bbf21 1284 int ret;
b8aa1682
JD
1285
1286 relay_connection = zmalloc(sizeof(struct relay_command));
1287 if (relay_connection == NULL) {
1288 PERROR("Relay command zmalloc");
9d1bbf21 1289 goto error;
b8aa1682
JD
1290 }
1291 ret = read(fd, relay_connection, sizeof(struct relay_command));
1292 if (ret < 0 || ret < sizeof(relay_connection)) {
1293 PERROR("read relay cmd pipe");
9d1bbf21 1294 goto error_read;
b8aa1682
JD
1295 }
1296
1297 lttng_ht_node_init_ulong(&relay_connection->sock_n,
1298 (unsigned long) relay_connection->sock->fd);
9d1bbf21 1299 rcu_read_lock();
b8aa1682
JD
1300 lttng_ht_add_unique_ulong(relay_connections_ht,
1301 &relay_connection->sock_n);
9d1bbf21
MD
1302 rcu_read_unlock();
1303 return lttng_poll_add(events,
b8aa1682
JD
1304 relay_connection->sock->fd,
1305 LPOLLIN | LPOLLRDHUP);
1306
9d1bbf21
MD
1307error_read:
1308 free(relay_connection);
1309error:
1310 return -1;
1311}
1312
1313static
1314void deferred_free_connection(struct rcu_head *head)
1315{
1316 struct relay_command *relay_connection =
1317 caa_container_of(head, struct relay_command, rcu_node);
1318 free(relay_connection);
1319}
1320
1321static
1322void relay_del_connection(struct lttng_ht *relay_connections_ht,
1323 struct lttng_ht *streams_ht, struct lttng_ht_iter *iter,
1324 struct relay_command *relay_connection)
1325{
1326 int ret;
1327
1328 ret = lttng_ht_del(relay_connections_ht, iter);
1329 assert(!ret);
1330 if (relay_connection->type == RELAY_CONTROL) {
1331 relay_delete_session(relay_connection, streams_ht);
1332 }
1333 call_rcu(&relay_connection->rcu_node,
1334 deferred_free_connection);
b8aa1682
JD
1335}
1336
1337/*
1338 * This thread does the actual work
1339 */
1340static
1341void *relay_thread_worker(void *data)
1342{
095a4ae5 1343 int i, ret, pollfd, err = -1;
b8aa1682
JD
1344 uint32_t revents, nb_fd;
1345 struct relay_command *relay_connection;
1346 struct lttng_poll_event events;
1347 struct lttng_ht *relay_connections_ht;
1348 struct lttng_ht_node_ulong *node;
1349 struct lttng_ht_iter iter;
1350 struct lttng_ht *streams_ht;
1351 struct lttcomm_relayd_hdr recv_hdr;
1352
1353 DBG("[thread] Relay worker started");
1354
9d1bbf21
MD
1355 rcu_register_thread();
1356
b8aa1682
JD
1357 /* table of connections indexed on socket */
1358 relay_connections_ht = lttng_ht_new(0, LTTNG_HT_TYPE_ULONG);
095a4ae5
MD
1359 if (!relay_connections_ht) {
1360 goto relay_connections_ht_error;
1361 }
b8aa1682
JD
1362
1363 /* tables of streams indexed by stream ID */
1364 streams_ht = lttng_ht_new(0, LTTNG_HT_TYPE_ULONG);
095a4ae5
MD
1365 if (!streams_ht) {
1366 goto streams_ht_error;
1367 }
b8aa1682
JD
1368
1369 ret = create_thread_poll_set(&events, 2);
1370 if (ret < 0) {
1371 goto error_poll_create;
1372 }
1373
1374 ret = lttng_poll_add(&events, relay_cmd_pipe[0], LPOLLIN | LPOLLRDHUP);
1375 if (ret < 0) {
1376 goto error;
1377 }
1378
1379 while (1) {
1380 /* Zeroed the events structure */
1381 lttng_poll_reset(&events);
1382
1383 nb_fd = LTTNG_POLL_GETNB(&events);
1384
1385 /* Infinite blocking call, waiting for transmission */
1386 restart:
87c1611d 1387 DBG3("Relayd worker thread polling...");
b8aa1682
JD
1388 ret = lttng_poll_wait(&events, -1);
1389 if (ret < 0) {
1390 /*
1391 * Restart interrupted system call.
1392 */
1393 if (errno == EINTR) {
1394 goto restart;
1395 }
1396 goto error;
1397 }
1398
1399 for (i = 0; i < nb_fd; i++) {
1400 /* Fetch once the poll data */
1401 revents = LTTNG_POLL_GETEV(&events, i);
1402 pollfd = LTTNG_POLL_GETFD(&events, i);
1403
1404 /* Thread quit pipe has been closed. Killing thread. */
1405 ret = check_thread_quit_pipe(pollfd, revents);
1406 if (ret) {
095a4ae5
MD
1407 err = 0;
1408 goto exit;
b8aa1682
JD
1409 }
1410
1411 /* Inspect the relay cmd pipe for new connection */
1412 if (pollfd == relay_cmd_pipe[0]) {
1413 if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
1414 ERR("Relay pipe error");
1415 goto error;
1416 } else if (revents & LPOLLIN) {
1417 DBG("Relay command received");
1418 ret = relay_add_connection(relay_cmd_pipe[0],
1419 &events, relay_connections_ht);
1420 if (ret < 0) {
1421 goto error;
1422 }
1423 }
1424 } else if (revents > 0) {
9d1bbf21 1425 rcu_read_lock();
b8aa1682
JD
1426 lttng_ht_lookup(relay_connections_ht,
1427 (void *)((unsigned long) pollfd),
1428 &iter);
1429 node = lttng_ht_iter_get_node_ulong(&iter);
1430 if (node == NULL) {
1431 DBG2("Relay sock %d not found", pollfd);
9d1bbf21 1432 rcu_read_unlock();
b8aa1682
JD
1433 goto error;
1434 }
1435 relay_connection = caa_container_of(node,
1436 struct relay_command, sock_n);
1437
1438 if (revents & (LPOLLERR)) {
1439 ERR("POLL ERROR");
9d1bbf21
MD
1440 relay_cleanup_poll_connection(&events, pollfd);
1441 relay_del_connection(relay_connections_ht,
1442 streams_ht, &iter,
1443 relay_connection);
b8aa1682
JD
1444 } else if (revents & (LPOLLHUP | LPOLLRDHUP)) {
1445 DBG("Socket %d hung up", pollfd);
9d1bbf21
MD
1446 relay_cleanup_poll_connection(&events, pollfd);
1447 relay_del_connection(relay_connections_ht,
1448 streams_ht, &iter,
1449 relay_connection);
b8aa1682
JD
1450 } else if (revents & LPOLLIN) {
1451 /* control socket */
1452 if (relay_connection->type == RELAY_CONTROL) {
1453 ret = relay_connection->sock->ops->recvmsg(
1454 relay_connection->sock, &recv_hdr,
1455 sizeof(struct lttcomm_relayd_hdr), MSG_WAITALL);
1456 /* connection closed */
1457 if (ret <= 0) {
9d1bbf21
MD
1458 relay_cleanup_poll_connection(&events, pollfd);
1459 relay_del_connection(relay_connections_ht,
1460 streams_ht, &iter,
1461 relay_connection);
b8aa1682
JD
1462 DBG("Control connection closed with %d", pollfd);
1463 } else {
1464 if (relay_connection->session) {
1465 DBG2("Relay worker receiving data for session : %lu",
1466 relay_connection->session->id);
1467 }
1468 ret = relay_process_control(&recv_hdr,
1469 relay_connection,
1470 streams_ht);
1471 /*
1472 * there was an error in processing a control
1473 * command: clear the session
1474 * */
1475 if (ret < 0) {
9d1bbf21
MD
1476 relay_cleanup_poll_connection(&events, pollfd);
1477 relay_del_connection(relay_connections_ht,
1478 streams_ht, &iter,
1479 relay_connection);
b8aa1682
JD
1480 DBG("Connection closed with %d", pollfd);
1481 }
1482 }
1483 /* data socket */
1484 } else if (relay_connection->type == RELAY_DATA) {
1485 ret = relay_process_data(relay_connection, streams_ht);
1486 /* connection closed */
1487 if (ret < 0) {
9d1bbf21
MD
1488 relay_cleanup_poll_connection(&events, pollfd);
1489 relay_del_connection(relay_connections_ht,
1490 streams_ht, &iter,
1491 relay_connection);
b8aa1682
JD
1492 DBG("Data connection closed with %d", pollfd);
1493 }
1494 }
1495 }
9d1bbf21 1496 rcu_read_unlock();
b8aa1682
JD
1497 }
1498 }
1499 }
1500
095a4ae5 1501exit:
b8aa1682
JD
1502error:
1503 lttng_poll_clean(&events);
1504
1505 /* empty the hash table and free the memory */
9d1bbf21 1506 rcu_read_lock();
b8aa1682
JD
1507 cds_lfht_for_each_entry(relay_connections_ht->ht, &iter.iter, node, node) {
1508 node = lttng_ht_iter_get_node_ulong(&iter);
1509 if (node) {
1510 relay_connection = caa_container_of(node,
1511 struct relay_command, sock_n);
9d1bbf21
MD
1512 relay_del_connection(relay_connections_ht,
1513 streams_ht, &iter,
1514 relay_connection);
b8aa1682 1515 }
b8aa1682 1516 }
9d1bbf21 1517 rcu_read_unlock();
b8aa1682 1518error_poll_create:
095a4ae5
MD
1519 lttng_ht_destroy(streams_ht);
1520streams_ht_error:
b8aa1682 1521 lttng_ht_destroy(relay_connections_ht);
095a4ae5
MD
1522relay_connections_ht_error:
1523 if (err) {
1524 DBG("Thread exited with error");
1525 }
b8aa1682 1526 DBG("Worker thread cleanup complete");
095a4ae5 1527 free(data_buffer);
b8aa1682 1528 stop_threads();
9d1bbf21 1529 rcu_unregister_thread();
b8aa1682
JD
1530 return NULL;
1531}
1532
1533/*
1534 * Create the relay command pipe to wake thread_manage_apps.
1535 * Closed in cleanup().
1536 */
1537static int create_relay_cmd_pipe(void)
1538{
a02de639 1539 int ret;
b8aa1682 1540
a02de639 1541 ret = utils_create_pipe_cloexec(relay_cmd_pipe);
b8aa1682 1542
b8aa1682
JD
1543 return ret;
1544}
1545
1546/*
1547 * main
1548 */
1549int main(int argc, char **argv)
1550{
1551 int ret = 0;
1552 void *status;
1553
1554 /* Create thread quit pipe */
1555 if ((ret = init_thread_quit_pipe()) < 0) {
1556 goto error;
1557 }
1558
1559 /* Parse arguments */
1560 progname = argv[0];
1561 if ((ret = parse_args(argc, argv) < 0)) {
a02de639 1562 goto exit;
b8aa1682
JD
1563 }
1564
1565 if ((ret = set_signal_handler()) < 0) {
1566 goto exit;
1567 }
1568
1569 /* Daemonize */
1570 if (opt_daemon) {
1571 ret = daemon(0, 0);
1572 if (ret < 0) {
1573 PERROR("daemon");
a02de639 1574 goto exit;
b8aa1682
JD
1575 }
1576 }
1577
1578 /* Check if daemon is UID = 0 */
1579 is_root = !getuid();
1580
1581 if (!is_root) {
1582 if (control_uri->port < 1024 || data_uri->port < 1024) {
1583 ERR("Need to be root to use ports < 1024");
1584 ret = -1;
a02de639 1585 goto exit;
b8aa1682
JD
1586 }
1587 }
1588
1589 /* Setup the thread apps communication pipe. */
1590 if ((ret = create_relay_cmd_pipe()) < 0) {
1591 goto exit;
1592 }
1593
1594 /* Init relay command queue. */
1595 cds_wfq_init(&relay_cmd_queue.queue);
1596
1597 /* Set up max poll set size */
1598 lttng_poll_set_max_size();
1599
1600 /* Setup the dispatcher thread */
1601 ret = pthread_create(&dispatcher_thread, NULL,
1602 relay_thread_dispatcher, (void *) NULL);
1603 if (ret != 0) {
1604 PERROR("pthread_create dispatcher");
1605 goto exit_dispatcher;
1606 }
1607
1608 /* Setup the worker thread */
1609 ret = pthread_create(&worker_thread, NULL,
1610 relay_thread_worker, (void *) NULL);
1611 if (ret != 0) {
1612 PERROR("pthread_create worker");
1613 goto exit_worker;
1614 }
1615
1616 /* Setup the listener thread */
1617 ret = pthread_create(&listener_thread, NULL,
1618 relay_thread_listener, (void *) NULL);
1619 if (ret != 0) {
1620 PERROR("pthread_create listener");
1621 goto exit_listener;
1622 }
1623
1624exit_listener:
1625 ret = pthread_join(listener_thread, &status);
1626 if (ret != 0) {
1627 PERROR("pthread_join");
1628 goto error; /* join error, exit without cleanup */
1629 }
1630
1631exit_worker:
1632 ret = pthread_join(worker_thread, &status);
1633 if (ret != 0) {
1634 PERROR("pthread_join");
1635 goto error; /* join error, exit without cleanup */
1636 }
1637
1638exit_dispatcher:
1639 ret = pthread_join(dispatcher_thread, &status);
1640 if (ret != 0) {
1641 PERROR("pthread_join");
1642 goto error; /* join error, exit without cleanup */
1643 }
1644
1645exit:
1646 cleanup();
1647 if (!ret) {
1648 exit(EXIT_SUCCESS);
1649 }
a02de639 1650
b8aa1682
JD
1651error:
1652 exit(EXIT_FAILURE);
1653}
This page took 0.151192 seconds and 4 git commands to generate.