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