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