06cfa5c7bb64da3321af4c251bf3f2a50580091e
[lttng-tools.git] / src / bin / lttng-relayd / live.c
1 /*
2 * Copyright (C) 2013 Julien Desfossez <jdesfossez@efficios.com>
3 * Copyright (C) 2013 David Goulet <dgoulet@efficios.com>
4 * Copyright (C) 2015 Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
5 *
6 * SPDX-License-Identifier: GPL-2.0-only
7 *
8 */
9
10 #define _LGPL_SOURCE
11 #include <fcntl.h>
12 #include <getopt.h>
13 #include <grp.h>
14 #include <inttypes.h>
15 #include <limits.h>
16 #include <pthread.h>
17 #include <signal.h>
18 #include <stdio.h>
19 #include <stdlib.h>
20 #include <string.h>
21 #include <sys/mman.h>
22 #include <sys/mount.h>
23 #include <sys/resource.h>
24 #include <sys/socket.h>
25 #include <sys/stat.h>
26 #include <sys/types.h>
27 #include <sys/wait.h>
28 #include <unistd.h>
29 #include <urcu/futex.h>
30 #include <urcu/rculist.h>
31 #include <urcu/uatomic.h>
32
33 #include <common/common.h>
34 #include <common/compat/endian.h>
35 #include <common/compat/poll.h>
36 #include <common/compat/socket.h>
37 #include <common/defaults.h>
38 #include <common/fd-tracker/utils.h>
39 #include <common/fs-handle.h>
40 #include <common/futex.h>
41 #include <common/index/index.h>
42 #include <common/sessiond-comm/inet.h>
43 #include <common/sessiond-comm/relayd.h>
44 #include <common/sessiond-comm/sessiond-comm.h>
45 #include <common/uri.h>
46 #include <common/utils.h>
47 #include <lttng/lttng.h>
48
49 #include "cmd.h"
50 #include "connection.h"
51 #include "ctf-trace.h"
52 #include "health-relayd.h"
53 #include "live.h"
54 #include "lttng-relayd.h"
55 #include "session.h"
56 #include "stream.h"
57 #include "testpoint.h"
58 #include "utils.h"
59 #include "viewer-session.h"
60 #include "viewer-stream.h"
61
62 #define SESSION_BUF_DEFAULT_COUNT 16
63
64 static struct lttng_uri *live_uri;
65
66 /*
67 * This pipe is used to inform the worker thread that a command is queued and
68 * ready to be processed.
69 */
70 static int live_conn_pipe[2] = { -1, -1 };
71
72 /* Shared between threads */
73 static int live_dispatch_thread_exit;
74
75 static pthread_t live_listener_thread;
76 static pthread_t live_dispatcher_thread;
77 static pthread_t live_worker_thread;
78
79 /*
80 * Relay command queue.
81 *
82 * The live_thread_listener and live_thread_dispatcher communicate with this
83 * queue.
84 */
85 static struct relay_conn_queue viewer_conn_queue;
86
87 static uint64_t last_relay_viewer_session_id;
88 static pthread_mutex_t last_relay_viewer_session_id_lock =
89 PTHREAD_MUTEX_INITIALIZER;
90
91 /*
92 * Cleanup the daemon
93 */
94 static
95 void cleanup_relayd_live(void)
96 {
97 DBG("Cleaning up");
98
99 free(live_uri);
100 }
101
102 /*
103 * Receive a request buffer using a given socket, destination allocated buffer
104 * of length size.
105 *
106 * Return the size of the received message or else a negative value on error
107 * with errno being set by recvmsg() syscall.
108 */
109 static
110 ssize_t recv_request(struct lttcomm_sock *sock, void *buf, size_t size)
111 {
112 ssize_t ret;
113
114 ret = sock->ops->recvmsg(sock, buf, size, 0);
115 if (ret < 0 || ret != size) {
116 if (ret == 0) {
117 /* Orderly shutdown. Not necessary to print an error. */
118 DBG("Socket %d did an orderly shutdown", sock->fd);
119 } else {
120 ERR("Relay failed to receive request.");
121 }
122 ret = -1;
123 }
124
125 return ret;
126 }
127
128 /*
129 * Send a response buffer using a given socket, source allocated buffer of
130 * length size.
131 *
132 * Return the size of the sent message or else a negative value on error with
133 * errno being set by sendmsg() syscall.
134 */
135 static
136 ssize_t send_response(struct lttcomm_sock *sock, void *buf, size_t size)
137 {
138 ssize_t ret;
139
140 ret = sock->ops->sendmsg(sock, buf, size, 0);
141 if (ret < 0) {
142 ERR("Relayd failed to send response.");
143 }
144
145 return ret;
146 }
147
148 /*
149 * Atomically check if new streams got added in one of the sessions attached
150 * and reset the flag to 0.
151 *
152 * Returns 1 if new streams got added, 0 if nothing changed, a negative value
153 * on error.
154 */
155 static
156 int check_new_streams(struct relay_connection *conn)
157 {
158 struct relay_session *session;
159 unsigned long current_val;
160 int ret = 0;
161
162 if (!conn->viewer_session) {
163 goto end;
164 }
165 rcu_read_lock();
166 cds_list_for_each_entry_rcu(session,
167 &conn->viewer_session->session_list,
168 viewer_session_node) {
169 if (!session_get(session)) {
170 continue;
171 }
172 current_val = uatomic_cmpxchg(&session->new_streams, 1, 0);
173 ret = current_val;
174 session_put(session);
175 if (ret == 1) {
176 goto end;
177 }
178 }
179 end:
180 rcu_read_unlock();
181 return ret;
182 }
183
184 /*
185 * Send viewer streams to the given socket. The ignore_sent_flag indicates if
186 * this function should ignore the sent flag or not.
187 *
188 * Return 0 on success or else a negative value.
189 */
190 static
191 ssize_t send_viewer_streams(struct lttcomm_sock *sock,
192 uint64_t session_id, unsigned int ignore_sent_flag)
193 {
194 ssize_t ret;
195 struct lttng_viewer_stream send_stream;
196 struct lttng_ht_iter iter;
197 struct relay_viewer_stream *vstream;
198
199 rcu_read_lock();
200
201 cds_lfht_for_each_entry(viewer_streams_ht->ht, &iter.iter, vstream,
202 stream_n.node) {
203 struct ctf_trace *ctf_trace;
204
205 health_code_update();
206
207 if (!viewer_stream_get(vstream)) {
208 continue;
209 }
210
211 pthread_mutex_lock(&vstream->stream->lock);
212 /* Ignore if not the same session. */
213 if (vstream->stream->trace->session->id != session_id ||
214 (!ignore_sent_flag && vstream->sent_flag)) {
215 pthread_mutex_unlock(&vstream->stream->lock);
216 viewer_stream_put(vstream);
217 continue;
218 }
219
220 ctf_trace = vstream->stream->trace;
221 send_stream.id = htobe64(vstream->stream->stream_handle);
222 send_stream.ctf_trace_id = htobe64(ctf_trace->id);
223 send_stream.metadata_flag = htobe32(
224 vstream->stream->is_metadata);
225 if (lttng_strncpy(send_stream.path_name, vstream->path_name,
226 sizeof(send_stream.path_name))) {
227 pthread_mutex_unlock(&vstream->stream->lock);
228 viewer_stream_put(vstream);
229 ret = -1; /* Error. */
230 goto end_unlock;
231 }
232 if (lttng_strncpy(send_stream.channel_name,
233 vstream->channel_name,
234 sizeof(send_stream.channel_name))) {
235 pthread_mutex_unlock(&vstream->stream->lock);
236 viewer_stream_put(vstream);
237 ret = -1; /* Error. */
238 goto end_unlock;
239 }
240
241 DBG("Sending stream %" PRIu64 " to viewer",
242 vstream->stream->stream_handle);
243 vstream->sent_flag = 1;
244 pthread_mutex_unlock(&vstream->stream->lock);
245
246 ret = send_response(sock, &send_stream, sizeof(send_stream));
247 viewer_stream_put(vstream);
248 if (ret < 0) {
249 goto end_unlock;
250 }
251 }
252
253 ret = 0;
254
255 end_unlock:
256 rcu_read_unlock();
257 return ret;
258 }
259
260 /*
261 * Create every viewer stream possible for the given session with the seek
262 * type. Three counters *can* be return which are in order the total amount of
263 * viewer stream of the session, the number of unsent stream and the number of
264 * stream created. Those counters can be NULL and thus will be ignored.
265 *
266 * session must be locked to ensure that we see either none or all initial
267 * streams for a session, but no intermediate state..
268 *
269 * Return 0 on success or else a negative value.
270 */
271 static int make_viewer_streams(struct relay_session *session,
272 struct lttng_trace_chunk *viewer_trace_chunk,
273 enum lttng_viewer_seek seek_t,
274 uint32_t *nb_total,
275 uint32_t *nb_unsent,
276 uint32_t *nb_created,
277 bool *closed)
278 {
279 int ret;
280 struct lttng_ht_iter iter;
281 struct ctf_trace *ctf_trace;
282
283 assert(session);
284 ASSERT_LOCKED(session->lock);
285
286 if (!viewer_trace_chunk) {
287 ERR("Internal error: viewer session associated with session \"%s\" has a NULL trace chunk",
288 session->session_name);
289 ret = -1;
290 goto error;
291 }
292
293 if (session->connection_closed) {
294 *closed = true;
295 }
296
297 /*
298 * Create viewer streams for relay streams that are ready to be
299 * used for a the given session id only.
300 */
301 rcu_read_lock();
302 cds_lfht_for_each_entry(session->ctf_traces_ht->ht, &iter.iter, ctf_trace,
303 node.node) {
304 bool trace_has_metadata_stream = false;
305 struct relay_stream *stream;
306
307 health_code_update();
308
309 if (!ctf_trace_get(ctf_trace)) {
310 continue;
311 }
312
313 /*
314 * Iterate over all the streams of the trace to see if we have a
315 * metadata stream.
316 */
317 cds_list_for_each_entry_rcu(
318 stream, &ctf_trace->stream_list, stream_node)
319 {
320 if (stream->is_metadata) {
321 trace_has_metadata_stream = true;
322 break;
323 }
324 }
325
326 /*
327 * If there is no metadata stream in this trace at the moment
328 * and we never sent one to the viewer, skip the trace. We
329 * accept that the viewer will not see this trace at all.
330 */
331 if (!trace_has_metadata_stream &&
332 !ctf_trace->metadata_stream_sent_to_viewer) {
333 ctf_trace_put(ctf_trace);
334 continue;
335 }
336
337 cds_list_for_each_entry_rcu(stream, &ctf_trace->stream_list, stream_node) {
338 struct relay_viewer_stream *vstream;
339
340 if (!stream_get(stream)) {
341 continue;
342 }
343 /*
344 * stream published is protected by the session lock.
345 */
346 if (!stream->published) {
347 goto next;
348 }
349 vstream = viewer_stream_get_by_id(stream->stream_handle);
350 if (!vstream) {
351 /*
352 * Save that we sent the metadata stream to the
353 * viewer. So that we know what trace the viewer
354 * is aware of.
355 */
356 if (stream->is_metadata) {
357 ctf_trace->metadata_stream_sent_to_viewer =
358 true;
359 }
360 vstream = viewer_stream_create(stream,
361 viewer_trace_chunk, seek_t);
362 if (!vstream) {
363 ret = -1;
364 ctf_trace_put(ctf_trace);
365 stream_put(stream);
366 goto error_unlock;
367 }
368
369 if (nb_created) {
370 /* Update number of created stream counter. */
371 (*nb_created)++;
372 }
373 /*
374 * Ensure a self-reference is preserved even
375 * after we have put our local reference.
376 */
377 if (!viewer_stream_get(vstream)) {
378 ERR("Unable to get self-reference on viewer stream, logic error.");
379 abort();
380 }
381 } else {
382 if (!vstream->sent_flag && nb_unsent) {
383 /* Update number of unsent stream counter. */
384 (*nb_unsent)++;
385 }
386 }
387 /* Update number of total stream counter. */
388 if (nb_total) {
389 if (stream->is_metadata) {
390 if (!stream->closed ||
391 stream->metadata_received > vstream->metadata_sent) {
392 (*nb_total)++;
393 }
394 } else {
395 if (!stream->closed ||
396 !(((int64_t) (stream->prev_data_seq - stream->last_net_seq_num)) >= 0)) {
397
398 (*nb_total)++;
399 }
400 }
401 }
402 /* Put local reference. */
403 viewer_stream_put(vstream);
404 next:
405 stream_put(stream);
406 }
407 ctf_trace_put(ctf_trace);
408 }
409
410 ret = 0;
411
412 error_unlock:
413 rcu_read_unlock();
414 error:
415 return ret;
416 }
417
418 int relayd_live_stop(void)
419 {
420 /* Stop dispatch thread */
421 CMM_STORE_SHARED(live_dispatch_thread_exit, 1);
422 futex_nto1_wake(&viewer_conn_queue.futex);
423 return 0;
424 }
425
426 /*
427 * Create a poll set with O_CLOEXEC and add the thread quit pipe to the set.
428 */
429 static
430 int create_named_thread_poll_set(struct lttng_poll_event *events,
431 int size, const char *name)
432 {
433 int ret;
434
435 if (events == NULL || size == 0) {
436 ret = -1;
437 goto error;
438 }
439
440 ret = fd_tracker_util_poll_create(the_fd_tracker,
441 name, events, 1, LTTNG_CLOEXEC);
442 if (ret) {
443 PERROR("Failed to create \"%s\" poll file descriptor", name);
444 goto error;
445 }
446
447 /* Add quit pipe */
448 ret = lttng_poll_add(events, thread_quit_pipe[0], LPOLLIN | LPOLLERR);
449 if (ret < 0) {
450 goto error;
451 }
452
453 return 0;
454
455 error:
456 return ret;
457 }
458
459 /*
460 * Check if the thread quit pipe was triggered.
461 *
462 * Return 1 if it was triggered else 0;
463 */
464 static
465 int check_thread_quit_pipe(int fd, uint32_t events)
466 {
467 if (fd == thread_quit_pipe[0] && (events & LPOLLIN)) {
468 return 1;
469 }
470
471 return 0;
472 }
473
474 static
475 int create_sock(void *data, int *out_fd)
476 {
477 int ret;
478 struct lttcomm_sock *sock = data;
479
480 ret = lttcomm_create_sock(sock);
481 if (ret < 0) {
482 goto end;
483 }
484
485 *out_fd = sock->fd;
486 end:
487 return ret;
488 }
489
490 static
491 int close_sock(void *data, int *in_fd)
492 {
493 struct lttcomm_sock *sock = data;
494
495 return sock->ops->close(sock);
496 }
497
498 static int accept_sock(void *data, int *out_fd)
499 {
500 int ret = 0;
501 /* Socks is an array of in_sock, out_sock. */
502 struct lttcomm_sock **socks = data;
503 struct lttcomm_sock *in_sock = socks[0];
504
505 socks[1] = in_sock->ops->accept(in_sock);
506 if (!socks[1]) {
507 ret = -1;
508 goto end;
509 }
510 *out_fd = socks[1]->fd;
511 end:
512 return ret;
513 }
514
515 static
516 struct lttcomm_sock *accept_live_sock(struct lttcomm_sock *listening_sock,
517 const char *name)
518 {
519 int out_fd, ret;
520 struct lttcomm_sock *socks[2] = { listening_sock, NULL };
521 struct lttcomm_sock *new_sock = NULL;
522
523 ret = fd_tracker_open_unsuspendable_fd(the_fd_tracker, &out_fd,
524 (const char **) &name, 1, accept_sock, &socks);
525 if (ret) {
526 goto end;
527 }
528 new_sock = socks[1];
529 DBG("%s accepted, socket %d", name, new_sock->fd);
530 end:
531 return new_sock;
532 }
533
534 /*
535 * Create and init socket from uri.
536 */
537 static
538 struct lttcomm_sock *init_socket(struct lttng_uri *uri, const char *name)
539 {
540 int ret, sock_fd;
541 struct lttcomm_sock *sock = NULL;
542 char uri_str[LTTNG_PATH_MAX];
543 char *formated_name = NULL;
544
545 sock = lttcomm_alloc_sock_from_uri(uri);
546 if (sock == NULL) {
547 ERR("Allocating socket");
548 goto error;
549 }
550
551 /*
552 * Don't fail to create the socket if the name can't be built as it is
553 * only used for debugging purposes.
554 */
555 ret = uri_to_str_url(uri, uri_str, sizeof(uri_str));
556 uri_str[sizeof(uri_str) - 1] = '\0';
557 if (ret >= 0) {
558 ret = asprintf(&formated_name, "%s socket @ %s", name,
559 uri_str);
560 if (ret < 0) {
561 formated_name = NULL;
562 }
563 }
564
565 ret = fd_tracker_open_unsuspendable_fd(the_fd_tracker, &sock_fd,
566 (const char **) (formated_name ? &formated_name : NULL),
567 1, create_sock, sock);
568 if (ret) {
569 PERROR("Failed to create \"%s\" socket",
570 formated_name ?: "Unknown");
571 goto error;
572 }
573 DBG("Listening on %s socket %d", name, sock->fd);
574
575 ret = sock->ops->bind(sock);
576 if (ret < 0) {
577 PERROR("Failed to bind lttng-live socket");
578 goto error;
579 }
580
581 ret = sock->ops->listen(sock, -1);
582 if (ret < 0) {
583 goto error;
584
585 }
586
587 free(formated_name);
588 return sock;
589
590 error:
591 if (sock) {
592 lttcomm_destroy_sock(sock);
593 }
594 free(formated_name);
595 return NULL;
596 }
597
598 /*
599 * This thread manages the listening for new connections on the network
600 */
601 static
602 void *thread_listener(void *data)
603 {
604 int i, ret, pollfd, err = -1;
605 uint32_t revents, nb_fd;
606 struct lttng_poll_event events;
607 struct lttcomm_sock *live_control_sock;
608
609 DBG("[thread] Relay live listener started");
610
611 rcu_register_thread();
612 health_register(health_relayd, HEALTH_RELAYD_TYPE_LIVE_LISTENER);
613
614 health_code_update();
615
616 live_control_sock = init_socket(live_uri, "Live listener");
617 if (!live_control_sock) {
618 goto error_sock_control;
619 }
620
621 /* Pass 2 as size here for the thread quit pipe and control sockets. */
622 ret = create_named_thread_poll_set(&events, 2,
623 "Live listener thread epoll");
624 if (ret < 0) {
625 goto error_create_poll;
626 }
627
628 /* Add the control socket */
629 ret = lttng_poll_add(&events, live_control_sock->fd, LPOLLIN | LPOLLRDHUP);
630 if (ret < 0) {
631 goto error_poll_add;
632 }
633
634 lttng_relay_notify_ready();
635
636 if (testpoint(relayd_thread_live_listener)) {
637 goto error_testpoint;
638 }
639
640 while (1) {
641 health_code_update();
642
643 DBG("Listener accepting live viewers connections");
644
645 restart:
646 health_poll_entry();
647 ret = lttng_poll_wait(&events, -1);
648 health_poll_exit();
649 if (ret < 0) {
650 /*
651 * Restart interrupted system call.
652 */
653 if (errno == EINTR) {
654 goto restart;
655 }
656 goto error;
657 }
658 nb_fd = ret;
659
660 DBG("Relay new viewer connection received");
661 for (i = 0; i < nb_fd; i++) {
662 health_code_update();
663
664 /* Fetch once the poll data */
665 revents = LTTNG_POLL_GETEV(&events, i);
666 pollfd = LTTNG_POLL_GETFD(&events, i);
667
668 /* Thread quit pipe has been closed. Killing thread. */
669 ret = check_thread_quit_pipe(pollfd, revents);
670 if (ret) {
671 err = 0;
672 goto exit;
673 }
674
675 if (revents & LPOLLIN) {
676 /*
677 * A new connection is requested, therefore a
678 * viewer connection is allocated in this
679 * thread, enqueued to a global queue and
680 * dequeued (and freed) in the worker thread.
681 */
682 int val = 1;
683 struct relay_connection *new_conn;
684 struct lttcomm_sock *newsock;
685
686 newsock = accept_live_sock(live_control_sock,
687 "Live socket to client");
688 if (!newsock) {
689 PERROR("accepting control sock");
690 goto error;
691 }
692 DBG("Relay viewer connection accepted socket %d", newsock->fd);
693
694 ret = setsockopt(newsock->fd, SOL_SOCKET, SO_REUSEADDR, &val,
695 sizeof(val));
696 if (ret < 0) {
697 PERROR("setsockopt inet");
698 lttcomm_destroy_sock(newsock);
699 goto error;
700 }
701 new_conn = connection_create(newsock, RELAY_CONNECTION_UNKNOWN);
702 if (!new_conn) {
703 lttcomm_destroy_sock(newsock);
704 goto error;
705 }
706 /* Ownership assumed by the connection. */
707 newsock = NULL;
708
709 /* Enqueue request for the dispatcher thread. */
710 cds_wfcq_enqueue(&viewer_conn_queue.head, &viewer_conn_queue.tail,
711 &new_conn->qnode);
712
713 /*
714 * Wake the dispatch queue futex.
715 * Implicit memory barrier with the
716 * exchange in cds_wfcq_enqueue.
717 */
718 futex_nto1_wake(&viewer_conn_queue.futex);
719 } else if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
720 ERR("socket poll error");
721 goto error;
722 } else {
723 ERR("Unexpected poll events %u for sock %d", revents, pollfd);
724 goto error;
725 }
726 }
727 }
728
729 exit:
730 error:
731 error_poll_add:
732 error_testpoint:
733 (void) fd_tracker_util_poll_clean(the_fd_tracker, &events);
734 error_create_poll:
735 if (live_control_sock->fd >= 0) {
736 int sock_fd = live_control_sock->fd;
737
738 ret = fd_tracker_close_unsuspendable_fd(the_fd_tracker,
739 &sock_fd, 1, close_sock,
740 live_control_sock);
741 if (ret) {
742 PERROR("close");
743 }
744 live_control_sock->fd = -1;
745 }
746 lttcomm_destroy_sock(live_control_sock);
747 error_sock_control:
748 if (err) {
749 health_error();
750 DBG("Live viewer listener thread exited with error");
751 }
752 health_unregister(health_relayd);
753 rcu_unregister_thread();
754 DBG("Live viewer listener thread cleanup complete");
755 if (lttng_relay_stop_threads()) {
756 ERR("Error stopping threads");
757 }
758 return NULL;
759 }
760
761 /*
762 * This thread manages the dispatching of the requests to worker threads
763 */
764 static
765 void *thread_dispatcher(void *data)
766 {
767 int err = -1;
768 ssize_t ret;
769 struct cds_wfcq_node *node;
770 struct relay_connection *conn = NULL;
771
772 DBG("[thread] Live viewer relay dispatcher started");
773
774 health_register(health_relayd, HEALTH_RELAYD_TYPE_LIVE_DISPATCHER);
775
776 if (testpoint(relayd_thread_live_dispatcher)) {
777 goto error_testpoint;
778 }
779
780 health_code_update();
781
782 for (;;) {
783 health_code_update();
784
785 /* Atomically prepare the queue futex */
786 futex_nto1_prepare(&viewer_conn_queue.futex);
787
788 if (CMM_LOAD_SHARED(live_dispatch_thread_exit)) {
789 break;
790 }
791
792 do {
793 health_code_update();
794
795 /* Dequeue commands */
796 node = cds_wfcq_dequeue_blocking(&viewer_conn_queue.head,
797 &viewer_conn_queue.tail);
798 if (node == NULL) {
799 DBG("Woken up but nothing in the live-viewer "
800 "relay command queue");
801 /* Continue thread execution */
802 break;
803 }
804 conn = caa_container_of(node, struct relay_connection, qnode);
805 DBG("Dispatching viewer request waiting on sock %d",
806 conn->sock->fd);
807
808 /*
809 * Inform worker thread of the new request. This
810 * call is blocking so we can be assured that
811 * the data will be read at some point in time
812 * or wait to the end of the world :)
813 */
814 ret = lttng_write(live_conn_pipe[1], &conn, sizeof(conn));
815 if (ret < 0) {
816 PERROR("write conn pipe");
817 connection_put(conn);
818 goto error;
819 }
820 } while (node != NULL);
821
822 /* Futex wait on queue. Blocking call on futex() */
823 health_poll_entry();
824 futex_nto1_wait(&viewer_conn_queue.futex);
825 health_poll_exit();
826 }
827
828 /* Normal exit, no error */
829 err = 0;
830
831 error:
832 error_testpoint:
833 if (err) {
834 health_error();
835 ERR("Health error occurred in %s", __func__);
836 }
837 health_unregister(health_relayd);
838 DBG("Live viewer dispatch thread dying");
839 if (lttng_relay_stop_threads()) {
840 ERR("Error stopping threads");
841 }
842 return NULL;
843 }
844
845 /*
846 * Establish connection with the viewer and check the versions.
847 *
848 * Return 0 on success or else negative value.
849 */
850 static
851 int viewer_connect(struct relay_connection *conn)
852 {
853 int ret;
854 struct lttng_viewer_connect reply, msg;
855
856 conn->version_check_done = 1;
857
858 health_code_update();
859
860 DBG("Viewer is establishing a connection to the relayd.");
861
862 ret = recv_request(conn->sock, &msg, sizeof(msg));
863 if (ret < 0) {
864 goto end;
865 }
866
867 health_code_update();
868
869 memset(&reply, 0, sizeof(reply));
870 reply.major = RELAYD_VERSION_COMM_MAJOR;
871 reply.minor = RELAYD_VERSION_COMM_MINOR;
872
873 /* Major versions must be the same */
874 if (reply.major != be32toh(msg.major)) {
875 DBG("Incompatible major versions ([relayd] %u vs [client] %u)",
876 reply.major, be32toh(msg.major));
877 ret = -1;
878 goto end;
879 }
880
881 conn->major = reply.major;
882 /* We adapt to the lowest compatible version */
883 if (reply.minor <= be32toh(msg.minor)) {
884 conn->minor = reply.minor;
885 } else {
886 conn->minor = be32toh(msg.minor);
887 }
888
889 if (be32toh(msg.type) == LTTNG_VIEWER_CLIENT_COMMAND) {
890 conn->type = RELAY_VIEWER_COMMAND;
891 } else if (be32toh(msg.type) == LTTNG_VIEWER_CLIENT_NOTIFICATION) {
892 conn->type = RELAY_VIEWER_NOTIFICATION;
893 } else {
894 ERR("Unknown connection type : %u", be32toh(msg.type));
895 ret = -1;
896 goto end;
897 }
898
899 reply.major = htobe32(reply.major);
900 reply.minor = htobe32(reply.minor);
901 if (conn->type == RELAY_VIEWER_COMMAND) {
902 /*
903 * Increment outside of htobe64 macro, because the argument can
904 * be used more than once within the macro, and thus the
905 * operation may be undefined.
906 */
907 pthread_mutex_lock(&last_relay_viewer_session_id_lock);
908 last_relay_viewer_session_id++;
909 pthread_mutex_unlock(&last_relay_viewer_session_id_lock);
910 reply.viewer_session_id = htobe64(last_relay_viewer_session_id);
911 }
912
913 health_code_update();
914
915 ret = send_response(conn->sock, &reply, sizeof(reply));
916 if (ret < 0) {
917 goto end;
918 }
919
920 health_code_update();
921
922 DBG("Version check done using protocol %u.%u", conn->major, conn->minor);
923 ret = 0;
924
925 end:
926 return ret;
927 }
928
929 /*
930 * Send the viewer the list of current sessions.
931 * We need to create a copy of the hash table content because otherwise
932 * we cannot assume the number of entries stays the same between getting
933 * the number of HT elements and iteration over the HT.
934 *
935 * Return 0 on success or else a negative value.
936 */
937 static
938 int viewer_list_sessions(struct relay_connection *conn)
939 {
940 int ret = 0;
941 struct lttng_viewer_list_sessions session_list;
942 struct lttng_ht_iter iter;
943 struct relay_session *session;
944 struct lttng_viewer_session *send_session_buf = NULL;
945 uint32_t buf_count = SESSION_BUF_DEFAULT_COUNT;
946 uint32_t count = 0;
947
948 DBG("List sessions received");
949
950 send_session_buf = zmalloc(SESSION_BUF_DEFAULT_COUNT * sizeof(*send_session_buf));
951 if (!send_session_buf) {
952 return -1;
953 }
954
955 rcu_read_lock();
956 cds_lfht_for_each_entry(sessions_ht->ht, &iter.iter, session,
957 session_n.node) {
958 struct lttng_viewer_session *send_session;
959
960 health_code_update();
961
962 pthread_mutex_lock(&session->lock);
963 if (session->connection_closed) {
964 /* Skip closed session */
965 goto next_session;
966 }
967 if (!session->current_trace_chunk) {
968 /*
969 * Skip un-attachable session. It is either
970 * being destroyed or has not had a trace
971 * chunk created against it yet.
972 */
973 goto next_session;
974 }
975
976 if (count >= buf_count) {
977 struct lttng_viewer_session *newbuf;
978 uint32_t new_buf_count = buf_count << 1;
979
980 newbuf = realloc(send_session_buf,
981 new_buf_count * sizeof(*send_session_buf));
982 if (!newbuf) {
983 ret = -1;
984 goto break_loop;
985 }
986 send_session_buf = newbuf;
987 buf_count = new_buf_count;
988 }
989 send_session = &send_session_buf[count];
990 if (lttng_strncpy(send_session->session_name,
991 session->session_name,
992 sizeof(send_session->session_name))) {
993 ret = -1;
994 goto break_loop;
995 }
996 if (lttng_strncpy(send_session->hostname, session->hostname,
997 sizeof(send_session->hostname))) {
998 ret = -1;
999 goto break_loop;
1000 }
1001 send_session->id = htobe64(session->id);
1002 send_session->live_timer = htobe32(session->live_timer);
1003 if (session->viewer_attached) {
1004 send_session->clients = htobe32(1);
1005 } else {
1006 send_session->clients = htobe32(0);
1007 }
1008 send_session->streams = htobe32(session->stream_count);
1009 count++;
1010 next_session:
1011 pthread_mutex_unlock(&session->lock);
1012 continue;
1013 break_loop:
1014 pthread_mutex_unlock(&session->lock);
1015 break;
1016 }
1017 rcu_read_unlock();
1018 if (ret < 0) {
1019 goto end_free;
1020 }
1021
1022 session_list.sessions_count = htobe32(count);
1023
1024 health_code_update();
1025
1026 ret = send_response(conn->sock, &session_list, sizeof(session_list));
1027 if (ret < 0) {
1028 goto end_free;
1029 }
1030
1031 health_code_update();
1032
1033 ret = send_response(conn->sock, send_session_buf,
1034 count * sizeof(*send_session_buf));
1035 if (ret < 0) {
1036 goto end_free;
1037 }
1038 health_code_update();
1039
1040 ret = 0;
1041 end_free:
1042 free(send_session_buf);
1043 return ret;
1044 }
1045
1046 /*
1047 * Send the viewer the list of current streams.
1048 */
1049 static
1050 int viewer_get_new_streams(struct relay_connection *conn)
1051 {
1052 int ret, send_streams = 0;
1053 uint32_t nb_created = 0, nb_unsent = 0, nb_streams = 0, nb_total = 0;
1054 struct lttng_viewer_new_streams_request request;
1055 struct lttng_viewer_new_streams_response response;
1056 struct relay_session *session = NULL;
1057 uint64_t session_id;
1058 bool closed = false;
1059
1060 assert(conn);
1061
1062 DBG("Get new streams received");
1063
1064 health_code_update();
1065
1066 /* Receive the request from the connected client. */
1067 ret = recv_request(conn->sock, &request, sizeof(request));
1068 if (ret < 0) {
1069 goto error;
1070 }
1071 session_id = be64toh(request.session_id);
1072
1073 health_code_update();
1074
1075 memset(&response, 0, sizeof(response));
1076
1077 session = session_get_by_id(session_id);
1078 if (!session) {
1079 DBG("Relay session %" PRIu64 " not found", session_id);
1080 response.status = htobe32(LTTNG_VIEWER_NEW_STREAMS_ERR);
1081 goto send_reply;
1082 }
1083
1084 if (!viewer_session_is_attached(conn->viewer_session, session)) {
1085 response.status = htobe32(LTTNG_VIEWER_NEW_STREAMS_ERR);
1086 goto send_reply;
1087 }
1088
1089 pthread_mutex_lock(&session->lock);
1090 ret = make_viewer_streams(session,
1091 conn->viewer_session->current_trace_chunk,
1092 LTTNG_VIEWER_SEEK_LAST, &nb_total, &nb_unsent,
1093 &nb_created, &closed);
1094 if (ret < 0) {
1095 goto error_unlock_session;
1096 }
1097 send_streams = 1;
1098 response.status = htobe32(LTTNG_VIEWER_NEW_STREAMS_OK);
1099
1100 /* Only send back the newly created streams with the unsent ones. */
1101 nb_streams = nb_created + nb_unsent;
1102 response.streams_count = htobe32(nb_streams);
1103
1104 /*
1105 * If the session is closed, HUP when there are no more streams
1106 * with data.
1107 */
1108 if (closed && nb_total == 0) {
1109 send_streams = 0;
1110 response.streams_count = 0;
1111 response.status = htobe32(LTTNG_VIEWER_NEW_STREAMS_HUP);
1112 goto send_reply_unlock;
1113 }
1114 send_reply_unlock:
1115 pthread_mutex_unlock(&session->lock);
1116
1117 send_reply:
1118 health_code_update();
1119 ret = send_response(conn->sock, &response, sizeof(response));
1120 if (ret < 0) {
1121 goto end_put_session;
1122 }
1123 health_code_update();
1124
1125 /*
1126 * Unknown or empty session, just return gracefully, the viewer
1127 * knows what is happening.
1128 */
1129 if (!send_streams || !nb_streams) {
1130 ret = 0;
1131 goto end_put_session;
1132 }
1133
1134 /*
1135 * Send stream and *DON'T* ignore the sent flag so every viewer
1136 * streams that were not sent from that point will be sent to
1137 * the viewer.
1138 */
1139 ret = send_viewer_streams(conn->sock, session_id, 0);
1140 if (ret < 0) {
1141 goto end_put_session;
1142 }
1143
1144 end_put_session:
1145 if (session) {
1146 session_put(session);
1147 }
1148 error:
1149 return ret;
1150 error_unlock_session:
1151 pthread_mutex_unlock(&session->lock);
1152 session_put(session);
1153 return ret;
1154 }
1155
1156 /*
1157 * Send the viewer the list of current sessions.
1158 */
1159 static
1160 int viewer_attach_session(struct relay_connection *conn)
1161 {
1162 int send_streams = 0;
1163 ssize_t ret;
1164 uint32_t nb_streams = 0;
1165 enum lttng_viewer_seek seek_type;
1166 struct lttng_viewer_attach_session_request request;
1167 struct lttng_viewer_attach_session_response response;
1168 struct relay_session *session = NULL;
1169 enum lttng_viewer_attach_return_code viewer_attach_status;
1170 bool closed = false;
1171 uint64_t session_id;
1172
1173 assert(conn);
1174
1175 health_code_update();
1176
1177 /* Receive the request from the connected client. */
1178 ret = recv_request(conn->sock, &request, sizeof(request));
1179 if (ret < 0) {
1180 goto error;
1181 }
1182
1183 session_id = be64toh(request.session_id);
1184 health_code_update();
1185
1186 memset(&response, 0, sizeof(response));
1187
1188 if (!conn->viewer_session) {
1189 DBG("Client trying to attach before creating a live viewer session");
1190 response.status = htobe32(LTTNG_VIEWER_ATTACH_NO_SESSION);
1191 goto send_reply;
1192 }
1193
1194 session = session_get_by_id(session_id);
1195 if (!session) {
1196 DBG("Relay session %" PRIu64 " not found", session_id);
1197 response.status = htobe32(LTTNG_VIEWER_ATTACH_UNK);
1198 goto send_reply;
1199 }
1200 DBG("Attach session ID %" PRIu64 " received", session_id);
1201
1202 pthread_mutex_lock(&session->lock);
1203 if (!session->current_trace_chunk) {
1204 /*
1205 * Session is either being destroyed or it never had a trace
1206 * chunk created against it.
1207 */
1208 DBG("Session requested by live client has no current trace chunk, returning unknown session");
1209 response.status = htobe32(LTTNG_VIEWER_ATTACH_UNK);
1210 goto send_reply;
1211 }
1212 if (session->live_timer == 0) {
1213 DBG("Not live session");
1214 response.status = htobe32(LTTNG_VIEWER_ATTACH_NOT_LIVE);
1215 goto send_reply;
1216 }
1217
1218 send_streams = 1;
1219 viewer_attach_status = viewer_session_attach(conn->viewer_session,
1220 session);
1221 if (viewer_attach_status != LTTNG_VIEWER_ATTACH_OK) {
1222 response.status = htobe32(viewer_attach_status);
1223 goto send_reply;
1224 }
1225
1226 switch (be32toh(request.seek)) {
1227 case LTTNG_VIEWER_SEEK_BEGINNING:
1228 case LTTNG_VIEWER_SEEK_LAST:
1229 response.status = htobe32(LTTNG_VIEWER_ATTACH_OK);
1230 seek_type = be32toh(request.seek);
1231 break;
1232 default:
1233 ERR("Wrong seek parameter");
1234 response.status = htobe32(LTTNG_VIEWER_ATTACH_SEEK_ERR);
1235 send_streams = 0;
1236 goto send_reply;
1237 }
1238
1239 ret = make_viewer_streams(session,
1240 conn->viewer_session->current_trace_chunk, seek_type,
1241 &nb_streams, NULL, NULL, &closed);
1242 if (ret < 0) {
1243 goto end_put_session;
1244 }
1245 pthread_mutex_unlock(&session->lock);
1246 session_put(session);
1247 session = NULL;
1248
1249 response.streams_count = htobe32(nb_streams);
1250 /*
1251 * If the session is closed when the viewer is attaching, it
1252 * means some of the streams may have been concurrently removed,
1253 * so we don't allow the viewer to attach, even if there are
1254 * streams available.
1255 */
1256 if (closed) {
1257 send_streams = 0;
1258 response.streams_count = 0;
1259 response.status = htobe32(LTTNG_VIEWER_ATTACH_UNK);
1260 goto send_reply;
1261 }
1262
1263 send_reply:
1264 health_code_update();
1265 ret = send_response(conn->sock, &response, sizeof(response));
1266 if (ret < 0) {
1267 goto end_put_session;
1268 }
1269 health_code_update();
1270
1271 /*
1272 * Unknown or empty session, just return gracefully, the viewer
1273 * knows what is happening.
1274 */
1275 if (!send_streams || !nb_streams) {
1276 ret = 0;
1277 goto end_put_session;
1278 }
1279
1280 /* Send stream and ignore the sent flag. */
1281 ret = send_viewer_streams(conn->sock, session_id, 1);
1282 if (ret < 0) {
1283 goto end_put_session;
1284 }
1285
1286 end_put_session:
1287 if (session) {
1288 pthread_mutex_unlock(&session->lock);
1289 session_put(session);
1290 }
1291 error:
1292 return ret;
1293 }
1294
1295 /*
1296 * Open the index file if needed for the given vstream.
1297 *
1298 * If an index file is successfully opened, the vstream will set it as its
1299 * current index file.
1300 *
1301 * Return 0 on success, a negative value on error (-ENOENT if not ready yet).
1302 *
1303 * Called with rstream lock held.
1304 */
1305 static int try_open_index(struct relay_viewer_stream *vstream,
1306 struct relay_stream *rstream)
1307 {
1308 int ret = 0;
1309 const uint32_t connection_major = rstream->trace->session->major;
1310 const uint32_t connection_minor = rstream->trace->session->minor;
1311 enum lttng_trace_chunk_status chunk_status;
1312
1313 if (vstream->index_file) {
1314 goto end;
1315 }
1316
1317 /*
1318 * First time, we open the index file and at least one index is ready.
1319 */
1320 if (rstream->index_received_seqcount == 0) {
1321 ret = -ENOENT;
1322 goto end;
1323 }
1324 chunk_status = lttng_index_file_create_from_trace_chunk_read_only(
1325 vstream->stream_file.trace_chunk, rstream->path_name,
1326 rstream->channel_name, rstream->tracefile_size,
1327 vstream->current_tracefile_id,
1328 lttng_to_index_major(connection_major, connection_minor),
1329 lttng_to_index_minor(connection_major, connection_minor),
1330 true, &vstream->index_file);
1331 if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
1332 if (chunk_status == LTTNG_TRACE_CHUNK_STATUS_NO_FILE) {
1333 ret = -ENOENT;
1334 } else {
1335 ret = -1;
1336 }
1337 }
1338
1339 end:
1340 return ret;
1341 }
1342
1343 /*
1344 * Check the status of the index for the given stream. This function
1345 * updates the index structure if needed and can put (close) the vstream
1346 * in the HUP situation.
1347 *
1348 * Return 0 means that we can proceed with the index. A value of 1 means
1349 * that the index has been updated and is ready to be sent to the
1350 * client. A negative value indicates an error that can't be handled.
1351 *
1352 * Called with rstream lock held.
1353 */
1354 static int check_index_status(struct relay_viewer_stream *vstream,
1355 struct relay_stream *rstream, struct ctf_trace *trace,
1356 struct lttng_viewer_index *index)
1357 {
1358 int ret;
1359
1360 DBG("Check index status: index_received_seqcount %" PRIu64 " "
1361 "index_sent_seqcount %" PRIu64 " "
1362 "for stream %" PRIu64,
1363 rstream->index_received_seqcount,
1364 vstream->index_sent_seqcount,
1365 vstream->stream->stream_handle);
1366 if ((trace->session->connection_closed || rstream->closed)
1367 && rstream->index_received_seqcount
1368 == vstream->index_sent_seqcount) {
1369 /*
1370 * Last index sent and session connection or relay
1371 * stream are closed.
1372 */
1373 index->status = htobe32(LTTNG_VIEWER_INDEX_HUP);
1374 goto hup;
1375 } else if (rstream->beacon_ts_end != -1ULL &&
1376 (rstream->index_received_seqcount == 0 ||
1377 (vstream->index_sent_seqcount != 0 &&
1378 rstream->index_received_seqcount
1379 <= vstream->index_sent_seqcount))) {
1380 /*
1381 * We've received a synchronization beacon and the last index
1382 * available has been sent, the index for now is inactive.
1383 *
1384 * In this case, we have received a beacon which allows us to
1385 * inform the client of a time interval during which we can
1386 * guarantee that there are no events to read (and never will
1387 * be).
1388 *
1389 * The sent seqcount can grow higher than receive seqcount on
1390 * clear because the rotation performed by clear will push
1391 * the index_sent_seqcount ahead (see
1392 * viewer_stream_sync_tracefile_array_tail) and skip over
1393 * packet sequence numbers.
1394 */
1395 index->status = htobe32(LTTNG_VIEWER_INDEX_INACTIVE);
1396 index->timestamp_end = htobe64(rstream->beacon_ts_end);
1397 index->stream_id = htobe64(rstream->ctf_stream_id);
1398 DBG("Check index status: inactive with beacon, for stream %" PRIu64,
1399 vstream->stream->stream_handle);
1400 goto index_ready;
1401 } else if (rstream->index_received_seqcount == 0 ||
1402 (vstream->index_sent_seqcount != 0 &&
1403 rstream->index_received_seqcount
1404 <= vstream->index_sent_seqcount)) {
1405 /*
1406 * This checks whether received <= sent seqcount. In
1407 * this case, we have not received a beacon. Therefore,
1408 * we can only ask the client to retry later.
1409 *
1410 * The sent seqcount can grow higher than receive seqcount on
1411 * clear because the rotation performed by clear will push
1412 * the index_sent_seqcount ahead (see
1413 * viewer_stream_sync_tracefile_array_tail) and skip over
1414 * packet sequence numbers.
1415 */
1416 index->status = htobe32(LTTNG_VIEWER_INDEX_RETRY);
1417 DBG("Check index status: retry for stream %" PRIu64,
1418 vstream->stream->stream_handle);
1419 goto index_ready;
1420 } else if (!tracefile_array_seq_in_file(rstream->tfa,
1421 vstream->current_tracefile_id,
1422 vstream->index_sent_seqcount)) {
1423 /*
1424 * The next index we want to send cannot be read either
1425 * because we need to perform a rotation, or due to
1426 * the producer having overwritten its trace file.
1427 */
1428 DBG("Viewer stream %" PRIu64 " rotation",
1429 vstream->stream->stream_handle);
1430 ret = viewer_stream_rotate(vstream);
1431 if (ret == 1) {
1432 /* EOF across entire stream. */
1433 index->status = htobe32(LTTNG_VIEWER_INDEX_HUP);
1434 goto hup;
1435 }
1436 /*
1437 * If we have been pushed due to overwrite, it
1438 * necessarily means there is data that can be read in
1439 * the stream. If we rotated because we reached the end
1440 * of a tracefile, it means the following tracefile
1441 * needs to contain at least one index, else we would
1442 * have already returned LTTNG_VIEWER_INDEX_RETRY to the
1443 * viewer. The updated index_sent_seqcount needs to
1444 * point to a readable index entry now.
1445 *
1446 * In the case where we "rotate" on a single file, we
1447 * can end up in a case where the requested index is
1448 * still unavailable.
1449 */
1450 if (rstream->tracefile_count == 1 &&
1451 !tracefile_array_seq_in_file(
1452 rstream->tfa,
1453 vstream->current_tracefile_id,
1454 vstream->index_sent_seqcount)) {
1455 index->status = htobe32(LTTNG_VIEWER_INDEX_RETRY);
1456 DBG("Check index status: retry: "
1457 "tracefile array sequence number %" PRIu64
1458 " not in file for stream %" PRIu64,
1459 vstream->index_sent_seqcount,
1460 vstream->stream->stream_handle);
1461 goto index_ready;
1462 }
1463 assert(tracefile_array_seq_in_file(rstream->tfa,
1464 vstream->current_tracefile_id,
1465 vstream->index_sent_seqcount));
1466 }
1467 /* ret == 0 means successful so we continue. */
1468 ret = 0;
1469 return ret;
1470
1471 hup:
1472 viewer_stream_put(vstream);
1473 index_ready:
1474 return 1;
1475 }
1476
1477 /*
1478 * Send the next index for a stream.
1479 *
1480 * Return 0 on success or else a negative value.
1481 */
1482 static
1483 int viewer_get_next_index(struct relay_connection *conn)
1484 {
1485 int ret;
1486 struct lttng_viewer_get_next_index request_index;
1487 struct lttng_viewer_index viewer_index;
1488 struct ctf_packet_index packet_index;
1489 struct relay_viewer_stream *vstream = NULL;
1490 struct relay_stream *rstream = NULL;
1491 struct ctf_trace *ctf_trace = NULL;
1492 struct relay_viewer_stream *metadata_viewer_stream = NULL;
1493
1494 assert(conn);
1495
1496 DBG("Viewer get next index");
1497
1498 memset(&viewer_index, 0, sizeof(viewer_index));
1499 health_code_update();
1500
1501 ret = recv_request(conn->sock, &request_index, sizeof(request_index));
1502 if (ret < 0) {
1503 goto end;
1504 }
1505 health_code_update();
1506
1507 vstream = viewer_stream_get_by_id(be64toh(request_index.stream_id));
1508 if (!vstream) {
1509 DBG("Client requested index of unknown stream id %" PRIu64,
1510 (uint64_t) be64toh(request_index.stream_id));
1511 viewer_index.status = htobe32(LTTNG_VIEWER_INDEX_ERR);
1512 goto send_reply;
1513 }
1514
1515 /* Use back. ref. Protected by refcounts. */
1516 rstream = vstream->stream;
1517 ctf_trace = rstream->trace;
1518
1519 /* metadata_viewer_stream may be NULL. */
1520 metadata_viewer_stream =
1521 ctf_trace_get_viewer_metadata_stream(ctf_trace);
1522
1523 pthread_mutex_lock(&rstream->lock);
1524
1525 /*
1526 * The viewer should not ask for index on metadata stream.
1527 */
1528 if (rstream->is_metadata) {
1529 viewer_index.status = htobe32(LTTNG_VIEWER_INDEX_HUP);
1530 goto send_reply;
1531 }
1532
1533 if (rstream->ongoing_rotation.is_set) {
1534 /* Rotation is ongoing, try again later. */
1535 viewer_index.status = htobe32(LTTNG_VIEWER_INDEX_RETRY);
1536 goto send_reply;
1537 }
1538
1539 if (rstream->trace->session->ongoing_rotation) {
1540 /* Rotation is ongoing, try again later. */
1541 viewer_index.status = htobe32(LTTNG_VIEWER_INDEX_RETRY);
1542 goto send_reply;
1543 }
1544
1545 if (rstream->trace_chunk) {
1546 uint64_t rchunk_id, vchunk_id;
1547
1548 /*
1549 * If the relay stream is not yet closed, ensure the viewer
1550 * chunk matches the relay chunk after clear.
1551 */
1552 if (lttng_trace_chunk_get_id(rstream->trace_chunk,
1553 &rchunk_id) != LTTNG_TRACE_CHUNK_STATUS_OK) {
1554 viewer_index.status = htobe32(LTTNG_VIEWER_INDEX_ERR);
1555 goto send_reply;
1556 }
1557 if (lttng_trace_chunk_get_id(
1558 conn->viewer_session->current_trace_chunk,
1559 &vchunk_id) != LTTNG_TRACE_CHUNK_STATUS_OK) {
1560 viewer_index.status = htobe32(LTTNG_VIEWER_INDEX_ERR);
1561 goto send_reply;
1562 }
1563
1564 if (rchunk_id != vchunk_id) {
1565 DBG("Relay and viewer chunk ids differ: "
1566 "rchunk_id %" PRIu64 " vchunk_id %" PRIu64,
1567 rchunk_id, vchunk_id);
1568
1569 lttng_trace_chunk_put(
1570 conn->viewer_session->current_trace_chunk);
1571 conn->viewer_session->current_trace_chunk = NULL;
1572 ret = viewer_session_set_trace_chunk_copy(
1573 conn->viewer_session,
1574 rstream->trace_chunk);
1575 if (ret) {
1576 viewer_index.status =
1577 htobe32(LTTNG_VIEWER_INDEX_ERR);
1578 goto send_reply;
1579 }
1580 }
1581 }
1582 if (conn->viewer_session->current_trace_chunk !=
1583 vstream->stream_file.trace_chunk) {
1584 bool acquired_reference;
1585
1586 DBG("Viewer session and viewer stream chunk differ: "
1587 "vsession chunk %p vstream chunk %p",
1588 conn->viewer_session->current_trace_chunk,
1589 vstream->stream_file.trace_chunk);
1590 lttng_trace_chunk_put(vstream->stream_file.trace_chunk);
1591 acquired_reference = lttng_trace_chunk_get(conn->viewer_session->current_trace_chunk);
1592 assert(acquired_reference);
1593 vstream->stream_file.trace_chunk =
1594 conn->viewer_session->current_trace_chunk;
1595 viewer_stream_sync_tracefile_array_tail(vstream);
1596 viewer_stream_close_files(vstream);
1597 }
1598
1599 ret = check_index_status(vstream, rstream, ctf_trace, &viewer_index);
1600 if (ret < 0) {
1601 goto error_put;
1602 } else if (ret == 1) {
1603 /*
1604 * We have no index to send and check_index_status has populated
1605 * viewer_index's status.
1606 */
1607 goto send_reply;
1608 }
1609 /* At this point, ret is 0 thus we will be able to read the index. */
1610 assert(!ret);
1611
1612 /* Try to open an index if one is needed for that stream. */
1613 ret = try_open_index(vstream, rstream);
1614 if (ret == -ENOENT) {
1615 if (rstream->closed) {
1616 viewer_index.status = htobe32(LTTNG_VIEWER_INDEX_HUP);
1617 goto send_reply;
1618 } else {
1619 viewer_index.status = htobe32(LTTNG_VIEWER_INDEX_RETRY);
1620 goto send_reply;
1621 }
1622 }
1623 if (ret < 0) {
1624 viewer_index.status = htobe32(LTTNG_VIEWER_INDEX_ERR);
1625 goto send_reply;
1626 }
1627
1628 /*
1629 * vstream->stream_fd may be NULL if it has been closed by
1630 * tracefile rotation, or if we are at the beginning of the
1631 * stream. We open the data stream file here to protect against
1632 * overwrite caused by tracefile rotation (in association with
1633 * unlink performed before overwrite).
1634 */
1635 if (!vstream->stream_file.handle) {
1636 char file_path[LTTNG_PATH_MAX];
1637 enum lttng_trace_chunk_status status;
1638 struct fs_handle *fs_handle;
1639
1640 ret = utils_stream_file_path(rstream->path_name,
1641 rstream->channel_name, rstream->tracefile_size,
1642 vstream->current_tracefile_id, NULL, file_path,
1643 sizeof(file_path));
1644 if (ret < 0) {
1645 goto error_put;
1646 }
1647
1648 /*
1649 * It is possible the the file we are trying to open is
1650 * missing if the stream has been closed (application exits with
1651 * per-pid buffers) and a clear command has been performed.
1652 */
1653 status = lttng_trace_chunk_open_fs_handle(
1654 vstream->stream_file.trace_chunk,
1655 file_path, O_RDONLY, 0, &fs_handle, true);
1656 if (status != LTTNG_TRACE_CHUNK_STATUS_OK) {
1657 if (status == LTTNG_TRACE_CHUNK_STATUS_NO_FILE &&
1658 rstream->closed) {
1659 viewer_index.status = htobe32(LTTNG_VIEWER_INDEX_HUP);
1660 goto send_reply;
1661 }
1662 PERROR("Failed to open trace file for viewer stream");
1663 goto error_put;
1664 }
1665 vstream->stream_file.handle = fs_handle;
1666 }
1667
1668 ret = check_new_streams(conn);
1669 if (ret < 0) {
1670 viewer_index.status = htobe32(LTTNG_VIEWER_INDEX_ERR);
1671 goto send_reply;
1672 } else if (ret == 1) {
1673 viewer_index.flags |= LTTNG_VIEWER_FLAG_NEW_STREAM;
1674 }
1675
1676 ret = lttng_index_file_read(vstream->index_file, &packet_index);
1677 if (ret) {
1678 ERR("Relay error reading index file");
1679 viewer_index.status = htobe32(LTTNG_VIEWER_INDEX_ERR);
1680 goto send_reply;
1681 } else {
1682 viewer_index.status = htobe32(LTTNG_VIEWER_INDEX_OK);
1683 vstream->index_sent_seqcount++;
1684 }
1685
1686 /*
1687 * Indexes are stored in big endian, no need to switch before sending.
1688 */
1689 DBG("Sending viewer index for stream %" PRIu64 " offset %" PRIu64,
1690 rstream->stream_handle,
1691 (uint64_t) be64toh(packet_index.offset));
1692 viewer_index.offset = packet_index.offset;
1693 viewer_index.packet_size = packet_index.packet_size;
1694 viewer_index.content_size = packet_index.content_size;
1695 viewer_index.timestamp_begin = packet_index.timestamp_begin;
1696 viewer_index.timestamp_end = packet_index.timestamp_end;
1697 viewer_index.events_discarded = packet_index.events_discarded;
1698 viewer_index.stream_id = packet_index.stream_id;
1699
1700 send_reply:
1701 if (rstream) {
1702 pthread_mutex_unlock(&rstream->lock);
1703 }
1704
1705 if (metadata_viewer_stream) {
1706 pthread_mutex_lock(&metadata_viewer_stream->stream->lock);
1707 DBG("get next index metadata check: recv %" PRIu64
1708 " sent %" PRIu64,
1709 metadata_viewer_stream->stream->metadata_received,
1710 metadata_viewer_stream->metadata_sent);
1711 if (!metadata_viewer_stream->stream->metadata_received ||
1712 metadata_viewer_stream->stream->metadata_received >
1713 metadata_viewer_stream->metadata_sent) {
1714 viewer_index.flags |= LTTNG_VIEWER_FLAG_NEW_METADATA;
1715 }
1716 pthread_mutex_unlock(&metadata_viewer_stream->stream->lock);
1717 }
1718
1719 viewer_index.flags = htobe32(viewer_index.flags);
1720 health_code_update();
1721
1722 ret = send_response(conn->sock, &viewer_index, sizeof(viewer_index));
1723 if (ret < 0) {
1724 goto end;
1725 }
1726 health_code_update();
1727
1728 if (vstream) {
1729 DBG("Index %" PRIu64 " for stream %" PRIu64 " sent",
1730 vstream->index_sent_seqcount,
1731 vstream->stream->stream_handle);
1732 }
1733 end:
1734 if (metadata_viewer_stream) {
1735 viewer_stream_put(metadata_viewer_stream);
1736 }
1737 if (vstream) {
1738 viewer_stream_put(vstream);
1739 }
1740 return ret;
1741
1742 error_put:
1743 pthread_mutex_unlock(&rstream->lock);
1744 if (metadata_viewer_stream) {
1745 viewer_stream_put(metadata_viewer_stream);
1746 }
1747 viewer_stream_put(vstream);
1748 return ret;
1749 }
1750
1751 /*
1752 * Send the next index for a stream
1753 *
1754 * Return 0 on success or else a negative value.
1755 */
1756 static
1757 int viewer_get_packet(struct relay_connection *conn)
1758 {
1759 int ret;
1760 off_t lseek_ret;
1761 char *reply = NULL;
1762 struct lttng_viewer_get_packet get_packet_info;
1763 struct lttng_viewer_trace_packet reply_header;
1764 struct relay_viewer_stream *vstream = NULL;
1765 uint32_t reply_size = sizeof(reply_header);
1766 uint32_t packet_data_len = 0;
1767 ssize_t read_len;
1768 uint64_t stream_id;
1769
1770 DBG2("Relay get data packet");
1771
1772 health_code_update();
1773
1774 ret = recv_request(conn->sock, &get_packet_info,
1775 sizeof(get_packet_info));
1776 if (ret < 0) {
1777 goto end;
1778 }
1779 health_code_update();
1780
1781 /* From this point on, the error label can be reached. */
1782 memset(&reply_header, 0, sizeof(reply_header));
1783 stream_id = (uint64_t) be64toh(get_packet_info.stream_id);
1784
1785 vstream = viewer_stream_get_by_id(stream_id);
1786 if (!vstream) {
1787 DBG("Client requested packet of unknown stream id %" PRIu64,
1788 stream_id);
1789 reply_header.status = htobe32(LTTNG_VIEWER_GET_PACKET_ERR);
1790 goto send_reply_nolock;
1791 } else {
1792 packet_data_len = be32toh(get_packet_info.len);
1793 reply_size += packet_data_len;
1794 }
1795
1796 reply = zmalloc(reply_size);
1797 if (!reply) {
1798 PERROR("packet reply zmalloc");
1799 reply_size = sizeof(reply_header);
1800 goto error;
1801 }
1802
1803 pthread_mutex_lock(&vstream->stream->lock);
1804 lseek_ret = fs_handle_seek(vstream->stream_file.handle,
1805 be64toh(get_packet_info.offset), SEEK_SET);
1806 if (lseek_ret < 0) {
1807 PERROR("Failed to seek file system handle of viewer stream %" PRIu64
1808 " to offset %" PRIu64,
1809 stream_id,
1810 (uint64_t) be64toh(get_packet_info.offset));
1811 goto error;
1812 }
1813 read_len = fs_handle_read(vstream->stream_file.handle,
1814 reply + sizeof(reply_header), packet_data_len);
1815 if (read_len < packet_data_len) {
1816 PERROR("Failed to read from file system handle of viewer stream id %" PRIu64
1817 ", offset: %" PRIu64,
1818 stream_id,
1819 (uint64_t) be64toh(get_packet_info.offset));
1820 goto error;
1821 }
1822 reply_header.status = htobe32(LTTNG_VIEWER_GET_PACKET_OK);
1823 reply_header.len = htobe32(packet_data_len);
1824 goto send_reply;
1825
1826 error:
1827 reply_header.status = htobe32(LTTNG_VIEWER_GET_PACKET_ERR);
1828
1829 send_reply:
1830 if (vstream) {
1831 pthread_mutex_unlock(&vstream->stream->lock);
1832 }
1833 send_reply_nolock:
1834
1835 health_code_update();
1836
1837 if (reply) {
1838 memcpy(reply, &reply_header, sizeof(reply_header));
1839 ret = send_response(conn->sock, reply, reply_size);
1840 } else {
1841 /* No reply to send. */
1842 ret = send_response(conn->sock, &reply_header,
1843 reply_size);
1844 }
1845
1846 health_code_update();
1847 if (ret < 0) {
1848 PERROR("sendmsg of packet data failed");
1849 goto end_free;
1850 }
1851
1852 DBG("Sent %u bytes for stream %" PRIu64, reply_size, stream_id);
1853
1854 end_free:
1855 free(reply);
1856 end:
1857 if (vstream) {
1858 viewer_stream_put(vstream);
1859 }
1860 return ret;
1861 }
1862
1863 /*
1864 * Send the session's metadata
1865 *
1866 * Return 0 on success else a negative value.
1867 */
1868 static
1869 int viewer_get_metadata(struct relay_connection *conn)
1870 {
1871 int ret = 0;
1872 int fd = -1;
1873 ssize_t read_len;
1874 uint64_t len = 0;
1875 char *data = NULL;
1876 struct lttng_viewer_get_metadata request;
1877 struct lttng_viewer_metadata_packet reply;
1878 struct relay_viewer_stream *vstream = NULL;
1879
1880 assert(conn);
1881
1882 DBG("Relay get metadata");
1883
1884 health_code_update();
1885
1886 ret = recv_request(conn->sock, &request, sizeof(request));
1887 if (ret < 0) {
1888 goto end;
1889 }
1890 health_code_update();
1891
1892 memset(&reply, 0, sizeof(reply));
1893
1894 vstream = viewer_stream_get_by_id(be64toh(request.stream_id));
1895 if (!vstream) {
1896 /*
1897 * The metadata stream can be closed by a CLOSE command
1898 * just before we attach. It can also be closed by
1899 * per-pid tracing during tracing. Therefore, it is
1900 * possible that we cannot find this viewer stream.
1901 * Reply back to the client with an error if we cannot
1902 * find it.
1903 */
1904 DBG("Client requested metadata of unknown stream id %" PRIu64,
1905 (uint64_t) be64toh(request.stream_id));
1906 reply.status = htobe32(LTTNG_VIEWER_METADATA_ERR);
1907 goto send_reply;
1908 }
1909 pthread_mutex_lock(&vstream->stream->lock);
1910 if (!vstream->stream->is_metadata) {
1911 ERR("Invalid metadata stream");
1912 goto error;
1913 }
1914
1915 if (vstream->metadata_sent >= vstream->stream->metadata_received) {
1916 /*
1917 * The live viewers expect to receive a NO_NEW_METADATA
1918 * status before a stream disappears, otherwise they abort the
1919 * entire live connection when receiving an error status.
1920 *
1921 * Clear feature resets the metadata_sent to 0 until the
1922 * same metadata is received again.
1923 */
1924 reply.status = htobe32(LTTNG_VIEWER_NO_NEW_METADATA);
1925 /*
1926 * The live viewer considers a closed 0 byte metadata stream as
1927 * an error.
1928 */
1929 if (vstream->metadata_sent > 0) {
1930 vstream->stream->no_new_metadata_notified = true;
1931 if (vstream->stream->closed) {
1932 /* Release ownership for the viewer metadata stream. */
1933 viewer_stream_put(vstream);
1934 }
1935 }
1936 goto send_reply;
1937 }
1938
1939 len = vstream->stream->metadata_received - vstream->metadata_sent;
1940
1941 /* first time, we open the metadata file */
1942 if (!vstream->stream_file.handle) {
1943 struct fs_handle *fs_handle;
1944 char file_path[LTTNG_PATH_MAX];
1945 enum lttng_trace_chunk_status status;
1946 struct relay_stream *rstream = vstream->stream;
1947
1948 ret = utils_stream_file_path(rstream->path_name,
1949 rstream->channel_name, rstream->tracefile_size,
1950 vstream->current_tracefile_id, NULL, file_path,
1951 sizeof(file_path));
1952 if (ret < 0) {
1953 goto error;
1954 }
1955
1956 /*
1957 * It is possible the the metadata file we are trying to open is
1958 * missing if the stream has been closed (application exits with
1959 * per-pid buffers) and a clear command has been performed.
1960 */
1961 status = lttng_trace_chunk_open_fs_handle(
1962 vstream->stream_file.trace_chunk,
1963 file_path, O_RDONLY, 0, &fs_handle, true);
1964 if (status != LTTNG_TRACE_CHUNK_STATUS_OK) {
1965 if (status == LTTNG_TRACE_CHUNK_STATUS_NO_FILE) {
1966 reply.status = htobe32(LTTNG_VIEWER_NO_NEW_METADATA);
1967 len = 0;
1968 if (vstream->stream->closed) {
1969 viewer_stream_put(vstream);
1970 }
1971 goto send_reply;
1972 }
1973 PERROR("Failed to open metadata file for viewer stream");
1974 goto error;
1975 }
1976 vstream->stream_file.handle = fs_handle;
1977 }
1978
1979 reply.len = htobe64(len);
1980 data = zmalloc(len);
1981 if (!data) {
1982 PERROR("viewer metadata zmalloc");
1983 goto error;
1984 }
1985
1986 fd = fs_handle_get_fd(vstream->stream_file.handle);
1987 if (fd < 0) {
1988 ERR("Failed to restore viewer stream file system handle");
1989 goto error;
1990 }
1991 read_len = lttng_read(fd, data, len);
1992 fs_handle_put_fd(vstream->stream_file.handle);
1993 fd = -1;
1994 if (read_len < len) {
1995 PERROR("Relay reading metadata file");
1996 goto error;
1997 }
1998 vstream->metadata_sent += read_len;
1999 reply.status = htobe32(LTTNG_VIEWER_METADATA_OK);
2000
2001 goto send_reply;
2002
2003 error:
2004 reply.status = htobe32(LTTNG_VIEWER_METADATA_ERR);
2005
2006 send_reply:
2007 health_code_update();
2008 if (vstream) {
2009 pthread_mutex_unlock(&vstream->stream->lock);
2010 }
2011 ret = send_response(conn->sock, &reply, sizeof(reply));
2012 if (ret < 0) {
2013 goto end_free;
2014 }
2015 health_code_update();
2016
2017 if (len > 0) {
2018 ret = send_response(conn->sock, data, len);
2019 if (ret < 0) {
2020 goto end_free;
2021 }
2022 }
2023
2024 DBG("Sent %" PRIu64 " bytes of metadata for stream %" PRIu64, len,
2025 (uint64_t) be64toh(request.stream_id));
2026
2027 DBG("Metadata sent");
2028
2029 end_free:
2030 free(data);
2031 end:
2032 if (vstream) {
2033 viewer_stream_put(vstream);
2034 }
2035 return ret;
2036 }
2037
2038 /*
2039 * Create a viewer session.
2040 *
2041 * Return 0 on success or else a negative value.
2042 */
2043 static
2044 int viewer_create_session(struct relay_connection *conn)
2045 {
2046 int ret;
2047 struct lttng_viewer_create_session_response resp;
2048
2049 DBG("Viewer create session received");
2050
2051 memset(&resp, 0, sizeof(resp));
2052 resp.status = htobe32(LTTNG_VIEWER_CREATE_SESSION_OK);
2053 conn->viewer_session = viewer_session_create();
2054 if (!conn->viewer_session) {
2055 ERR("Allocation viewer session");
2056 resp.status = htobe32(LTTNG_VIEWER_CREATE_SESSION_ERR);
2057 goto send_reply;
2058 }
2059
2060 send_reply:
2061 health_code_update();
2062 ret = send_response(conn->sock, &resp, sizeof(resp));
2063 if (ret < 0) {
2064 goto end;
2065 }
2066 health_code_update();
2067 ret = 0;
2068
2069 end:
2070 return ret;
2071 }
2072
2073 /*
2074 * Detach a viewer session.
2075 *
2076 * Return 0 on success or else a negative value.
2077 */
2078 static
2079 int viewer_detach_session(struct relay_connection *conn)
2080 {
2081 int ret;
2082 struct lttng_viewer_detach_session_response response;
2083 struct lttng_viewer_detach_session_request request;
2084 struct relay_session *session = NULL;
2085 uint64_t viewer_session_to_close;
2086
2087 DBG("Viewer detach session received");
2088
2089 assert(conn);
2090
2091 health_code_update();
2092
2093 /* Receive the request from the connected client. */
2094 ret = recv_request(conn->sock, &request, sizeof(request));
2095 if (ret < 0) {
2096 goto end;
2097 }
2098 viewer_session_to_close = be64toh(request.session_id);
2099
2100 if (!conn->viewer_session) {
2101 DBG("Client trying to detach before creating a live viewer session");
2102 response.status = htobe32(LTTNG_VIEWER_DETACH_SESSION_ERR);
2103 goto send_reply;
2104 }
2105
2106 health_code_update();
2107
2108 memset(&response, 0, sizeof(response));
2109 DBG("Detaching from session ID %" PRIu64, viewer_session_to_close);
2110
2111 session = session_get_by_id(be64toh(request.session_id));
2112 if (!session) {
2113 DBG("Relay session %" PRIu64 " not found",
2114 (uint64_t) be64toh(request.session_id));
2115 response.status = htobe32(LTTNG_VIEWER_DETACH_SESSION_UNK);
2116 goto send_reply;
2117 }
2118
2119 ret = viewer_session_is_attached(conn->viewer_session, session);
2120 if (ret != 1) {
2121 DBG("Not attached to this session");
2122 response.status = htobe32(LTTNG_VIEWER_DETACH_SESSION_ERR);
2123 goto send_reply_put;
2124 }
2125
2126 viewer_session_close_one_session(conn->viewer_session, session);
2127 response.status = htobe32(LTTNG_VIEWER_DETACH_SESSION_OK);
2128 DBG("Session %" PRIu64 " detached.", viewer_session_to_close);
2129
2130 send_reply_put:
2131 session_put(session);
2132
2133 send_reply:
2134 health_code_update();
2135 ret = send_response(conn->sock, &response, sizeof(response));
2136 if (ret < 0) {
2137 goto end;
2138 }
2139 health_code_update();
2140 ret = 0;
2141
2142 end:
2143 return ret;
2144 }
2145
2146 /*
2147 * live_relay_unknown_command: send -1 if received unknown command
2148 */
2149 static
2150 void live_relay_unknown_command(struct relay_connection *conn)
2151 {
2152 struct lttcomm_relayd_generic_reply reply;
2153
2154 memset(&reply, 0, sizeof(reply));
2155 reply.ret_code = htobe32(LTTNG_ERR_UNK);
2156 (void) send_response(conn->sock, &reply, sizeof(reply));
2157 }
2158
2159 /*
2160 * Process the commands received on the control socket
2161 */
2162 static
2163 int process_control(struct lttng_viewer_cmd *recv_hdr,
2164 struct relay_connection *conn)
2165 {
2166 int ret = 0;
2167 uint32_t msg_value;
2168
2169 msg_value = be32toh(recv_hdr->cmd);
2170
2171 /*
2172 * Make sure we've done the version check before any command other then a
2173 * new client connection.
2174 */
2175 if (msg_value != LTTNG_VIEWER_CONNECT && !conn->version_check_done) {
2176 ERR("Viewer conn value %" PRIu32 " before version check", msg_value);
2177 ret = -1;
2178 goto end;
2179 }
2180
2181 switch (msg_value) {
2182 case LTTNG_VIEWER_CONNECT:
2183 ret = viewer_connect(conn);
2184 break;
2185 case LTTNG_VIEWER_LIST_SESSIONS:
2186 ret = viewer_list_sessions(conn);
2187 break;
2188 case LTTNG_VIEWER_ATTACH_SESSION:
2189 ret = viewer_attach_session(conn);
2190 break;
2191 case LTTNG_VIEWER_GET_NEXT_INDEX:
2192 ret = viewer_get_next_index(conn);
2193 break;
2194 case LTTNG_VIEWER_GET_PACKET:
2195 ret = viewer_get_packet(conn);
2196 break;
2197 case LTTNG_VIEWER_GET_METADATA:
2198 ret = viewer_get_metadata(conn);
2199 break;
2200 case LTTNG_VIEWER_GET_NEW_STREAMS:
2201 ret = viewer_get_new_streams(conn);
2202 break;
2203 case LTTNG_VIEWER_CREATE_SESSION:
2204 ret = viewer_create_session(conn);
2205 break;
2206 case LTTNG_VIEWER_DETACH_SESSION:
2207 ret = viewer_detach_session(conn);
2208 break;
2209 default:
2210 ERR("Received unknown viewer command (%u)",
2211 be32toh(recv_hdr->cmd));
2212 live_relay_unknown_command(conn);
2213 ret = -1;
2214 goto end;
2215 }
2216
2217 end:
2218 return ret;
2219 }
2220
2221 static
2222 void cleanup_connection_pollfd(struct lttng_poll_event *events, int pollfd)
2223 {
2224 int ret;
2225
2226 (void) lttng_poll_del(events, pollfd);
2227
2228 ret = fd_tracker_close_unsuspendable_fd(the_fd_tracker, &pollfd, 1,
2229 fd_tracker_util_close_fd, NULL);
2230 if (ret < 0) {
2231 ERR("Closing pollfd %d", pollfd);
2232 }
2233 }
2234
2235 /*
2236 * This thread does the actual work
2237 */
2238 static
2239 void *thread_worker(void *data)
2240 {
2241 int ret, err = -1;
2242 uint32_t nb_fd;
2243 struct lttng_poll_event events;
2244 struct lttng_ht *viewer_connections_ht;
2245 struct lttng_ht_iter iter;
2246 struct lttng_viewer_cmd recv_hdr;
2247 struct relay_connection *destroy_conn;
2248
2249 DBG("[thread] Live viewer relay worker started");
2250
2251 rcu_register_thread();
2252
2253 health_register(health_relayd, HEALTH_RELAYD_TYPE_LIVE_WORKER);
2254
2255 if (testpoint(relayd_thread_live_worker)) {
2256 goto error_testpoint;
2257 }
2258
2259 /* table of connections indexed on socket */
2260 viewer_connections_ht = lttng_ht_new(0, LTTNG_HT_TYPE_ULONG);
2261 if (!viewer_connections_ht) {
2262 goto viewer_connections_ht_error;
2263 }
2264
2265 ret = create_named_thread_poll_set(&events, 2,
2266 "Live viewer worker thread epoll");
2267 if (ret < 0) {
2268 goto error_poll_create;
2269 }
2270
2271 ret = lttng_poll_add(&events, live_conn_pipe[0], LPOLLIN | LPOLLRDHUP);
2272 if (ret < 0) {
2273 goto error;
2274 }
2275
2276 restart:
2277 while (1) {
2278 int i;
2279
2280 health_code_update();
2281
2282 /* Infinite blocking call, waiting for transmission */
2283 DBG3("Relayd live viewer worker thread polling...");
2284 health_poll_entry();
2285 ret = lttng_poll_wait(&events, -1);
2286 health_poll_exit();
2287 if (ret < 0) {
2288 /*
2289 * Restart interrupted system call.
2290 */
2291 if (errno == EINTR) {
2292 goto restart;
2293 }
2294 goto error;
2295 }
2296
2297 nb_fd = ret;
2298
2299 /*
2300 * Process control. The control connection is prioritised so we don't
2301 * starve it with high throughput tracing data on the data
2302 * connection.
2303 */
2304 for (i = 0; i < nb_fd; i++) {
2305 /* Fetch once the poll data */
2306 uint32_t revents = LTTNG_POLL_GETEV(&events, i);
2307 int pollfd = LTTNG_POLL_GETFD(&events, i);
2308
2309 health_code_update();
2310
2311 /* Thread quit pipe has been closed. Killing thread. */
2312 ret = check_thread_quit_pipe(pollfd, revents);
2313 if (ret) {
2314 err = 0;
2315 goto exit;
2316 }
2317
2318 /* Inspect the relay conn pipe for new connection. */
2319 if (pollfd == live_conn_pipe[0]) {
2320 if (revents & LPOLLIN) {
2321 struct relay_connection *conn;
2322
2323 ret = lttng_read(live_conn_pipe[0],
2324 &conn, sizeof(conn));
2325 if (ret < 0) {
2326 goto error;
2327 }
2328 ret = lttng_poll_add(&events,
2329 conn->sock->fd,
2330 LPOLLIN | LPOLLRDHUP);
2331 if (ret) {
2332 ERR("Failed to add new live connection file descriptor to poll set");
2333 goto error;
2334 }
2335 connection_ht_add(viewer_connections_ht, conn);
2336 DBG("Connection socket %d added to poll", conn->sock->fd);
2337 } else if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
2338 ERR("Relay live pipe error");
2339 goto error;
2340 } else {
2341 ERR("Unexpected poll events %u for sock %d", revents, pollfd);
2342 goto error;
2343 }
2344 } else {
2345 /* Connection activity. */
2346 struct relay_connection *conn;
2347
2348 conn = connection_get_by_sock(viewer_connections_ht, pollfd);
2349 if (!conn) {
2350 continue;
2351 }
2352
2353 if (revents & LPOLLIN) {
2354 ret = conn->sock->ops->recvmsg(conn->sock, &recv_hdr,
2355 sizeof(recv_hdr), 0);
2356 if (ret <= 0) {
2357 /* Connection closed. */
2358 cleanup_connection_pollfd(&events, pollfd);
2359 /* Put "create" ownership reference. */
2360 connection_put(conn);
2361 DBG("Viewer control conn closed with %d", pollfd);
2362 } else {
2363 ret = process_control(&recv_hdr, conn);
2364 if (ret < 0) {
2365 /* Clear the session on error. */
2366 cleanup_connection_pollfd(&events, pollfd);
2367 /* Put "create" ownership reference. */
2368 connection_put(conn);
2369 DBG("Viewer connection closed with %d", pollfd);
2370 }
2371 }
2372 } else if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
2373 cleanup_connection_pollfd(&events, pollfd);
2374 /* Put "create" ownership reference. */
2375 connection_put(conn);
2376 } else {
2377 ERR("Unexpected poll events %u for sock %d", revents, pollfd);
2378 connection_put(conn);
2379 goto error;
2380 }
2381 /* Put local "get_by_sock" reference. */
2382 connection_put(conn);
2383 }
2384 }
2385 }
2386
2387 exit:
2388 error:
2389 (void) fd_tracker_util_poll_clean(the_fd_tracker, &events);
2390
2391 /* Cleanup remaining connection object. */
2392 rcu_read_lock();
2393 cds_lfht_for_each_entry(viewer_connections_ht->ht, &iter.iter,
2394 destroy_conn,
2395 sock_n.node) {
2396 health_code_update();
2397 connection_put(destroy_conn);
2398 }
2399 rcu_read_unlock();
2400 error_poll_create:
2401 lttng_ht_destroy(viewer_connections_ht);
2402 viewer_connections_ht_error:
2403 /* Close relay conn pipes */
2404 (void) fd_tracker_util_pipe_close(the_fd_tracker, live_conn_pipe);
2405 if (err) {
2406 DBG("Viewer worker thread exited with error");
2407 }
2408 DBG("Viewer worker thread cleanup complete");
2409 error_testpoint:
2410 if (err) {
2411 health_error();
2412 ERR("Health error occurred in %s", __func__);
2413 }
2414 health_unregister(health_relayd);
2415 if (lttng_relay_stop_threads()) {
2416 ERR("Error stopping threads");
2417 }
2418 rcu_unregister_thread();
2419 return NULL;
2420 }
2421
2422 /*
2423 * Create the relay command pipe to wake thread_manage_apps.
2424 * Closed in cleanup().
2425 */
2426 static int create_conn_pipe(void)
2427 {
2428 return fd_tracker_util_pipe_open_cloexec(the_fd_tracker,
2429 "Live connection pipe", live_conn_pipe);
2430 }
2431
2432 int relayd_live_join(void)
2433 {
2434 int ret, retval = 0;
2435 void *status;
2436
2437 ret = pthread_join(live_listener_thread, &status);
2438 if (ret) {
2439 errno = ret;
2440 PERROR("pthread_join live listener");
2441 retval = -1;
2442 }
2443
2444 ret = pthread_join(live_worker_thread, &status);
2445 if (ret) {
2446 errno = ret;
2447 PERROR("pthread_join live worker");
2448 retval = -1;
2449 }
2450
2451 ret = pthread_join(live_dispatcher_thread, &status);
2452 if (ret) {
2453 errno = ret;
2454 PERROR("pthread_join live dispatcher");
2455 retval = -1;
2456 }
2457
2458 cleanup_relayd_live();
2459
2460 return retval;
2461 }
2462
2463 /*
2464 * main
2465 */
2466 int relayd_live_create(struct lttng_uri *uri)
2467 {
2468 int ret = 0, retval = 0;
2469 void *status;
2470 int is_root;
2471
2472 if (!uri) {
2473 retval = -1;
2474 goto exit_init_data;
2475 }
2476 live_uri = uri;
2477
2478 /* Check if daemon is UID = 0 */
2479 is_root = !getuid();
2480
2481 if (!is_root) {
2482 if (live_uri->port < 1024) {
2483 ERR("Need to be root to use ports < 1024");
2484 retval = -1;
2485 goto exit_init_data;
2486 }
2487 }
2488
2489 /* Setup the thread apps communication pipe. */
2490 if (create_conn_pipe()) {
2491 retval = -1;
2492 goto exit_init_data;
2493 }
2494
2495 /* Init relay command queue. */
2496 cds_wfcq_init(&viewer_conn_queue.head, &viewer_conn_queue.tail);
2497
2498 /* Set up max poll set size */
2499 if (lttng_poll_set_max_size()) {
2500 retval = -1;
2501 goto exit_init_data;
2502 }
2503
2504 /* Setup the dispatcher thread */
2505 ret = pthread_create(&live_dispatcher_thread, default_pthread_attr(),
2506 thread_dispatcher, (void *) NULL);
2507 if (ret) {
2508 errno = ret;
2509 PERROR("pthread_create viewer dispatcher");
2510 retval = -1;
2511 goto exit_dispatcher_thread;
2512 }
2513
2514 /* Setup the worker thread */
2515 ret = pthread_create(&live_worker_thread, default_pthread_attr(),
2516 thread_worker, NULL);
2517 if (ret) {
2518 errno = ret;
2519 PERROR("pthread_create viewer worker");
2520 retval = -1;
2521 goto exit_worker_thread;
2522 }
2523
2524 /* Setup the listener thread */
2525 ret = pthread_create(&live_listener_thread, default_pthread_attr(),
2526 thread_listener, (void *) NULL);
2527 if (ret) {
2528 errno = ret;
2529 PERROR("pthread_create viewer listener");
2530 retval = -1;
2531 goto exit_listener_thread;
2532 }
2533
2534 /*
2535 * All OK, started all threads.
2536 */
2537 return retval;
2538
2539 /*
2540 * Join on the live_listener_thread should anything be added after
2541 * the live_listener thread's creation.
2542 */
2543
2544 exit_listener_thread:
2545
2546 ret = pthread_join(live_worker_thread, &status);
2547 if (ret) {
2548 errno = ret;
2549 PERROR("pthread_join live worker");
2550 retval = -1;
2551 }
2552 exit_worker_thread:
2553
2554 ret = pthread_join(live_dispatcher_thread, &status);
2555 if (ret) {
2556 errno = ret;
2557 PERROR("pthread_join live dispatcher");
2558 retval = -1;
2559 }
2560 exit_dispatcher_thread:
2561
2562 exit_init_data:
2563 cleanup_relayd_live();
2564
2565 return retval;
2566 }
This page took 0.140652 seconds and 3 git commands to generate.