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