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