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