Fix: return error if sendmsg fails on relayd
[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 || ret != 1) {
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 /*
459 * We are about to close the stream so set the data pending flag to 1
460 * which will make the end data pending command skip the stream which
461 * is now closed and ready. Note that after proceeding to a file close,
462 * the written file is ready for reading.
463 */
464 stream->data_pending_check_done = 1;
465 return 1;
466 }
467 return 0;
468 }
469
470 /*
471 * This thread manages the listening for new connections on the network
472 */
473 static
474 void *relay_thread_listener(void *data)
475 {
476 int i, ret, pollfd, err = -1;
477 int val = 1;
478 uint32_t revents, nb_fd;
479 struct lttng_poll_event events;
480 struct lttcomm_sock *control_sock, *data_sock;
481
482 DBG("[thread] Relay listener started");
483
484 control_sock = relay_init_sock(control_uri);
485 if (!control_sock) {
486 goto error_sock_control;
487 }
488
489 data_sock = relay_init_sock(data_uri);
490 if (!data_sock) {
491 goto error_sock_relay;
492 }
493
494 /*
495 * Pass 3 as size here for the thread quit pipe, control and data socket.
496 */
497 ret = create_thread_poll_set(&events, 3);
498 if (ret < 0) {
499 goto error_create_poll;
500 }
501
502 /* Add the control socket */
503 ret = lttng_poll_add(&events, control_sock->fd, LPOLLIN | LPOLLRDHUP);
504 if (ret < 0) {
505 goto error_poll_add;
506 }
507
508 /* Add the data socket */
509 ret = lttng_poll_add(&events, data_sock->fd, LPOLLIN | LPOLLRDHUP);
510 if (ret < 0) {
511 goto error_poll_add;
512 }
513
514 while (1) {
515 DBG("Listener accepting connections");
516
517 restart:
518 ret = lttng_poll_wait(&events, -1);
519 if (ret < 0) {
520 /*
521 * Restart interrupted system call.
522 */
523 if (errno == EINTR) {
524 goto restart;
525 }
526 goto error;
527 }
528
529 nb_fd = ret;
530
531 DBG("Relay new connection received");
532 for (i = 0; i < nb_fd; i++) {
533 /* Fetch once the poll data */
534 revents = LTTNG_POLL_GETEV(&events, i);
535 pollfd = LTTNG_POLL_GETFD(&events, i);
536
537 /* Thread quit pipe has been closed. Killing thread. */
538 ret = check_thread_quit_pipe(pollfd, revents);
539 if (ret) {
540 err = 0;
541 goto exit;
542 }
543
544 if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
545 ERR("socket poll error");
546 goto error;
547 } else if (revents & LPOLLIN) {
548 /*
549 * Get allocated in this thread,
550 * enqueued to a global queue, dequeued
551 * and freed in the worker thread.
552 */
553 struct relay_command *relay_cmd;
554 struct lttcomm_sock *newsock;
555
556 relay_cmd = zmalloc(sizeof(struct relay_command));
557 if (relay_cmd == NULL) {
558 PERROR("relay command zmalloc");
559 goto error;
560 }
561
562 if (pollfd == data_sock->fd) {
563 newsock = data_sock->ops->accept(data_sock);
564 if (!newsock) {
565 PERROR("accepting data sock");
566 free(relay_cmd);
567 goto error;
568 }
569 relay_cmd->type = RELAY_DATA;
570 DBG("Relay data connection accepted, socket %d", newsock->fd);
571 } else {
572 assert(pollfd == control_sock->fd);
573 newsock = control_sock->ops->accept(control_sock);
574 if (!newsock) {
575 PERROR("accepting control sock");
576 free(relay_cmd);
577 goto error;
578 }
579 relay_cmd->type = RELAY_CONTROL;
580 DBG("Relay control connection accepted, socket %d", newsock->fd);
581 }
582 ret = setsockopt(newsock->fd, SOL_SOCKET, SO_REUSEADDR,
583 &val, sizeof(int));
584 if (ret < 0) {
585 PERROR("setsockopt inet");
586 lttcomm_destroy_sock(newsock);
587 free(relay_cmd);
588 goto error;
589 }
590 relay_cmd->sock = newsock;
591 /*
592 * Lock free enqueue the request.
593 */
594 cds_wfq_enqueue(&relay_cmd_queue.queue, &relay_cmd->node);
595
596 /*
597 * Wake the dispatch queue futex. Implicit memory
598 * barrier with the exchange in cds_wfq_enqueue.
599 */
600 futex_nto1_wake(&relay_cmd_queue.futex);
601 }
602 }
603 }
604
605 exit:
606 error:
607 error_poll_add:
608 lttng_poll_clean(&events);
609 error_create_poll:
610 if (data_sock->fd >= 0) {
611 ret = data_sock->ops->close(data_sock);
612 if (ret) {
613 PERROR("close");
614 }
615 }
616 lttcomm_destroy_sock(data_sock);
617 error_sock_relay:
618 if (control_sock->fd >= 0) {
619 ret = control_sock->ops->close(control_sock);
620 if (ret) {
621 PERROR("close");
622 }
623 }
624 lttcomm_destroy_sock(control_sock);
625 error_sock_control:
626 if (err) {
627 DBG("Thread exited with error");
628 }
629 DBG("Relay listener thread cleanup complete");
630 stop_threads();
631 return NULL;
632 }
633
634 /*
635 * This thread manages the dispatching of the requests to worker threads
636 */
637 static
638 void *relay_thread_dispatcher(void *data)
639 {
640 int ret;
641 struct cds_wfq_node *node;
642 struct relay_command *relay_cmd = NULL;
643
644 DBG("[thread] Relay dispatcher started");
645
646 while (!CMM_LOAD_SHARED(dispatch_thread_exit)) {
647 /* Atomically prepare the queue futex */
648 futex_nto1_prepare(&relay_cmd_queue.futex);
649
650 do {
651 /* Dequeue commands */
652 node = cds_wfq_dequeue_blocking(&relay_cmd_queue.queue);
653 if (node == NULL) {
654 DBG("Woken up but nothing in the relay command queue");
655 /* Continue thread execution */
656 break;
657 }
658
659 relay_cmd = caa_container_of(node, struct relay_command, node);
660 DBG("Dispatching request waiting on sock %d", relay_cmd->sock->fd);
661
662 /*
663 * Inform worker thread of the new request. This
664 * call is blocking so we can be assured that the data will be read
665 * at some point in time or wait to the end of the world :)
666 */
667 do {
668 ret = write(relay_cmd_pipe[1], relay_cmd,
669 sizeof(struct relay_command));
670 } while (ret < 0 && errno == EINTR);
671 free(relay_cmd);
672 if (ret < 0 || ret != sizeof(struct relay_command)) {
673 PERROR("write cmd pipe");
674 goto error;
675 }
676 } while (node != NULL);
677
678 /* Futex wait on queue. Blocking call on futex() */
679 futex_nto1_wait(&relay_cmd_queue.futex);
680 }
681
682 error:
683 DBG("Dispatch thread dying");
684 stop_threads();
685 return NULL;
686 }
687
688 /*
689 * Return the realpath(3) of the path even if the last directory token does not
690 * exist. For example, with /tmp/test1/test2, if test2/ does not exist but the
691 * /tmp/test1 does, the real path is returned. In normal time, realpath(3)
692 * fails if the end point directory does not exist.
693 */
694 static
695 char *expand_full_path(const char *path)
696 {
697 const char *end_path = path;
698 char *next, *cut_path, *expanded_path, *respath;
699
700 /* Find last token delimited by '/' */
701 while ((next = strpbrk(end_path + 1, "/"))) {
702 end_path = next;
703 }
704
705 /* Cut last token from original path */
706 cut_path = strndup(path, end_path - path);
707
708 expanded_path = malloc(PATH_MAX);
709 if (expanded_path == NULL) {
710 respath = NULL;
711 goto end;
712 }
713
714 respath = realpath(cut_path, expanded_path);
715 if (respath == NULL) {
716 switch (errno) {
717 case ENOENT:
718 ERR("%s: No such file or directory", cut_path);
719 break;
720 default:
721 PERROR("realpath");
722 break;
723 }
724 free(expanded_path);
725 } else {
726 /* Add end part to expanded path */
727 strcat(respath, end_path);
728 }
729 end:
730 free(cut_path);
731 return respath;
732 }
733
734
735 /*
736 * config_get_default_path
737 *
738 * Returns the HOME directory path. Caller MUST NOT free(3) the return pointer.
739 */
740 static
741 char *config_get_default_path(void)
742 {
743 return getenv("HOME");
744 }
745
746 /*
747 * Create recursively directory using the FULL path.
748 */
749 static
750 int mkdir_recursive(char *path, mode_t mode)
751 {
752 char *p, tmp[PATH_MAX];
753 struct stat statbuf;
754 size_t len;
755 int ret;
756
757 ret = snprintf(tmp, sizeof(tmp), "%s", path);
758 if (ret < 0) {
759 PERROR("snprintf mkdir");
760 goto error;
761 }
762
763 len = ret;
764 if (tmp[len - 1] == '/') {
765 tmp[len - 1] = 0;
766 }
767
768 for (p = tmp + 1; *p; p++) {
769 if (*p == '/') {
770 *p = 0;
771 if (tmp[strlen(tmp) - 1] == '.' &&
772 tmp[strlen(tmp) - 2] == '.' &&
773 tmp[strlen(tmp) - 3] == '/') {
774 ERR("Using '/../' is not permitted in the trace path (%s)",
775 tmp);
776 ret = -1;
777 goto error;
778 }
779 ret = stat(tmp, &statbuf);
780 if (ret < 0) {
781 ret = mkdir(tmp, mode);
782 if (ret < 0) {
783 if (errno != EEXIST) {
784 PERROR("mkdir recursive");
785 ret = -errno;
786 goto error;
787 }
788 }
789 }
790 *p = '/';
791 }
792 }
793
794 ret = mkdir(tmp, mode);
795 if (ret < 0) {
796 if (errno != EEXIST) {
797 PERROR("mkdir recursive last piece");
798 ret = -errno;
799 } else {
800 ret = 0;
801 }
802 }
803
804 error:
805 return ret;
806 }
807
808 static
809 char *create_output_path_auto(char *path_name)
810 {
811 int ret;
812 char *traces_path = NULL;
813 char *alloc_path = NULL;
814 char *default_path;
815
816 default_path = config_get_default_path();
817 if (default_path == NULL) {
818 ERR("Home path not found.\n \
819 Please specify an output path using -o, --output PATH");
820 goto exit;
821 }
822 alloc_path = strdup(default_path);
823 if (alloc_path == NULL) {
824 PERROR("Path allocation");
825 goto exit;
826 }
827 ret = asprintf(&traces_path, "%s/" DEFAULT_TRACE_DIR_NAME
828 "/%s", alloc_path, path_name);
829 if (ret < 0) {
830 PERROR("asprintf trace dir name");
831 goto exit;
832 }
833 exit:
834 free(alloc_path);
835 return traces_path;
836 }
837
838 static
839 char *create_output_path_noauto(char *path_name)
840 {
841 int ret;
842 char *traces_path = NULL;
843 char *full_path;
844
845 full_path = expand_full_path(opt_output_path);
846 ret = asprintf(&traces_path, "%s/%s", full_path, path_name);
847 if (ret < 0) {
848 PERROR("asprintf trace dir name");
849 goto exit;
850 }
851 exit:
852 free(full_path);
853 return traces_path;
854 }
855
856 /*
857 * create_output_path: create the output trace directory
858 */
859 static
860 char *create_output_path(char *path_name)
861 {
862 if (opt_output_path == NULL) {
863 return create_output_path_auto(path_name);
864 } else {
865 return create_output_path_noauto(path_name);
866 }
867 }
868
869 static
870 void deferred_free_stream(struct rcu_head *head)
871 {
872 struct relay_stream *stream =
873 caa_container_of(head, struct relay_stream, rcu_node);
874 free(stream);
875 }
876
877 /*
878 * relay_delete_session: Free all memory associated with a session and
879 * close all the FDs
880 */
881 static
882 void relay_delete_session(struct relay_command *cmd, struct lttng_ht *streams_ht)
883 {
884 struct lttng_ht_iter iter;
885 struct lttng_ht_node_ulong *node;
886 struct relay_stream *stream;
887 int ret;
888
889 if (!cmd->session) {
890 return;
891 }
892
893 DBG("Relay deleting session %" PRIu64, cmd->session->id);
894
895 rcu_read_lock();
896 cds_lfht_for_each_entry(streams_ht->ht, &iter.iter, node, node) {
897 node = lttng_ht_iter_get_node_ulong(&iter);
898 if (node) {
899 stream = caa_container_of(node,
900 struct relay_stream, stream_n);
901 if (stream->session == cmd->session) {
902 ret = close(stream->fd);
903 if (ret < 0) {
904 PERROR("close stream fd on delete session");
905 }
906 ret = lttng_ht_del(streams_ht, &iter);
907 assert(!ret);
908 call_rcu(&stream->rcu_node,
909 deferred_free_stream);
910 }
911 }
912 }
913 rcu_read_unlock();
914
915 free(cmd->session);
916 }
917
918 /*
919 * Handle the RELAYD_CREATE_SESSION command.
920 *
921 * On success, send back the session id or else return a negative value.
922 */
923 static
924 int relay_create_session(struct lttcomm_relayd_hdr *recv_hdr,
925 struct relay_command *cmd)
926 {
927 int ret = 0, send_ret;
928 struct relay_session *session;
929 struct lttcomm_relayd_status_session reply;
930
931 assert(recv_hdr);
932 assert(cmd);
933
934 memset(&reply, 0, sizeof(reply));
935
936 session = zmalloc(sizeof(struct relay_session));
937 if (session == NULL) {
938 PERROR("relay session zmalloc");
939 ret = -1;
940 goto error;
941 }
942
943 session->id = ++last_relay_session_id;
944 session->sock = cmd->sock;
945 cmd->session = session;
946
947 reply.session_id = htobe64(session->id);
948
949 DBG("Created session %" PRIu64, session->id);
950
951 error:
952 if (ret < 0) {
953 reply.ret_code = htobe32(LTTNG_ERR_FATAL);
954 } else {
955 reply.ret_code = htobe32(LTTNG_OK);
956 }
957
958 send_ret = cmd->sock->ops->sendmsg(cmd->sock, &reply, sizeof(reply), 0);
959 if (send_ret < 0) {
960 ERR("Relayd sending session id");
961 ret = send_ret;
962 }
963
964 return ret;
965 }
966
967 /*
968 * relay_add_stream: allocate a new stream for a session
969 */
970 static
971 int relay_add_stream(struct lttcomm_relayd_hdr *recv_hdr,
972 struct relay_command *cmd, struct lttng_ht *streams_ht)
973 {
974 struct relay_session *session = cmd->session;
975 struct lttcomm_relayd_add_stream stream_info;
976 struct relay_stream *stream = NULL;
977 struct lttcomm_relayd_status_stream reply;
978 char *path = NULL, *root_path = NULL;
979 int ret, send_ret;
980
981 if (!session || cmd->version_check_done == 0) {
982 ERR("Trying to add a stream before version check");
983 ret = -1;
984 goto end_no_session;
985 }
986
987 ret = cmd->sock->ops->recvmsg(cmd->sock, &stream_info,
988 sizeof(struct lttcomm_relayd_add_stream), 0);
989 if (ret < sizeof(struct lttcomm_relayd_add_stream)) {
990 if (ret == 0) {
991 /* Orderly shutdown. Not necessary to print an error. */
992 DBG("Socket %d did an orderly shutdown", cmd->sock->fd);
993 } else {
994 ERR("Relay didn't receive valid add_stream struct size : %d", ret);
995 }
996 ret = -1;
997 goto end_no_session;
998 }
999 stream = zmalloc(sizeof(struct relay_stream));
1000 if (stream == NULL) {
1001 PERROR("relay stream zmalloc");
1002 ret = -1;
1003 goto end_no_session;
1004 }
1005
1006 rcu_read_lock();
1007 stream->stream_handle = ++last_relay_stream_id;
1008 stream->prev_seq = -1ULL;
1009 stream->session = session;
1010
1011 root_path = create_output_path(stream_info.pathname);
1012 if (!root_path) {
1013 ret = -1;
1014 goto end;
1015 }
1016 ret = mkdir_recursive(root_path, S_IRWXU | S_IRWXG);
1017 if (ret < 0) {
1018 ERR("relay creating output directory");
1019 goto end;
1020 }
1021
1022 ret = asprintf(&path, "%s/%s", root_path, stream_info.channel_name);
1023 if (ret < 0) {
1024 PERROR("asprintf stream path");
1025 goto end;
1026 }
1027
1028 ret = open(path, O_WRONLY|O_CREAT|O_TRUNC, S_IRWXU|S_IRWXG|S_IRWXO);
1029 if (ret < 0) {
1030 PERROR("Relay creating trace file");
1031 goto end;
1032 }
1033
1034 stream->fd = ret;
1035 DBG("Tracefile %s created", path);
1036
1037 lttng_ht_node_init_ulong(&stream->stream_n,
1038 (unsigned long) stream->stream_handle);
1039 lttng_ht_add_unique_ulong(streams_ht,
1040 &stream->stream_n);
1041
1042 DBG("Relay new stream added %s", stream_info.channel_name);
1043
1044 end:
1045 free(path);
1046 free(root_path);
1047 /* send the session id to the client or a negative return code on error */
1048 if (ret < 0) {
1049 reply.ret_code = htobe32(LTTNG_ERR_UNK);
1050 } else {
1051 reply.ret_code = htobe32(LTTNG_OK);
1052 }
1053 reply.handle = htobe64(stream->stream_handle);
1054 send_ret = cmd->sock->ops->sendmsg(cmd->sock, &reply,
1055 sizeof(struct lttcomm_relayd_status_stream), 0);
1056 if (send_ret < 0) {
1057 ERR("Relay sending stream id");
1058 ret = send_ret;
1059 }
1060 rcu_read_unlock();
1061
1062 end_no_session:
1063 return ret;
1064 }
1065
1066 /*
1067 * relay_close_stream: close a specific stream
1068 */
1069 static
1070 int relay_close_stream(struct lttcomm_relayd_hdr *recv_hdr,
1071 struct relay_command *cmd, struct lttng_ht *streams_ht)
1072 {
1073 struct relay_session *session = cmd->session;
1074 struct lttcomm_relayd_close_stream stream_info;
1075 struct lttcomm_relayd_generic_reply reply;
1076 struct relay_stream *stream;
1077 int ret, send_ret;
1078 struct lttng_ht_node_ulong *node;
1079 struct lttng_ht_iter iter;
1080
1081 DBG("Close stream received");
1082
1083 if (!session || cmd->version_check_done == 0) {
1084 ERR("Trying to close a stream before version check");
1085 ret = -1;
1086 goto end_no_session;
1087 }
1088
1089 ret = cmd->sock->ops->recvmsg(cmd->sock, &stream_info,
1090 sizeof(struct lttcomm_relayd_close_stream), 0);
1091 if (ret < sizeof(struct lttcomm_relayd_close_stream)) {
1092 if (ret == 0) {
1093 /* Orderly shutdown. Not necessary to print an error. */
1094 DBG("Socket %d did an orderly shutdown", cmd->sock->fd);
1095 } else {
1096 ERR("Relay didn't receive valid add_stream struct size : %d", ret);
1097 }
1098 ret = -1;
1099 goto end_no_session;
1100 }
1101
1102 rcu_read_lock();
1103 lttng_ht_lookup(streams_ht,
1104 (void *)((unsigned long) be64toh(stream_info.stream_id)),
1105 &iter);
1106 node = lttng_ht_iter_get_node_ulong(&iter);
1107 if (node == NULL) {
1108 DBG("Relay stream %" PRIu64 " not found", be64toh(stream_info.stream_id));
1109 ret = -1;
1110 goto end_unlock;
1111 }
1112
1113 stream = caa_container_of(node, struct relay_stream, stream_n);
1114 if (!stream) {
1115 ret = -1;
1116 goto end_unlock;
1117 }
1118
1119 stream->last_net_seq_num = be64toh(stream_info.last_net_seq_num);
1120 stream->close_flag = 1;
1121
1122 if (close_stream_check(stream)) {
1123 int delret;
1124
1125 delret = close(stream->fd);
1126 if (delret < 0) {
1127 PERROR("close stream");
1128 }
1129 delret = lttng_ht_del(streams_ht, &iter);
1130 assert(!delret);
1131 call_rcu(&stream->rcu_node,
1132 deferred_free_stream);
1133 DBG("Closed tracefile %d from close stream", stream->fd);
1134 }
1135
1136 end_unlock:
1137 rcu_read_unlock();
1138
1139 if (ret < 0) {
1140 reply.ret_code = htobe32(LTTNG_ERR_UNK);
1141 } else {
1142 reply.ret_code = htobe32(LTTNG_OK);
1143 }
1144 send_ret = cmd->sock->ops->sendmsg(cmd->sock, &reply,
1145 sizeof(struct lttcomm_relayd_generic_reply), 0);
1146 if (send_ret < 0) {
1147 ERR("Relay sending stream id");
1148 ret = send_ret;
1149 }
1150
1151 end_no_session:
1152 return ret;
1153 }
1154
1155 /*
1156 * relay_unknown_command: send -1 if received unknown command
1157 */
1158 static
1159 void relay_unknown_command(struct relay_command *cmd)
1160 {
1161 struct lttcomm_relayd_generic_reply reply;
1162 int ret;
1163
1164 reply.ret_code = htobe32(LTTNG_ERR_UNK);
1165 ret = cmd->sock->ops->sendmsg(cmd->sock, &reply,
1166 sizeof(struct lttcomm_relayd_generic_reply), 0);
1167 if (ret < 0) {
1168 ERR("Relay sending unknown command");
1169 }
1170 }
1171
1172 /*
1173 * relay_start: send an acknowledgment to the client to tell if we are
1174 * ready to receive data. We are ready if a session is established.
1175 */
1176 static
1177 int relay_start(struct lttcomm_relayd_hdr *recv_hdr,
1178 struct relay_command *cmd)
1179 {
1180 int ret = htobe32(LTTNG_OK);
1181 struct lttcomm_relayd_generic_reply reply;
1182 struct relay_session *session = cmd->session;
1183
1184 if (!session) {
1185 DBG("Trying to start the streaming without a session established");
1186 ret = htobe32(LTTNG_ERR_UNK);
1187 }
1188
1189 reply.ret_code = ret;
1190 ret = cmd->sock->ops->sendmsg(cmd->sock, &reply,
1191 sizeof(struct lttcomm_relayd_generic_reply), 0);
1192 if (ret < 0) {
1193 ERR("Relay sending start ack");
1194 }
1195
1196 return ret;
1197 }
1198
1199 /*
1200 * Get stream from stream id.
1201 * Need to be called with RCU read-side lock held.
1202 */
1203 static
1204 struct relay_stream *relay_stream_from_stream_id(uint64_t stream_id,
1205 struct lttng_ht *streams_ht)
1206 {
1207 struct lttng_ht_node_ulong *node;
1208 struct lttng_ht_iter iter;
1209 struct relay_stream *ret;
1210
1211 lttng_ht_lookup(streams_ht,
1212 (void *)((unsigned long) stream_id),
1213 &iter);
1214 node = lttng_ht_iter_get_node_ulong(&iter);
1215 if (node == NULL) {
1216 DBG("Relay stream %" PRIu64 " not found", stream_id);
1217 ret = NULL;
1218 goto end;
1219 }
1220
1221 ret = caa_container_of(node, struct relay_stream, stream_n);
1222
1223 end:
1224 return ret;
1225 }
1226
1227 /*
1228 * Append padding to the file pointed by the file descriptor fd.
1229 */
1230 static int write_padding_to_file(int fd, uint32_t size)
1231 {
1232 int ret = 0;
1233 char *zeros;
1234
1235 if (size == 0) {
1236 goto end;
1237 }
1238
1239 zeros = zmalloc(size);
1240 if (zeros == NULL) {
1241 PERROR("zmalloc zeros for padding");
1242 ret = -1;
1243 goto end;
1244 }
1245
1246 do {
1247 ret = write(fd, zeros, size);
1248 } while (ret < 0 && errno == EINTR);
1249 if (ret < 0 || ret != size) {
1250 PERROR("write padding to file");
1251 }
1252
1253 free(zeros);
1254
1255 end:
1256 return ret;
1257 }
1258
1259 /*
1260 * relay_recv_metadata: receive the metada for the session.
1261 */
1262 static
1263 int relay_recv_metadata(struct lttcomm_relayd_hdr *recv_hdr,
1264 struct relay_command *cmd, struct lttng_ht *streams_ht)
1265 {
1266 int ret = htobe32(LTTNG_OK);
1267 struct relay_session *session = cmd->session;
1268 struct lttcomm_relayd_metadata_payload *metadata_struct;
1269 struct relay_stream *metadata_stream;
1270 uint64_t data_size, payload_size;
1271
1272 if (!session) {
1273 ERR("Metadata sent before version check");
1274 ret = -1;
1275 goto end;
1276 }
1277
1278 data_size = payload_size = be64toh(recv_hdr->data_size);
1279 if (data_size < sizeof(struct lttcomm_relayd_metadata_payload)) {
1280 ERR("Incorrect data size");
1281 ret = -1;
1282 goto end;
1283 }
1284 payload_size -= sizeof(struct lttcomm_relayd_metadata_payload);
1285
1286 if (data_buffer_size < data_size) {
1287 /* In case the realloc fails, we can free the memory */
1288 char *tmp_data_ptr;
1289
1290 tmp_data_ptr = realloc(data_buffer, data_size);
1291 if (!tmp_data_ptr) {
1292 ERR("Allocating data buffer");
1293 free(data_buffer);
1294 ret = -1;
1295 goto end;
1296 }
1297 data_buffer = tmp_data_ptr;
1298 data_buffer_size = data_size;
1299 }
1300 memset(data_buffer, 0, data_size);
1301 DBG2("Relay receiving metadata, waiting for %" PRIu64 " bytes", data_size);
1302 ret = cmd->sock->ops->recvmsg(cmd->sock, data_buffer, data_size, 0);
1303 if (ret < 0 || ret != data_size) {
1304 if (ret == 0) {
1305 /* Orderly shutdown. Not necessary to print an error. */
1306 DBG("Socket %d did an orderly shutdown", cmd->sock->fd);
1307 } else {
1308 ERR("Relay didn't receive the whole metadata");
1309 }
1310 ret = -1;
1311 goto end;
1312 }
1313 metadata_struct = (struct lttcomm_relayd_metadata_payload *) data_buffer;
1314
1315 rcu_read_lock();
1316 metadata_stream = relay_stream_from_stream_id(
1317 be64toh(metadata_struct->stream_id), streams_ht);
1318 if (!metadata_stream) {
1319 ret = -1;
1320 goto end_unlock;
1321 }
1322
1323 do {
1324 ret = write(metadata_stream->fd, metadata_struct->payload,
1325 payload_size);
1326 } while (ret < 0 && errno == EINTR);
1327 if (ret < 0 || ret != payload_size) {
1328 ERR("Relay error writing metadata on file");
1329 ret = -1;
1330 goto end_unlock;
1331 }
1332
1333 ret = write_padding_to_file(metadata_stream->fd,
1334 be32toh(metadata_struct->padding_size));
1335 if (ret < 0) {
1336 goto end_unlock;
1337 }
1338
1339 DBG2("Relay metadata written");
1340
1341 end_unlock:
1342 rcu_read_unlock();
1343 end:
1344 return ret;
1345 }
1346
1347 /*
1348 * relay_send_version: send relayd version number
1349 */
1350 static
1351 int relay_send_version(struct lttcomm_relayd_hdr *recv_hdr,
1352 struct relay_command *cmd)
1353 {
1354 int ret;
1355 struct lttcomm_relayd_version reply, msg;
1356
1357 assert(cmd);
1358
1359 cmd->version_check_done = 1;
1360
1361 /* Get version from the other side. */
1362 ret = cmd->sock->ops->recvmsg(cmd->sock, &msg, sizeof(msg), 0);
1363 if (ret < 0 || ret != sizeof(msg)) {
1364 if (ret == 0) {
1365 /* Orderly shutdown. Not necessary to print an error. */
1366 DBG("Socket %d did an orderly shutdown", cmd->sock->fd);
1367 } else {
1368 ERR("Relay failed to receive the version values.");
1369 }
1370 ret = -1;
1371 goto end;
1372 }
1373
1374 /*
1375 * For now, we just ignore the received version but after 2.1 stable
1376 * release, a check must be done to see if we either adapt to the other
1377 * side version (which MUST be lower than us) or keep the latest data
1378 * structure considering that the other side will adapt.
1379 */
1380
1381 ret = sscanf(VERSION, "%10u.%10u", &reply.major, &reply.minor);
1382 if (ret < 2) {
1383 ERR("Error in scanning version");
1384 ret = -1;
1385 goto end;
1386 }
1387 reply.major = htobe32(reply.major);
1388 reply.minor = htobe32(reply.minor);
1389 ret = cmd->sock->ops->sendmsg(cmd->sock, &reply,
1390 sizeof(struct lttcomm_relayd_version), 0);
1391 if (ret < 0) {
1392 ERR("Relay sending version");
1393 }
1394 DBG("Version check done (%u.%u)", be32toh(reply.major),
1395 be32toh(reply.minor));
1396
1397 end:
1398 return ret;
1399 }
1400
1401 /*
1402 * Check for data pending for a given stream id from the session daemon.
1403 */
1404 static
1405 int relay_data_pending(struct lttcomm_relayd_hdr *recv_hdr,
1406 struct relay_command *cmd, struct lttng_ht *streams_ht)
1407 {
1408 struct relay_session *session = cmd->session;
1409 struct lttcomm_relayd_data_pending msg;
1410 struct lttcomm_relayd_generic_reply reply;
1411 struct relay_stream *stream;
1412 int ret;
1413 struct lttng_ht_node_ulong *node;
1414 struct lttng_ht_iter iter;
1415 uint64_t last_net_seq_num, stream_id;
1416
1417 DBG("Data pending command received");
1418
1419 if (!session || cmd->version_check_done == 0) {
1420 ERR("Trying to check for data before version check");
1421 ret = -1;
1422 goto end_no_session;
1423 }
1424
1425 ret = cmd->sock->ops->recvmsg(cmd->sock, &msg, sizeof(msg), 0);
1426 if (ret < sizeof(msg)) {
1427 if (ret == 0) {
1428 /* Orderly shutdown. Not necessary to print an error. */
1429 DBG("Socket %d did an orderly shutdown", cmd->sock->fd);
1430 } else {
1431 ERR("Relay didn't receive valid data_pending struct size : %d",
1432 ret);
1433 }
1434 ret = -1;
1435 goto end_no_session;
1436 }
1437
1438 stream_id = be64toh(msg.stream_id);
1439 last_net_seq_num = be64toh(msg.last_net_seq_num);
1440
1441 rcu_read_lock();
1442 lttng_ht_lookup(streams_ht, (void *)((unsigned long) stream_id), &iter);
1443 node = lttng_ht_iter_get_node_ulong(&iter);
1444 if (node == NULL) {
1445 DBG("Relay stream %" PRIu64 " not found", stream_id);
1446 ret = -1;
1447 goto end_unlock;
1448 }
1449
1450 stream = caa_container_of(node, struct relay_stream, stream_n);
1451 assert(stream);
1452
1453 DBG("Data pending for stream id %" PRIu64 " prev_seq %" PRIu64
1454 " and last_seq %" PRIu64, stream_id, stream->prev_seq,
1455 last_net_seq_num);
1456
1457 /* Avoid wrapping issue */
1458 if (((int64_t) (stream->prev_seq - last_net_seq_num)) >= 0) {
1459 /* Data has in fact been written and is NOT pending */
1460 ret = 0;
1461 } else {
1462 /* Data still being streamed thus pending */
1463 ret = 1;
1464 }
1465
1466 /* Pending check is now done. */
1467 stream->data_pending_check_done = 1;
1468
1469 end_unlock:
1470 rcu_read_unlock();
1471
1472 reply.ret_code = htobe32(ret);
1473 ret = cmd->sock->ops->sendmsg(cmd->sock, &reply, sizeof(reply), 0);
1474 if (ret < 0) {
1475 ERR("Relay data pending ret code failed");
1476 }
1477
1478 end_no_session:
1479 return ret;
1480 }
1481
1482 /*
1483 * Wait for the control socket to reach a quiescent state.
1484 *
1485 * Note that for now, when receiving this command from the session daemon, this
1486 * means that every subsequent commands or data received on the control socket
1487 * has been handled. So, this is why we simply return OK here.
1488 */
1489 static
1490 int relay_quiescent_control(struct lttcomm_relayd_hdr *recv_hdr,
1491 struct relay_command *cmd, struct lttng_ht *streams_ht)
1492 {
1493 int ret;
1494 uint64_t stream_id;
1495 struct relay_stream *stream;
1496 struct lttng_ht_iter iter;
1497 struct lttcomm_relayd_quiescent_control msg;
1498 struct lttcomm_relayd_generic_reply reply;
1499
1500 DBG("Checking quiescent state on control socket");
1501
1502 if (!cmd->session || cmd->version_check_done == 0) {
1503 ERR("Trying to check for data before version check");
1504 ret = -1;
1505 goto end_no_session;
1506 }
1507
1508 ret = cmd->sock->ops->recvmsg(cmd->sock, &msg, sizeof(msg), 0);
1509 if (ret < sizeof(msg)) {
1510 if (ret == 0) {
1511 /* Orderly shutdown. Not necessary to print an error. */
1512 DBG("Socket %d did an orderly shutdown", cmd->sock->fd);
1513 } else {
1514 ERR("Relay didn't receive valid begin data_pending struct size: %d",
1515 ret);
1516 }
1517 ret = -1;
1518 goto end_no_session;
1519 }
1520
1521 stream_id = be64toh(msg.stream_id);
1522
1523 rcu_read_lock();
1524 cds_lfht_for_each_entry(streams_ht->ht, &iter.iter, stream, stream_n.node) {
1525 if (stream->stream_handle == stream_id) {
1526 stream->data_pending_check_done = 1;
1527 DBG("Relay quiescent control pending flag set to %" PRIu64,
1528 stream_id);
1529 break;
1530 }
1531 }
1532 rcu_read_unlock();
1533
1534 reply.ret_code = htobe32(LTTNG_OK);
1535 ret = cmd->sock->ops->sendmsg(cmd->sock, &reply, sizeof(reply), 0);
1536 if (ret < 0) {
1537 ERR("Relay data quiescent control ret code failed");
1538 }
1539
1540 end_no_session:
1541 return ret;
1542 }
1543
1544 /*
1545 * Initialize a data pending command. This means that a client is about to ask
1546 * for data pending for each stream he/she holds. Simply iterate over all
1547 * streams of a session and set the data_pending_check_done flag.
1548 *
1549 * This command returns to the client a LTTNG_OK code.
1550 */
1551 static
1552 int relay_begin_data_pending(struct lttcomm_relayd_hdr *recv_hdr,
1553 struct relay_command *cmd, struct lttng_ht *streams_ht)
1554 {
1555 int ret;
1556 struct lttng_ht_iter iter;
1557 struct lttcomm_relayd_begin_data_pending msg;
1558 struct lttcomm_relayd_generic_reply reply;
1559 struct relay_stream *stream;
1560 uint64_t session_id;
1561
1562 assert(recv_hdr);
1563 assert(cmd);
1564 assert(streams_ht);
1565
1566 DBG("Init streams for data pending");
1567
1568 if (!cmd->session || cmd->version_check_done == 0) {
1569 ERR("Trying to check for data before version check");
1570 ret = -1;
1571 goto end_no_session;
1572 }
1573
1574 ret = cmd->sock->ops->recvmsg(cmd->sock, &msg, sizeof(msg), 0);
1575 if (ret < sizeof(msg)) {
1576 if (ret == 0) {
1577 /* Orderly shutdown. Not necessary to print an error. */
1578 DBG("Socket %d did an orderly shutdown", cmd->sock->fd);
1579 } else {
1580 ERR("Relay didn't receive valid begin data_pending struct size: %d",
1581 ret);
1582 }
1583 ret = -1;
1584 goto end_no_session;
1585 }
1586
1587 session_id = be64toh(msg.session_id);
1588
1589 /*
1590 * Iterate over all streams to set the begin data pending flag. For now, the
1591 * streams are indexed by stream handle so we have to iterate over all
1592 * streams to find the one associated with the right session_id.
1593 */
1594 rcu_read_lock();
1595 cds_lfht_for_each_entry(streams_ht->ht, &iter.iter, stream, stream_n.node) {
1596 if (stream->session->id == session_id) {
1597 stream->data_pending_check_done = 0;
1598 DBG("Set begin data pending flag to stream %" PRIu64,
1599 stream->stream_handle);
1600 }
1601 }
1602 rcu_read_unlock();
1603
1604 /* All good, send back reply. */
1605 reply.ret_code = htobe32(LTTNG_OK);
1606
1607 ret = cmd->sock->ops->sendmsg(cmd->sock, &reply, sizeof(reply), 0);
1608 if (ret < 0) {
1609 ERR("Relay begin data pending send reply failed");
1610 }
1611
1612 end_no_session:
1613 return ret;
1614 }
1615
1616 /*
1617 * End data pending command. This will check, for a given session id, if each
1618 * stream associated with it has its data_pending_check_done flag set. If not,
1619 * this means that the client lost track of the stream but the data is still
1620 * being streamed on our side. In this case, we inform the client that data is
1621 * inflight.
1622 *
1623 * Return to the client if there is data in flight or not with a ret_code.
1624 */
1625 static
1626 int relay_end_data_pending(struct lttcomm_relayd_hdr *recv_hdr,
1627 struct relay_command *cmd, struct lttng_ht *streams_ht)
1628 {
1629 int ret;
1630 struct lttng_ht_iter iter;
1631 struct lttcomm_relayd_end_data_pending msg;
1632 struct lttcomm_relayd_generic_reply reply;
1633 struct relay_stream *stream;
1634 uint64_t session_id;
1635 uint32_t is_data_inflight = 0;
1636
1637 assert(recv_hdr);
1638 assert(cmd);
1639 assert(streams_ht);
1640
1641 DBG("End data pending command");
1642
1643 if (!cmd->session || cmd->version_check_done == 0) {
1644 ERR("Trying to check for data before version check");
1645 ret = -1;
1646 goto end_no_session;
1647 }
1648
1649 ret = cmd->sock->ops->recvmsg(cmd->sock, &msg, sizeof(msg), 0);
1650 if (ret < sizeof(msg)) {
1651 if (ret == 0) {
1652 /* Orderly shutdown. Not necessary to print an error. */
1653 DBG("Socket %d did an orderly shutdown", cmd->sock->fd);
1654 } else {
1655 ERR("Relay didn't receive valid end data_pending struct size: %d",
1656 ret);
1657 }
1658 ret = -1;
1659 goto end_no_session;
1660 }
1661
1662 session_id = be64toh(msg.session_id);
1663
1664 /* Iterate over all streams to see if the begin data pending flag is set. */
1665 rcu_read_lock();
1666 cds_lfht_for_each_entry(streams_ht->ht, &iter.iter, stream, stream_n.node) {
1667 if (stream->session->id == session_id &&
1668 !stream->data_pending_check_done) {
1669 is_data_inflight = 1;
1670 DBG("Data is still in flight for stream %" PRIu64,
1671 stream->stream_handle);
1672 break;
1673 }
1674 }
1675 rcu_read_unlock();
1676
1677 /* All good, send back reply. */
1678 reply.ret_code = htobe32(is_data_inflight);
1679
1680 ret = cmd->sock->ops->sendmsg(cmd->sock, &reply, sizeof(reply), 0);
1681 if (ret < 0) {
1682 ERR("Relay end data pending send reply failed");
1683 }
1684
1685 end_no_session:
1686 return ret;
1687 }
1688
1689 /*
1690 * relay_process_control: Process the commands received on the control socket
1691 */
1692 static
1693 int relay_process_control(struct lttcomm_relayd_hdr *recv_hdr,
1694 struct relay_command *cmd, struct lttng_ht *streams_ht)
1695 {
1696 int ret = 0;
1697
1698 switch (be32toh(recv_hdr->cmd)) {
1699 case RELAYD_CREATE_SESSION:
1700 ret = relay_create_session(recv_hdr, cmd);
1701 break;
1702 case RELAYD_ADD_STREAM:
1703 ret = relay_add_stream(recv_hdr, cmd, streams_ht);
1704 break;
1705 case RELAYD_START_DATA:
1706 ret = relay_start(recv_hdr, cmd);
1707 break;
1708 case RELAYD_SEND_METADATA:
1709 ret = relay_recv_metadata(recv_hdr, cmd, streams_ht);
1710 break;
1711 case RELAYD_VERSION:
1712 ret = relay_send_version(recv_hdr, cmd);
1713 break;
1714 case RELAYD_CLOSE_STREAM:
1715 ret = relay_close_stream(recv_hdr, cmd, streams_ht);
1716 break;
1717 case RELAYD_DATA_PENDING:
1718 ret = relay_data_pending(recv_hdr, cmd, streams_ht);
1719 break;
1720 case RELAYD_QUIESCENT_CONTROL:
1721 ret = relay_quiescent_control(recv_hdr, cmd, streams_ht);
1722 break;
1723 case RELAYD_BEGIN_DATA_PENDING:
1724 ret = relay_begin_data_pending(recv_hdr, cmd, streams_ht);
1725 break;
1726 case RELAYD_END_DATA_PENDING:
1727 ret = relay_end_data_pending(recv_hdr, cmd, streams_ht);
1728 break;
1729 case RELAYD_UPDATE_SYNC_INFO:
1730 default:
1731 ERR("Received unknown command (%u)", be32toh(recv_hdr->cmd));
1732 relay_unknown_command(cmd);
1733 ret = -1;
1734 goto end;
1735 }
1736
1737 end:
1738 return ret;
1739 }
1740
1741 /*
1742 * relay_process_data: Process the data received on the data socket
1743 */
1744 static
1745 int relay_process_data(struct relay_command *cmd, struct lttng_ht *streams_ht)
1746 {
1747 int ret = 0;
1748 struct relay_stream *stream;
1749 struct lttcomm_relayd_data_hdr data_hdr;
1750 uint64_t stream_id;
1751 uint64_t net_seq_num;
1752 uint32_t data_size;
1753
1754 ret = cmd->sock->ops->recvmsg(cmd->sock, &data_hdr,
1755 sizeof(struct lttcomm_relayd_data_hdr), 0);
1756 if (ret <= 0) {
1757 if (ret == 0) {
1758 /* Orderly shutdown. Not necessary to print an error. */
1759 DBG("Socket %d did an orderly shutdown", cmd->sock->fd);
1760 } else {
1761 ERR("Unable to receive data header on sock %d", cmd->sock->fd);
1762 }
1763 ret = -1;
1764 goto end;
1765 }
1766
1767 stream_id = be64toh(data_hdr.stream_id);
1768
1769 rcu_read_lock();
1770 stream = relay_stream_from_stream_id(stream_id, streams_ht);
1771 if (!stream) {
1772 ret = -1;
1773 goto end_unlock;
1774 }
1775
1776 data_size = be32toh(data_hdr.data_size);
1777 if (data_buffer_size < data_size) {
1778 char *tmp_data_ptr;
1779
1780 tmp_data_ptr = realloc(data_buffer, data_size);
1781 if (!tmp_data_ptr) {
1782 ERR("Allocating data buffer");
1783 free(data_buffer);
1784 ret = -1;
1785 goto end_unlock;
1786 }
1787 data_buffer = tmp_data_ptr;
1788 data_buffer_size = data_size;
1789 }
1790 memset(data_buffer, 0, data_size);
1791
1792 net_seq_num = be64toh(data_hdr.net_seq_num);
1793
1794 DBG3("Receiving data of size %u for stream id %" PRIu64 " seqnum %" PRIu64,
1795 data_size, stream_id, net_seq_num);
1796 ret = cmd->sock->ops->recvmsg(cmd->sock, data_buffer, data_size, 0);
1797 if (ret <= 0) {
1798 if (ret == 0) {
1799 /* Orderly shutdown. Not necessary to print an error. */
1800 DBG("Socket %d did an orderly shutdown", cmd->sock->fd);
1801 }
1802 ret = -1;
1803 goto end_unlock;
1804 }
1805
1806 do {
1807 ret = write(stream->fd, data_buffer, data_size);
1808 } while (ret < 0 && errno == EINTR);
1809 if (ret < 0 || ret != data_size) {
1810 ERR("Relay error writing data to file");
1811 ret = -1;
1812 goto end_unlock;
1813 }
1814
1815 DBG2("Relay wrote %d bytes to tracefile for stream id %" PRIu64,
1816 ret, stream->stream_handle);
1817
1818 ret = write_padding_to_file(stream->fd, be32toh(data_hdr.padding_size));
1819 if (ret < 0) {
1820 goto end_unlock;
1821 }
1822
1823 stream->prev_seq = net_seq_num;
1824
1825 /* Check if we need to close the FD */
1826 if (close_stream_check(stream)) {
1827 int cret;
1828 struct lttng_ht_iter iter;
1829
1830 cret = close(stream->fd);
1831 if (cret < 0) {
1832 PERROR("close stream process data");
1833 }
1834 iter.iter.node = &stream->stream_n.node;
1835 ret = lttng_ht_del(streams_ht, &iter);
1836 assert(!ret);
1837 call_rcu(&stream->rcu_node,
1838 deferred_free_stream);
1839 DBG("Closed tracefile %d after recv data", stream->fd);
1840 }
1841
1842 end_unlock:
1843 rcu_read_unlock();
1844 end:
1845 return ret;
1846 }
1847
1848 static
1849 void relay_cleanup_poll_connection(struct lttng_poll_event *events, int pollfd)
1850 {
1851 int ret;
1852
1853 lttng_poll_del(events, pollfd);
1854
1855 ret = close(pollfd);
1856 if (ret < 0) {
1857 ERR("Closing pollfd %d", pollfd);
1858 }
1859 }
1860
1861 static
1862 int relay_add_connection(int fd, struct lttng_poll_event *events,
1863 struct lttng_ht *relay_connections_ht)
1864 {
1865 struct relay_command *relay_connection;
1866 int ret;
1867
1868 relay_connection = zmalloc(sizeof(struct relay_command));
1869 if (relay_connection == NULL) {
1870 PERROR("Relay command zmalloc");
1871 goto error;
1872 }
1873 do {
1874 ret = read(fd, relay_connection, sizeof(struct relay_command));
1875 } while (ret < 0 && errno == EINTR);
1876 if (ret < 0 || ret < sizeof(struct relay_command)) {
1877 PERROR("read relay cmd pipe");
1878 goto error_read;
1879 }
1880
1881 lttng_ht_node_init_ulong(&relay_connection->sock_n,
1882 (unsigned long) relay_connection->sock->fd);
1883 rcu_read_lock();
1884 lttng_ht_add_unique_ulong(relay_connections_ht,
1885 &relay_connection->sock_n);
1886 rcu_read_unlock();
1887 return lttng_poll_add(events,
1888 relay_connection->sock->fd,
1889 LPOLLIN | LPOLLRDHUP);
1890
1891 error_read:
1892 free(relay_connection);
1893 error:
1894 return -1;
1895 }
1896
1897 static
1898 void deferred_free_connection(struct rcu_head *head)
1899 {
1900 struct relay_command *relay_connection =
1901 caa_container_of(head, struct relay_command, rcu_node);
1902
1903 lttcomm_destroy_sock(relay_connection->sock);
1904 free(relay_connection);
1905 }
1906
1907 static
1908 void relay_del_connection(struct lttng_ht *relay_connections_ht,
1909 struct lttng_ht *streams_ht, struct lttng_ht_iter *iter,
1910 struct relay_command *relay_connection)
1911 {
1912 int ret;
1913
1914 ret = lttng_ht_del(relay_connections_ht, iter);
1915 assert(!ret);
1916 if (relay_connection->type == RELAY_CONTROL) {
1917 relay_delete_session(relay_connection, streams_ht);
1918 }
1919
1920 call_rcu(&relay_connection->rcu_node,
1921 deferred_free_connection);
1922 }
1923
1924 /*
1925 * This thread does the actual work
1926 */
1927 static
1928 void *relay_thread_worker(void *data)
1929 {
1930 int ret, err = -1, last_seen_data_fd = -1;
1931 uint32_t nb_fd;
1932 struct relay_command *relay_connection;
1933 struct lttng_poll_event events;
1934 struct lttng_ht *relay_connections_ht;
1935 struct lttng_ht_node_ulong *node;
1936 struct lttng_ht_iter iter;
1937 struct lttng_ht *streams_ht;
1938 struct lttcomm_relayd_hdr recv_hdr;
1939
1940 DBG("[thread] Relay worker started");
1941
1942 rcu_register_thread();
1943
1944 /* table of connections indexed on socket */
1945 relay_connections_ht = lttng_ht_new(0, LTTNG_HT_TYPE_ULONG);
1946 if (!relay_connections_ht) {
1947 goto relay_connections_ht_error;
1948 }
1949
1950 /* tables of streams indexed by stream ID */
1951 streams_ht = lttng_ht_new(0, LTTNG_HT_TYPE_ULONG);
1952 if (!streams_ht) {
1953 goto streams_ht_error;
1954 }
1955
1956 ret = create_thread_poll_set(&events, 2);
1957 if (ret < 0) {
1958 goto error_poll_create;
1959 }
1960
1961 ret = lttng_poll_add(&events, relay_cmd_pipe[0], LPOLLIN | LPOLLRDHUP);
1962 if (ret < 0) {
1963 goto error;
1964 }
1965
1966 restart:
1967 while (1) {
1968 int idx = -1, i, seen_control = 0, last_notdel_data_fd = -1;
1969
1970 /* Infinite blocking call, waiting for transmission */
1971 DBG3("Relayd worker thread polling...");
1972 ret = lttng_poll_wait(&events, -1);
1973 if (ret < 0) {
1974 /*
1975 * Restart interrupted system call.
1976 */
1977 if (errno == EINTR) {
1978 goto restart;
1979 }
1980 goto error;
1981 }
1982
1983 nb_fd = ret;
1984
1985 /*
1986 * Process control. The control connection is prioritised so we don't
1987 * starve it with high throughout put tracing data on the data
1988 * connection.
1989 */
1990 for (i = 0; i < nb_fd; i++) {
1991 /* Fetch once the poll data */
1992 uint32_t revents = LTTNG_POLL_GETEV(&events, i);
1993 int pollfd = LTTNG_POLL_GETFD(&events, i);
1994
1995 /* Thread quit pipe has been closed. Killing thread. */
1996 ret = check_thread_quit_pipe(pollfd, revents);
1997 if (ret) {
1998 err = 0;
1999 goto exit;
2000 }
2001
2002 /* Inspect the relay cmd pipe for new connection */
2003 if (pollfd == relay_cmd_pipe[0]) {
2004 if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
2005 ERR("Relay pipe error");
2006 goto error;
2007 } else if (revents & LPOLLIN) {
2008 DBG("Relay command received");
2009 ret = relay_add_connection(relay_cmd_pipe[0],
2010 &events, relay_connections_ht);
2011 if (ret < 0) {
2012 goto error;
2013 }
2014 }
2015 } else if (revents) {
2016 rcu_read_lock();
2017 lttng_ht_lookup(relay_connections_ht,
2018 (void *)((unsigned long) pollfd),
2019 &iter);
2020 node = lttng_ht_iter_get_node_ulong(&iter);
2021 if (node == NULL) {
2022 DBG2("Relay sock %d not found", pollfd);
2023 rcu_read_unlock();
2024 goto error;
2025 }
2026 relay_connection = caa_container_of(node,
2027 struct relay_command, sock_n);
2028
2029 if (revents & (LPOLLERR)) {
2030 ERR("POLL ERROR");
2031 relay_cleanup_poll_connection(&events, pollfd);
2032 relay_del_connection(relay_connections_ht,
2033 streams_ht, &iter,
2034 relay_connection);
2035 if (last_seen_data_fd == pollfd) {
2036 last_seen_data_fd = last_notdel_data_fd;
2037 }
2038 } else if (revents & (LPOLLHUP | LPOLLRDHUP)) {
2039 DBG("Socket %d hung up", pollfd);
2040 relay_cleanup_poll_connection(&events, pollfd);
2041 relay_del_connection(relay_connections_ht,
2042 streams_ht, &iter,
2043 relay_connection);
2044 if (last_seen_data_fd == pollfd) {
2045 last_seen_data_fd = last_notdel_data_fd;
2046 }
2047 } else if (revents & LPOLLIN) {
2048 /* control socket */
2049 if (relay_connection->type == RELAY_CONTROL) {
2050 ret = relay_connection->sock->ops->recvmsg(
2051 relay_connection->sock, &recv_hdr,
2052 sizeof(struct lttcomm_relayd_hdr), 0);
2053 /* connection closed */
2054 if (ret <= 0) {
2055 relay_cleanup_poll_connection(&events, pollfd);
2056 relay_del_connection(relay_connections_ht,
2057 streams_ht, &iter,
2058 relay_connection);
2059 DBG("Control connection closed with %d", pollfd);
2060 } else {
2061 if (relay_connection->session) {
2062 DBG2("Relay worker receiving data for session : %" PRIu64,
2063 relay_connection->session->id);
2064 }
2065 ret = relay_process_control(&recv_hdr,
2066 relay_connection,
2067 streams_ht);
2068 if (ret < 0) {
2069 /* Clear the session on error. */
2070 relay_cleanup_poll_connection(&events, pollfd);
2071 relay_del_connection(relay_connections_ht,
2072 streams_ht, &iter,
2073 relay_connection);
2074 DBG("Connection closed with %d", pollfd);
2075 }
2076 seen_control = 1;
2077 }
2078 } else {
2079 /*
2080 * Flag the last seen data fd not deleted. It will be
2081 * used as the last seen fd if any fd gets deleted in
2082 * this first loop.
2083 */
2084 last_notdel_data_fd = pollfd;
2085 }
2086 }
2087 rcu_read_unlock();
2088 }
2089 }
2090
2091 /*
2092 * The last loop handled a control request, go back to poll to make
2093 * sure we prioritise the control socket.
2094 */
2095 if (seen_control) {
2096 continue;
2097 }
2098
2099 if (last_seen_data_fd >= 0) {
2100 for (i = 0; i < nb_fd; i++) {
2101 int pollfd = LTTNG_POLL_GETFD(&events, i);
2102 if (last_seen_data_fd == pollfd) {
2103 idx = i;
2104 break;
2105 }
2106 }
2107 }
2108
2109 /* Process data connection. */
2110 for (i = idx + 1; i < nb_fd; i++) {
2111 /* Fetch the poll data. */
2112 uint32_t revents = LTTNG_POLL_GETEV(&events, i);
2113 int pollfd = LTTNG_POLL_GETFD(&events, i);
2114
2115 /* Skip the command pipe. It's handled in the first loop. */
2116 if (pollfd == relay_cmd_pipe[0]) {
2117 continue;
2118 }
2119
2120 if (revents) {
2121 rcu_read_lock();
2122 lttng_ht_lookup(relay_connections_ht,
2123 (void *)((unsigned long) pollfd),
2124 &iter);
2125 node = lttng_ht_iter_get_node_ulong(&iter);
2126 if (node == NULL) {
2127 /* Skip it. Might be removed before. */
2128 rcu_read_unlock();
2129 continue;
2130 }
2131 relay_connection = caa_container_of(node,
2132 struct relay_command, sock_n);
2133
2134 if (revents & LPOLLIN) {
2135 if (relay_connection->type != RELAY_DATA) {
2136 continue;
2137 }
2138
2139 ret = relay_process_data(relay_connection, streams_ht);
2140 /* connection closed */
2141 if (ret < 0) {
2142 relay_cleanup_poll_connection(&events, pollfd);
2143 relay_del_connection(relay_connections_ht,
2144 streams_ht, &iter,
2145 relay_connection);
2146 DBG("Data connection closed with %d", pollfd);
2147 /*
2148 * Every goto restart call sets the last seen fd where
2149 * here we don't really care since we gracefully
2150 * continue the loop after the connection is deleted.
2151 */
2152 } else {
2153 /* Keep last seen port. */
2154 last_seen_data_fd = pollfd;
2155 rcu_read_unlock();
2156 goto restart;
2157 }
2158 }
2159 rcu_read_unlock();
2160 }
2161 }
2162 last_seen_data_fd = -1;
2163 }
2164
2165 exit:
2166 error:
2167 lttng_poll_clean(&events);
2168
2169 /* empty the hash table and free the memory */
2170 rcu_read_lock();
2171 cds_lfht_for_each_entry(relay_connections_ht->ht, &iter.iter, node, node) {
2172 node = lttng_ht_iter_get_node_ulong(&iter);
2173 if (node) {
2174 relay_connection = caa_container_of(node,
2175 struct relay_command, sock_n);
2176 relay_del_connection(relay_connections_ht,
2177 streams_ht, &iter,
2178 relay_connection);
2179 }
2180 }
2181 rcu_read_unlock();
2182 error_poll_create:
2183 lttng_ht_destroy(streams_ht);
2184 streams_ht_error:
2185 lttng_ht_destroy(relay_connections_ht);
2186 relay_connections_ht_error:
2187 /* Close relay cmd pipes */
2188 utils_close_pipe(relay_cmd_pipe);
2189 if (err) {
2190 DBG("Thread exited with error");
2191 }
2192 DBG("Worker thread cleanup complete");
2193 free(data_buffer);
2194 stop_threads();
2195 rcu_unregister_thread();
2196 return NULL;
2197 }
2198
2199 /*
2200 * Create the relay command pipe to wake thread_manage_apps.
2201 * Closed in cleanup().
2202 */
2203 static int create_relay_cmd_pipe(void)
2204 {
2205 int ret;
2206
2207 ret = utils_create_pipe_cloexec(relay_cmd_pipe);
2208
2209 return ret;
2210 }
2211
2212 /*
2213 * main
2214 */
2215 int main(int argc, char **argv)
2216 {
2217 int ret = 0;
2218 void *status;
2219
2220 /* Create thread quit pipe */
2221 if ((ret = init_thread_quit_pipe()) < 0) {
2222 goto error;
2223 }
2224
2225 /* Parse arguments */
2226 progname = argv[0];
2227 if ((ret = parse_args(argc, argv)) < 0) {
2228 goto exit;
2229 }
2230
2231 if ((ret = set_signal_handler()) < 0) {
2232 goto exit;
2233 }
2234
2235 /* Daemonize */
2236 if (opt_daemon) {
2237 ret = daemon(0, 0);
2238 if (ret < 0) {
2239 PERROR("daemon");
2240 goto exit;
2241 }
2242 }
2243
2244 /* Check if daemon is UID = 0 */
2245 is_root = !getuid();
2246
2247 if (!is_root) {
2248 if (control_uri->port < 1024 || data_uri->port < 1024) {
2249 ERR("Need to be root to use ports < 1024");
2250 ret = -1;
2251 goto exit;
2252 }
2253 }
2254
2255 /* Setup the thread apps communication pipe. */
2256 if ((ret = create_relay_cmd_pipe()) < 0) {
2257 goto exit;
2258 }
2259
2260 /* Init relay command queue. */
2261 cds_wfq_init(&relay_cmd_queue.queue);
2262
2263 /* Set up max poll set size */
2264 lttng_poll_set_max_size();
2265
2266 /* Setup the dispatcher thread */
2267 ret = pthread_create(&dispatcher_thread, NULL,
2268 relay_thread_dispatcher, (void *) NULL);
2269 if (ret != 0) {
2270 PERROR("pthread_create dispatcher");
2271 goto exit_dispatcher;
2272 }
2273
2274 /* Setup the worker thread */
2275 ret = pthread_create(&worker_thread, NULL,
2276 relay_thread_worker, (void *) NULL);
2277 if (ret != 0) {
2278 PERROR("pthread_create worker");
2279 goto exit_worker;
2280 }
2281
2282 /* Setup the listener thread */
2283 ret = pthread_create(&listener_thread, NULL,
2284 relay_thread_listener, (void *) NULL);
2285 if (ret != 0) {
2286 PERROR("pthread_create listener");
2287 goto exit_listener;
2288 }
2289
2290 exit_listener:
2291 ret = pthread_join(listener_thread, &status);
2292 if (ret != 0) {
2293 PERROR("pthread_join");
2294 goto error; /* join error, exit without cleanup */
2295 }
2296
2297 exit_worker:
2298 ret = pthread_join(worker_thread, &status);
2299 if (ret != 0) {
2300 PERROR("pthread_join");
2301 goto error; /* join error, exit without cleanup */
2302 }
2303
2304 exit_dispatcher:
2305 ret = pthread_join(dispatcher_thread, &status);
2306 if (ret != 0) {
2307 PERROR("pthread_join");
2308 goto error; /* join error, exit without cleanup */
2309 }
2310
2311 exit:
2312 cleanup();
2313 if (!ret) {
2314 exit(EXIT_SUCCESS);
2315 }
2316
2317 error:
2318 exit(EXIT_FAILURE);
2319 }
This page took 0.135926 seconds and 4 git commands to generate.