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