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