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