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