Fix: relayd metadata size
[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
f6416125
MD
1073 data_size = payload_size = be64toh(recv_hdr->data_size);
1074 if (data_size < sizeof(struct lttcomm_relayd_metadata_payload)) {
1075 ERR("Incorrect data size");
1076 ret = -1;
1077 goto end;
1078 }
1079 payload_size -= sizeof(struct lttcomm_relayd_metadata_payload);
1080
b8aa1682
JD
1081 if (data_buffer_size < data_size) {
1082 data_buffer = realloc(data_buffer, data_size);
1083 if (!data_buffer) {
1084 ERR("Allocating data buffer");
1085 ret = -1;
1086 goto end;
1087 }
1088 data_buffer_size = data_size;
1089 }
1090 memset(data_buffer, 0, data_size);
1091 DBG2("Relay receiving metadata, waiting for %lu bytes", data_size);
87c1611d
DG
1092 ret = cmd->sock->ops->recvmsg(cmd->sock, data_buffer, data_size,
1093 MSG_WAITALL);
b8aa1682
JD
1094 if (ret < 0 || ret != data_size) {
1095 ret = -1;
1096 ERR("Relay didn't receive the whole metadata");
1097 goto end;
1098 }
1099 metadata_struct = (struct lttcomm_relayd_metadata_payload *) data_buffer;
9d1bbf21
MD
1100
1101 rcu_read_lock();
b8aa1682
JD
1102 metadata_stream = relay_stream_from_stream_id(
1103 be64toh(metadata_struct->stream_id), streams_ht);
1104 if (!metadata_stream) {
1105 ret = -1;
9d1bbf21 1106 goto end_unlock;
b8aa1682
JD
1107 }
1108
6f94560a
MD
1109 do {
1110 ret = write(metadata_stream->fd, metadata_struct->payload,
1111 payload_size);
1112 } while (ret < 0 && errno == EINTR);
87c1611d 1113 if (ret < payload_size) {
b8aa1682
JD
1114 ERR("Relay error writing metadata on file");
1115 ret = -1;
9d1bbf21 1116 goto end_unlock;
b8aa1682
JD
1117 }
1118 DBG2("Relay metadata written");
1119
9d1bbf21 1120end_unlock:
6e3c5836 1121 rcu_read_unlock();
b8aa1682
JD
1122end:
1123 return ret;
1124}
1125
1126/*
1127 * relay_send_version: send relayd version number
1128 */
1129static
1130int relay_send_version(struct lttcomm_relayd_hdr *recv_hdr,
1131 struct relay_command *cmd)
1132{
1133 int ret = htobe32(LTTCOMM_OK);
1134 struct lttcomm_relayd_version reply;
1135 struct relay_session *session = NULL;
1136
1137 if (cmd->session == NULL) {
1138 session = zmalloc(sizeof(struct relay_session));
1139 if (session == NULL) {
1140 PERROR("relay session zmalloc");
1141 ret = -1;
1142 goto end;
1143 }
1144 session->id = ++last_relay_session_id;
1145 DBG("Created session %lu", session->id);
1146 cmd->session = session;
1147 }
1148 session->version_check_done = 1;
1149
1150 sscanf(VERSION, "%u.%u", &reply.major, &reply.minor);
1151 reply.major = htobe32(reply.major);
1152 reply.minor = htobe32(reply.minor);
1153 ret = cmd->sock->ops->sendmsg(cmd->sock, &reply,
1154 sizeof(struct lttcomm_relayd_version), 0);
1155 if (ret < 0) {
1156 ERR("Relay sending version");
1157 }
1158 DBG("Version check done");
1159
1160end:
1161 return ret;
1162}
1163
1164/*
1165 * relay_process_control: Process the commands received on the control socket
1166 */
1167static
1168int relay_process_control(struct lttcomm_relayd_hdr *recv_hdr,
1169 struct relay_command *cmd, struct lttng_ht *streams_ht)
1170{
1171 int ret = 0;
1172
1173 switch (be32toh(recv_hdr->cmd)) {
1174 /*
1175 case RELAYD_CREATE_SESSION:
1176 ret = relay_create_session(recv_hdr, cmd);
1177 break;
1178 */
1179 case RELAYD_ADD_STREAM:
1180 ret = relay_add_stream(recv_hdr, cmd, streams_ht);
1181 break;
1182 case RELAYD_START_DATA:
1183 ret = relay_start(recv_hdr, cmd);
1184 break;
1185 case RELAYD_SEND_METADATA:
1186 ret = relay_recv_metadata(recv_hdr, cmd, streams_ht);
1187 break;
1188 case RELAYD_VERSION:
1189 ret = relay_send_version(recv_hdr, cmd);
1190 break;
1191 case RELAYD_UPDATE_SYNC_INFO:
1192 default:
1193 ERR("Received unknown command (%u)", be32toh(recv_hdr->cmd));
1194 relay_unknown_command(cmd);
1195 ret = -1;
1196 goto end;
1197 }
1198
1199end:
1200 return ret;
1201}
1202
1203/*
1204 * relay_process_data: Process the data received on the data socket
1205 */
1206static
1207int relay_process_data(struct relay_command *cmd, struct lttng_ht *streams_ht)
1208{
1209 int ret = 0;
1210 struct relay_stream *stream;
1211 struct lttcomm_relayd_data_hdr data_hdr;
1212 uint64_t stream_id;
1213 uint32_t data_size;
1214
1215 ret = cmd->sock->ops->recvmsg(cmd->sock, &data_hdr,
1216 sizeof(struct lttcomm_relayd_data_hdr), MSG_WAITALL);
1217 if (ret <= 0) {
1218 ERR("Connections seems to be closed");
1219 ret = -1;
1220 goto end;
1221 }
1222
1223 stream_id = be64toh(data_hdr.stream_id);
9d1bbf21
MD
1224
1225 rcu_read_lock();
b8aa1682
JD
1226 stream = relay_stream_from_stream_id(stream_id, streams_ht);
1227 if (!stream) {
1228 ret = -1;
9d1bbf21 1229 goto end_unlock;
b8aa1682
JD
1230 }
1231
1232 data_size = be32toh(data_hdr.data_size);
1233 if (data_buffer_size < data_size) {
1234 data_buffer = realloc(data_buffer, data_size);
1235 if (!data_buffer) {
1236 ERR("Allocating data buffer");
1237 ret = -1;
9d1bbf21 1238 goto end_unlock;
b8aa1682
JD
1239 }
1240 data_buffer_size = data_size;
1241 }
1242 memset(data_buffer, 0, data_size);
1243
87c1611d 1244 DBG3("Receiving data of size %u for stream id %zu", data_size, stream_id);
b8aa1682
JD
1245 ret = cmd->sock->ops->recvmsg(cmd->sock, data_buffer, data_size, MSG_WAITALL);
1246 if (ret <= 0) {
1247 ret = -1;
9d1bbf21 1248 goto end_unlock;
b8aa1682
JD
1249 }
1250
6f94560a
MD
1251 do {
1252 ret = write(stream->fd, data_buffer, data_size);
1253 } while (ret < 0 && errno == EINTR);
b8aa1682
JD
1254 if (ret < data_size) {
1255 ERR("Relay error writing data to file");
1256 ret = -1;
9d1bbf21 1257 goto end_unlock;
b8aa1682
JD
1258 }
1259 DBG2("Relay wrote %d bytes to tracefile for stream id %lu", ret, stream->stream_handle);
1260
9d1bbf21
MD
1261end_unlock:
1262 rcu_read_unlock();
b8aa1682
JD
1263end:
1264 return ret;
1265}
1266
1267static
9d1bbf21 1268void relay_cleanup_poll_connection(struct lttng_poll_event *events, int pollfd)
b8aa1682
JD
1269{
1270 int ret;
1271
b8aa1682
JD
1272 lttng_poll_del(events, pollfd);
1273
1274 ret = close(pollfd);
1275 if (ret < 0) {
1276 ERR("Closing pollfd %d", pollfd);
1277 }
1278}
1279
1280static
1281int relay_add_connection(int fd, struct lttng_poll_event *events,
1282 struct lttng_ht *relay_connections_ht)
1283{
b8aa1682 1284 struct relay_command *relay_connection;
9d1bbf21 1285 int ret;
b8aa1682
JD
1286
1287 relay_connection = zmalloc(sizeof(struct relay_command));
1288 if (relay_connection == NULL) {
1289 PERROR("Relay command zmalloc");
9d1bbf21 1290 goto error;
b8aa1682
JD
1291 }
1292 ret = read(fd, relay_connection, sizeof(struct relay_command));
1293 if (ret < 0 || ret < sizeof(relay_connection)) {
1294 PERROR("read relay cmd pipe");
9d1bbf21 1295 goto error_read;
b8aa1682
JD
1296 }
1297
1298 lttng_ht_node_init_ulong(&relay_connection->sock_n,
1299 (unsigned long) relay_connection->sock->fd);
9d1bbf21 1300 rcu_read_lock();
b8aa1682
JD
1301 lttng_ht_add_unique_ulong(relay_connections_ht,
1302 &relay_connection->sock_n);
9d1bbf21
MD
1303 rcu_read_unlock();
1304 return lttng_poll_add(events,
b8aa1682
JD
1305 relay_connection->sock->fd,
1306 LPOLLIN | LPOLLRDHUP);
1307
9d1bbf21
MD
1308error_read:
1309 free(relay_connection);
1310error:
1311 return -1;
1312}
1313
1314static
1315void deferred_free_connection(struct rcu_head *head)
1316{
1317 struct relay_command *relay_connection =
1318 caa_container_of(head, struct relay_command, rcu_node);
1319 free(relay_connection);
1320}
1321
1322static
1323void relay_del_connection(struct lttng_ht *relay_connections_ht,
1324 struct lttng_ht *streams_ht, struct lttng_ht_iter *iter,
1325 struct relay_command *relay_connection)
1326{
1327 int ret;
1328
1329 ret = lttng_ht_del(relay_connections_ht, iter);
1330 assert(!ret);
1331 if (relay_connection->type == RELAY_CONTROL) {
1332 relay_delete_session(relay_connection, streams_ht);
1333 }
1334 call_rcu(&relay_connection->rcu_node,
1335 deferred_free_connection);
b8aa1682
JD
1336}
1337
1338/*
1339 * This thread does the actual work
1340 */
1341static
1342void *relay_thread_worker(void *data)
1343{
095a4ae5 1344 int i, ret, pollfd, err = -1;
b8aa1682
JD
1345 uint32_t revents, nb_fd;
1346 struct relay_command *relay_connection;
1347 struct lttng_poll_event events;
1348 struct lttng_ht *relay_connections_ht;
1349 struct lttng_ht_node_ulong *node;
1350 struct lttng_ht_iter iter;
1351 struct lttng_ht *streams_ht;
1352 struct lttcomm_relayd_hdr recv_hdr;
1353
1354 DBG("[thread] Relay worker started");
1355
9d1bbf21
MD
1356 rcu_register_thread();
1357
b8aa1682
JD
1358 /* table of connections indexed on socket */
1359 relay_connections_ht = lttng_ht_new(0, LTTNG_HT_TYPE_ULONG);
095a4ae5
MD
1360 if (!relay_connections_ht) {
1361 goto relay_connections_ht_error;
1362 }
b8aa1682
JD
1363
1364 /* tables of streams indexed by stream ID */
1365 streams_ht = lttng_ht_new(0, LTTNG_HT_TYPE_ULONG);
095a4ae5
MD
1366 if (!streams_ht) {
1367 goto streams_ht_error;
1368 }
b8aa1682
JD
1369
1370 ret = create_thread_poll_set(&events, 2);
1371 if (ret < 0) {
1372 goto error_poll_create;
1373 }
1374
1375 ret = lttng_poll_add(&events, relay_cmd_pipe[0], LPOLLIN | LPOLLRDHUP);
1376 if (ret < 0) {
1377 goto error;
1378 }
1379
1380 while (1) {
1381 /* Zeroed the events structure */
1382 lttng_poll_reset(&events);
1383
1384 nb_fd = LTTNG_POLL_GETNB(&events);
1385
1386 /* Infinite blocking call, waiting for transmission */
1387 restart:
87c1611d 1388 DBG3("Relayd worker thread polling...");
b8aa1682
JD
1389 ret = lttng_poll_wait(&events, -1);
1390 if (ret < 0) {
1391 /*
1392 * Restart interrupted system call.
1393 */
1394 if (errno == EINTR) {
1395 goto restart;
1396 }
1397 goto error;
1398 }
1399
1400 for (i = 0; i < nb_fd; i++) {
1401 /* Fetch once the poll data */
1402 revents = LTTNG_POLL_GETEV(&events, i);
1403 pollfd = LTTNG_POLL_GETFD(&events, i);
1404
1405 /* Thread quit pipe has been closed. Killing thread. */
1406 ret = check_thread_quit_pipe(pollfd, revents);
1407 if (ret) {
095a4ae5
MD
1408 err = 0;
1409 goto exit;
b8aa1682
JD
1410 }
1411
1412 /* Inspect the relay cmd pipe for new connection */
1413 if (pollfd == relay_cmd_pipe[0]) {
1414 if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
1415 ERR("Relay pipe error");
1416 goto error;
1417 } else if (revents & LPOLLIN) {
1418 DBG("Relay command received");
1419 ret = relay_add_connection(relay_cmd_pipe[0],
1420 &events, relay_connections_ht);
1421 if (ret < 0) {
1422 goto error;
1423 }
1424 }
1425 } else if (revents > 0) {
9d1bbf21 1426 rcu_read_lock();
b8aa1682
JD
1427 lttng_ht_lookup(relay_connections_ht,
1428 (void *)((unsigned long) pollfd),
1429 &iter);
1430 node = lttng_ht_iter_get_node_ulong(&iter);
1431 if (node == NULL) {
1432 DBG2("Relay sock %d not found", pollfd);
9d1bbf21 1433 rcu_read_unlock();
b8aa1682
JD
1434 goto error;
1435 }
1436 relay_connection = caa_container_of(node,
1437 struct relay_command, sock_n);
1438
1439 if (revents & (LPOLLERR)) {
1440 ERR("POLL ERROR");
9d1bbf21
MD
1441 relay_cleanup_poll_connection(&events, pollfd);
1442 relay_del_connection(relay_connections_ht,
1443 streams_ht, &iter,
1444 relay_connection);
b8aa1682
JD
1445 } else if (revents & (LPOLLHUP | LPOLLRDHUP)) {
1446 DBG("Socket %d hung up", pollfd);
9d1bbf21
MD
1447 relay_cleanup_poll_connection(&events, pollfd);
1448 relay_del_connection(relay_connections_ht,
1449 streams_ht, &iter,
1450 relay_connection);
b8aa1682
JD
1451 } else if (revents & LPOLLIN) {
1452 /* control socket */
1453 if (relay_connection->type == RELAY_CONTROL) {
1454 ret = relay_connection->sock->ops->recvmsg(
1455 relay_connection->sock, &recv_hdr,
1456 sizeof(struct lttcomm_relayd_hdr), MSG_WAITALL);
1457 /* connection closed */
1458 if (ret <= 0) {
9d1bbf21
MD
1459 relay_cleanup_poll_connection(&events, pollfd);
1460 relay_del_connection(relay_connections_ht,
1461 streams_ht, &iter,
1462 relay_connection);
b8aa1682
JD
1463 DBG("Control connection closed with %d", pollfd);
1464 } else {
1465 if (relay_connection->session) {
1466 DBG2("Relay worker receiving data for session : %lu",
1467 relay_connection->session->id);
1468 }
1469 ret = relay_process_control(&recv_hdr,
1470 relay_connection,
1471 streams_ht);
1472 /*
1473 * there was an error in processing a control
1474 * command: clear the session
1475 * */
1476 if (ret < 0) {
9d1bbf21
MD
1477 relay_cleanup_poll_connection(&events, pollfd);
1478 relay_del_connection(relay_connections_ht,
1479 streams_ht, &iter,
1480 relay_connection);
b8aa1682
JD
1481 DBG("Connection closed with %d", pollfd);
1482 }
1483 }
1484 /* data socket */
1485 } else if (relay_connection->type == RELAY_DATA) {
1486 ret = relay_process_data(relay_connection, streams_ht);
1487 /* connection closed */
1488 if (ret < 0) {
9d1bbf21
MD
1489 relay_cleanup_poll_connection(&events, pollfd);
1490 relay_del_connection(relay_connections_ht,
1491 streams_ht, &iter,
1492 relay_connection);
b8aa1682
JD
1493 DBG("Data connection closed with %d", pollfd);
1494 }
1495 }
1496 }
9d1bbf21 1497 rcu_read_unlock();
b8aa1682
JD
1498 }
1499 }
1500 }
1501
095a4ae5 1502exit:
b8aa1682
JD
1503error:
1504 lttng_poll_clean(&events);
1505
1506 /* empty the hash table and free the memory */
9d1bbf21 1507 rcu_read_lock();
b8aa1682
JD
1508 cds_lfht_for_each_entry(relay_connections_ht->ht, &iter.iter, node, node) {
1509 node = lttng_ht_iter_get_node_ulong(&iter);
1510 if (node) {
1511 relay_connection = caa_container_of(node,
1512 struct relay_command, sock_n);
9d1bbf21
MD
1513 relay_del_connection(relay_connections_ht,
1514 streams_ht, &iter,
1515 relay_connection);
b8aa1682 1516 }
b8aa1682 1517 }
9d1bbf21 1518 rcu_read_unlock();
b8aa1682 1519error_poll_create:
095a4ae5
MD
1520 lttng_ht_destroy(streams_ht);
1521streams_ht_error:
b8aa1682 1522 lttng_ht_destroy(relay_connections_ht);
095a4ae5
MD
1523relay_connections_ht_error:
1524 if (err) {
1525 DBG("Thread exited with error");
1526 }
b8aa1682 1527 DBG("Worker thread cleanup complete");
095a4ae5 1528 free(data_buffer);
b8aa1682 1529 stop_threads();
9d1bbf21 1530 rcu_unregister_thread();
b8aa1682
JD
1531 return NULL;
1532}
1533
1534/*
1535 * Create the relay command pipe to wake thread_manage_apps.
1536 * Closed in cleanup().
1537 */
1538static int create_relay_cmd_pipe(void)
1539{
a02de639 1540 int ret;
b8aa1682 1541
a02de639 1542 ret = utils_create_pipe_cloexec(relay_cmd_pipe);
b8aa1682 1543
b8aa1682
JD
1544 return ret;
1545}
1546
1547/*
1548 * main
1549 */
1550int main(int argc, char **argv)
1551{
1552 int ret = 0;
1553 void *status;
1554
1555 /* Create thread quit pipe */
1556 if ((ret = init_thread_quit_pipe()) < 0) {
1557 goto error;
1558 }
1559
1560 /* Parse arguments */
1561 progname = argv[0];
1562 if ((ret = parse_args(argc, argv) < 0)) {
a02de639 1563 goto exit;
b8aa1682
JD
1564 }
1565
1566 if ((ret = set_signal_handler()) < 0) {
1567 goto exit;
1568 }
1569
1570 /* Daemonize */
1571 if (opt_daemon) {
1572 ret = daemon(0, 0);
1573 if (ret < 0) {
1574 PERROR("daemon");
a02de639 1575 goto exit;
b8aa1682
JD
1576 }
1577 }
1578
1579 /* Check if daemon is UID = 0 */
1580 is_root = !getuid();
1581
1582 if (!is_root) {
1583 if (control_uri->port < 1024 || data_uri->port < 1024) {
1584 ERR("Need to be root to use ports < 1024");
1585 ret = -1;
a02de639 1586 goto exit;
b8aa1682
JD
1587 }
1588 }
1589
1590 /* Setup the thread apps communication pipe. */
1591 if ((ret = create_relay_cmd_pipe()) < 0) {
1592 goto exit;
1593 }
1594
1595 /* Init relay command queue. */
1596 cds_wfq_init(&relay_cmd_queue.queue);
1597
1598 /* Set up max poll set size */
1599 lttng_poll_set_max_size();
1600
1601 /* Setup the dispatcher thread */
1602 ret = pthread_create(&dispatcher_thread, NULL,
1603 relay_thread_dispatcher, (void *) NULL);
1604 if (ret != 0) {
1605 PERROR("pthread_create dispatcher");
1606 goto exit_dispatcher;
1607 }
1608
1609 /* Setup the worker thread */
1610 ret = pthread_create(&worker_thread, NULL,
1611 relay_thread_worker, (void *) NULL);
1612 if (ret != 0) {
1613 PERROR("pthread_create worker");
1614 goto exit_worker;
1615 }
1616
1617 /* Setup the listener thread */
1618 ret = pthread_create(&listener_thread, NULL,
1619 relay_thread_listener, (void *) NULL);
1620 if (ret != 0) {
1621 PERROR("pthread_create listener");
1622 goto exit_listener;
1623 }
1624
1625exit_listener:
1626 ret = pthread_join(listener_thread, &status);
1627 if (ret != 0) {
1628 PERROR("pthread_join");
1629 goto error; /* join error, exit without cleanup */
1630 }
1631
1632exit_worker:
1633 ret = pthread_join(worker_thread, &status);
1634 if (ret != 0) {
1635 PERROR("pthread_join");
1636 goto error; /* join error, exit without cleanup */
1637 }
1638
1639exit_dispatcher:
1640 ret = pthread_join(dispatcher_thread, &status);
1641 if (ret != 0) {
1642 PERROR("pthread_join");
1643 goto error; /* join error, exit without cleanup */
1644 }
1645
1646exit:
1647 cleanup();
1648 if (!ret) {
1649 exit(EXIT_SUCCESS);
1650 }
a02de639 1651
b8aa1682
JD
1652error:
1653 exit(EXIT_FAILURE);
1654}
This page took 0.150257 seconds and 4 git commands to generate.