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