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