Map session id of relayd and sessiond in consumer
[lttng-tools.git] / src / bin / lttng-relayd / main.c
CommitLineData
b8aa1682
JD
1/*
2 * Copyright (C) 2012 - Julien Desfossez <jdesfossez@efficios.com>
3 * David Goulet <dgoulet@efficios.com>
4 *
5 * This program is free software; you can redistribute it and/or modify
6 * it under the terms of the GNU General Public License, version 2 only,
7 * as published by the Free Software Foundation.
8 *
9 * This program is distributed in the hope that it will be useful, but WITHOUT
10 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
11 * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
12 * more details.
13 *
14 * You should have received a copy of the GNU General Public License along
15 * with this program; if not, write to the Free Software Foundation, Inc.,
16 * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
17 */
18
19#define _GNU_SOURCE
20#include <getopt.h>
21#include <grp.h>
22#include <limits.h>
23#include <pthread.h>
24#include <signal.h>
25#include <stdio.h>
26#include <stdlib.h>
27#include <string.h>
28#include <sys/mman.h>
29#include <sys/mount.h>
30#include <sys/resource.h>
31#include <sys/socket.h>
32#include <sys/stat.h>
33#include <sys/types.h>
34#include <sys/wait.h>
173af62f 35#include <inttypes.h>
b8aa1682
JD
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>
a02de639 53#include <common/utils.h>
b8aa1682
JD
54
55#include "lttng-relayd.h"
56
57/* command line options */
58static int opt_daemon;
59static char *opt_output_path;
095a4ae5
MD
60static struct lttng_uri *control_uri;
61static struct lttng_uri *data_uri;
b8aa1682
JD
62
63const char *progname;
64static 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 */
70static 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 */
76static int relay_cmd_pipe[2] = { -1, -1 };
77
26c9d55e 78/* Shared between threads */
b8aa1682
JD
79static int dispatch_thread_exit;
80
81static pthread_t listener_thread;
82static pthread_t dispatcher_thread;
83static pthread_t worker_thread;
84
095a4ae5
MD
85static uint64_t last_relay_stream_id;
86static uint64_t last_relay_session_id;
b8aa1682
JD
87
88/*
89 * Relay command queue.
90 *
91 * The relay_thread_listener and relay_thread_dispatcher communicate with this
92 * queue.
93 */
94static struct relay_cmd_queue relay_cmd_queue;
95
96/* buffer allocated at startup, used to store the trace data */
095a4ae5
MD
97static char *data_buffer;
98static unsigned int data_buffer_size;
b8aa1682
JD
99
100/*
101 * usage function on stderr
102 */
103static
104void 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");
25672d02 111 fprintf(stderr, " -o, --output Output path for traces (PATH)\n");
b8aa1682
JD
112 fprintf(stderr, " -v, --verbose Verbose mode. Activate DBG() macro.\n");
113}
114
115static
116int 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[] = {
e3678fd8
MD
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', },
095a4ae5 129 { NULL, 0, 0, 0, },
b8aa1682
JD
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
224exit:
225 return ret;
226}
227
228/*
229 * Cleanup the daemon
230 */
231static
232void cleanup(void)
233{
b8aa1682
JD
234 DBG("Cleaning up");
235
095a4ae5
MD
236 /* free the dynamically allocated opt_output_path */
237 free(opt_output_path);
238
a02de639
CB
239 /* Close thread quit pipes */
240 utils_close_pipe(thread_quit_pipe);
241
710c1f73
DG
242 uri_free(control_uri);
243 uri_free(data_uri);
b8aa1682
JD
244}
245
246/*
247 * Write to writable pipe used to notify a thread.
248 */
249static
250int notify_thread_pipe(int wpipe)
251{
252 int ret;
253
6f94560a
MD
254 do {
255 ret = write(wpipe, "!", 1);
256 } while (ret < 0 && errno == EINTR);
b8aa1682
JD
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 */
267static
268void 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 */
26c9d55e 280 CMM_STORE_SHARED(dispatch_thread_exit, 1);
b8aa1682
JD
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 */
290static
291void 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 */
314static
315int 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 */
354static
355int init_thread_quit_pipe(void)
356{
a02de639 357 int ret;
b8aa1682 358
a02de639 359 ret = utils_create_pipe_cloexec(thread_quit_pipe);
b8aa1682 360
b8aa1682
JD
361 return ret;
362}
363
364/*
365 * Create a poll set with O_CLOEXEC and add the thread quit pipe to the set.
366 */
367static
368int 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
390error:
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 */
399static
400int 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 */
412static
413struct 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
443error:
444 if (sock) {
445 lttcomm_destroy_sock(sock);
446 }
447 return NULL;
448}
449
173af62f
DG
450/*
451 * Return nonzero if stream needs to be closed.
452 */
453static
454int 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
b8aa1682
JD
463/*
464 * This thread manages the listening for new connections on the network
465 */
466static
467void *relay_thread_listener(void *data)
468{
095a4ae5 469 int i, ret, pollfd, err = -1;
b8aa1682
JD
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
b8aa1682
JD
475 DBG("[thread] Relay listener started");
476
477 control_sock = relay_init_sock(control_uri);
478 if (!control_sock) {
095a4ae5 479 goto error_sock_control;
b8aa1682
JD
480 }
481
482 data_sock = relay_init_sock(data_uri);
483 if (!data_sock) {
095a4ae5 484 goto error_sock_relay;
b8aa1682
JD
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
512restart:
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) {
095a4ae5
MD
533 err = 0;
534 goto exit;
b8aa1682
JD
535 }
536
537 if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
538 ERR("socket poll error");
539 goto error;
540 } else if (revents & LPOLLIN) {
4b7f17b2
MD
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;
b8aa1682
JD
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);
4b7f17b2 557 if (!newsock) {
b8aa1682 558 PERROR("accepting data sock");
4b7f17b2 559 free(relay_cmd);
b8aa1682
JD
560 goto error;
561 }
562 relay_cmd->type = RELAY_DATA;
563 DBG("Relay data connection accepted, socket %d", newsock->fd);
4b7f17b2
MD
564 } else {
565 assert(pollfd == control_sock->fd);
b8aa1682 566 newsock = control_sock->ops->accept(control_sock);
4b7f17b2 567 if (!newsock) {
b8aa1682 568 PERROR("accepting control sock");
4b7f17b2 569 free(relay_cmd);
b8aa1682
JD
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");
4b7f17b2
MD
579 lttcomm_destroy_sock(newsock);
580 free(relay_cmd);
b8aa1682
JD
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
095a4ae5 598exit:
b8aa1682
JD
599error:
600error_poll_add:
601 lttng_poll_clean(&events);
602error_create_poll:
095a4ae5
MD
603 if (data_sock->fd >= 0) {
604 ret = data_sock->ops->close(data_sock);
b8aa1682
JD
605 if (ret) {
606 PERROR("close");
607 }
b8aa1682 608 }
095a4ae5
MD
609 lttcomm_destroy_sock(data_sock);
610error_sock_relay:
611 if (control_sock->fd >= 0) {
612 ret = control_sock->ops->close(control_sock);
b8aa1682
JD
613 if (ret) {
614 PERROR("close");
615 }
b8aa1682 616 }
095a4ae5
MD
617 lttcomm_destroy_sock(control_sock);
618error_sock_control:
619 if (err) {
620 DBG("Thread exited with error");
621 }
b8aa1682
JD
622 DBG("Relay listener thread cleanup complete");
623 stop_threads();
b8aa1682
JD
624 return NULL;
625}
626
627/*
628 * This thread manages the dispatching of the requests to worker threads
629 */
630static
631void *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
26c9d55e 639 while (!CMM_LOAD_SHARED(dispatch_thread_exit)) {
b8aa1682
JD
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 */
6f94560a
MD
660 do {
661 ret = write(relay_cmd_pipe[1], relay_cmd,
662 sizeof(struct relay_command));
663 } while (ret < 0 && errno == EINTR);
b8aa1682
JD
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
675error:
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 */
687static
688char *expand_full_path(const char *path)
689{
690 const char *end_path = path;
095a4ae5 691 char *next, *cut_path, *expanded_path, *respath;
b8aa1682
JD
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) {
095a4ae5
MD
703 respath = NULL;
704 goto end;
b8aa1682
JD
705 }
706
095a4ae5
MD
707 respath = realpath(cut_path, expanded_path);
708 if (respath == NULL) {
b8aa1682
JD
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 }
095a4ae5
MD
717 free(expanded_path);
718 } else {
719 /* Add end part to expanded path */
720 strcat(respath, end_path);
b8aa1682 721 }
095a4ae5 722end:
b8aa1682 723 free(cut_path);
095a4ae5 724 return respath;
b8aa1682
JD
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 */
733static
734char *config_get_default_path(void)
735{
736 return getenv("HOME");
737}
738
739/*
740 * Create recursively directory using the FULL path.
741 */
742static
743int 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) {
095a4ae5 776 if (errno != EEXIST) {
b8aa1682
JD
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) {
095a4ae5 789 if (errno != EEXIST) {
b8aa1682
JD
790 PERROR("mkdir recursive last piece");
791 ret = -errno;
792 } else {
793 ret = 0;
794 }
795 }
796
797error:
798 return ret;
799}
800
b8aa1682 801static
095a4ae5 802char *create_output_path_auto(char *path_name)
b8aa1682 803{
095a4ae5 804 int ret;
b8aa1682 805 char *traces_path = NULL;
095a4ae5
MD
806 char *alloc_path = NULL;
807 char *default_path;
b8aa1682 808
095a4ae5
MD
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;
b8aa1682 825 }
095a4ae5 826exit:
b8aa1682 827 free(alloc_path);
095a4ae5
MD
828 return traces_path;
829}
830
831static
832char *create_output_path_noauto(char *path_name)
833{
834 int ret;
835 char *traces_path = NULL;
836 char *full_path;
b8aa1682 837
095a4ae5
MD
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 }
b8aa1682 844exit:
095a4ae5 845 free(full_path);
b8aa1682
JD
846 return traces_path;
847}
848
095a4ae5
MD
849/*
850 * create_output_path: create the output trace directory
851 */
852static
853char *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
9d1bbf21
MD
862static
863void 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
b8aa1682
JD
870/*
871 * relay_delete_session: Free all memory associated with a session and
872 * close all the FDs
873 */
874static
875void 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
095a4ae5 882 if (!cmd->session) {
b8aa1682 883 return;
095a4ae5 884 }
b8aa1682 885
77c7c900 886 DBG("Relay deleting session %" PRIu64, cmd->session->id);
5b6d8097 887
9d1bbf21 888 rcu_read_lock();
b8aa1682
JD
889 cds_lfht_for_each_entry(streams_ht->ht, &iter.iter, node, node) {
890 node = lttng_ht_iter_get_node_ulong(&iter);
891 if (node) {
892 stream = caa_container_of(node,
893 struct relay_stream, stream_n);
894 if (stream->session == cmd->session) {
f66c074c
DG
895 ret = close(stream->fd);
896 if (ret < 0) {
897 PERROR("close stream fd on delete session");
898 }
b8aa1682
JD
899 ret = lttng_ht_del(streams_ht, &iter);
900 assert(!ret);
9d1bbf21
MD
901 call_rcu(&stream->rcu_node,
902 deferred_free_stream);
b8aa1682
JD
903 }
904 }
905 }
9d1bbf21 906 rcu_read_unlock();
5b6d8097
DG
907
908 free(cmd->session);
b8aa1682
JD
909}
910
c5b6f4f0
DG
911/*
912 * Handle the RELAYD_CREATE_SESSION command.
913 *
914 * On success, send back the session id or else return a negative value.
915 */
916static
917int relay_create_session(struct lttcomm_relayd_hdr *recv_hdr,
918 struct relay_command *cmd)
919{
920 int ret = 0, send_ret;
921 struct relay_session *session;
922 struct lttcomm_relayd_status_session reply;
923
924 assert(recv_hdr);
925 assert(cmd);
926
927 memset(&reply, 0, sizeof(reply));
928
929 session = zmalloc(sizeof(struct relay_session));
930 if (session == NULL) {
931 PERROR("relay session zmalloc");
932 ret = -1;
933 goto error;
934 }
935
936 session->id = ++last_relay_session_id;
937 session->sock = cmd->sock;
938 cmd->session = session;
939
940 reply.session_id = htobe64(session->id);
941
942 DBG("Created session %" PRIu64, session->id);
943
944error:
945 if (ret < 0) {
946 reply.ret_code = htobe32(LTTNG_ERR_FATAL);
947 } else {
948 reply.ret_code = htobe32(LTTNG_OK);
949 }
950
951 send_ret = cmd->sock->ops->sendmsg(cmd->sock, &reply, sizeof(reply), 0);
952 if (send_ret < 0) {
953 ERR("Relayd sending session id");
954 }
955
956 return ret;
957}
958
b8aa1682
JD
959/*
960 * relay_add_stream: allocate a new stream for a session
961 */
962static
963int relay_add_stream(struct lttcomm_relayd_hdr *recv_hdr,
964 struct relay_command *cmd, struct lttng_ht *streams_ht)
965{
966 struct relay_session *session = cmd->session;
967 struct lttcomm_relayd_add_stream stream_info;
968 struct relay_stream *stream = NULL;
969 struct lttcomm_relayd_status_stream reply;
970 char *path = NULL, *root_path = NULL;
971 int ret, send_ret;
972
c5b6f4f0 973 if (!session || cmd->version_check_done == 0) {
b8aa1682
JD
974 ERR("Trying to add a stream before version check");
975 ret = -1;
976 goto end_no_session;
977 }
978
979 /* FIXME : use data_size for something ? */
980 ret = cmd->sock->ops->recvmsg(cmd->sock, &stream_info,
7c5aef62 981 sizeof(struct lttcomm_relayd_add_stream), 0);
b8aa1682
JD
982 if (ret < sizeof(struct lttcomm_relayd_add_stream)) {
983 ERR("Relay didn't receive valid add_stream struct size : %d", ret);
984 ret = -1;
985 goto end_no_session;
986 }
987 stream = zmalloc(sizeof(struct relay_stream));
988 if (stream == NULL) {
989 PERROR("relay stream zmalloc");
990 ret = -1;
991 goto end_no_session;
992 }
993
9d1bbf21 994 rcu_read_lock();
b8aa1682 995 stream->stream_handle = ++last_relay_stream_id;
173af62f 996 stream->prev_seq = -1ULL;
b8aa1682
JD
997 stream->session = session;
998
999 root_path = create_output_path(stream_info.pathname);
1000 if (!root_path) {
1001 ret = -1;
1002 goto end;
1003 }
1004 ret = mkdir_recursive(root_path, S_IRWXU | S_IRWXG);
1005 if (ret < 0) {
b8aa1682
JD
1006 ERR("relay creating output directory");
1007 goto end;
1008 }
1009
1010 ret = asprintf(&path, "%s/%s", root_path, stream_info.channel_name);
1011 if (ret < 0) {
1012 PERROR("asprintf stream path");
1013 goto end;
1014 }
1015
1016 ret = open(path, O_WRONLY|O_CREAT|O_TRUNC, S_IRWXU|S_IRWXG|S_IRWXO);
1017 if (ret < 0) {
1018 PERROR("Relay creating trace file");
1019 goto end;
1020 }
1021
1022 stream->fd = ret;
1023 DBG("Tracefile %s created", path);
1024
1025 lttng_ht_node_init_ulong(&stream->stream_n,
1026 (unsigned long) stream->stream_handle);
1027 lttng_ht_add_unique_ulong(streams_ht,
1028 &stream->stream_n);
1029
1030 DBG("Relay new stream added %s", stream_info.channel_name);
1031
1032end:
1033 free(path);
1034 free(root_path);
1035 /* send the session id to the client or a negative return code on error */
1036 if (ret < 0) {
f73fabfd 1037 reply.ret_code = htobe32(LTTNG_ERR_UNK);
b8aa1682 1038 } else {
f73fabfd 1039 reply.ret_code = htobe32(LTTNG_OK);
b8aa1682
JD
1040 }
1041 reply.handle = htobe64(stream->stream_handle);
1042 send_ret = cmd->sock->ops->sendmsg(cmd->sock, &reply,
1043 sizeof(struct lttcomm_relayd_status_stream), 0);
1044 if (send_ret < 0) {
1045 ERR("Relay sending stream id");
1046 }
9d1bbf21 1047 rcu_read_unlock();
b8aa1682
JD
1048
1049end_no_session:
1050 return ret;
1051}
1052
173af62f
DG
1053/*
1054 * relay_close_stream: close a specific stream
1055 */
1056static
1057int relay_close_stream(struct lttcomm_relayd_hdr *recv_hdr,
1058 struct relay_command *cmd, struct lttng_ht *streams_ht)
1059{
1060 struct relay_session *session = cmd->session;
1061 struct lttcomm_relayd_close_stream stream_info;
1062 struct lttcomm_relayd_generic_reply reply;
1063 struct relay_stream *stream;
1064 int ret, send_ret;
1065 struct lttng_ht_node_ulong *node;
1066 struct lttng_ht_iter iter;
1067
1068 DBG("Close stream received");
1069
c5b6f4f0 1070 if (!session || cmd->version_check_done == 0) {
173af62f
DG
1071 ERR("Trying to close a stream before version check");
1072 ret = -1;
1073 goto end_no_session;
1074 }
1075
1076 ret = cmd->sock->ops->recvmsg(cmd->sock, &stream_info,
7c5aef62 1077 sizeof(struct lttcomm_relayd_close_stream), 0);
173af62f
DG
1078 if (ret < sizeof(struct lttcomm_relayd_close_stream)) {
1079 ERR("Relay didn't receive valid add_stream struct size : %d", ret);
1080 ret = -1;
1081 goto end_no_session;
1082 }
1083
1084 rcu_read_lock();
1085 lttng_ht_lookup(streams_ht,
1086 (void *)((unsigned long) be64toh(stream_info.stream_id)),
1087 &iter);
1088 node = lttng_ht_iter_get_node_ulong(&iter);
1089 if (node == NULL) {
77c7c900 1090 DBG("Relay stream %" PRIu64 " not found", be64toh(stream_info.stream_id));
173af62f
DG
1091 ret = -1;
1092 goto end_unlock;
1093 }
1094
1095 stream = caa_container_of(node, struct relay_stream, stream_n);
1096 if (!stream) {
1097 ret = -1;
1098 goto end_unlock;
1099 }
1100
8e2583a4 1101 stream->last_net_seq_num = be64toh(stream_info.last_net_seq_num);
173af62f
DG
1102 stream->close_flag = 1;
1103
1104 if (close_stream_check(stream)) {
1105 int delret;
1106
f66c074c
DG
1107 delret = close(stream->fd);
1108 if (delret < 0) {
1109 PERROR("close stream");
1110 }
173af62f
DG
1111 delret = lttng_ht_del(streams_ht, &iter);
1112 assert(!delret);
1113 call_rcu(&stream->rcu_node,
1114 deferred_free_stream);
1115 DBG("Closed tracefile %d from close stream", stream->fd);
1116 }
1117
1118end_unlock:
1119 rcu_read_unlock();
1120
1121 if (ret < 0) {
f73fabfd 1122 reply.ret_code = htobe32(LTTNG_ERR_UNK);
173af62f 1123 } else {
f73fabfd 1124 reply.ret_code = htobe32(LTTNG_OK);
173af62f
DG
1125 }
1126 send_ret = cmd->sock->ops->sendmsg(cmd->sock, &reply,
1127 sizeof(struct lttcomm_relayd_generic_reply), 0);
1128 if (send_ret < 0) {
1129 ERR("Relay sending stream id");
1130 }
1131
1132end_no_session:
1133 return ret;
1134}
1135
b8aa1682
JD
1136/*
1137 * relay_unknown_command: send -1 if received unknown command
1138 */
1139static
1140void relay_unknown_command(struct relay_command *cmd)
1141{
1142 struct lttcomm_relayd_generic_reply reply;
1143 int ret;
1144
f73fabfd 1145 reply.ret_code = htobe32(LTTNG_ERR_UNK);
b8aa1682
JD
1146 ret = cmd->sock->ops->sendmsg(cmd->sock, &reply,
1147 sizeof(struct lttcomm_relayd_generic_reply), 0);
1148 if (ret < 0) {
1149 ERR("Relay sending unknown command");
1150 }
1151}
1152
1153/*
1154 * relay_start: send an acknowledgment to the client to tell if we are
1155 * ready to receive data. We are ready if a session is established.
1156 */
1157static
1158int relay_start(struct lttcomm_relayd_hdr *recv_hdr,
1159 struct relay_command *cmd)
1160{
f73fabfd 1161 int ret = htobe32(LTTNG_OK);
b8aa1682
JD
1162 struct lttcomm_relayd_generic_reply reply;
1163 struct relay_session *session = cmd->session;
1164
1165 if (!session) {
1166 DBG("Trying to start the streaming without a session established");
f73fabfd 1167 ret = htobe32(LTTNG_ERR_UNK);
b8aa1682
JD
1168 }
1169
1170 reply.ret_code = ret;
1171 ret = cmd->sock->ops->sendmsg(cmd->sock, &reply,
1172 sizeof(struct lttcomm_relayd_generic_reply), 0);
1173 if (ret < 0) {
1174 ERR("Relay sending start ack");
1175 }
1176
1177 return ret;
1178}
1179
1180/*
1181 * Get stream from stream id.
9d1bbf21 1182 * Need to be called with RCU read-side lock held.
b8aa1682
JD
1183 */
1184static
1185struct relay_stream *relay_stream_from_stream_id(uint64_t stream_id,
1186 struct lttng_ht *streams_ht)
1187{
1188 struct lttng_ht_node_ulong *node;
1189 struct lttng_ht_iter iter;
1190 struct relay_stream *ret;
1191
1192 lttng_ht_lookup(streams_ht,
1193 (void *)((unsigned long) stream_id),
1194 &iter);
1195 node = lttng_ht_iter_get_node_ulong(&iter);
1196 if (node == NULL) {
77c7c900 1197 DBG("Relay stream %" PRIu64 " not found", stream_id);
b8aa1682
JD
1198 ret = NULL;
1199 goto end;
1200 }
1201
1202 ret = caa_container_of(node, struct relay_stream, stream_n);
1203
1204end:
1205 return ret;
1206}
1207
1d4dfdef
DG
1208/*
1209 * Append padding to the file pointed by the file descriptor fd.
1210 */
1211static int write_padding_to_file(int fd, uint32_t size)
1212{
1213 int ret = 0;
1214 char *zeros;
1215
1216 if (size == 0) {
1217 goto end;
1218 }
1219
1220 zeros = zmalloc(size);
1221 if (zeros == NULL) {
1222 PERROR("zmalloc zeros for padding");
1223 ret = -1;
1224 goto end;
1225 }
1226
1227 do {
1228 ret = write(fd, zeros, size);
1229 } while (ret < 0 && errno == EINTR);
1230 if (ret < 0) {
1231 PERROR("write padding to file");
1232 }
1233
e986c7a1
DG
1234 free(zeros);
1235
1d4dfdef
DG
1236end:
1237 return ret;
1238}
1239
b8aa1682
JD
1240/*
1241 * relay_recv_metadata: receive the metada for the session.
1242 */
1243static
1244int relay_recv_metadata(struct lttcomm_relayd_hdr *recv_hdr,
1245 struct relay_command *cmd, struct lttng_ht *streams_ht)
1246{
f73fabfd 1247 int ret = htobe32(LTTNG_OK);
b8aa1682
JD
1248 struct relay_session *session = cmd->session;
1249 struct lttcomm_relayd_metadata_payload *metadata_struct;
1250 struct relay_stream *metadata_stream;
1251 uint64_t data_size, payload_size;
1252
1253 if (!session) {
1254 ERR("Metadata sent before version check");
1255 ret = -1;
1256 goto end;
1257 }
1258
f6416125
MD
1259 data_size = payload_size = be64toh(recv_hdr->data_size);
1260 if (data_size < sizeof(struct lttcomm_relayd_metadata_payload)) {
1261 ERR("Incorrect data size");
1262 ret = -1;
1263 goto end;
1264 }
1265 payload_size -= sizeof(struct lttcomm_relayd_metadata_payload);
1266
b8aa1682 1267 if (data_buffer_size < data_size) {
d7b3776f
DG
1268 /* In case the realloc fails, we can free the memory */
1269 char *tmp_data_ptr = data_buffer;
b8aa1682
JD
1270 data_buffer = realloc(data_buffer, data_size);
1271 if (!data_buffer) {
1272 ERR("Allocating data buffer");
d7b3776f 1273 free(tmp_data_ptr);
b8aa1682
JD
1274 ret = -1;
1275 goto end;
1276 }
1277 data_buffer_size = data_size;
1278 }
1279 memset(data_buffer, 0, data_size);
77c7c900 1280 DBG2("Relay receiving metadata, waiting for %" PRIu64 " bytes", data_size);
7c5aef62 1281 ret = cmd->sock->ops->recvmsg(cmd->sock, data_buffer, data_size, 0);
b8aa1682
JD
1282 if (ret < 0 || ret != data_size) {
1283 ret = -1;
1284 ERR("Relay didn't receive the whole metadata");
1285 goto end;
1286 }
1287 metadata_struct = (struct lttcomm_relayd_metadata_payload *) data_buffer;
9d1bbf21
MD
1288
1289 rcu_read_lock();
b8aa1682
JD
1290 metadata_stream = relay_stream_from_stream_id(
1291 be64toh(metadata_struct->stream_id), streams_ht);
1292 if (!metadata_stream) {
1293 ret = -1;
9d1bbf21 1294 goto end_unlock;
b8aa1682
JD
1295 }
1296
6f94560a
MD
1297 do {
1298 ret = write(metadata_stream->fd, metadata_struct->payload,
1299 payload_size);
1300 } while (ret < 0 && errno == EINTR);
87c1611d 1301 if (ret < payload_size) {
b8aa1682
JD
1302 ERR("Relay error writing metadata on file");
1303 ret = -1;
9d1bbf21 1304 goto end_unlock;
b8aa1682 1305 }
1d4dfdef
DG
1306
1307 ret = write_padding_to_file(metadata_stream->fd,
1308 be32toh(metadata_struct->padding_size));
1309 if (ret < 0) {
1310 goto end_unlock;
1311 }
1312
b8aa1682
JD
1313 DBG2("Relay metadata written");
1314
9d1bbf21 1315end_unlock:
6e3c5836 1316 rcu_read_unlock();
b8aa1682
JD
1317end:
1318 return ret;
1319}
1320
1321/*
1322 * relay_send_version: send relayd version number
1323 */
1324static
1325int relay_send_version(struct lttcomm_relayd_hdr *recv_hdr,
1326 struct relay_command *cmd)
1327{
7f51dcba 1328 int ret;
092b6259 1329 struct lttcomm_relayd_version reply, msg;
b8aa1682 1330
c5b6f4f0
DG
1331 assert(cmd);
1332
1333 cmd->version_check_done = 1;
b8aa1682 1334
092b6259 1335 /* Get version from the other side. */
7c5aef62 1336 ret = cmd->sock->ops->recvmsg(cmd->sock, &msg, sizeof(msg), 0);
092b6259
DG
1337 if (ret < 0 || ret != sizeof(msg)) {
1338 ret = -1;
1339 ERR("Relay failed to receive the version values.");
1340 goto end;
1341 }
1342
1343 /*
1344 * For now, we just ignore the received version but after 2.1 stable
1345 * release, a check must be done to see if we either adapt to the other
1346 * side version (which MUST be lower than us) or keep the latest data
1347 * structure considering that the other side will adapt.
1348 */
1349
7f51dcba
MD
1350 ret = sscanf(VERSION, "%u.%u", &reply.major, &reply.minor);
1351 if (ret < 2) {
1352 ERR("Error in scanning version");
1353 ret = -1;
1354 goto end;
1355 }
b8aa1682
JD
1356 reply.major = htobe32(reply.major);
1357 reply.minor = htobe32(reply.minor);
1358 ret = cmd->sock->ops->sendmsg(cmd->sock, &reply,
1359 sizeof(struct lttcomm_relayd_version), 0);
1360 if (ret < 0) {
1361 ERR("Relay sending version");
1362 }
0a6b5085
DG
1363 DBG("Version check done (%u.%u)", be32toh(reply.major),
1364 be32toh(reply.minor));
b8aa1682
JD
1365
1366end:
1367 return ret;
1368}
1369
c8f59ee5 1370/*
6d805429 1371 * Check for data pending for a given stream id from the session daemon.
c8f59ee5
DG
1372 */
1373static
6d805429 1374int relay_data_pending(struct lttcomm_relayd_hdr *recv_hdr,
c8f59ee5
DG
1375 struct relay_command *cmd, struct lttng_ht *streams_ht)
1376{
1377 struct relay_session *session = cmd->session;
6d805429 1378 struct lttcomm_relayd_data_pending msg;
c8f59ee5
DG
1379 struct lttcomm_relayd_generic_reply reply;
1380 struct relay_stream *stream;
1381 int ret;
1382 struct lttng_ht_node_ulong *node;
1383 struct lttng_ht_iter iter;
1384 uint64_t last_net_seq_num, stream_id;
1385
6d805429 1386 DBG("Data pending command received");
c8f59ee5 1387
c5b6f4f0 1388 if (!session || cmd->version_check_done == 0) {
c8f59ee5
DG
1389 ERR("Trying to check for data before version check");
1390 ret = -1;
1391 goto end_no_session;
1392 }
1393
7c5aef62 1394 ret = cmd->sock->ops->recvmsg(cmd->sock, &msg, sizeof(msg), 0);
c8f59ee5 1395 if (ret < sizeof(msg)) {
6d805429 1396 ERR("Relay didn't receive valid data_pending struct size : %d", ret);
c8f59ee5
DG
1397 ret = -1;
1398 goto end_no_session;
1399 }
1400
1401 stream_id = be64toh(msg.stream_id);
1402 last_net_seq_num = be64toh(msg.last_net_seq_num);
1403
1404 rcu_read_lock();
1405 lttng_ht_lookup(streams_ht, (void *)((unsigned long) stream_id), &iter);
1406 node = lttng_ht_iter_get_node_ulong(&iter);
1407 if (node == NULL) {
1408 DBG("Relay stream %" PRIu64 " not found", stream_id);
1409 ret = -1;
1410 goto end_unlock;
1411 }
1412
1413 stream = caa_container_of(node, struct relay_stream, stream_n);
1414 assert(stream);
1415
6d805429 1416 DBG("Data pending for stream id %" PRIu64 " prev_seq %" PRIu64
c8f59ee5
DG
1417 " and last_seq %" PRIu64, stream_id, stream->prev_seq,
1418 last_net_seq_num);
1419
33832e64
DG
1420 /* Avoid wrapping issue */
1421 if (((int64_t) (stream->prev_seq - last_net_seq_num)) <= 0) {
6d805429 1422 /* Data has in fact been written and is NOT pending */
c8f59ee5 1423 ret = 0;
6d805429
DG
1424 } else {
1425 /* Data still being streamed thus pending */
1426 ret = 1;
c8f59ee5
DG
1427 }
1428
1429end_unlock:
1430 rcu_read_unlock();
1431
1432 reply.ret_code = htobe32(ret);
1433 ret = cmd->sock->ops->sendmsg(cmd->sock, &reply, sizeof(reply), 0);
1434 if (ret < 0) {
6d805429 1435 ERR("Relay data pending ret code failed");
c8f59ee5
DG
1436 }
1437
1438end_no_session:
1439 return ret;
1440}
1441
1442/*
1443 * Wait for the control socket to reach a quiescent state.
1444 *
1445 * Note that for now, when receiving this command from the session daemon, this
1446 * means that every subsequent commands or data received on the control socket
1447 * has been handled. So, this is why we simply return OK here.
1448 */
1449static
1450int relay_quiescent_control(struct lttcomm_relayd_hdr *recv_hdr,
1451 struct relay_command *cmd)
1452{
1453 int ret;
1454 struct lttcomm_relayd_generic_reply reply;
1455
1456 DBG("Checking quiescent state on control socket");
1457
1458 reply.ret_code = htobe32(LTTNG_OK);
1459 ret = cmd->sock->ops->sendmsg(cmd->sock, &reply, sizeof(reply), 0);
1460 if (ret < 0) {
6d805429 1461 ERR("Relay data quiescent control ret code failed");
c8f59ee5
DG
1462 }
1463
1464 return ret;
1465}
1466
b8aa1682
JD
1467/*
1468 * relay_process_control: Process the commands received on the control socket
1469 */
1470static
1471int relay_process_control(struct lttcomm_relayd_hdr *recv_hdr,
1472 struct relay_command *cmd, struct lttng_ht *streams_ht)
1473{
1474 int ret = 0;
1475
1476 switch (be32toh(recv_hdr->cmd)) {
b8aa1682
JD
1477 case RELAYD_CREATE_SESSION:
1478 ret = relay_create_session(recv_hdr, cmd);
1479 break;
b8aa1682
JD
1480 case RELAYD_ADD_STREAM:
1481 ret = relay_add_stream(recv_hdr, cmd, streams_ht);
1482 break;
1483 case RELAYD_START_DATA:
1484 ret = relay_start(recv_hdr, cmd);
1485 break;
1486 case RELAYD_SEND_METADATA:
1487 ret = relay_recv_metadata(recv_hdr, cmd, streams_ht);
1488 break;
1489 case RELAYD_VERSION:
1490 ret = relay_send_version(recv_hdr, cmd);
1491 break;
173af62f
DG
1492 case RELAYD_CLOSE_STREAM:
1493 ret = relay_close_stream(recv_hdr, cmd, streams_ht);
1494 break;
6d805429
DG
1495 case RELAYD_DATA_PENDING:
1496 ret = relay_data_pending(recv_hdr, cmd, streams_ht);
c8f59ee5
DG
1497 break;
1498 case RELAYD_QUIESCENT_CONTROL:
1499 ret = relay_quiescent_control(recv_hdr, cmd);
1500 break;
b8aa1682
JD
1501 case RELAYD_UPDATE_SYNC_INFO:
1502 default:
1503 ERR("Received unknown command (%u)", be32toh(recv_hdr->cmd));
1504 relay_unknown_command(cmd);
1505 ret = -1;
1506 goto end;
1507 }
1508
1509end:
1510 return ret;
1511}
1512
1513/*
1514 * relay_process_data: Process the data received on the data socket
1515 */
1516static
1517int relay_process_data(struct relay_command *cmd, struct lttng_ht *streams_ht)
1518{
1519 int ret = 0;
1520 struct relay_stream *stream;
1521 struct lttcomm_relayd_data_hdr data_hdr;
1522 uint64_t stream_id;
173af62f 1523 uint64_t net_seq_num;
b8aa1682
JD
1524 uint32_t data_size;
1525
1526 ret = cmd->sock->ops->recvmsg(cmd->sock, &data_hdr,
7c5aef62 1527 sizeof(struct lttcomm_relayd_data_hdr), 0);
b8aa1682
JD
1528 if (ret <= 0) {
1529 ERR("Connections seems to be closed");
1530 ret = -1;
1531 goto end;
1532 }
1533
1534 stream_id = be64toh(data_hdr.stream_id);
9d1bbf21
MD
1535
1536 rcu_read_lock();
b8aa1682
JD
1537 stream = relay_stream_from_stream_id(stream_id, streams_ht);
1538 if (!stream) {
1539 ret = -1;
9d1bbf21 1540 goto end_unlock;
b8aa1682
JD
1541 }
1542
1543 data_size = be32toh(data_hdr.data_size);
1544 if (data_buffer_size < data_size) {
d7b3776f 1545 char *tmp_data_ptr = data_buffer;
b8aa1682
JD
1546 data_buffer = realloc(data_buffer, data_size);
1547 if (!data_buffer) {
1548 ERR("Allocating data buffer");
d7b3776f 1549 free(tmp_data_ptr);
b8aa1682 1550 ret = -1;
9d1bbf21 1551 goto end_unlock;
b8aa1682
JD
1552 }
1553 data_buffer_size = data_size;
1554 }
1555 memset(data_buffer, 0, data_size);
1556
173af62f
DG
1557 net_seq_num = be64toh(data_hdr.net_seq_num);
1558
77c7c900 1559 DBG3("Receiving data of size %u for stream id %" PRIu64 " seqnum %" PRIu64,
173af62f 1560 data_size, stream_id, net_seq_num);
7c5aef62 1561 ret = cmd->sock->ops->recvmsg(cmd->sock, data_buffer, data_size, 0);
b8aa1682
JD
1562 if (ret <= 0) {
1563 ret = -1;
9d1bbf21 1564 goto end_unlock;
b8aa1682
JD
1565 }
1566
6f94560a
MD
1567 do {
1568 ret = write(stream->fd, data_buffer, data_size);
1569 } while (ret < 0 && errno == EINTR);
b8aa1682
JD
1570 if (ret < data_size) {
1571 ERR("Relay error writing data to file");
1572 ret = -1;
9d1bbf21 1573 goto end_unlock;
b8aa1682 1574 }
1d4dfdef 1575
5ab7344e
JD
1576 DBG2("Relay wrote %d bytes to tracefile for stream id %" PRIu64,
1577 ret, stream->stream_handle);
1578
1d4dfdef
DG
1579 ret = write_padding_to_file(stream->fd, be32toh(data_hdr.padding_size));
1580 if (ret < 0) {
1581 goto end_unlock;
1582 }
1583
173af62f
DG
1584 stream->prev_seq = net_seq_num;
1585
1586 /* Check if we need to close the FD */
1587 if (close_stream_check(stream)) {
f66c074c 1588 int cret;
173af62f
DG
1589 struct lttng_ht_iter iter;
1590
f66c074c
DG
1591 cret = close(stream->fd);
1592 if (cret < 0) {
1593 PERROR("close stream process data");
1594 }
173af62f
DG
1595 iter.iter.node = &stream->stream_n.node;
1596 ret = lttng_ht_del(streams_ht, &iter);
1597 assert(!ret);
1598 call_rcu(&stream->rcu_node,
1599 deferred_free_stream);
1600 DBG("Closed tracefile %d after recv data", stream->fd);
1601 }
1602
9d1bbf21
MD
1603end_unlock:
1604 rcu_read_unlock();
b8aa1682
JD
1605end:
1606 return ret;
1607}
1608
1609static
9d1bbf21 1610void relay_cleanup_poll_connection(struct lttng_poll_event *events, int pollfd)
b8aa1682
JD
1611{
1612 int ret;
1613
b8aa1682
JD
1614 lttng_poll_del(events, pollfd);
1615
1616 ret = close(pollfd);
1617 if (ret < 0) {
1618 ERR("Closing pollfd %d", pollfd);
1619 }
1620}
1621
1622static
1623int relay_add_connection(int fd, struct lttng_poll_event *events,
1624 struct lttng_ht *relay_connections_ht)
1625{
b8aa1682 1626 struct relay_command *relay_connection;
9d1bbf21 1627 int ret;
b8aa1682
JD
1628
1629 relay_connection = zmalloc(sizeof(struct relay_command));
1630 if (relay_connection == NULL) {
1631 PERROR("Relay command zmalloc");
9d1bbf21 1632 goto error;
b8aa1682
JD
1633 }
1634 ret = read(fd, relay_connection, sizeof(struct relay_command));
cdf0f8fc 1635 if (ret < 0 || ret < sizeof(struct relay_command)) {
b8aa1682 1636 PERROR("read relay cmd pipe");
9d1bbf21 1637 goto error_read;
b8aa1682
JD
1638 }
1639
1640 lttng_ht_node_init_ulong(&relay_connection->sock_n,
1641 (unsigned long) relay_connection->sock->fd);
9d1bbf21 1642 rcu_read_lock();
b8aa1682
JD
1643 lttng_ht_add_unique_ulong(relay_connections_ht,
1644 &relay_connection->sock_n);
9d1bbf21
MD
1645 rcu_read_unlock();
1646 return lttng_poll_add(events,
b8aa1682
JD
1647 relay_connection->sock->fd,
1648 LPOLLIN | LPOLLRDHUP);
1649
9d1bbf21
MD
1650error_read:
1651 free(relay_connection);
1652error:
1653 return -1;
1654}
1655
1656static
1657void deferred_free_connection(struct rcu_head *head)
1658{
1659 struct relay_command *relay_connection =
1660 caa_container_of(head, struct relay_command, rcu_node);
5b6d8097
DG
1661
1662 lttcomm_destroy_sock(relay_connection->sock);
9d1bbf21
MD
1663 free(relay_connection);
1664}
1665
1666static
1667void relay_del_connection(struct lttng_ht *relay_connections_ht,
1668 struct lttng_ht *streams_ht, struct lttng_ht_iter *iter,
1669 struct relay_command *relay_connection)
1670{
1671 int ret;
1672
1673 ret = lttng_ht_del(relay_connections_ht, iter);
1674 assert(!ret);
1675 if (relay_connection->type == RELAY_CONTROL) {
1676 relay_delete_session(relay_connection, streams_ht);
1677 }
5b6d8097 1678
9d1bbf21
MD
1679 call_rcu(&relay_connection->rcu_node,
1680 deferred_free_connection);
b8aa1682
JD
1681}
1682
1683/*
1684 * This thread does the actual work
1685 */
1686static
1687void *relay_thread_worker(void *data)
1688{
095a4ae5 1689 int i, ret, pollfd, err = -1;
b8aa1682
JD
1690 uint32_t revents, nb_fd;
1691 struct relay_command *relay_connection;
1692 struct lttng_poll_event events;
1693 struct lttng_ht *relay_connections_ht;
1694 struct lttng_ht_node_ulong *node;
1695 struct lttng_ht_iter iter;
1696 struct lttng_ht *streams_ht;
1697 struct lttcomm_relayd_hdr recv_hdr;
1698
1699 DBG("[thread] Relay worker started");
1700
9d1bbf21
MD
1701 rcu_register_thread();
1702
b8aa1682
JD
1703 /* table of connections indexed on socket */
1704 relay_connections_ht = lttng_ht_new(0, LTTNG_HT_TYPE_ULONG);
095a4ae5
MD
1705 if (!relay_connections_ht) {
1706 goto relay_connections_ht_error;
1707 }
b8aa1682
JD
1708
1709 /* tables of streams indexed by stream ID */
1710 streams_ht = lttng_ht_new(0, LTTNG_HT_TYPE_ULONG);
095a4ae5
MD
1711 if (!streams_ht) {
1712 goto streams_ht_error;
1713 }
b8aa1682
JD
1714
1715 ret = create_thread_poll_set(&events, 2);
1716 if (ret < 0) {
1717 goto error_poll_create;
1718 }
1719
1720 ret = lttng_poll_add(&events, relay_cmd_pipe[0], LPOLLIN | LPOLLRDHUP);
1721 if (ret < 0) {
1722 goto error;
1723 }
1724
1725 while (1) {
1726 /* Zeroed the events structure */
1727 lttng_poll_reset(&events);
1728
1729 nb_fd = LTTNG_POLL_GETNB(&events);
1730
1731 /* Infinite blocking call, waiting for transmission */
1732 restart:
87c1611d 1733 DBG3("Relayd worker thread polling...");
b8aa1682
JD
1734 ret = lttng_poll_wait(&events, -1);
1735 if (ret < 0) {
1736 /*
1737 * Restart interrupted system call.
1738 */
1739 if (errno == EINTR) {
1740 goto restart;
1741 }
1742 goto error;
1743 }
1744
1745 for (i = 0; i < nb_fd; i++) {
1746 /* Fetch once the poll data */
1747 revents = LTTNG_POLL_GETEV(&events, i);
1748 pollfd = LTTNG_POLL_GETFD(&events, i);
1749
1750 /* Thread quit pipe has been closed. Killing thread. */
1751 ret = check_thread_quit_pipe(pollfd, revents);
1752 if (ret) {
095a4ae5
MD
1753 err = 0;
1754 goto exit;
b8aa1682
JD
1755 }
1756
1757 /* Inspect the relay cmd pipe for new connection */
1758 if (pollfd == relay_cmd_pipe[0]) {
1759 if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
1760 ERR("Relay pipe error");
1761 goto error;
1762 } else if (revents & LPOLLIN) {
1763 DBG("Relay command received");
1764 ret = relay_add_connection(relay_cmd_pipe[0],
1765 &events, relay_connections_ht);
1766 if (ret < 0) {
1767 goto error;
1768 }
1769 }
1770 } else if (revents > 0) {
9d1bbf21 1771 rcu_read_lock();
b8aa1682
JD
1772 lttng_ht_lookup(relay_connections_ht,
1773 (void *)((unsigned long) pollfd),
1774 &iter);
1775 node = lttng_ht_iter_get_node_ulong(&iter);
1776 if (node == NULL) {
1777 DBG2("Relay sock %d not found", pollfd);
9d1bbf21 1778 rcu_read_unlock();
b8aa1682
JD
1779 goto error;
1780 }
1781 relay_connection = caa_container_of(node,
1782 struct relay_command, sock_n);
1783
1784 if (revents & (LPOLLERR)) {
1785 ERR("POLL ERROR");
9d1bbf21
MD
1786 relay_cleanup_poll_connection(&events, pollfd);
1787 relay_del_connection(relay_connections_ht,
1788 streams_ht, &iter,
1789 relay_connection);
b8aa1682
JD
1790 } else if (revents & (LPOLLHUP | LPOLLRDHUP)) {
1791 DBG("Socket %d hung up", pollfd);
9d1bbf21
MD
1792 relay_cleanup_poll_connection(&events, pollfd);
1793 relay_del_connection(relay_connections_ht,
1794 streams_ht, &iter,
1795 relay_connection);
b8aa1682
JD
1796 } else if (revents & LPOLLIN) {
1797 /* control socket */
1798 if (relay_connection->type == RELAY_CONTROL) {
1799 ret = relay_connection->sock->ops->recvmsg(
1800 relay_connection->sock, &recv_hdr,
7c5aef62 1801 sizeof(struct lttcomm_relayd_hdr), 0);
b8aa1682
JD
1802 /* connection closed */
1803 if (ret <= 0) {
9d1bbf21
MD
1804 relay_cleanup_poll_connection(&events, pollfd);
1805 relay_del_connection(relay_connections_ht,
1806 streams_ht, &iter,
1807 relay_connection);
b8aa1682
JD
1808 DBG("Control connection closed with %d", pollfd);
1809 } else {
1810 if (relay_connection->session) {
77c7c900 1811 DBG2("Relay worker receiving data for session : %" PRIu64,
b8aa1682
JD
1812 relay_connection->session->id);
1813 }
1814 ret = relay_process_control(&recv_hdr,
1815 relay_connection,
1816 streams_ht);
1817 /*
1818 * there was an error in processing a control
1819 * command: clear the session
1820 * */
1821 if (ret < 0) {
9d1bbf21
MD
1822 relay_cleanup_poll_connection(&events, pollfd);
1823 relay_del_connection(relay_connections_ht,
1824 streams_ht, &iter,
1825 relay_connection);
b8aa1682
JD
1826 DBG("Connection closed with %d", pollfd);
1827 }
1828 }
1829 /* data socket */
1830 } else if (relay_connection->type == RELAY_DATA) {
1831 ret = relay_process_data(relay_connection, streams_ht);
1832 /* connection closed */
1833 if (ret < 0) {
9d1bbf21
MD
1834 relay_cleanup_poll_connection(&events, pollfd);
1835 relay_del_connection(relay_connections_ht,
1836 streams_ht, &iter,
1837 relay_connection);
b8aa1682
JD
1838 DBG("Data connection closed with %d", pollfd);
1839 }
1840 }
1841 }
9d1bbf21 1842 rcu_read_unlock();
b8aa1682
JD
1843 }
1844 }
1845 }
1846
095a4ae5 1847exit:
b8aa1682
JD
1848error:
1849 lttng_poll_clean(&events);
1850
1851 /* empty the hash table and free the memory */
9d1bbf21 1852 rcu_read_lock();
b8aa1682
JD
1853 cds_lfht_for_each_entry(relay_connections_ht->ht, &iter.iter, node, node) {
1854 node = lttng_ht_iter_get_node_ulong(&iter);
1855 if (node) {
1856 relay_connection = caa_container_of(node,
1857 struct relay_command, sock_n);
9d1bbf21
MD
1858 relay_del_connection(relay_connections_ht,
1859 streams_ht, &iter,
1860 relay_connection);
b8aa1682 1861 }
b8aa1682 1862 }
9d1bbf21 1863 rcu_read_unlock();
b8aa1682 1864error_poll_create:
095a4ae5
MD
1865 lttng_ht_destroy(streams_ht);
1866streams_ht_error:
b8aa1682 1867 lttng_ht_destroy(relay_connections_ht);
095a4ae5 1868relay_connections_ht_error:
6620da75
DG
1869 /* Close relay cmd pipes */
1870 utils_close_pipe(relay_cmd_pipe);
095a4ae5
MD
1871 if (err) {
1872 DBG("Thread exited with error");
1873 }
b8aa1682 1874 DBG("Worker thread cleanup complete");
095a4ae5 1875 free(data_buffer);
b8aa1682 1876 stop_threads();
9d1bbf21 1877 rcu_unregister_thread();
b8aa1682
JD
1878 return NULL;
1879}
1880
1881/*
1882 * Create the relay command pipe to wake thread_manage_apps.
1883 * Closed in cleanup().
1884 */
1885static int create_relay_cmd_pipe(void)
1886{
a02de639 1887 int ret;
b8aa1682 1888
a02de639 1889 ret = utils_create_pipe_cloexec(relay_cmd_pipe);
b8aa1682 1890
b8aa1682
JD
1891 return ret;
1892}
1893
1894/*
1895 * main
1896 */
1897int main(int argc, char **argv)
1898{
1899 int ret = 0;
1900 void *status;
1901
1902 /* Create thread quit pipe */
1903 if ((ret = init_thread_quit_pipe()) < 0) {
1904 goto error;
1905 }
1906
1907 /* Parse arguments */
1908 progname = argv[0];
1909 if ((ret = parse_args(argc, argv) < 0)) {
a02de639 1910 goto exit;
b8aa1682
JD
1911 }
1912
1913 if ((ret = set_signal_handler()) < 0) {
1914 goto exit;
1915 }
1916
1917 /* Daemonize */
1918 if (opt_daemon) {
1919 ret = daemon(0, 0);
1920 if (ret < 0) {
1921 PERROR("daemon");
a02de639 1922 goto exit;
b8aa1682
JD
1923 }
1924 }
1925
1926 /* Check if daemon is UID = 0 */
1927 is_root = !getuid();
1928
1929 if (!is_root) {
1930 if (control_uri->port < 1024 || data_uri->port < 1024) {
1931 ERR("Need to be root to use ports < 1024");
1932 ret = -1;
a02de639 1933 goto exit;
b8aa1682
JD
1934 }
1935 }
1936
1937 /* Setup the thread apps communication pipe. */
1938 if ((ret = create_relay_cmd_pipe()) < 0) {
1939 goto exit;
1940 }
1941
1942 /* Init relay command queue. */
1943 cds_wfq_init(&relay_cmd_queue.queue);
1944
1945 /* Set up max poll set size */
1946 lttng_poll_set_max_size();
1947
1948 /* Setup the dispatcher thread */
1949 ret = pthread_create(&dispatcher_thread, NULL,
1950 relay_thread_dispatcher, (void *) NULL);
1951 if (ret != 0) {
1952 PERROR("pthread_create dispatcher");
1953 goto exit_dispatcher;
1954 }
1955
1956 /* Setup the worker thread */
1957 ret = pthread_create(&worker_thread, NULL,
1958 relay_thread_worker, (void *) NULL);
1959 if (ret != 0) {
1960 PERROR("pthread_create worker");
1961 goto exit_worker;
1962 }
1963
1964 /* Setup the listener thread */
1965 ret = pthread_create(&listener_thread, NULL,
1966 relay_thread_listener, (void *) NULL);
1967 if (ret != 0) {
1968 PERROR("pthread_create listener");
1969 goto exit_listener;
1970 }
1971
1972exit_listener:
1973 ret = pthread_join(listener_thread, &status);
1974 if (ret != 0) {
1975 PERROR("pthread_join");
1976 goto error; /* join error, exit without cleanup */
1977 }
1978
1979exit_worker:
1980 ret = pthread_join(worker_thread, &status);
1981 if (ret != 0) {
1982 PERROR("pthread_join");
1983 goto error; /* join error, exit without cleanup */
1984 }
1985
1986exit_dispatcher:
1987 ret = pthread_join(dispatcher_thread, &status);
1988 if (ret != 0) {
1989 PERROR("pthread_join");
1990 goto error; /* join error, exit without cleanup */
1991 }
1992
1993exit:
1994 cleanup();
1995 if (!ret) {
1996 exit(EXIT_SUCCESS);
1997 }
a02de639 1998
b8aa1682
JD
1999error:
2000 exit(EXIT_FAILURE);
2001}
This page took 0.112529 seconds and 5 git commands to generate.