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