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