projects / lttng-tools.git / blob
? search:
summary | shortlog | log | commit | commitdiff | tree
blame | history | raw | HEAD
Fix: Relayd and sessiond version check
[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 <inttypes.h>
36 #include <urcu/futex.h>
37 #include <urcu/uatomic.h>
38 #include <unistd.h>
39 #include <fcntl.h>
40 #include <config.h>
41
42 #include <lttng/lttng.h>
43 #include <common/common.h>
44 #include <common/compat/poll.h>
45 #include <common/compat/socket.h>
46 #include <common/defaults.h>
47 #include <common/futex.h>
48 #include <common/sessiond-comm/sessiond-comm.h>
49 #include <common/sessiond-comm/inet.h>
50 #include <common/hashtable/hashtable.h>
51 #include <common/sessiond-comm/relayd.h>
52 #include <common/uri.h>
53 #include <common/utils.h>
54
55 #include "lttng-relayd.h"
56
57 /* command line options */
58 static int opt_daemon;
59 static char *opt_output_path;
60 static struct lttng_uri *control_uri;
61 static struct lttng_uri *data_uri;
62
63 const char *progname;
64 static int is_root; /* Set to 1 if the daemon is running as root */
65
66 /*
67 * Quit pipe for all threads. This permits a single cancellation point
68 * for all threads when receiving an event on the pipe.
69 */
70 static int thread_quit_pipe[2] = { -1, -1 };
71
72 /*
73 * This pipe is used to inform the worker thread that a command is queued and
74 * ready to be processed.
75 */
76 static int relay_cmd_pipe[2] = { -1, -1 };
77
78 /* Shared between threads */
79 static int dispatch_thread_exit;
80
81 static pthread_t listener_thread;
82 static pthread_t dispatcher_thread;
83 static pthread_t worker_thread;
84
85 static uint64_t last_relay_stream_id;
86 static uint64_t last_relay_session_id;
87
88 /*
89 * Relay command queue.
90 *
91 * The relay_thread_listener and relay_thread_dispatcher communicate with this
92 * queue.
93 */
94 static struct relay_cmd_queue relay_cmd_queue;
95
96 /* buffer allocated at startup, used to store the trace data */
97 static char *data_buffer;
98 static unsigned int data_buffer_size;
99
100 /*
101 * usage function on stderr
102 */
103 static
104 void usage(void)
105 {
106 fprintf(stderr, "Usage: %s OPTIONS\n\nOptions:\n", progname);
107 fprintf(stderr, " -h, --help Display this usage.\n");
108 fprintf(stderr, " -d, --daemonize Start as a daemon.\n");
109 fprintf(stderr, " -C, --control-port Control port listening (URI)\n");
110 fprintf(stderr, " -D, --data-port Data port listening (URI)\n");
111 fprintf(stderr, " -o, --output Output path for traces (PATH)\n");
112 fprintf(stderr, " -v, --verbose Verbose mode. Activate DBG() macro.\n");
113 }
114
115 static
116 int parse_args(int argc, char **argv)
117 {
118 int c;
119 int ret = 0;
120 char *default_address;
121
122 static struct option long_options[] = {
123 { "control-port", 1, 0, 'C', },
124 { "data-port", 1, 0, 'D', },
125 { "daemonize", 0, 0, 'd', },
126 { "help", 0, 0, 'h', },
127 { "output", 1, 0, 'o', },
128 { "verbose", 0, 0, 'v', },
129 { NULL, 0, 0, 0, },
130 };
131
132 while (1) {
133 int option_index = 0;
134 c = getopt_long(argc, argv, "dhv" "C:D:o:",
135 long_options, &option_index);
136 if (c == -1) {
137 break;
138 }
139
140 switch (c) {
141 case 0:
142 fprintf(stderr, "option %s", long_options[option_index].name);
143 if (optarg) {
144 fprintf(stderr, " with arg %s\n", optarg);
145 }
146 break;
147 case 'C':
148 ret = uri_parse(optarg, &control_uri);
149 if (ret < 0) {
150 ERR("Invalid control URI specified");
151 goto exit;
152 }
153 if (control_uri->port == 0) {
154 control_uri->port = DEFAULT_NETWORK_CONTROL_PORT;
155 }
156 break;
157 case 'D':
158 ret = uri_parse(optarg, &data_uri);
159 if (ret < 0) {
160 ERR("Invalid data URI specified");
161 goto exit;
162 }
163 if (data_uri->port == 0) {
164 data_uri->port = DEFAULT_NETWORK_DATA_PORT;
165 }
166 break;
167 case 'd':
168 opt_daemon = 1;
169 break;
170 case 'h':
171 usage();
172 exit(EXIT_FAILURE);
173 case 'o':
174 ret = asprintf(&opt_output_path, "%s", optarg);
175 if (ret < 0) {
176 PERROR("asprintf opt_output_path");
177 goto exit;
178 }
179 break;
180 case 'v':
181 /* Verbose level can increase using multiple -v */
182 lttng_opt_verbose += 1;
183 break;
184 default:
185 /* Unknown option or other error.
186 * Error is printed by getopt, just return */
187 ret = -1;
188 goto exit;
189 }
190 }
191
192 /* assign default values */
193 if (control_uri == NULL) {
194 ret = asprintf(&default_address, "tcp://0.0.0.0:%d",
195 DEFAULT_NETWORK_CONTROL_PORT);
196 if (ret < 0) {
197 PERROR("asprintf default data address");
198 goto exit;
199 }
200
201 ret = uri_parse(default_address, &control_uri);
202 free(default_address);
203 if (ret < 0) {
204 ERR("Invalid control URI specified");
205 goto exit;
206 }
207 }
208 if (data_uri == NULL) {
209 ret = asprintf(&default_address, "tcp://0.0.0.0:%d",
210 DEFAULT_NETWORK_DATA_PORT);
211 if (ret < 0) {
212 PERROR("asprintf default data address");
213 goto exit;
214 }
215
216 ret = uri_parse(default_address, &data_uri);
217 free(default_address);
218 if (ret < 0) {
219 ERR("Invalid data URI specified");
220 goto exit;
221 }
222 }
223
224 exit:
225 return ret;
226 }
227
228 /*
229 * Cleanup the daemon
230 */
231 static
232 void cleanup(void)
233 {
234 DBG("Cleaning up");
235
236 /* free the dynamically allocated opt_output_path */
237 free(opt_output_path);
238
239 /* Close thread quit pipes */
240 utils_close_pipe(thread_quit_pipe);
241
242 uri_free(control_uri);
243 uri_free(data_uri);
244 }
245
246 /*
247 * Write to writable pipe used to notify a thread.
248 */
249 static
250 int notify_thread_pipe(int wpipe)
251 {
252 int ret;
253
254 do {
255 ret = write(wpipe, "!", 1);
256 } while (ret < 0 && errno == EINTR);
257 if (ret < 0) {
258 PERROR("write poll pipe");
259 }
260
261 return ret;
262 }
263
264 /*
265 * Stop all threads by closing the thread quit pipe.
266 */
267 static
268 void stop_threads(void)
269 {
270 int ret;
271
272 /* Stopping all threads */
273 DBG("Terminating all threads");
274 ret = notify_thread_pipe(thread_quit_pipe[1]);
275 if (ret < 0) {
276 ERR("write error on thread quit pipe");
277 }
278
279 /* Dispatch thread */
280 CMM_STORE_SHARED(dispatch_thread_exit, 1);
281 futex_nto1_wake(&relay_cmd_queue.futex);
282 }
283
284 /*
285 * Signal handler for the daemon
286 *
287 * Simply stop all worker threads, leaving main() return gracefully after
288 * joining all threads and calling cleanup().
289 */
290 static
291 void sighandler(int sig)
292 {
293 switch (sig) {
294 case SIGPIPE:
295 DBG("SIGPIPE caught");
296 return;
297 case SIGINT:
298 DBG("SIGINT caught");
299 stop_threads();
300 break;
301 case SIGTERM:
302 DBG("SIGTERM caught");
303 stop_threads();
304 break;
305 default:
306 break;
307 }
308 }
309
310 /*
311 * Setup signal handler for :
312 * SIGINT, SIGTERM, SIGPIPE
313 */
314 static
315 int set_signal_handler(void)
316 {
317 int ret = 0;
318 struct sigaction sa;
319 sigset_t sigset;
320
321 if ((ret = sigemptyset(&sigset)) < 0) {
322 PERROR("sigemptyset");
323 return ret;
324 }
325
326 sa.sa_handler = sighandler;
327 sa.sa_mask = sigset;
328 sa.sa_flags = 0;
329 if ((ret = sigaction(SIGTERM, &sa, NULL)) < 0) {
330 PERROR("sigaction");
331 return ret;
332 }
333
334 if ((ret = sigaction(SIGINT, &sa, NULL)) < 0) {
335 PERROR("sigaction");
336 return ret;
337 }
338
339 if ((ret = sigaction(SIGPIPE, &sa, NULL)) < 0) {
340 PERROR("sigaction");
341 return ret;
342 }
343
344 DBG("Signal handler set for SIGTERM, SIGPIPE and SIGINT");
345
346 return ret;
347 }
348
349 /*
350 * Init thread quit pipe.
351 *
352 * Return -1 on error or 0 if all pipes are created.
353 */
354 static
355 int init_thread_quit_pipe(void)
356 {
357 int ret;
358
359 ret = utils_create_pipe_cloexec(thread_quit_pipe);
360
361 return ret;
362 }
363
364 /*
365 * Create a poll set with O_CLOEXEC and add the thread quit pipe to the set.
366 */
367 static
368 int create_thread_poll_set(struct lttng_poll_event *events, int size)
369 {
370 int ret;
371
372 if (events == NULL || size == 0) {
373 ret = -1;
374 goto error;
375 }
376
377 ret = lttng_poll_create(events, size, LTTNG_CLOEXEC);
378 if (ret < 0) {
379 goto error;
380 }
381
382 /* Add quit pipe */
383 ret = lttng_poll_add(events, thread_quit_pipe[0], LPOLLIN);
384 if (ret < 0) {
385 goto error;
386 }
387
388 return 0;
389
390 error:
391 return ret;
392 }
393
394 /*
395 * Check if the thread quit pipe was triggered.
396 *
397 * Return 1 if it was triggered else 0;
398 */
399 static
400 int check_thread_quit_pipe(int fd, uint32_t events)
401 {
402 if (fd == thread_quit_pipe[0] && (events & LPOLLIN)) {
403 return 1;
404 }
405
406 return 0;
407 }
408
409 /*
410 * Create and init socket from uri.
411 */
412 static
413 struct lttcomm_sock *relay_init_sock(struct lttng_uri *uri)
414 {
415 int ret;
416 struct lttcomm_sock *sock = NULL;
417
418 sock = lttcomm_alloc_sock_from_uri(uri);
419 if (sock == NULL) {
420 ERR("Allocating socket");
421 goto error;
422 }
423
424 ret = lttcomm_create_sock(sock);
425 if (ret < 0) {
426 goto error;
427 }
428 DBG("Listening on sock %d", sock->fd);
429
430 ret = sock->ops->bind(sock);
431 if (ret < 0) {
432 goto error;
433 }
434
435 ret = sock->ops->listen(sock, -1);
436 if (ret < 0) {
437 goto error;
438
439 }
440
441 return sock;
442
443 error:
444 if (sock) {
445 lttcomm_destroy_sock(sock);
446 }
447 return NULL;
448 }
449
450 /*
451 * Return nonzero if stream needs to be closed.
452 */
453 static
454 int close_stream_check(struct relay_stream *stream)
455 {
456
457 if (stream->close_flag && stream->prev_seq == stream->last_net_seq_num) {
458 return 1;
459 }
460 return 0;
461 }
462
463 /*
464 * This thread manages the listening for new connections on the network
465 */
466 static
467 void *relay_thread_listener(void *data)
468 {
469 int i, ret, pollfd, err = -1;
470 int val = 1;
471 uint32_t revents, nb_fd;
472 struct lttng_poll_event events;
473 struct lttcomm_sock *control_sock, *data_sock;
474
475 DBG("[thread] Relay listener started");
476
477 control_sock = relay_init_sock(control_uri);
478 if (!control_sock) {
479 goto error_sock_control;
480 }
481
482 data_sock = relay_init_sock(data_uri);
483 if (!data_sock) {
484 goto error_sock_relay;
485 }
486
487 /*
488 * Pass 3 as size here for the thread quit pipe, control and data socket.
489 */
490 ret = create_thread_poll_set(&events, 3);
491 if (ret < 0) {
492 goto error_create_poll;
493 }
494
495 /* Add the control socket */
496 ret = lttng_poll_add(&events, control_sock->fd, LPOLLIN | LPOLLRDHUP);
497 if (ret < 0) {
498 goto error_poll_add;
499 }
500
501 /* Add the data socket */
502 ret = lttng_poll_add(&events, data_sock->fd, LPOLLIN | LPOLLRDHUP);
503 if (ret < 0) {
504 goto error_poll_add;
505 }
506
507 while (1) {
508 DBG("Listener accepting connections");
509
510 nb_fd = LTTNG_POLL_GETNB(&events);
511
512 restart:
513 ret = lttng_poll_wait(&events, -1);
514 if (ret < 0) {
515 /*
516 * Restart interrupted system call.
517 */
518 if (errno == EINTR) {
519 goto restart;
520 }
521 goto error;
522 }
523
524 DBG("Relay new connection received");
525 for (i = 0; i < nb_fd; i++) {
526 /* Fetch once the poll data */
527 revents = LTTNG_POLL_GETEV(&events, i);
528 pollfd = LTTNG_POLL_GETFD(&events, i);
529
530 /* Thread quit pipe has been closed. Killing thread. */
531 ret = check_thread_quit_pipe(pollfd, revents);
532 if (ret) {
533 err = 0;
534 goto exit;
535 }
536
537 if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
538 ERR("socket poll error");
539 goto error;
540 } else if (revents & LPOLLIN) {
541 /*
542 * Get allocated in this thread,
543 * enqueued to a global queue, dequeued
544 * and freed in the worker thread.
545 */
546 struct relay_command *relay_cmd;
547 struct lttcomm_sock *newsock;
548
549 relay_cmd = zmalloc(sizeof(struct relay_command));
550 if (relay_cmd == NULL) {
551 PERROR("relay command zmalloc");
552 goto error;
553 }
554
555 if (pollfd == data_sock->fd) {
556 newsock = data_sock->ops->accept(data_sock);
557 if (!newsock) {
558 PERROR("accepting data sock");
559 free(relay_cmd);
560 goto error;
561 }
562 relay_cmd->type = RELAY_DATA;
563 DBG("Relay data connection accepted, socket %d", newsock->fd);
564 } else {
565 assert(pollfd == control_sock->fd);
566 newsock = control_sock->ops->accept(control_sock);
567 if (!newsock) {
568 PERROR("accepting control sock");
569 free(relay_cmd);
570 goto error;
571 }
572 relay_cmd->type = RELAY_CONTROL;
573 DBG("Relay control connection accepted, socket %d", newsock->fd);
574 }
575 ret = setsockopt(newsock->fd, SOL_SOCKET, SO_REUSEADDR,
576 &val, sizeof(int));
577 if (ret < 0) {
578 PERROR("setsockopt inet");
579 lttcomm_destroy_sock(newsock);
580 free(relay_cmd);
581 goto error;
582 }
583 relay_cmd->sock = newsock;
584 /*
585 * Lock free enqueue the request.
586 */
587 cds_wfq_enqueue(&relay_cmd_queue.queue, &relay_cmd->node);
588
589 /*
590 * Wake the dispatch queue futex. Implicit memory
591 * barrier with the exchange in cds_wfq_enqueue.
592 */
593 futex_nto1_wake(&relay_cmd_queue.futex);
594 }
595 }
596 }
597
598 exit:
599 error:
600 error_poll_add:
601 lttng_poll_clean(&events);
602 error_create_poll:
603 if (data_sock->fd >= 0) {
604 ret = data_sock->ops->close(data_sock);
605 if (ret) {
606 PERROR("close");
607 }
608 }
609 lttcomm_destroy_sock(data_sock);
610 error_sock_relay:
611 if (control_sock->fd >= 0) {
612 ret = control_sock->ops->close(control_sock);
613 if (ret) {
614 PERROR("close");
615 }
616 }
617 lttcomm_destroy_sock(control_sock);
618 error_sock_control:
619 if (err) {
620 DBG("Thread exited with error");
621 }
622 DBG("Relay listener thread cleanup complete");
623 stop_threads();
624 return NULL;
625 }
626
627 /*
628 * This thread manages the dispatching of the requests to worker threads
629 */
630 static
631 void *relay_thread_dispatcher(void *data)
632 {
633 int ret;
634 struct cds_wfq_node *node;
635 struct relay_command *relay_cmd = NULL;
636
637 DBG("[thread] Relay dispatcher started");
638
639 while (!CMM_LOAD_SHARED(dispatch_thread_exit)) {
640 /* Atomically prepare the queue futex */
641 futex_nto1_prepare(&relay_cmd_queue.futex);
642
643 do {
644 /* Dequeue commands */
645 node = cds_wfq_dequeue_blocking(&relay_cmd_queue.queue);
646 if (node == NULL) {
647 DBG("Woken up but nothing in the relay command queue");
648 /* Continue thread execution */
649 break;
650 }
651
652 relay_cmd = caa_container_of(node, struct relay_command, node);
653 DBG("Dispatching request waiting on sock %d", relay_cmd->sock->fd);
654
655 /*
656 * Inform worker thread of the new request. This
657 * call is blocking so we can be assured that the data will be read
658 * at some point in time or wait to the end of the world :)
659 */
660 do {
661 ret = write(relay_cmd_pipe[1], relay_cmd,
662 sizeof(struct relay_command));
663 } while (ret < 0 && errno == EINTR);
664 free(relay_cmd);
665 if (ret < 0) {
666 PERROR("write cmd pipe");
667 goto error;
668 }
669 } while (node != NULL);
670
671 /* Futex wait on queue. Blocking call on futex() */
672 futex_nto1_wait(&relay_cmd_queue.futex);
673 }
674
675 error:
676 DBG("Dispatch thread dying");
677 stop_threads();
678 return NULL;
679 }
680
681 /*
682 * Return the realpath(3) of the path even if the last directory token does not
683 * exist. For example, with /tmp/test1/test2, if test2/ does not exist but the
684 * /tmp/test1 does, the real path is returned. In normal time, realpath(3)
685 * fails if the end point directory does not exist.
686 */
687 static
688 char *expand_full_path(const char *path)
689 {
690 const char *end_path = path;
691 char *next, *cut_path, *expanded_path, *respath;
692
693 /* Find last token delimited by '/' */
694 while ((next = strpbrk(end_path + 1, "/"))) {
695 end_path = next;
696 }
697
698 /* Cut last token from original path */
699 cut_path = strndup(path, end_path - path);
700
701 expanded_path = malloc(PATH_MAX);
702 if (expanded_path == NULL) {
703 respath = NULL;
704 goto end;
705 }
706
707 respath = realpath(cut_path, expanded_path);
708 if (respath == NULL) {
709 switch (errno) {
710 case ENOENT:
711 ERR("%s: No such file or directory", cut_path);
712 break;
713 default:
714 PERROR("realpath");
715 break;
716 }
717 free(expanded_path);
718 } else {
719 /* Add end part to expanded path */
720 strcat(respath, end_path);
721 }
722 end:
723 free(cut_path);
724 return respath;
725 }
726
727
728 /*
729 * config_get_default_path
730 *
731 * Returns the HOME directory path. Caller MUST NOT free(3) the return pointer.
732 */
733 static
734 char *config_get_default_path(void)
735 {
736 return getenv("HOME");
737 }
738
739 /*
740 * Create recursively directory using the FULL path.
741 */
742 static
743 int mkdir_recursive(char *path, mode_t mode)
744 {
745 char *p, tmp[PATH_MAX];
746 struct stat statbuf;
747 size_t len;
748 int ret;
749
750 ret = snprintf(tmp, sizeof(tmp), "%s", path);
751 if (ret < 0) {
752 PERROR("snprintf mkdir");
753 goto error;
754 }
755
756 len = ret;
757 if (tmp[len - 1] == '/') {
758 tmp[len - 1] = 0;
759 }
760
761 for (p = tmp + 1; *p; p++) {
762 if (*p == '/') {
763 *p = 0;
764 if (tmp[strlen(tmp) - 1] == '.' &&
765 tmp[strlen(tmp) - 2] == '.' &&
766 tmp[strlen(tmp) - 3] == '/') {
767 ERR("Using '/../' is not permitted in the trace path (%s)",
768 tmp);
769 ret = -1;
770 goto error;
771 }
772 ret = stat(tmp, &statbuf);
773 if (ret < 0) {
774 ret = mkdir(tmp, mode);
775 if (ret < 0) {
776 if (errno != EEXIST) {
777 PERROR("mkdir recursive");
778 ret = -errno;
779 goto error;
780 }
781 }
782 }
783 *p = '/';
784 }
785 }
786
787 ret = mkdir(tmp, mode);
788 if (ret < 0) {
789 if (errno != EEXIST) {
790 PERROR("mkdir recursive last piece");
791 ret = -errno;
792 } else {
793 ret = 0;
794 }
795 }
796
797 error:
798 return ret;
799 }
800
801 static
802 char *create_output_path_auto(char *path_name)
803 {
804 int ret;
805 char *traces_path = NULL;
806 char *alloc_path = NULL;
807 char *default_path;
808
809 default_path = config_get_default_path();
810 if (default_path == NULL) {
811 ERR("Home path not found.\n \
812 Please specify an output path using -o, --output PATH");
813 goto exit;
814 }
815 alloc_path = strdup(default_path);
816 if (alloc_path == NULL) {
817 PERROR("Path allocation");
818 goto exit;
819 }
820 ret = asprintf(&traces_path, "%s/" DEFAULT_TRACE_DIR_NAME
821 "/%s", alloc_path, path_name);
822 if (ret < 0) {
823 PERROR("asprintf trace dir name");
824 goto exit;
825 }
826 exit:
827 free(alloc_path);
828 return traces_path;
829 }
830
831 static
832 char *create_output_path_noauto(char *path_name)
833 {
834 int ret;
835 char *traces_path = NULL;
836 char *full_path;
837
838 full_path = expand_full_path(opt_output_path);
839 ret = asprintf(&traces_path, "%s/%s", full_path, path_name);
840 if (ret < 0) {
841 PERROR("asprintf trace dir name");
842 goto exit;
843 }
844 exit:
845 free(full_path);
846 return traces_path;
847 }
848
849 /*
850 * create_output_path: create the output trace directory
851 */
852 static
853 char *create_output_path(char *path_name)
854 {
855 if (opt_output_path == NULL) {
856 return create_output_path_auto(path_name);
857 } else {
858 return create_output_path_noauto(path_name);
859 }
860 }
861
862 static
863 void deferred_free_stream(struct rcu_head *head)
864 {
865 struct relay_stream *stream =
866 caa_container_of(head, struct relay_stream, rcu_node);
867 free(stream);
868 }
869
870 /*
871 * relay_delete_session: Free all memory associated with a session and
872 * close all the FDs
873 */
874 static
875 void relay_delete_session(struct relay_command *cmd, struct lttng_ht *streams_ht)
876 {
877 struct lttng_ht_iter iter;
878 struct lttng_ht_node_ulong *node;
879 struct relay_stream *stream;
880 int ret;
881
882 if (!cmd->session) {
883 return;
884 }
885
886 DBG("Relay deleting session %" PRIu64, cmd->session->id);
887
888 lttcomm_destroy_sock(cmd->session->sock);
889
890 rcu_read_lock();
891 cds_lfht_for_each_entry(streams_ht->ht, &iter.iter, node, node) {
892 node = lttng_ht_iter_get_node_ulong(&iter);
893 if (node) {
894 stream = caa_container_of(node,
895 struct relay_stream, stream_n);
896 if (stream->session == cmd->session) {
897 ret = close(stream->fd);
898 if (ret < 0) {
899 PERROR("close stream fd on delete session");
900 }
901 ret = lttng_ht_del(streams_ht, &iter);
902 assert(!ret);
903 call_rcu(&stream->rcu_node,
904 deferred_free_stream);
905 }
906 }
907 }
908 rcu_read_unlock();
909
910 free(cmd->session);
911 }
912
913 /*
914 * relay_add_stream: allocate a new stream for a session
915 */
916 static
917 int relay_add_stream(struct lttcomm_relayd_hdr *recv_hdr,
918 struct relay_command *cmd, struct lttng_ht *streams_ht)
919 {
920 struct relay_session *session = cmd->session;
921 struct lttcomm_relayd_add_stream stream_info;
922 struct relay_stream *stream = NULL;
923 struct lttcomm_relayd_status_stream reply;
924 char *path = NULL, *root_path = NULL;
925 int ret, send_ret;
926
927 if (!session || session->version_check_done == 0) {
928 ERR("Trying to add a stream before version check");
929 ret = -1;
930 goto end_no_session;
931 }
932
933 /* FIXME : use data_size for something ? */
934 ret = cmd->sock->ops->recvmsg(cmd->sock, &stream_info,
935 sizeof(struct lttcomm_relayd_add_stream), MSG_WAITALL);
936 if (ret < sizeof(struct lttcomm_relayd_add_stream)) {
937 ERR("Relay didn't receive valid add_stream struct size : %d", ret);
938 ret = -1;
939 goto end_no_session;
940 }
941 stream = zmalloc(sizeof(struct relay_stream));
942 if (stream == NULL) {
943 PERROR("relay stream zmalloc");
944 ret = -1;
945 goto end_no_session;
946 }
947
948 rcu_read_lock();
949 stream->stream_handle = ++last_relay_stream_id;
950 stream->prev_seq = -1ULL;
951 stream->session = session;
952
953 root_path = create_output_path(stream_info.pathname);
954 if (!root_path) {
955 ret = -1;
956 goto end;
957 }
958 ret = mkdir_recursive(root_path, S_IRWXU | S_IRWXG);
959 if (ret < 0) {
960 ERR("relay creating output directory");
961 goto end;
962 }
963
964 ret = asprintf(&path, "%s/%s", root_path, stream_info.channel_name);
965 if (ret < 0) {
966 PERROR("asprintf stream path");
967 goto end;
968 }
969
970 ret = open(path, O_WRONLY|O_CREAT|O_TRUNC, S_IRWXU|S_IRWXG|S_IRWXO);
971 if (ret < 0) {
972 PERROR("Relay creating trace file");
973 goto end;
974 }
975
976 stream->fd = ret;
977 DBG("Tracefile %s created", path);
978
979 lttng_ht_node_init_ulong(&stream->stream_n,
980 (unsigned long) stream->stream_handle);
981 lttng_ht_add_unique_ulong(streams_ht,
982 &stream->stream_n);
983
984 DBG("Relay new stream added %s", stream_info.channel_name);
985
986 end:
987 free(path);
988 free(root_path);
989 /* send the session id to the client or a negative return code on error */
990 if (ret < 0) {
991 reply.ret_code = htobe32(LTTNG_ERR_UNK);
992 } else {
993 reply.ret_code = htobe32(LTTNG_OK);
994 }
995 reply.handle = htobe64(stream->stream_handle);
996 send_ret = cmd->sock->ops->sendmsg(cmd->sock, &reply,
997 sizeof(struct lttcomm_relayd_status_stream), 0);
998 if (send_ret < 0) {
999 ERR("Relay sending stream id");
1000 }
1001 rcu_read_unlock();
1002
1003 end_no_session:
1004 return ret;
1005 }
1006
1007 /*
1008 * relay_close_stream: close a specific stream
1009 */
1010 static
1011 int relay_close_stream(struct lttcomm_relayd_hdr *recv_hdr,
1012 struct relay_command *cmd, struct lttng_ht *streams_ht)
1013 {
1014 struct relay_session *session = cmd->session;
1015 struct lttcomm_relayd_close_stream stream_info;
1016 struct lttcomm_relayd_generic_reply reply;
1017 struct relay_stream *stream;
1018 int ret, send_ret;
1019 struct lttng_ht_node_ulong *node;
1020 struct lttng_ht_iter iter;
1021
1022 DBG("Close stream received");
1023
1024 if (!session || session->version_check_done == 0) {
1025 ERR("Trying to close a stream before version check");
1026 ret = -1;
1027 goto end_no_session;
1028 }
1029
1030 ret = cmd->sock->ops->recvmsg(cmd->sock, &stream_info,
1031 sizeof(struct lttcomm_relayd_close_stream), MSG_WAITALL);
1032 if (ret < sizeof(struct lttcomm_relayd_close_stream)) {
1033 ERR("Relay didn't receive valid add_stream struct size : %d", ret);
1034 ret = -1;
1035 goto end_no_session;
1036 }
1037
1038 rcu_read_lock();
1039 lttng_ht_lookup(streams_ht,
1040 (void *)((unsigned long) be64toh(stream_info.stream_id)),
1041 &iter);
1042 node = lttng_ht_iter_get_node_ulong(&iter);
1043 if (node == NULL) {
1044 DBG("Relay stream %" PRIu64 " not found", be64toh(stream_info.stream_id));
1045 ret = -1;
1046 goto end_unlock;
1047 }
1048
1049 stream = caa_container_of(node, struct relay_stream, stream_n);
1050 if (!stream) {
1051 ret = -1;
1052 goto end_unlock;
1053 }
1054
1055 stream->last_net_seq_num = be64toh(stream_info.last_net_seq_num);
1056 stream->close_flag = 1;
1057
1058 if (close_stream_check(stream)) {
1059 int delret;
1060
1061 delret = close(stream->fd);
1062 if (delret < 0) {
1063 PERROR("close stream");
1064 }
1065 delret = lttng_ht_del(streams_ht, &iter);
1066 assert(!delret);
1067 call_rcu(&stream->rcu_node,
1068 deferred_free_stream);
1069 DBG("Closed tracefile %d from close stream", stream->fd);
1070 }
1071
1072 end_unlock:
1073 rcu_read_unlock();
1074
1075 if (ret < 0) {
1076 reply.ret_code = htobe32(LTTNG_ERR_UNK);
1077 } else {
1078 reply.ret_code = htobe32(LTTNG_OK);
1079 }
1080 send_ret = cmd->sock->ops->sendmsg(cmd->sock, &reply,
1081 sizeof(struct lttcomm_relayd_generic_reply), 0);
1082 if (send_ret < 0) {
1083 ERR("Relay sending stream id");
1084 }
1085
1086 end_no_session:
1087 return ret;
1088 }
1089
1090 /*
1091 * relay_unknown_command: send -1 if received unknown command
1092 */
1093 static
1094 void relay_unknown_command(struct relay_command *cmd)
1095 {
1096 struct lttcomm_relayd_generic_reply reply;
1097 int ret;
1098
1099 reply.ret_code = htobe32(LTTNG_ERR_UNK);
1100 ret = cmd->sock->ops->sendmsg(cmd->sock, &reply,
1101 sizeof(struct lttcomm_relayd_generic_reply), 0);
1102 if (ret < 0) {
1103 ERR("Relay sending unknown command");
1104 }
1105 }
1106
1107 /*
1108 * relay_start: send an acknowledgment to the client to tell if we are
1109 * ready to receive data. We are ready if a session is established.
1110 */
1111 static
1112 int relay_start(struct lttcomm_relayd_hdr *recv_hdr,
1113 struct relay_command *cmd)
1114 {
1115 int ret = htobe32(LTTNG_OK);
1116 struct lttcomm_relayd_generic_reply reply;
1117 struct relay_session *session = cmd->session;
1118
1119 if (!session) {
1120 DBG("Trying to start the streaming without a session established");
1121 ret = htobe32(LTTNG_ERR_UNK);
1122 }
1123
1124 reply.ret_code = ret;
1125 ret = cmd->sock->ops->sendmsg(cmd->sock, &reply,
1126 sizeof(struct lttcomm_relayd_generic_reply), 0);
1127 if (ret < 0) {
1128 ERR("Relay sending start ack");
1129 }
1130
1131 return ret;
1132 }
1133
1134 /*
1135 * Get stream from stream id.
1136 * Need to be called with RCU read-side lock held.
1137 */
1138 static
1139 struct relay_stream *relay_stream_from_stream_id(uint64_t stream_id,
1140 struct lttng_ht *streams_ht)
1141 {
1142 struct lttng_ht_node_ulong *node;
1143 struct lttng_ht_iter iter;
1144 struct relay_stream *ret;
1145
1146 lttng_ht_lookup(streams_ht,
1147 (void *)((unsigned long) stream_id),
1148 &iter);
1149 node = lttng_ht_iter_get_node_ulong(&iter);
1150 if (node == NULL) {
1151 DBG("Relay stream %" PRIu64 " not found", stream_id);
1152 ret = NULL;
1153 goto end;
1154 }
1155
1156 ret = caa_container_of(node, struct relay_stream, stream_n);
1157
1158 end:
1159 return ret;
1160 }
1161
1162 /*
1163 * Append padding to the file pointed by the file descriptor fd.
1164 */
1165 static int write_padding_to_file(int fd, uint32_t size)
1166 {
1167 int ret = 0;
1168 char *zeros;
1169
1170 if (size == 0) {
1171 goto end;
1172 }
1173
1174 zeros = zmalloc(size);
1175 if (zeros == NULL) {
1176 PERROR("zmalloc zeros for padding");
1177 ret = -1;
1178 goto end;
1179 }
1180
1181 do {
1182 ret = write(fd, zeros, size);
1183 } while (ret < 0 && errno == EINTR);
1184 if (ret < 0) {
1185 PERROR("write padding to file");
1186 }
1187
1188 free(zeros);
1189
1190 end:
1191 return ret;
1192 }
1193
1194 /*
1195 * relay_recv_metadata: receive the metada for the session.
1196 */
1197 static
1198 int relay_recv_metadata(struct lttcomm_relayd_hdr *recv_hdr,
1199 struct relay_command *cmd, struct lttng_ht *streams_ht)
1200 {
1201 int ret = htobe32(LTTNG_OK);
1202 struct relay_session *session = cmd->session;
1203 struct lttcomm_relayd_metadata_payload *metadata_struct;
1204 struct relay_stream *metadata_stream;
1205 uint64_t data_size, payload_size;
1206
1207 if (!session) {
1208 ERR("Metadata sent before version check");
1209 ret = -1;
1210 goto end;
1211 }
1212
1213 data_size = payload_size = be64toh(recv_hdr->data_size);
1214 if (data_size < sizeof(struct lttcomm_relayd_metadata_payload)) {
1215 ERR("Incorrect data size");
1216 ret = -1;
1217 goto end;
1218 }
1219 payload_size -= sizeof(struct lttcomm_relayd_metadata_payload);
1220
1221 if (data_buffer_size < data_size) {
1222 /* In case the realloc fails, we can free the memory */
1223 char *tmp_data_ptr = data_buffer;
1224 data_buffer = realloc(data_buffer, data_size);
1225 if (!data_buffer) {
1226 ERR("Allocating data buffer");
1227 free(tmp_data_ptr);
1228 ret = -1;
1229 goto end;
1230 }
1231 data_buffer_size = data_size;
1232 }
1233 memset(data_buffer, 0, data_size);
1234 DBG2("Relay receiving metadata, waiting for %" PRIu64 " bytes", data_size);
1235 ret = cmd->sock->ops->recvmsg(cmd->sock, data_buffer, data_size,
1236 MSG_WAITALL);
1237 if (ret < 0 || ret != data_size) {
1238 ret = -1;
1239 ERR("Relay didn't receive the whole metadata");
1240 goto end;
1241 }
1242 metadata_struct = (struct lttcomm_relayd_metadata_payload *) data_buffer;
1243
1244 rcu_read_lock();
1245 metadata_stream = relay_stream_from_stream_id(
1246 be64toh(metadata_struct->stream_id), streams_ht);
1247 if (!metadata_stream) {
1248 ret = -1;
1249 goto end_unlock;
1250 }
1251
1252 do {
1253 ret = write(metadata_stream->fd, metadata_struct->payload,
1254 payload_size);
1255 } while (ret < 0 && errno == EINTR);
1256 if (ret < payload_size) {
1257 ERR("Relay error writing metadata on file");
1258 ret = -1;
1259 goto end_unlock;
1260 }
1261
1262 ret = write_padding_to_file(metadata_stream->fd,
1263 be32toh(metadata_struct->padding_size));
1264 if (ret < 0) {
1265 goto end_unlock;
1266 }
1267
1268 DBG2("Relay metadata written");
1269
1270 end_unlock:
1271 rcu_read_unlock();
1272 end:
1273 return ret;
1274 }
1275
1276 /*
1277 * relay_send_version: send relayd version number
1278 */
1279 static
1280 int relay_send_version(struct lttcomm_relayd_hdr *recv_hdr,
1281 struct relay_command *cmd)
1282 {
1283 int ret;
1284 struct lttcomm_relayd_version reply, msg;
1285 struct relay_session *session;
1286
1287 if (cmd->session == NULL) {
1288 session = zmalloc(sizeof(struct relay_session));
1289 if (session == NULL) {
1290 PERROR("relay session zmalloc");
1291 ret = -1;
1292 goto end;
1293 }
1294 session->id = ++last_relay_session_id;
1295 DBG("Created session %" PRIu64, session->id);
1296 cmd->session = session;
1297 } else {
1298 session = cmd->session;
1299 }
1300 session->version_check_done = 1;
1301
1302 /* Get version from the other side. */
1303 ret = cmd->sock->ops->recvmsg(cmd->sock, &msg, sizeof(msg), MSG_WAITALL);
1304 if (ret < 0 || ret != sizeof(msg)) {
1305 ret = -1;
1306 ERR("Relay failed to receive the version values.");
1307 goto end;
1308 }
1309
1310 /*
1311 * For now, we just ignore the received version but after 2.1 stable
1312 * release, a check must be done to see if we either adapt to the other
1313 * side version (which MUST be lower than us) or keep the latest data
1314 * structure considering that the other side will adapt.
1315 */
1316
1317 ret = sscanf(VERSION, "%u.%u", &reply.major, &reply.minor);
1318 if (ret < 2) {
1319 ERR("Error in scanning version");
1320 ret = -1;
1321 goto end;
1322 }
1323 reply.major = htobe32(reply.major);
1324 reply.minor = htobe32(reply.minor);
1325 ret = cmd->sock->ops->sendmsg(cmd->sock, &reply,
1326 sizeof(struct lttcomm_relayd_version), 0);
1327 if (ret < 0) {
1328 ERR("Relay sending version");
1329 }
1330 DBG("Version check done (%u.%u)", be32toh(reply.major),
1331 be32toh(reply.minor));
1332
1333 end:
1334 return ret;
1335 }
1336
1337 /*
1338 * Check for data pending for a given stream id from the session daemon.
1339 */
1340 static
1341 int relay_data_pending(struct lttcomm_relayd_hdr *recv_hdr,
1342 struct relay_command *cmd, struct lttng_ht *streams_ht)
1343 {
1344 struct relay_session *session = cmd->session;
1345 struct lttcomm_relayd_data_pending msg;
1346 struct lttcomm_relayd_generic_reply reply;
1347 struct relay_stream *stream;
1348 int ret;
1349 struct lttng_ht_node_ulong *node;
1350 struct lttng_ht_iter iter;
1351 uint64_t last_net_seq_num, stream_id;
1352
1353 DBG("Data pending command received");
1354
1355 if (!session || session->version_check_done == 0) {
1356 ERR("Trying to check for data before version check");
1357 ret = -1;
1358 goto end_no_session;
1359 }
1360
1361 ret = cmd->sock->ops->recvmsg(cmd->sock, &msg, sizeof(msg), MSG_WAITALL);
1362 if (ret < sizeof(msg)) {
1363 ERR("Relay didn't receive valid data_pending struct size : %d", ret);
1364 ret = -1;
1365 goto end_no_session;
1366 }
1367
1368 stream_id = be64toh(msg.stream_id);
1369 last_net_seq_num = be64toh(msg.last_net_seq_num);
1370
1371 rcu_read_lock();
1372 lttng_ht_lookup(streams_ht, (void *)((unsigned long) stream_id), &iter);
1373 node = lttng_ht_iter_get_node_ulong(&iter);
1374 if (node == NULL) {
1375 DBG("Relay stream %" PRIu64 " not found", stream_id);
1376 ret = -1;
1377 goto end_unlock;
1378 }
1379
1380 stream = caa_container_of(node, struct relay_stream, stream_n);
1381 assert(stream);
1382
1383 DBG("Data pending for stream id %" PRIu64 " prev_seq %" PRIu64
1384 " and last_seq %" PRIu64, stream_id, stream->prev_seq,
1385 last_net_seq_num);
1386
1387 /* Avoid wrapping issue */
1388 if (((int64_t) (stream->prev_seq - last_net_seq_num)) <= 0) {
1389 /* Data has in fact been written and is NOT pending */
1390 ret = 0;
1391 } else {
1392 /* Data still being streamed thus pending */
1393 ret = 1;
1394 }
1395
1396 end_unlock:
1397 rcu_read_unlock();
1398
1399 reply.ret_code = htobe32(ret);
1400 ret = cmd->sock->ops->sendmsg(cmd->sock, &reply, sizeof(reply), 0);
1401 if (ret < 0) {
1402 ERR("Relay data pending ret code failed");
1403 }
1404
1405 end_no_session:
1406 return ret;
1407 }
1408
1409 /*
1410 * Wait for the control socket to reach a quiescent state.
1411 *
1412 * Note that for now, when receiving this command from the session daemon, this
1413 * means that every subsequent commands or data received on the control socket
1414 * has been handled. So, this is why we simply return OK here.
1415 */
1416 static
1417 int relay_quiescent_control(struct lttcomm_relayd_hdr *recv_hdr,
1418 struct relay_command *cmd)
1419 {
1420 int ret;
1421 struct lttcomm_relayd_generic_reply reply;
1422
1423 DBG("Checking quiescent state on control socket");
1424
1425 reply.ret_code = htobe32(LTTNG_OK);
1426 ret = cmd->sock->ops->sendmsg(cmd->sock, &reply, sizeof(reply), 0);
1427 if (ret < 0) {
1428 ERR("Relay data quiescent control ret code failed");
1429 }
1430
1431 return ret;
1432 }
1433
1434 /*
1435 * relay_process_control: Process the commands received on the control socket
1436 */
1437 static
1438 int relay_process_control(struct lttcomm_relayd_hdr *recv_hdr,
1439 struct relay_command *cmd, struct lttng_ht *streams_ht)
1440 {
1441 int ret = 0;
1442
1443 switch (be32toh(recv_hdr->cmd)) {
1444 /*
1445 case RELAYD_CREATE_SESSION:
1446 ret = relay_create_session(recv_hdr, cmd);
1447 break;
1448 */
1449 case RELAYD_ADD_STREAM:
1450 ret = relay_add_stream(recv_hdr, cmd, streams_ht);
1451 break;
1452 case RELAYD_START_DATA:
1453 ret = relay_start(recv_hdr, cmd);
1454 break;
1455 case RELAYD_SEND_METADATA:
1456 ret = relay_recv_metadata(recv_hdr, cmd, streams_ht);
1457 break;
1458 case RELAYD_VERSION:
1459 ret = relay_send_version(recv_hdr, cmd);
1460 break;
1461 case RELAYD_CLOSE_STREAM:
1462 ret = relay_close_stream(recv_hdr, cmd, streams_ht);
1463 break;
1464 case RELAYD_DATA_PENDING:
1465 ret = relay_data_pending(recv_hdr, cmd, streams_ht);
1466 break;
1467 case RELAYD_QUIESCENT_CONTROL:
1468 ret = relay_quiescent_control(recv_hdr, cmd);
1469 break;
1470 case RELAYD_UPDATE_SYNC_INFO:
1471 default:
1472 ERR("Received unknown command (%u)", be32toh(recv_hdr->cmd));
1473 relay_unknown_command(cmd);
1474 ret = -1;
1475 goto end;
1476 }
1477
1478 end:
1479 return ret;
1480 }
1481
1482 /*
1483 * relay_process_data: Process the data received on the data socket
1484 */
1485 static
1486 int relay_process_data(struct relay_command *cmd, struct lttng_ht *streams_ht)
1487 {
1488 int ret = 0;
1489 struct relay_stream *stream;
1490 struct lttcomm_relayd_data_hdr data_hdr;
1491 uint64_t stream_id;
1492 uint64_t net_seq_num;
1493 uint32_t data_size;
1494
1495 ret = cmd->sock->ops->recvmsg(cmd->sock, &data_hdr,
1496 sizeof(struct lttcomm_relayd_data_hdr), MSG_WAITALL);
1497 if (ret <= 0) {
1498 ERR("Connections seems to be closed");
1499 ret = -1;
1500 goto end;
1501 }
1502
1503 stream_id = be64toh(data_hdr.stream_id);
1504
1505 rcu_read_lock();
1506 stream = relay_stream_from_stream_id(stream_id, streams_ht);
1507 if (!stream) {
1508 ret = -1;
1509 goto end_unlock;
1510 }
1511
1512 data_size = be32toh(data_hdr.data_size);
1513 if (data_buffer_size < data_size) {
1514 char *tmp_data_ptr = data_buffer;
1515 data_buffer = realloc(data_buffer, data_size);
1516 if (!data_buffer) {
1517 ERR("Allocating data buffer");
1518 free(tmp_data_ptr);
1519 ret = -1;
1520 goto end_unlock;
1521 }
1522 data_buffer_size = data_size;
1523 }
1524 memset(data_buffer, 0, data_size);
1525
1526 net_seq_num = be64toh(data_hdr.net_seq_num);
1527
1528 DBG3("Receiving data of size %u for stream id %" PRIu64 " seqnum %" PRIu64,
1529 data_size, stream_id, net_seq_num);
1530 ret = cmd->sock->ops->recvmsg(cmd->sock, data_buffer, data_size, MSG_WAITALL);
1531 if (ret <= 0) {
1532 ret = -1;
1533 goto end_unlock;
1534 }
1535
1536 do {
1537 ret = write(stream->fd, data_buffer, data_size);
1538 } while (ret < 0 && errno == EINTR);
1539 if (ret < data_size) {
1540 ERR("Relay error writing data to file");
1541 ret = -1;
1542 goto end_unlock;
1543 }
1544
1545 DBG2("Relay wrote %d bytes to tracefile for stream id %" PRIu64,
1546 ret, stream->stream_handle);
1547
1548 ret = write_padding_to_file(stream->fd, be32toh(data_hdr.padding_size));
1549 if (ret < 0) {
1550 goto end_unlock;
1551 }
1552
1553 stream->prev_seq = net_seq_num;
1554
1555 /* Check if we need to close the FD */
1556 if (close_stream_check(stream)) {
1557 int cret;
1558 struct lttng_ht_iter iter;
1559
1560 cret = close(stream->fd);
1561 if (cret < 0) {
1562 PERROR("close stream process data");
1563 }
1564 iter.iter.node = &stream->stream_n.node;
1565 ret = lttng_ht_del(streams_ht, &iter);
1566 assert(!ret);
1567 call_rcu(&stream->rcu_node,
1568 deferred_free_stream);
1569 DBG("Closed tracefile %d after recv data", stream->fd);
1570 }
1571
1572 end_unlock:
1573 rcu_read_unlock();
1574 end:
1575 return ret;
1576 }
1577
1578 static
1579 void relay_cleanup_poll_connection(struct lttng_poll_event *events, int pollfd)
1580 {
1581 int ret;
1582
1583 lttng_poll_del(events, pollfd);
1584
1585 ret = close(pollfd);
1586 if (ret < 0) {
1587 ERR("Closing pollfd %d", pollfd);
1588 }
1589 }
1590
1591 static
1592 int relay_add_connection(int fd, struct lttng_poll_event *events,
1593 struct lttng_ht *relay_connections_ht)
1594 {
1595 struct relay_command *relay_connection;
1596 int ret;
1597
1598 relay_connection = zmalloc(sizeof(struct relay_command));
1599 if (relay_connection == NULL) {
1600 PERROR("Relay command zmalloc");
1601 goto error;
1602 }
1603 ret = read(fd, relay_connection, sizeof(struct relay_command));
1604 if (ret < 0 || ret < sizeof(struct relay_command)) {
1605 PERROR("read relay cmd pipe");
1606 goto error_read;
1607 }
1608
1609 lttng_ht_node_init_ulong(&relay_connection->sock_n,
1610 (unsigned long) relay_connection->sock->fd);
1611 rcu_read_lock();
1612 lttng_ht_add_unique_ulong(relay_connections_ht,
1613 &relay_connection->sock_n);
1614 rcu_read_unlock();
1615 return lttng_poll_add(events,
1616 relay_connection->sock->fd,
1617 LPOLLIN | LPOLLRDHUP);
1618
1619 error_read:
1620 free(relay_connection);
1621 error:
1622 return -1;
1623 }
1624
1625 static
1626 void deferred_free_connection(struct rcu_head *head)
1627 {
1628 struct relay_command *relay_connection =
1629 caa_container_of(head, struct relay_command, rcu_node);
1630
1631 lttcomm_destroy_sock(relay_connection->sock);
1632 free(relay_connection);
1633 }
1634
1635 static
1636 void relay_del_connection(struct lttng_ht *relay_connections_ht,
1637 struct lttng_ht *streams_ht, struct lttng_ht_iter *iter,
1638 struct relay_command *relay_connection)
1639 {
1640 int ret;
1641
1642 ret = lttng_ht_del(relay_connections_ht, iter);
1643 assert(!ret);
1644 if (relay_connection->type == RELAY_CONTROL) {
1645 relay_delete_session(relay_connection, streams_ht);
1646 }
1647
1648 call_rcu(&relay_connection->rcu_node,
1649 deferred_free_connection);
1650 }
1651
1652 /*
1653 * This thread does the actual work
1654 */
1655 static
1656 void *relay_thread_worker(void *data)
1657 {
1658 int i, ret, pollfd, err = -1;
1659 uint32_t revents, nb_fd;
1660 struct relay_command *relay_connection;
1661 struct lttng_poll_event events;
1662 struct lttng_ht *relay_connections_ht;
1663 struct lttng_ht_node_ulong *node;
1664 struct lttng_ht_iter iter;
1665 struct lttng_ht *streams_ht;
1666 struct lttcomm_relayd_hdr recv_hdr;
1667
1668 DBG("[thread] Relay worker started");
1669
1670 rcu_register_thread();
1671
1672 /* table of connections indexed on socket */
1673 relay_connections_ht = lttng_ht_new(0, LTTNG_HT_TYPE_ULONG);
1674 if (!relay_connections_ht) {
1675 goto relay_connections_ht_error;
1676 }
1677
1678 /* tables of streams indexed by stream ID */
1679 streams_ht = lttng_ht_new(0, LTTNG_HT_TYPE_ULONG);
1680 if (!streams_ht) {
1681 goto streams_ht_error;
1682 }
1683
1684 ret = create_thread_poll_set(&events, 2);
1685 if (ret < 0) {
1686 goto error_poll_create;
1687 }
1688
1689 ret = lttng_poll_add(&events, relay_cmd_pipe[0], LPOLLIN | LPOLLRDHUP);
1690 if (ret < 0) {
1691 goto error;
1692 }
1693
1694 while (1) {
1695 /* Zeroed the events structure */
1696 lttng_poll_reset(&events);
1697
1698 nb_fd = LTTNG_POLL_GETNB(&events);
1699
1700 /* Infinite blocking call, waiting for transmission */
1701 restart:
1702 DBG3("Relayd worker thread polling...");
1703 ret = lttng_poll_wait(&events, -1);
1704 if (ret < 0) {
1705 /*
1706 * Restart interrupted system call.
1707 */
1708 if (errno == EINTR) {
1709 goto restart;
1710 }
1711 goto error;
1712 }
1713
1714 for (i = 0; i < nb_fd; i++) {
1715 /* Fetch once the poll data */
1716 revents = LTTNG_POLL_GETEV(&events, i);
1717 pollfd = LTTNG_POLL_GETFD(&events, i);
1718
1719 /* Thread quit pipe has been closed. Killing thread. */
1720 ret = check_thread_quit_pipe(pollfd, revents);
1721 if (ret) {
1722 err = 0;
1723 goto exit;
1724 }
1725
1726 /* Inspect the relay cmd pipe for new connection */
1727 if (pollfd == relay_cmd_pipe[0]) {
1728 if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
1729 ERR("Relay pipe error");
1730 goto error;
1731 } else if (revents & LPOLLIN) {
1732 DBG("Relay command received");
1733 ret = relay_add_connection(relay_cmd_pipe[0],
1734 &events, relay_connections_ht);
1735 if (ret < 0) {
1736 goto error;
1737 }
1738 }
1739 } else if (revents > 0) {
1740 rcu_read_lock();
1741 lttng_ht_lookup(relay_connections_ht,
1742 (void *)((unsigned long) pollfd),
1743 &iter);
1744 node = lttng_ht_iter_get_node_ulong(&iter);
1745 if (node == NULL) {
1746 DBG2("Relay sock %d not found", pollfd);
1747 rcu_read_unlock();
1748 goto error;
1749 }
1750 relay_connection = caa_container_of(node,
1751 struct relay_command, sock_n);
1752
1753 if (revents & (LPOLLERR)) {
1754 ERR("POLL ERROR");
1755 relay_cleanup_poll_connection(&events, pollfd);
1756 relay_del_connection(relay_connections_ht,
1757 streams_ht, &iter,
1758 relay_connection);
1759 } else if (revents & (LPOLLHUP | LPOLLRDHUP)) {
1760 DBG("Socket %d hung up", pollfd);
1761 relay_cleanup_poll_connection(&events, pollfd);
1762 relay_del_connection(relay_connections_ht,
1763 streams_ht, &iter,
1764 relay_connection);
1765 } else if (revents & LPOLLIN) {
1766 /* control socket */
1767 if (relay_connection->type == RELAY_CONTROL) {
1768 ret = relay_connection->sock->ops->recvmsg(
1769 relay_connection->sock, &recv_hdr,
1770 sizeof(struct lttcomm_relayd_hdr), MSG_WAITALL);
1771 /* connection closed */
1772 if (ret <= 0) {
1773 relay_cleanup_poll_connection(&events, pollfd);
1774 relay_del_connection(relay_connections_ht,
1775 streams_ht, &iter,
1776 relay_connection);
1777 DBG("Control connection closed with %d", pollfd);
1778 } else {
1779 if (relay_connection->session) {
1780 DBG2("Relay worker receiving data for session : %" PRIu64,
1781 relay_connection->session->id);
1782 }
1783 ret = relay_process_control(&recv_hdr,
1784 relay_connection,
1785 streams_ht);
1786 /*
1787 * there was an error in processing a control
1788 * command: clear the session
1789 * */
1790 if (ret < 0) {
1791 relay_cleanup_poll_connection(&events, pollfd);
1792 relay_del_connection(relay_connections_ht,
1793 streams_ht, &iter,
1794 relay_connection);
1795 DBG("Connection closed with %d", pollfd);
1796 }
1797 }
1798 /* data socket */
1799 } else if (relay_connection->type == RELAY_DATA) {
1800 ret = relay_process_data(relay_connection, streams_ht);
1801 /* connection closed */
1802 if (ret < 0) {
1803 relay_cleanup_poll_connection(&events, pollfd);
1804 relay_del_connection(relay_connections_ht,
1805 streams_ht, &iter,
1806 relay_connection);
1807 DBG("Data connection closed with %d", pollfd);
1808 }
1809 }
1810 }
1811 rcu_read_unlock();
1812 }
1813 }
1814 }
1815
1816 exit:
1817 error:
1818 lttng_poll_clean(&events);
1819
1820 /* empty the hash table and free the memory */
1821 rcu_read_lock();
1822 cds_lfht_for_each_entry(relay_connections_ht->ht, &iter.iter, node, node) {
1823 node = lttng_ht_iter_get_node_ulong(&iter);
1824 if (node) {
1825 relay_connection = caa_container_of(node,
1826 struct relay_command, sock_n);
1827 relay_del_connection(relay_connections_ht,
1828 streams_ht, &iter,
1829 relay_connection);
1830 }
1831 }
1832 rcu_read_unlock();
1833 error_poll_create:
1834 lttng_ht_destroy(streams_ht);
1835 streams_ht_error:
1836 lttng_ht_destroy(relay_connections_ht);
1837 relay_connections_ht_error:
1838 /* Close relay cmd pipes */
1839 utils_close_pipe(relay_cmd_pipe);
1840 if (err) {
1841 DBG("Thread exited with error");
1842 }
1843 DBG("Worker thread cleanup complete");
1844 free(data_buffer);
1845 stop_threads();
1846 rcu_unregister_thread();
1847 return NULL;
1848 }
1849
1850 /*
1851 * Create the relay command pipe to wake thread_manage_apps.
1852 * Closed in cleanup().
1853 */
1854 static int create_relay_cmd_pipe(void)
1855 {
1856 int ret;
1857
1858 ret = utils_create_pipe_cloexec(relay_cmd_pipe);
1859
1860 return ret;
1861 }
1862
1863 /*
1864 * main
1865 */
1866 int main(int argc, char **argv)
1867 {
1868 int ret = 0;
1869 void *status;
1870
1871 /* Create thread quit pipe */
1872 if ((ret = init_thread_quit_pipe()) < 0) {
1873 goto error;
1874 }
1875
1876 /* Parse arguments */
1877 progname = argv[0];
1878 if ((ret = parse_args(argc, argv) < 0)) {
1879 goto exit;
1880 }
1881
1882 if ((ret = set_signal_handler()) < 0) {
1883 goto exit;
1884 }
1885
1886 /* Daemonize */
1887 if (opt_daemon) {
1888 ret = daemon(0, 0);
1889 if (ret < 0) {
1890 PERROR("daemon");
1891 goto exit;
1892 }
1893 }
1894
1895 /* Check if daemon is UID = 0 */
1896 is_root = !getuid();
1897
1898 if (!is_root) {
1899 if (control_uri->port < 1024 || data_uri->port < 1024) {
1900 ERR("Need to be root to use ports < 1024");
1901 ret = -1;
1902 goto exit;
1903 }
1904 }
1905
1906 /* Setup the thread apps communication pipe. */
1907 if ((ret = create_relay_cmd_pipe()) < 0) {
1908 goto exit;
1909 }
1910
1911 /* Init relay command queue. */
1912 cds_wfq_init(&relay_cmd_queue.queue);
1913
1914 /* Set up max poll set size */
1915 lttng_poll_set_max_size();
1916
1917 /* Setup the dispatcher thread */
1918 ret = pthread_create(&dispatcher_thread, NULL,
1919 relay_thread_dispatcher, (void *) NULL);
1920 if (ret != 0) {
1921 PERROR("pthread_create dispatcher");
1922 goto exit_dispatcher;
1923 }
1924
1925 /* Setup the worker thread */
1926 ret = pthread_create(&worker_thread, NULL,
1927 relay_thread_worker, (void *) NULL);
1928 if (ret != 0) {
1929 PERROR("pthread_create worker");
1930 goto exit_worker;
1931 }
1932
1933 /* Setup the listener thread */
1934 ret = pthread_create(&listener_thread, NULL,
1935 relay_thread_listener, (void *) NULL);
1936 if (ret != 0) {
1937 PERROR("pthread_create listener");
1938 goto exit_listener;
1939 }
1940
1941 exit_listener:
1942 ret = pthread_join(listener_thread, &status);
1943 if (ret != 0) {
1944 PERROR("pthread_join");
1945 goto error; /* join error, exit without cleanup */
1946 }
1947
1948 exit_worker:
1949 ret = pthread_join(worker_thread, &status);
1950 if (ret != 0) {
1951 PERROR("pthread_join");
1952 goto error; /* join error, exit without cleanup */
1953 }
1954
1955 exit_dispatcher:
1956 ret = pthread_join(dispatcher_thread, &status);
1957 if (ret != 0) {
1958 PERROR("pthread_join");
1959 goto error; /* join error, exit without cleanup */
1960 }
1961
1962 exit:
1963 cleanup();
1964 if (!ret) {
1965 exit(EXIT_SUCCESS);
1966 }
1967
1968 error:
1969 exit(EXIT_FAILURE);
1970 }
LTTng 2.0 tools and control repository
This page took 0.11196 seconds and 4 git commands to generate.