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