c312e78e976e89bb6949e0aee54cf48e2bdca99c
[lttng-tools.git] / src / bin / lttng-relayd / live.c
1 /*
2 * Copyright (C) 2013 - Julien Desfossez <jdesfossez@efficios.com>
3 * David Goulet <dgoulet@efficios.com>
4 *
5 * This program is free software; you can redistribute it and/or modify
6 * it under the terms of the GNU General Public License, version 2 only,
7 * as published by the Free Software Foundation.
8 *
9 * This program is distributed in the hope that it will be useful, but WITHOUT
10 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
11 * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
12 * more details.
13 *
14 * You should have received a copy of the GNU General Public License along
15 * with this program; if not, write to the Free Software Foundation, Inc.,
16 * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
17 */
18
19 #define _GNU_SOURCE
20 #include <getopt.h>
21 #include <grp.h>
22 #include <limits.h>
23 #include <pthread.h>
24 #include <signal.h>
25 #include <stdio.h>
26 #include <stdlib.h>
27 #include <string.h>
28 #include <sys/mman.h>
29 #include <sys/mount.h>
30 #include <sys/resource.h>
31 #include <sys/socket.h>
32 #include <sys/stat.h>
33 #include <sys/types.h>
34 #include <sys/wait.h>
35 #include <inttypes.h>
36 #include <urcu/futex.h>
37 #include <urcu/uatomic.h>
38 #include <unistd.h>
39 #include <fcntl.h>
40 #include <config.h>
41
42 #include <lttng/lttng.h>
43 #include <common/common.h>
44 #include <common/compat/poll.h>
45 #include <common/compat/socket.h>
46 #include <common/defaults.h>
47 #include <common/futex.h>
48 #include <common/sessiond-comm/sessiond-comm.h>
49 #include <common/sessiond-comm/inet.h>
50 #include <common/sessiond-comm/relayd.h>
51 #include <common/uri.h>
52 #include <common/utils.h>
53
54 #include "cmd.h"
55 #include "live.h"
56 #include "lttng-relayd.h"
57 #include "lttng-viewer.h"
58 #include "utils.h"
59 #include "health-relayd.h"
60
61 static struct lttng_uri *live_uri;
62
63 /*
64 * Quit pipe for all threads. This permits a single cancellation point
65 * for all threads when receiving an event on the pipe.
66 */
67 static int live_thread_quit_pipe[2] = { -1, -1 };
68
69 /*
70 * This pipe is used to inform the worker thread that a command is queued and
71 * ready to be processed.
72 */
73 static int live_relay_cmd_pipe[2] = { -1, -1 };
74
75 /* Shared between threads */
76 static int live_dispatch_thread_exit;
77
78 static pthread_t live_listener_thread;
79 static pthread_t live_dispatcher_thread;
80 static pthread_t live_worker_thread;
81
82 /*
83 * Relay command queue.
84 *
85 * The live_thread_listener and live_thread_dispatcher communicate with this
86 * queue.
87 */
88 static struct relay_cmd_queue viewer_cmd_queue;
89
90 static uint64_t last_relay_viewer_session_id;
91
92 /*
93 * Cleanup the daemon
94 */
95 static
96 void cleanup(void)
97 {
98 DBG("Cleaning up");
99
100 free(live_uri);
101 }
102
103 /*
104 * Write to writable pipe used to notify a thread.
105 */
106 static
107 int notify_thread_pipe(int wpipe)
108 {
109 ssize_t ret;
110
111 ret = lttng_write(wpipe, "!", 1);
112 if (ret < 1) {
113 PERROR("write poll pipe");
114 }
115
116 return (int) ret;
117 }
118
119 /*
120 * Stop all threads by closing the thread quit pipe.
121 */
122 static
123 void stop_threads(void)
124 {
125 int ret;
126
127 /* Stopping all threads */
128 DBG("Terminating all live threads");
129 ret = notify_thread_pipe(live_thread_quit_pipe[1]);
130 if (ret < 0) {
131 ERR("write error on thread quit pipe");
132 }
133
134 /* Dispatch thread */
135 CMM_STORE_SHARED(live_dispatch_thread_exit, 1);
136 futex_nto1_wake(&viewer_cmd_queue.futex);
137 }
138
139 /*
140 * Create a poll set with O_CLOEXEC and add the thread quit pipe to the set.
141 */
142 static
143 int create_thread_poll_set(struct lttng_poll_event *events, int size)
144 {
145 int ret;
146
147 if (events == NULL || size == 0) {
148 ret = -1;
149 goto error;
150 }
151
152 ret = lttng_poll_create(events, size, LTTNG_CLOEXEC);
153 if (ret < 0) {
154 goto error;
155 }
156
157 /* Add quit pipe */
158 ret = lttng_poll_add(events, live_thread_quit_pipe[0], LPOLLIN);
159 if (ret < 0) {
160 goto error;
161 }
162
163 return 0;
164
165 error:
166 return ret;
167 }
168
169 /*
170 * Check if the thread quit pipe was triggered.
171 *
172 * Return 1 if it was triggered else 0;
173 */
174 static
175 int check_thread_quit_pipe(int fd, uint32_t events)
176 {
177 if (fd == live_thread_quit_pipe[0] && (events & LPOLLIN)) {
178 return 1;
179 }
180
181 return 0;
182 }
183
184 /*
185 * Create and init socket from uri.
186 */
187 static
188 struct lttcomm_sock *init_socket(struct lttng_uri *uri)
189 {
190 int ret;
191 struct lttcomm_sock *sock = NULL;
192
193 sock = lttcomm_alloc_sock_from_uri(uri);
194 if (sock == NULL) {
195 ERR("Allocating socket");
196 goto error;
197 }
198
199 ret = lttcomm_create_sock(sock);
200 if (ret < 0) {
201 goto error;
202 }
203 DBG("Listening on sock %d for live", sock->fd);
204
205 ret = sock->ops->bind(sock);
206 if (ret < 0) {
207 goto error;
208 }
209
210 ret = sock->ops->listen(sock, -1);
211 if (ret < 0) {
212 goto error;
213
214 }
215
216 return sock;
217
218 error:
219 if (sock) {
220 lttcomm_destroy_sock(sock);
221 }
222 return NULL;
223 }
224
225 /*
226 * This thread manages the listening for new connections on the network
227 */
228 static
229 void *thread_listener(void *data)
230 {
231 int i, ret, pollfd, err = -1;
232 int val = 1;
233 uint32_t revents, nb_fd;
234 struct lttng_poll_event events;
235 struct lttcomm_sock *live_control_sock;
236
237 DBG("[thread] Relay live listener started");
238
239 health_register(health_relayd, HEALTH_RELAYD_TYPE_LIVE_LISTENER);
240
241 health_code_update();
242
243 live_control_sock = init_socket(live_uri);
244 if (!live_control_sock) {
245 goto error_sock_control;
246 }
247
248 /*
249 * Pass 3 as size here for the thread quit pipe, control and data socket.
250 */
251 ret = create_thread_poll_set(&events, 2);
252 if (ret < 0) {
253 goto error_create_poll;
254 }
255
256 /* Add the control socket */
257 ret = lttng_poll_add(&events, live_control_sock->fd, LPOLLIN | LPOLLRDHUP);
258 if (ret < 0) {
259 goto error_poll_add;
260 }
261
262 while (1) {
263 health_code_update();
264
265 DBG("Listener accepting live viewers connections");
266
267 restart:
268 health_poll_entry();
269 ret = lttng_poll_wait(&events, -1);
270 health_poll_exit();
271 if (ret < 0) {
272 /*
273 * Restart interrupted system call.
274 */
275 if (errno == EINTR) {
276 goto restart;
277 }
278 goto error;
279 }
280 nb_fd = ret;
281
282 DBG("Relay new viewer connection received");
283 for (i = 0; i < nb_fd; i++) {
284 health_code_update();
285
286 /* Fetch once the poll data */
287 revents = LTTNG_POLL_GETEV(&events, i);
288 pollfd = LTTNG_POLL_GETFD(&events, i);
289
290 /* Thread quit pipe has been closed. Killing thread. */
291 ret = check_thread_quit_pipe(pollfd, revents);
292 if (ret) {
293 err = 0;
294 goto exit;
295 }
296
297 if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
298 ERR("socket poll error");
299 goto error;
300 } else if (revents & LPOLLIN) {
301 /*
302 * Get allocated in this thread, enqueued to a global queue,
303 * dequeued and freed in the worker thread.
304 */
305 struct relay_command *relay_cmd;
306 struct lttcomm_sock *newsock;
307
308 relay_cmd = zmalloc(sizeof(*relay_cmd));
309 if (!relay_cmd) {
310 PERROR("relay command zmalloc");
311 goto error;
312 }
313
314 assert(pollfd == live_control_sock->fd);
315 newsock = live_control_sock->ops->accept(live_control_sock);
316 if (!newsock) {
317 PERROR("accepting control sock");
318 free(relay_cmd);
319 goto error;
320 }
321 DBG("Relay viewer connection accepted socket %d", newsock->fd);
322 ret = setsockopt(newsock->fd, SOL_SOCKET, SO_REUSEADDR, &val,
323 sizeof(int));
324 if (ret < 0) {
325 PERROR("setsockopt inet");
326 lttcomm_destroy_sock(newsock);
327 free(relay_cmd);
328 goto error;
329 }
330 relay_cmd->sock = newsock;
331
332 /*
333 * Lock free enqueue the request.
334 */
335 cds_wfq_enqueue(&viewer_cmd_queue.queue, &relay_cmd->node);
336
337 /*
338 * Wake the dispatch queue futex. Implicit memory
339 * barrier with the exchange in cds_wfq_enqueue.
340 */
341 futex_nto1_wake(&viewer_cmd_queue.futex);
342 }
343 }
344 }
345
346 exit:
347 error:
348 error_poll_add:
349 lttng_poll_clean(&events);
350 error_create_poll:
351 if (live_control_sock->fd >= 0) {
352 ret = live_control_sock->ops->close(live_control_sock);
353 if (ret) {
354 PERROR("close");
355 }
356 }
357 lttcomm_destroy_sock(live_control_sock);
358 error_sock_control:
359 if (err) {
360 health_error();
361 DBG("Live viewer listener thread exited with error");
362 }
363 health_unregister(health_relayd);
364 DBG("Live viewer listener thread cleanup complete");
365 stop_threads();
366 return NULL;
367 }
368
369 /*
370 * This thread manages the dispatching of the requests to worker threads
371 */
372 static
373 void *thread_dispatcher(void *data)
374 {
375 int err = -1;
376 ssize_t ret;
377 struct cds_wfq_node *node;
378 struct relay_command *relay_cmd = NULL;
379
380 DBG("[thread] Live viewer relay dispatcher started");
381
382 health_register(health_relayd, HEALTH_RELAYD_TYPE_LIVE_DISPATCHER);
383
384 health_code_update();
385
386 while (!CMM_LOAD_SHARED(live_dispatch_thread_exit)) {
387 health_code_update();
388
389 /* Atomically prepare the queue futex */
390 futex_nto1_prepare(&viewer_cmd_queue.futex);
391
392 do {
393 health_code_update();
394
395 /* Dequeue commands */
396 node = cds_wfq_dequeue_blocking(&viewer_cmd_queue.queue);
397 if (node == NULL) {
398 DBG("Woken up but nothing in the live-viewer "
399 "relay command queue");
400 /* Continue thread execution */
401 break;
402 }
403
404 relay_cmd = caa_container_of(node, struct relay_command, node);
405 DBG("Dispatching viewer request waiting on sock %d",
406 relay_cmd->sock->fd);
407
408 /*
409 * Inform worker thread of the new request. This call is blocking
410 * so we can be assured that the data will be read at some point in
411 * time or wait to the end of the world :)
412 */
413 ret = lttng_write(live_relay_cmd_pipe[1], relay_cmd,
414 sizeof(*relay_cmd));
415 free(relay_cmd);
416 if (ret < sizeof(struct relay_command)) {
417 PERROR("write cmd pipe");
418 goto error;
419 }
420 } while (node != NULL);
421
422 /* Futex wait on queue. Blocking call on futex() */
423 health_poll_entry();
424 futex_nto1_wait(&viewer_cmd_queue.futex);
425 health_poll_exit();
426 }
427
428 /* Normal exit, no error */
429 err = 0;
430
431 error:
432 if (err) {
433 health_error();
434 ERR("Health error occurred in %s", __func__);
435 }
436 health_unregister(health_relayd);
437 DBG("Live viewer dispatch thread dying");
438 stop_threads();
439 return NULL;
440 }
441
442 /*
443 * Establish connection with the viewer and check the versions.
444 *
445 * Return 0 on success or else negative value.
446 */
447 static
448 int viewer_connect(struct relay_command *cmd)
449 {
450 int ret;
451 struct lttng_viewer_connect reply, msg;
452
453 assert(cmd);
454
455 cmd->version_check_done = 1;
456
457 health_code_update();
458
459 /* Get version from the other side. */
460 ret = cmd->sock->ops->recvmsg(cmd->sock, &msg, sizeof(msg), 0);
461 if (ret < 0 || ret != sizeof(msg)) {
462 if (ret == 0) {
463 /* Orderly shutdown. Not necessary to print an error. */
464 DBG("Socket %d did an orderly shutdown", cmd->sock->fd);
465 } else {
466 ERR("Relay failed to receive the version values.");
467 }
468 ret = -1;
469 goto end;
470 }
471
472 health_code_update();
473
474 reply.major = RELAYD_VERSION_COMM_MAJOR;
475 reply.minor = RELAYD_VERSION_COMM_MINOR;
476
477 /* Major versions must be the same */
478 if (reply.major != be32toh(msg.major)) {
479 DBG("Incompatible major versions (%u vs %u)", reply.major,
480 be32toh(msg.major));
481 ret = -1;
482 goto end;
483 }
484
485 cmd->major = reply.major;
486 /* We adapt to the lowest compatible version */
487 if (reply.minor <= be32toh(msg.minor)) {
488 cmd->minor = reply.minor;
489 } else {
490 cmd->minor = be32toh(msg.minor);
491 }
492
493 if (be32toh(msg.type) == VIEWER_CLIENT_COMMAND) {
494 cmd->type = RELAY_VIEWER_COMMAND;
495 } else if (be32toh(msg.type) == VIEWER_CLIENT_NOTIFICATION) {
496 cmd->type = RELAY_VIEWER_NOTIFICATION;
497 } else {
498 ERR("Unknown connection type : %u", be32toh(msg.type));
499 ret = -1;
500 goto end;
501 }
502
503 reply.major = htobe32(reply.major);
504 reply.minor = htobe32(reply.minor);
505 if (cmd->type == RELAY_VIEWER_COMMAND) {
506 reply.viewer_session_id = htobe64(++last_relay_viewer_session_id);
507 }
508
509 health_code_update();
510
511 ret = cmd->sock->ops->sendmsg(cmd->sock, &reply,
512 sizeof(struct lttng_viewer_connect), 0);
513 if (ret < 0) {
514 ERR("Relay sending version");
515 }
516
517 health_code_update();
518
519 DBG("Version check done using protocol %u.%u", cmd->major, cmd->minor);
520 ret = 0;
521
522 end:
523 return ret;
524 }
525
526 /*
527 * Send the viewer the list of current sessions.
528 *
529 * Return 0 on success or else a negative value.
530 */
531 static
532 int viewer_list_sessions(struct relay_command *cmd,
533 struct lttng_ht *sessions_ht)
534 {
535 int ret;
536 struct lttng_viewer_list_sessions session_list;
537 unsigned long count;
538 long approx_before, approx_after;
539 struct lttng_ht_node_ulong *node;
540 struct lttng_ht_iter iter;
541 struct lttng_viewer_session send_session;
542 struct relay_session *session;
543
544 DBG("List sessions received");
545
546 if (cmd->version_check_done == 0) {
547 ERR("Trying to list sessions before version check");
548 ret = -1;
549 goto end_no_session;
550 }
551
552 rcu_read_lock();
553 cds_lfht_count_nodes(sessions_ht->ht, &approx_before, &count, &approx_after);
554 session_list.sessions_count = htobe32(count);
555
556 health_code_update();
557
558 ret = cmd->sock->ops->sendmsg(cmd->sock, &session_list,
559 sizeof(session_list), 0);
560 if (ret < 0) {
561 ERR("Relay sending sessions list");
562 goto end_unlock;
563 }
564
565 health_code_update();
566
567 cds_lfht_for_each_entry(sessions_ht->ht, &iter.iter, node, node) {
568 health_code_update();
569
570 node = lttng_ht_iter_get_node_ulong(&iter);
571 if (!node) {
572 goto end_unlock;
573 }
574 session = caa_container_of(node, struct relay_session, session_n);
575
576 strncpy(send_session.session_name, session->session_name,
577 sizeof(send_session.session_name));
578 strncpy(send_session.hostname, session->hostname,
579 sizeof(send_session.hostname));
580 send_session.id = htobe64(session->id);
581 send_session.live_timer = htobe32(session->live_timer);
582 send_session.clients = htobe32(session->viewer_attached);
583 send_session.streams = htobe32(session->stream_count);
584
585 health_code_update();
586
587 ret = cmd->sock->ops->sendmsg(cmd->sock, &send_session,
588 sizeof(send_session), 0);
589 if (ret < 0) {
590 ERR("Relay sending session info");
591 goto end_unlock;
592 }
593 }
594 health_code_update();
595
596 rcu_read_unlock();
597 ret = 0;
598 goto end;
599
600 end_unlock:
601 rcu_read_unlock();
602
603 end:
604 end_no_session:
605 return ret;
606 }
607
608 /*
609 * Open index file using a given viewer stream.
610 *
611 * Return 0 on success or else a negative value.
612 */
613 static int open_index(struct relay_viewer_stream *stream)
614 {
615 int ret;
616 char fullpath[PATH_MAX];
617 struct ctf_packet_index_file_hdr hdr;
618
619 if (stream->tracefile_count > 0) {
620 ret = snprintf(fullpath, sizeof(fullpath), "%s/" DEFAULT_INDEX_DIR "/%s_%"
621 PRIu64 DEFAULT_INDEX_FILE_SUFFIX, stream->path_name,
622 stream->channel_name, stream->tracefile_count_current);
623 } else {
624 ret = snprintf(fullpath, sizeof(fullpath), "%s/" DEFAULT_INDEX_DIR "/%s"
625 DEFAULT_INDEX_FILE_SUFFIX, stream->path_name,
626 stream->channel_name);
627 }
628 if (ret < 0) {
629 PERROR("snprintf index path");
630 goto error;
631 }
632
633 DBG("Opening index file %s in read only", fullpath);
634 ret = open(fullpath, O_RDONLY);
635 if (ret < 0) {
636 if (errno == ENOENT) {
637 ret = -ENOENT;
638 goto error;
639 } else {
640 PERROR("opening index in read-only");
641 }
642 goto error;
643 }
644 stream->index_read_fd = ret;
645 DBG("Opening index file %s in read only, (fd: %d)", fullpath, ret);
646
647 ret = lttng_read(stream->index_read_fd, &hdr, sizeof(hdr));
648 if (ret < sizeof(hdr)) {
649 PERROR("Reading index header");
650 goto error;
651 }
652 if (be32toh(hdr.magic) != CTF_INDEX_MAGIC) {
653 ERR("Invalid header magic");
654 ret = -1;
655 goto error;
656 }
657 if (be32toh(hdr.index_major) != CTF_INDEX_MAJOR ||
658 be32toh(hdr.index_minor) != CTF_INDEX_MINOR) {
659 ERR("Invalid header version");
660 ret = -1;
661 goto error;
662 }
663 ret = 0;
664
665 error:
666 return ret;
667 }
668
669 /*
670 * Allocate and init a new viewer_stream.
671 *
672 * Copies the values from the stream passed in parameter and insert the new
673 * stream in the viewer_streams_ht.
674 *
675 * MUST be called with rcu_read_lock held.
676 *
677 * Returns 0 on success or a negative value on error.
678 */
679 static
680 int init_viewer_stream(struct relay_stream *stream, int seek_last)
681 {
682 int ret;
683 struct relay_viewer_stream *viewer_stream;
684
685 assert(stream);
686
687 viewer_stream = zmalloc(sizeof(*viewer_stream));
688 if (!viewer_stream) {
689 PERROR("relay viewer stream zmalloc");
690 ret = -1;
691 goto error;
692 }
693 viewer_stream->session_id = stream->session->id;
694 viewer_stream->stream_handle = stream->stream_handle;
695 viewer_stream->path_name = strndup(stream->path_name,
696 LTTNG_VIEWER_PATH_MAX);
697 viewer_stream->channel_name = strndup(stream->channel_name,
698 LTTNG_VIEWER_NAME_MAX);
699 viewer_stream->tracefile_count = stream->tracefile_count;
700 viewer_stream->metadata_flag = stream->metadata_flag;
701 viewer_stream->tracefile_count_last = -1ULL;
702 if (seek_last) {
703 viewer_stream->tracefile_count_current =
704 stream->tracefile_count_current;
705 } else {
706 viewer_stream->tracefile_count_current =
707 stream->oldest_tracefile_id;
708 }
709
710 viewer_stream->ctf_trace = stream->ctf_trace;
711 if (viewer_stream->metadata_flag) {
712 viewer_stream->ctf_trace->viewer_metadata_stream =
713 viewer_stream;
714 }
715 uatomic_inc(&viewer_stream->ctf_trace->refcount);
716
717 lttng_ht_node_init_u64(&viewer_stream->stream_n, stream->stream_handle);
718 lttng_ht_add_unique_u64(viewer_streams_ht, &viewer_stream->stream_n);
719
720 viewer_stream->index_read_fd = -1;
721 viewer_stream->read_fd = -1;
722
723 /*
724 * This is to avoid a race between the initialization of this object and
725 * the close of the given stream. If the stream is unable to find this
726 * viewer stream when closing, this copy will at least take the latest
727 * value.
728 * We also need that for the seek_last.
729 */
730 viewer_stream->total_index_received = stream->total_index_received;
731
732 /*
733 * If we never received an index for the current stream, delay
734 * the opening of the index, otherwise open it right now.
735 */
736 if (viewer_stream->tracefile_count_current ==
737 stream->tracefile_count_current &&
738 viewer_stream->total_index_received == 0) {
739 viewer_stream->index_read_fd = -1;
740 } else {
741 ret = open_index(viewer_stream);
742 if (ret < 0) {
743 goto error;
744 }
745 }
746
747 if (seek_last && viewer_stream->index_read_fd > 0) {
748 ret = lseek(viewer_stream->index_read_fd,
749 viewer_stream->total_index_received *
750 sizeof(struct ctf_packet_index),
751 SEEK_CUR);
752 if (ret < 0) {
753 goto error;
754 }
755 viewer_stream->last_sent_index =
756 viewer_stream->total_index_received;
757 }
758
759 ret = 0;
760
761 error:
762 return ret;
763 }
764
765 /*
766 * Rotate a stream to the next tracefile.
767 *
768 * Returns 0 on success, 1 on EOF, a negative value on error.
769 */
770 static
771 int rotate_viewer_stream(struct relay_viewer_stream *viewer_stream,
772 struct relay_stream *stream)
773 {
774 int ret;
775 uint64_t tracefile_id;
776
777 assert(viewer_stream);
778
779 tracefile_id = (viewer_stream->tracefile_count_current + 1) %
780 viewer_stream->tracefile_count;
781 /*
782 * Detect the last tracefile to open.
783 */
784 if (viewer_stream->tracefile_count_last != -1ULL &&
785 viewer_stream->tracefile_count_last ==
786 viewer_stream->tracefile_count_current) {
787 ret = 1;
788 goto end;
789 }
790
791 if (stream) {
792 pthread_mutex_lock(&stream->viewer_stream_rotation_lock);
793 }
794 /*
795 * The writer and the reader are not working in the same
796 * tracefile, we can read up to EOF, we don't care about the
797 * total_index_received.
798 */
799 if (!stream || (stream->tracefile_count_current != tracefile_id)) {
800 viewer_stream->close_write_flag = 1;
801 } else {
802 /*
803 * We are opening a file that is still open in write, make
804 * sure we limit our reading to the number of indexes
805 * received.
806 */
807 viewer_stream->close_write_flag = 0;
808 if (stream) {
809 viewer_stream->total_index_received =
810 stream->total_index_received;
811 }
812 }
813 viewer_stream->tracefile_count_current = tracefile_id;
814
815 ret = close(viewer_stream->index_read_fd);
816 if (ret < 0) {
817 PERROR("close index file %d",
818 viewer_stream->index_read_fd);
819 }
820 viewer_stream->index_read_fd = -1;
821 ret = close(viewer_stream->read_fd);
822 if (ret < 0) {
823 PERROR("close tracefile %d",
824 viewer_stream->read_fd);
825 }
826 viewer_stream->read_fd = -1;
827
828 pthread_mutex_lock(&viewer_stream->overwrite_lock);
829 viewer_stream->abort_flag = 0;
830 pthread_mutex_unlock(&viewer_stream->overwrite_lock);
831
832 viewer_stream->index_read_fd = -1;
833 viewer_stream->read_fd = -1;
834
835 if (stream) {
836 pthread_mutex_unlock(&stream->viewer_stream_rotation_lock);
837 }
838 ret = open_index(viewer_stream);
839 if (ret < 0) {
840 goto error;
841 }
842
843 ret = 0;
844
845 end:
846 error:
847 return ret;
848 }
849
850 /*
851 * Send the viewer the list of current sessions.
852 */
853 static
854 int viewer_attach_session(struct relay_command *cmd,
855 struct lttng_ht *sessions_ht)
856 {
857 int ret, send_streams = 0, nb_streams = 0;
858 struct lttng_viewer_attach_session_request request;
859 struct lttng_viewer_attach_session_response response;
860 struct lttng_viewer_stream send_stream;
861 struct relay_stream *stream;
862 struct relay_viewer_stream *viewer_stream;
863 struct lttng_ht_node_ulong *node;
864 struct lttng_ht_node_u64 *node64;
865 struct lttng_ht_iter iter;
866 struct relay_session *session;
867 int seek_last = 0;
868
869 assert(cmd);
870 assert(sessions_ht);
871
872 DBG("Attach session received");
873
874 if (cmd->version_check_done == 0) {
875 ERR("Trying to attach session before version check");
876 ret = -1;
877 goto end_no_session;
878 }
879
880 health_code_update();
881
882 ret = cmd->sock->ops->recvmsg(cmd->sock, &request, sizeof(request), 0);
883 if (ret < 0 || ret != sizeof(request)) {
884 if (ret == 0) {
885 /* Orderly shutdown. Not necessary to print an error. */
886 DBG("Socket %d did an orderly shutdown", cmd->sock->fd);
887 } else {
888 ERR("Relay failed to receive the attach parameters.");
889 }
890 ret = -1;
891 goto error;
892 }
893
894 health_code_update();
895
896 rcu_read_lock();
897 lttng_ht_lookup(sessions_ht,
898 (void *)((unsigned long) be64toh(request.session_id)), &iter);
899 node = lttng_ht_iter_get_node_ulong(&iter);
900 if (node == NULL) {
901 DBG("Relay session %" PRIu64 " not found",
902 be64toh(request.session_id));
903 response.status = htobe32(VIEWER_ATTACH_UNK);
904 goto send_reply;
905 }
906
907 session = caa_container_of(node, struct relay_session, session_n);
908 if (cmd->session_id == session->id) {
909 /* Same viewer already attached, just send the stream list. */
910 send_streams = 1;
911 response.status = htobe32(VIEWER_ATTACH_OK);
912 } else if (session->viewer_attached != 0) {
913 DBG("Already a viewer attached");
914 response.status = htobe32(VIEWER_ATTACH_ALREADY);
915 goto send_reply;
916 } else if (session->live_timer == 0) {
917 DBG("Not live session");
918 response.status = htobe32(VIEWER_ATTACH_NOT_LIVE);
919 goto send_reply;
920 } else {
921 session->viewer_attached++;
922 send_streams = 1;
923 response.status = htobe32(VIEWER_ATTACH_OK);
924 cmd->session_id = session->id;
925 cmd->session = session;
926 }
927
928 switch (be32toh(request.seek)) {
929 case VIEWER_SEEK_BEGINNING:
930 /* Default behaviour. */
931 break;
932 case VIEWER_SEEK_LAST:
933 seek_last = 1;
934 break;
935 default:
936 ERR("Wrong seek parameter");
937 response.status = htobe32(VIEWER_ATTACH_SEEK_ERR);
938 send_streams = 0;
939 goto send_reply;
940 }
941
942 if (send_streams) {
943 /* We should only be there if we have a session to attach to. */
944 assert(session);
945
946 /*
947 * Fill the viewer_streams_ht to count the number of streams
948 * ready to be sent and avoid concurrency issues on the
949 * relay_streams_ht and don't rely on a total session stream count.
950 */
951 cds_lfht_for_each_entry(relay_streams_ht->ht, &iter.iter, node, node) {
952 struct relay_viewer_stream *vstream;
953
954 health_code_update();
955
956 node = lttng_ht_iter_get_node_ulong(&iter);
957 if (!node) {
958 continue;
959 }
960 stream = caa_container_of(node, struct relay_stream, stream_n);
961 if (stream->session != cmd->session) {
962 continue;
963 }
964
965 /*
966 * Don't send streams with no ctf_trace, they are not ready to be
967 * read.
968 */
969 if (!stream->ctf_trace) {
970 continue;
971 }
972
973 vstream = live_find_viewer_stream_by_id(stream->stream_handle);
974 if (!vstream) {
975 ret = init_viewer_stream(stream, seek_last);
976 if (ret < 0) {
977 goto end_unlock;
978 }
979 }
980 nb_streams++;
981 }
982 response.streams_count = htobe32(nb_streams);
983 }
984
985 send_reply:
986 health_code_update();
987 ret = cmd->sock->ops->sendmsg(cmd->sock, &response, sizeof(response), 0);
988 if (ret < 0) {
989 ERR("Relay sending viewer attach response");
990 goto end_unlock;
991 }
992 health_code_update();
993
994 /*
995 * Unknown or empty session, just return gracefully, the viewer knows what
996 * is happening.
997 */
998 if (!send_streams || !nb_streams) {
999 ret = 0;
1000 goto end_unlock;
1001 }
1002
1003 /* We should only be there if we have a session to attach to. */
1004 assert(session);
1005 cds_lfht_for_each_entry(viewer_streams_ht->ht, &iter.iter, node, node) {
1006 health_code_update();
1007
1008 node64 = lttng_ht_iter_get_node_u64(&iter);
1009 if (!node64) {
1010 continue;
1011 }
1012 viewer_stream = caa_container_of(node64, struct relay_viewer_stream,
1013 stream_n);
1014 if (viewer_stream->session_id != cmd->session->id) {
1015 continue;
1016 }
1017
1018 send_stream.id = htobe64(viewer_stream->stream_handle);
1019 send_stream.ctf_trace_id = htobe64(viewer_stream->ctf_trace->id);
1020 send_stream.metadata_flag = htobe32(viewer_stream->metadata_flag);
1021 strncpy(send_stream.path_name, viewer_stream->path_name,
1022 sizeof(send_stream.path_name));
1023 strncpy(send_stream.channel_name, viewer_stream->channel_name,
1024 sizeof(send_stream.channel_name));
1025
1026 ret = cmd->sock->ops->sendmsg(cmd->sock, &send_stream,
1027 sizeof(send_stream), 0);
1028 if (ret < 0) {
1029 ERR("Relay sending stream %" PRIu64, viewer_stream->stream_handle);
1030 goto end_unlock;
1031 }
1032 DBG("Sent stream %" PRIu64 " to viewer", viewer_stream->stream_handle);
1033 }
1034 ret = 0;
1035
1036 end_unlock:
1037 rcu_read_unlock();
1038 end_no_session:
1039 error:
1040 return ret;
1041 }
1042
1043 /*
1044 * Get viewer stream from stream id.
1045 *
1046 * RCU read side lock MUST be acquired.
1047 */
1048 struct relay_viewer_stream *live_find_viewer_stream_by_id(uint64_t stream_id)
1049 {
1050 struct lttng_ht_node_u64 *node;
1051 struct lttng_ht_iter iter;
1052 struct relay_viewer_stream *stream = NULL;
1053
1054 lttng_ht_lookup(viewer_streams_ht, &stream_id, &iter);
1055 node = lttng_ht_iter_get_node_u64(&iter);
1056 if (node == NULL) {
1057 DBG("Relay viewer stream %" PRIu64 " not found", stream_id);
1058 goto end;
1059 }
1060 stream = caa_container_of(node, struct relay_viewer_stream, stream_n);
1061
1062 end:
1063 return stream;
1064 }
1065
1066 static
1067 void deferred_free_viewer_stream(struct rcu_head *head)
1068 {
1069 struct relay_viewer_stream *stream =
1070 caa_container_of(head, struct relay_viewer_stream, rcu_node);
1071
1072 free(stream->path_name);
1073 free(stream->channel_name);
1074 free(stream);
1075 }
1076
1077 static
1078 void delete_viewer_stream(struct relay_viewer_stream *vstream)
1079 {
1080 int delret;
1081 struct lttng_ht_iter iter;
1082
1083 iter.iter.node = &vstream->stream_n.node;
1084 delret = lttng_ht_del(viewer_streams_ht, &iter);
1085 assert(!delret);
1086 }
1087
1088 static
1089 void destroy_viewer_stream(struct relay_viewer_stream *vstream)
1090 {
1091 unsigned long ret_ref;
1092 int ret;
1093
1094 assert(vstream);
1095 ret_ref = uatomic_add_return(&vstream->ctf_trace->refcount, -1);
1096 assert(ret_ref >= 0);
1097
1098 if (vstream->read_fd >= 0) {
1099 ret = close(vstream->read_fd);
1100 if (ret < 0) {
1101 PERROR("close read_fd");
1102 }
1103 }
1104 if (vstream->index_read_fd >= 0) {
1105 ret = close(vstream->index_read_fd);
1106 if (ret < 0) {
1107 PERROR("close index_read_fd");
1108 }
1109 }
1110
1111 /*
1112 * If the only stream left in the HT is the metadata stream,
1113 * we need to remove it because we won't detect a EOF for this
1114 * stream.
1115 */
1116 if (ret_ref == 1 && vstream->ctf_trace->metadata_stream) {
1117 destroy_viewer_stream(vstream->ctf_trace->viewer_metadata_stream);
1118 vstream->ctf_trace->metadata_stream = NULL;
1119 DBG("Freeing ctf_trace %" PRIu64, vstream->ctf_trace->id);
1120 /*
1121 * The streaming-side is already closed and we can't receive a new
1122 * stream concurrently at this point (since the session is being
1123 * destroyed), so when we detect the refcount equals 0, we are the
1124 * only owners of the ctf_trace and we can free it ourself.
1125 */
1126 free(vstream->ctf_trace);
1127 }
1128
1129 call_rcu(&vstream->rcu_node, deferred_free_viewer_stream);
1130 }
1131
1132 /*
1133 * Send the next index for a stream.
1134 *
1135 * Return 0 on success or else a negative value.
1136 */
1137 static
1138 int viewer_get_next_index(struct relay_command *cmd,
1139 struct lttng_ht *sessions_ht)
1140 {
1141 int ret;
1142 struct lttng_viewer_get_next_index request_index;
1143 struct lttng_viewer_index viewer_index;
1144 struct ctf_packet_index packet_index;
1145 struct relay_viewer_stream *vstream;
1146 struct relay_stream *rstream;
1147
1148 assert(cmd);
1149 assert(sessions_ht);
1150
1151 DBG("Viewer get next index");
1152
1153 if (cmd->version_check_done == 0) {
1154 ERR("Trying to request index before version check");
1155 ret = -1;
1156 goto end_no_session;
1157 }
1158
1159 health_code_update();
1160 ret = cmd->sock->ops->recvmsg(cmd->sock, &request_index,
1161 sizeof(request_index), 0);
1162 if (ret < 0 || ret != sizeof(request_index)) {
1163 ret = -1;
1164 ERR("Relay didn't receive the whole packet");
1165 goto end;
1166 }
1167 health_code_update();
1168
1169 rcu_read_lock();
1170 vstream = live_find_viewer_stream_by_id(be64toh(request_index.stream_id));
1171 if (!vstream) {
1172 ret = -1;
1173 goto end_unlock;
1174 }
1175
1176 memset(&viewer_index, 0, sizeof(viewer_index));
1177
1178 /*
1179 * The viewer should not ask for index on metadata stream.
1180 */
1181 if (vstream->metadata_flag) {
1182 viewer_index.status = htobe32(VIEWER_INDEX_HUP);
1183 goto send_reply;
1184 }
1185
1186 /* First time, we open the index file */
1187 if (vstream->index_read_fd < 0) {
1188 ret = open_index(vstream);
1189 if (ret == -ENOENT) {
1190 /*
1191 * The index is created only when the first data packet arrives, it
1192 * might not be ready at the beginning of the session
1193 */
1194 viewer_index.status = htobe32(VIEWER_INDEX_RETRY);
1195 goto send_reply;
1196 } else if (ret < 0) {
1197 viewer_index.status = htobe32(VIEWER_INDEX_ERR);
1198 goto send_reply;
1199 }
1200 }
1201
1202 rstream = relay_stream_find_by_id(vstream->stream_handle);
1203 if (rstream) {
1204 if (vstream->abort_flag) {
1205 /* Rotate on abort (overwrite). */
1206 DBG("Viewer rotate because of overwrite");
1207 ret = rotate_viewer_stream(vstream, rstream);
1208 if (ret < 0) {
1209 goto end_unlock;
1210 } else if (ret == 1) {
1211 viewer_index.status = htobe32(VIEWER_INDEX_HUP);
1212 delete_viewer_stream(vstream);
1213 destroy_viewer_stream(vstream);
1214 goto send_reply;
1215 }
1216 }
1217 pthread_mutex_lock(&rstream->viewer_stream_rotation_lock);
1218 if (rstream->tracefile_count_current == vstream->tracefile_count_current) {
1219 if (rstream->beacon_ts_end != -1ULL &&
1220 vstream->last_sent_index == rstream->total_index_received) {
1221 viewer_index.status = htobe32(VIEWER_INDEX_INACTIVE);
1222 viewer_index.timestamp_end = htobe64(rstream->beacon_ts_end);
1223 pthread_mutex_unlock(&rstream->viewer_stream_rotation_lock);
1224 goto send_reply;
1225 /*
1226 * Reader and writer are working in the same tracefile, so we care
1227 * about the number of index received and sent. Otherwise, we read
1228 * up to EOF.
1229 */
1230 } else if (rstream->total_index_received <= vstream->last_sent_index
1231 && !vstream->close_write_flag) {
1232 pthread_mutex_unlock(&rstream->viewer_stream_rotation_lock);
1233 /* No new index to send, retry later. */
1234 viewer_index.status = htobe32(VIEWER_INDEX_RETRY);
1235 goto send_reply;
1236 }
1237 }
1238 pthread_mutex_unlock(&rstream->viewer_stream_rotation_lock);
1239 } else if (!rstream && vstream->close_write_flag &&
1240 vstream->total_index_received == vstream->last_sent_index) {
1241 /* Last index sent and current tracefile closed in write */
1242 viewer_index.status = htobe32(VIEWER_INDEX_HUP);
1243 delete_viewer_stream(vstream);
1244 destroy_viewer_stream(vstream);
1245 goto send_reply;
1246 } else {
1247 vstream->close_write_flag = 1;
1248 }
1249
1250 if (!vstream->ctf_trace->metadata_received ||
1251 vstream->ctf_trace->metadata_received >
1252 vstream->ctf_trace->metadata_sent) {
1253 viewer_index.flags |= LTTNG_VIEWER_FLAG_NEW_METADATA;
1254 }
1255
1256 pthread_mutex_lock(&vstream->overwrite_lock);
1257 if (vstream->abort_flag) {
1258 /*
1259 * The file is being overwritten by the writer, we cannot
1260 * use it.
1261 */
1262 viewer_index.status = htobe32(VIEWER_INDEX_RETRY);
1263 pthread_mutex_unlock(&vstream->overwrite_lock);
1264 ret = rotate_viewer_stream(vstream, rstream);
1265 if (ret < 0) {
1266 goto end_unlock;
1267 } else if (ret == 1) {
1268 viewer_index.status = htobe32(VIEWER_INDEX_HUP);
1269 delete_viewer_stream(vstream);
1270 destroy_viewer_stream(vstream);
1271 goto send_reply;
1272 }
1273 goto send_reply;
1274 }
1275 ret = lttng_read(vstream->index_read_fd, &packet_index,
1276 sizeof(packet_index));
1277 pthread_mutex_unlock(&vstream->overwrite_lock);
1278 if (ret < sizeof(packet_index)) {
1279 /*
1280 * The tracefile is closed in write, so we read up to EOF.
1281 */
1282 if (vstream->close_write_flag == 1) {
1283 viewer_index.status = htobe32(VIEWER_INDEX_RETRY);
1284 /* Rotate on normal EOF */
1285 ret = rotate_viewer_stream(vstream, rstream);
1286 if (ret < 0) {
1287 goto end_unlock;
1288 } else if (ret == 1) {
1289 viewer_index.status = htobe32(VIEWER_INDEX_HUP);
1290 delete_viewer_stream(vstream);
1291 destroy_viewer_stream(vstream);
1292 goto send_reply;
1293 }
1294 } else {
1295 PERROR("Relay reading index file %d",
1296 vstream->index_read_fd);
1297 viewer_index.status = htobe32(VIEWER_INDEX_ERR);
1298 }
1299 goto send_reply;
1300 } else {
1301 viewer_index.status = htobe32(VIEWER_INDEX_OK);
1302 vstream->last_sent_index++;
1303 }
1304
1305 /*
1306 * Indexes are stored in big endian, no need to switch before sending.
1307 */
1308 viewer_index.offset = packet_index.offset;
1309 viewer_index.packet_size = packet_index.packet_size;
1310 viewer_index.content_size = packet_index.content_size;
1311 viewer_index.timestamp_begin = packet_index.timestamp_begin;
1312 viewer_index.timestamp_end = packet_index.timestamp_end;
1313 viewer_index.events_discarded = packet_index.events_discarded;
1314 viewer_index.stream_id = packet_index.stream_id;
1315
1316 send_reply:
1317 viewer_index.flags = htobe32(viewer_index.flags);
1318 health_code_update();
1319 ret = cmd->sock->ops->sendmsg(cmd->sock, &viewer_index,
1320 sizeof(viewer_index), 0);
1321 if (ret < 0) {
1322 ERR("Relay index to viewer");
1323 goto end_unlock;
1324 }
1325 health_code_update();
1326
1327 DBG("Index %" PRIu64 "for stream %" PRIu64 "sent",
1328 vstream->last_sent_index, vstream->stream_handle);
1329
1330 end_unlock:
1331 rcu_read_unlock();
1332
1333 end_no_session:
1334 end:
1335 return ret;
1336 }
1337
1338 /*
1339 * Send the next index for a stream
1340 *
1341 * Return 0 on success or else a negative value.
1342 */
1343 static
1344 int viewer_get_packet(struct relay_command *cmd)
1345 {
1346 int ret, send_data = 0;
1347 char *data = NULL;
1348 uint32_t len = 0;
1349 ssize_t read_len;
1350 struct lttng_viewer_get_packet get_packet_info;
1351 struct lttng_viewer_trace_packet reply;
1352 struct relay_viewer_stream *stream;
1353
1354 assert(cmd);
1355
1356 DBG2("Relay get data packet");
1357
1358 if (cmd->version_check_done == 0) {
1359 ERR("Trying to get packet before version check");
1360 ret = -1;
1361 goto end;
1362 }
1363
1364 health_code_update();
1365 ret = cmd->sock->ops->recvmsg(cmd->sock, &get_packet_info,
1366 sizeof(get_packet_info), 0);
1367 if (ret < 0 || ret != sizeof(get_packet_info)) {
1368 ret = -1;
1369 ERR("Relay didn't receive the whole packet");
1370 goto end;
1371 }
1372 health_code_update();
1373
1374 /* From this point on, the error label can be reached. */
1375 memset(&reply, 0, sizeof(reply));
1376
1377 rcu_read_lock();
1378 stream = live_find_viewer_stream_by_id(be64toh(get_packet_info.stream_id));
1379 if (!stream) {
1380 goto error;
1381 }
1382 assert(stream->ctf_trace);
1383
1384 /*
1385 * First time we read this stream, we need open the tracefile, we should
1386 * only arrive here if an index has already been sent to the viewer, so the
1387 * tracefile must exist, if it does not it is a fatal error.
1388 */
1389 if (stream->read_fd < 0) {
1390 char fullpath[PATH_MAX];
1391
1392 if (stream->tracefile_count > 0) {
1393 ret = snprintf(fullpath, PATH_MAX, "%s/%s_%" PRIu64, stream->path_name,
1394 stream->channel_name,
1395 stream->tracefile_count_current);
1396 } else {
1397 ret = snprintf(fullpath, PATH_MAX, "%s/%s", stream->path_name,
1398 stream->channel_name);
1399 }
1400 if (ret < 0) {
1401 goto error;
1402 }
1403 ret = open(fullpath, O_RDONLY);
1404 if (ret < 0) {
1405 PERROR("Relay opening trace file");
1406 goto error;
1407 }
1408 stream->read_fd = ret;
1409 }
1410
1411 if (!stream->ctf_trace->metadata_received ||
1412 stream->ctf_trace->metadata_received >
1413 stream->ctf_trace->metadata_sent) {
1414 reply.status = htobe32(VIEWER_GET_PACKET_ERR);
1415 reply.flags |= LTTNG_VIEWER_FLAG_NEW_METADATA;
1416 goto send_reply;
1417 }
1418
1419 len = be32toh(get_packet_info.len);
1420 data = zmalloc(len);
1421 if (!data) {
1422 PERROR("relay data zmalloc");
1423 goto error;
1424 }
1425
1426 ret = lseek(stream->read_fd, be64toh(get_packet_info.offset), SEEK_SET);
1427 if (ret < 0) {
1428 /*
1429 * If the read fd was closed by the streaming side, the
1430 * abort_flag will be set to 1, otherwise it is an error.
1431 */
1432 if (stream->abort_flag == 0) {
1433 PERROR("lseek");
1434 goto error;
1435 }
1436 reply.status = htobe32(VIEWER_GET_PACKET_EOF);
1437 goto send_reply;
1438 }
1439 read_len = lttng_read(stream->read_fd, data, len);
1440 if (read_len < len) {
1441 /*
1442 * If the read fd was closed by the streaming side, the
1443 * abort_flag will be set to 1, otherwise it is an error.
1444 */
1445 if (stream->abort_flag == 0) {
1446 PERROR("Relay reading trace file, fd: %d, offset: %" PRIu64,
1447 stream->read_fd,
1448 be64toh(get_packet_info.offset));
1449 goto error;
1450 } else {
1451 reply.status = htobe32(VIEWER_GET_PACKET_EOF);
1452 goto send_reply;
1453 }
1454 }
1455 reply.status = htobe32(VIEWER_GET_PACKET_OK);
1456 reply.len = htobe32(len);
1457 send_data = 1;
1458 goto send_reply;
1459
1460 error:
1461 reply.status = htobe32(VIEWER_GET_PACKET_ERR);
1462
1463 send_reply:
1464 reply.flags = htobe32(reply.flags);
1465
1466 health_code_update();
1467 ret = cmd->sock->ops->sendmsg(cmd->sock, &reply, sizeof(reply), 0);
1468 if (ret < 0) {
1469 ERR("Relay data header to viewer");
1470 goto end_unlock;
1471 }
1472 health_code_update();
1473
1474 if (send_data) {
1475 health_code_update();
1476 ret = cmd->sock->ops->sendmsg(cmd->sock, data, len, 0);
1477 if (ret < 0) {
1478 ERR("Relay send data to viewer");
1479 goto end_unlock;
1480 }
1481 health_code_update();
1482 }
1483
1484 DBG("Sent %u bytes for stream %" PRIu64, len,
1485 be64toh(get_packet_info.stream_id));
1486
1487 end_unlock:
1488 free(data);
1489 rcu_read_unlock();
1490
1491 end:
1492 return ret;
1493 }
1494
1495 /*
1496 * Send the session's metadata
1497 *
1498 * Return 0 on success else a negative value.
1499 */
1500 static
1501 int viewer_get_metadata(struct relay_command *cmd)
1502 {
1503 int ret = 0;
1504 ssize_t read_len;
1505 uint64_t len = 0;
1506 char *data = NULL;
1507 struct lttng_viewer_get_metadata request;
1508 struct lttng_viewer_metadata_packet reply;
1509 struct relay_viewer_stream *stream;
1510
1511 assert(cmd);
1512
1513 DBG("Relay get metadata");
1514
1515 if (cmd->version_check_done == 0) {
1516 ERR("Trying to get metadata before version check");
1517 ret = -1;
1518 goto end;
1519 }
1520
1521 health_code_update();
1522 ret = cmd->sock->ops->recvmsg(cmd->sock, &request,
1523 sizeof(request), 0);
1524 if (ret < 0 || ret != sizeof(request)) {
1525 ret = -1;
1526 ERR("Relay didn't receive the whole packet");
1527 goto end;
1528 }
1529 health_code_update();
1530
1531 rcu_read_lock();
1532 stream = live_find_viewer_stream_by_id(be64toh(request.stream_id));
1533 if (!stream || !stream->metadata_flag) {
1534 ERR("Invalid metadata stream");
1535 goto error;
1536 }
1537 assert(stream->ctf_trace);
1538 assert(stream->ctf_trace->metadata_sent <=
1539 stream->ctf_trace->metadata_received);
1540
1541 len = stream->ctf_trace->metadata_received -
1542 stream->ctf_trace->metadata_sent;
1543 if (len == 0) {
1544 reply.status = htobe32(VIEWER_NO_NEW_METADATA);
1545 goto send_reply;
1546 }
1547
1548 /* first time, we open the metadata file */
1549 if (stream->read_fd < 0) {
1550 char fullpath[PATH_MAX];
1551
1552 ret = snprintf(fullpath, PATH_MAX, "%s/%s", stream->path_name,
1553 stream->channel_name);
1554 if (ret < 0) {
1555 goto error;
1556 }
1557 ret = open(fullpath, O_RDONLY);
1558 if (ret < 0) {
1559 PERROR("Relay opening metadata file");
1560 goto error;
1561 }
1562 stream->read_fd = ret;
1563 }
1564
1565 reply.len = htobe64(len);
1566 data = zmalloc(len);
1567 if (!data) {
1568 PERROR("viewer metadata zmalloc");
1569 goto error;
1570 }
1571
1572 read_len = lttng_read(stream->read_fd, data, len);
1573 if (read_len < len) {
1574 PERROR("Relay reading metadata file");
1575 goto error;
1576 }
1577 stream->ctf_trace->metadata_sent += read_len;
1578 reply.status = htobe32(VIEWER_METADATA_OK);
1579 goto send_reply;
1580
1581 error:
1582 reply.status = htobe32(VIEWER_METADATA_ERR);
1583
1584 send_reply:
1585 health_code_update();
1586 ret = cmd->sock->ops->sendmsg(cmd->sock, &reply, sizeof(reply), 0);
1587 if (ret < 0) {
1588 ERR("Relay data header to viewer");
1589 goto end_unlock;
1590 }
1591 health_code_update();
1592
1593 if (len > 0) {
1594 ret = cmd->sock->ops->sendmsg(cmd->sock, data, len, 0);
1595 if (ret < 0) {
1596 ERR("Relay send data to viewer");
1597 goto end_unlock;
1598 }
1599 }
1600
1601 DBG("Sent %" PRIu64 " bytes of metadata for stream %" PRIu64, len,
1602 be64toh(request.stream_id));
1603
1604 DBG("Metadata sent");
1605
1606 end_unlock:
1607 free(data);
1608 rcu_read_unlock();
1609 end:
1610 return ret;
1611 }
1612
1613 /*
1614 * live_relay_unknown_command: send -1 if received unknown command
1615 */
1616 static
1617 void live_relay_unknown_command(struct relay_command *cmd)
1618 {
1619 struct lttcomm_relayd_generic_reply reply;
1620 int ret;
1621
1622 reply.ret_code = htobe32(LTTNG_ERR_UNK);
1623 ret = cmd->sock->ops->sendmsg(cmd->sock, &reply,
1624 sizeof(struct lttcomm_relayd_generic_reply), 0);
1625 if (ret < 0) {
1626 ERR("Relay sending unknown command");
1627 }
1628 }
1629
1630 /*
1631 * Process the commands received on the control socket
1632 */
1633 static
1634 int process_control(struct lttng_viewer_cmd *recv_hdr,
1635 struct relay_command *cmd, struct lttng_ht *sessions_ht)
1636 {
1637 int ret = 0;
1638
1639 switch (be32toh(recv_hdr->cmd)) {
1640 case VIEWER_CONNECT:
1641 ret = viewer_connect(cmd);
1642 break;
1643 case VIEWER_LIST_SESSIONS:
1644 ret = viewer_list_sessions(cmd, sessions_ht);
1645 break;
1646 case VIEWER_ATTACH_SESSION:
1647 ret = viewer_attach_session(cmd, sessions_ht);
1648 break;
1649 case VIEWER_GET_NEXT_INDEX:
1650 ret = viewer_get_next_index(cmd, sessions_ht);
1651 break;
1652 case VIEWER_GET_PACKET:
1653 ret = viewer_get_packet(cmd);
1654 break;
1655 case VIEWER_GET_METADATA:
1656 ret = viewer_get_metadata(cmd);
1657 break;
1658 default:
1659 ERR("Received unknown viewer command (%u)", be32toh(recv_hdr->cmd));
1660 live_relay_unknown_command(cmd);
1661 ret = -1;
1662 goto end;
1663 }
1664
1665 end:
1666 return ret;
1667 }
1668
1669 static
1670 void cleanup_poll_connection(struct lttng_poll_event *events, int pollfd)
1671 {
1672 int ret;
1673
1674 assert(events);
1675
1676 lttng_poll_del(events, pollfd);
1677
1678 ret = close(pollfd);
1679 if (ret < 0) {
1680 ERR("Closing pollfd %d", pollfd);
1681 }
1682 }
1683
1684 /*
1685 * Create and add connection to the given hash table.
1686 *
1687 * Return poll add value or else -1 on error.
1688 */
1689 static
1690 int add_connection(int fd, struct lttng_poll_event *events,
1691 struct lttng_ht *relay_connections_ht)
1692 {
1693 int ret;
1694 struct relay_command *relay_connection;
1695
1696 assert(events);
1697 assert(relay_connections_ht);
1698
1699 relay_connection = zmalloc(sizeof(struct relay_command));
1700 if (relay_connection == NULL) {
1701 PERROR("Relay command zmalloc");
1702 goto error;
1703 }
1704
1705 ret = lttng_read(fd, relay_connection, sizeof(*relay_connection));
1706 if (ret < sizeof(*relay_connection)) {
1707 PERROR("read relay cmd pipe");
1708 goto error_read;
1709 }
1710
1711 lttng_ht_node_init_ulong(&relay_connection->sock_n,
1712 (unsigned long) relay_connection->sock->fd);
1713 rcu_read_lock();
1714 lttng_ht_add_unique_ulong(relay_connections_ht,
1715 &relay_connection->sock_n);
1716 rcu_read_unlock();
1717
1718 return lttng_poll_add(events, relay_connection->sock->fd,
1719 LPOLLIN | LPOLLRDHUP);
1720
1721 error_read:
1722 free(relay_connection);
1723 error:
1724 return -1;
1725 }
1726
1727 static
1728 void deferred_free_connection(struct rcu_head *head)
1729 {
1730 struct relay_command *relay_connection =
1731 caa_container_of(head, struct relay_command, rcu_node);
1732
1733 if (relay_connection->session &&
1734 relay_connection->session->viewer_attached > 0) {
1735 relay_connection->session->viewer_attached--;
1736 }
1737 lttcomm_destroy_sock(relay_connection->sock);
1738 free(relay_connection);
1739 }
1740
1741 /*
1742 * Delete all streams for a specific session ID.
1743 */
1744 static
1745 void viewer_del_streams(uint64_t session_id)
1746 {
1747 struct relay_viewer_stream *stream;
1748 struct lttng_ht_iter iter;
1749
1750 rcu_read_lock();
1751 cds_lfht_for_each_entry(viewer_streams_ht->ht, &iter.iter, stream,
1752 stream_n.node) {
1753 health_code_update();
1754
1755 if (stream->session_id != session_id) {
1756 continue;
1757 }
1758
1759 delete_viewer_stream(stream);
1760 assert(stream->ctf_trace);
1761
1762 if (stream->metadata_flag) {
1763 /*
1764 * The metadata viewer stream is destroyed once the refcount on the
1765 * ctf trace goes to 0 in the destroy stream function thus there is
1766 * no explicit call to that function here.
1767 */
1768 stream->ctf_trace->metadata_sent = 0;
1769 stream->ctf_trace->viewer_metadata_stream = NULL;
1770 } else {
1771 destroy_viewer_stream(stream);
1772 }
1773 }
1774 rcu_read_unlock();
1775 }
1776
1777 /*
1778 * Delete and free a connection.
1779 *
1780 * RCU read side lock MUST be acquired.
1781 */
1782 static
1783 void del_connection(struct lttng_ht *relay_connections_ht,
1784 struct lttng_ht_iter *iter, struct relay_command *relay_connection)
1785 {
1786 int ret;
1787
1788 assert(relay_connections_ht);
1789 assert(iter);
1790 assert(relay_connection);
1791
1792 DBG("Cleaning connection of session ID %" PRIu64,
1793 relay_connection->session_id);
1794
1795 ret = lttng_ht_del(relay_connections_ht, iter);
1796 assert(!ret);
1797
1798 viewer_del_streams(relay_connection->session_id);
1799
1800 call_rcu(&relay_connection->rcu_node, deferred_free_connection);
1801 }
1802
1803 /*
1804 * This thread does the actual work
1805 */
1806 static
1807 void *thread_worker(void *data)
1808 {
1809 int ret, err = -1;
1810 uint32_t nb_fd;
1811 struct relay_command *relay_connection;
1812 struct lttng_poll_event events;
1813 struct lttng_ht *relay_connections_ht;
1814 struct lttng_ht_node_ulong *node;
1815 struct lttng_ht_iter iter;
1816 struct lttng_viewer_cmd recv_hdr;
1817 struct relay_local_data *relay_ctx = (struct relay_local_data *) data;
1818 struct lttng_ht *sessions_ht = relay_ctx->sessions_ht;
1819
1820 DBG("[thread] Live viewer relay worker started");
1821
1822 rcu_register_thread();
1823
1824 health_register(health_relayd, HEALTH_RELAYD_TYPE_LIVE_WORKER);
1825
1826 /* table of connections indexed on socket */
1827 relay_connections_ht = lttng_ht_new(0, LTTNG_HT_TYPE_ULONG);
1828 if (!relay_connections_ht) {
1829 goto relay_connections_ht_error;
1830 }
1831
1832 ret = create_thread_poll_set(&events, 2);
1833 if (ret < 0) {
1834 goto error_poll_create;
1835 }
1836
1837 ret = lttng_poll_add(&events, live_relay_cmd_pipe[0], LPOLLIN | LPOLLRDHUP);
1838 if (ret < 0) {
1839 goto error;
1840 }
1841
1842 restart:
1843 while (1) {
1844 int i;
1845
1846 health_code_update();
1847
1848 /* Infinite blocking call, waiting for transmission */
1849 DBG3("Relayd live viewer worker thread polling...");
1850 health_poll_entry();
1851 ret = lttng_poll_wait(&events, -1);
1852 health_poll_exit();
1853 if (ret < 0) {
1854 /*
1855 * Restart interrupted system call.
1856 */
1857 if (errno == EINTR) {
1858 goto restart;
1859 }
1860 goto error;
1861 }
1862
1863 nb_fd = ret;
1864
1865 /*
1866 * Process control. The control connection is prioritised so we don't
1867 * starve it with high throughput tracing data on the data
1868 * connection.
1869 */
1870 for (i = 0; i < nb_fd; i++) {
1871 /* Fetch once the poll data */
1872 uint32_t revents = LTTNG_POLL_GETEV(&events, i);
1873 int pollfd = LTTNG_POLL_GETFD(&events, i);
1874
1875 health_code_update();
1876
1877 /* Thread quit pipe has been closed. Killing thread. */
1878 ret = check_thread_quit_pipe(pollfd, revents);
1879 if (ret) {
1880 err = 0;
1881 goto exit;
1882 }
1883
1884 /* Inspect the relay cmd pipe for new connection */
1885 if (pollfd == live_relay_cmd_pipe[0]) {
1886 if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
1887 ERR("Relay live pipe error");
1888 goto error;
1889 } else if (revents & LPOLLIN) {
1890 DBG("Relay live viewer command received");
1891 ret = add_connection(live_relay_cmd_pipe[0],
1892 &events, relay_connections_ht);
1893 if (ret < 0) {
1894 goto error;
1895 }
1896 }
1897 } else if (revents) {
1898 rcu_read_lock();
1899 lttng_ht_lookup(relay_connections_ht,
1900 (void *)((unsigned long) pollfd), &iter);
1901 node = lttng_ht_iter_get_node_ulong(&iter);
1902 if (node == NULL) {
1903 DBG2("Relay viewer sock %d not found", pollfd);
1904 rcu_read_unlock();
1905 goto error;
1906 }
1907 relay_connection = caa_container_of(node, struct relay_command,
1908 sock_n);
1909
1910 if (revents & (LPOLLERR)) {
1911 cleanup_poll_connection(&events, pollfd);
1912 del_connection(relay_connections_ht, &iter,
1913 relay_connection);
1914 } else if (revents & (LPOLLHUP | LPOLLRDHUP)) {
1915 DBG("Viewer socket %d hung up", pollfd);
1916 cleanup_poll_connection(&events, pollfd);
1917 del_connection(relay_connections_ht, &iter,
1918 relay_connection);
1919 } else if (revents & LPOLLIN) {
1920 ret = relay_connection->sock->ops->recvmsg(
1921 relay_connection->sock, &recv_hdr,
1922 sizeof(struct lttng_viewer_cmd),
1923 0);
1924 /* connection closed */
1925 if (ret <= 0) {
1926 cleanup_poll_connection(&events, pollfd);
1927 del_connection(relay_connections_ht, &iter,
1928 relay_connection);
1929 DBG("Viewer control connection closed with %d",
1930 pollfd);
1931 } else {
1932 if (relay_connection->session) {
1933 DBG2("Relay viewer worker receiving data for "
1934 "session: %" PRIu64,
1935 relay_connection->session->id);
1936 }
1937 ret = process_control(&recv_hdr, relay_connection,
1938 sessions_ht);
1939 if (ret < 0) {
1940 /* Clear the session on error. */
1941 cleanup_poll_connection(&events, pollfd);
1942 del_connection(relay_connections_ht, &iter,
1943 relay_connection);
1944 DBG("Viewer connection closed with %d", pollfd);
1945 }
1946 }
1947 }
1948 rcu_read_unlock();
1949 }
1950 }
1951 }
1952
1953 exit:
1954 error:
1955 lttng_poll_clean(&events);
1956
1957 /* empty the hash table and free the memory */
1958 rcu_read_lock();
1959 cds_lfht_for_each_entry(relay_connections_ht->ht, &iter.iter, node, node) {
1960 health_code_update();
1961
1962 node = lttng_ht_iter_get_node_ulong(&iter);
1963 if (!node) {
1964 continue;
1965 }
1966
1967 relay_connection = caa_container_of(node, struct relay_command,
1968 sock_n);
1969 del_connection(relay_connections_ht, &iter, relay_connection);
1970 }
1971 rcu_read_unlock();
1972 error_poll_create:
1973 lttng_ht_destroy(relay_connections_ht);
1974 relay_connections_ht_error:
1975 /* Close relay cmd pipes */
1976 utils_close_pipe(live_relay_cmd_pipe);
1977 if (err) {
1978 DBG("Viewer worker thread exited with error");
1979 }
1980 DBG("Viewer worker thread cleanup complete");
1981 if (err) {
1982 health_error();
1983 ERR("Health error occurred in %s", __func__);
1984 }
1985 health_unregister(health_relayd);
1986 stop_threads();
1987 rcu_unregister_thread();
1988 return NULL;
1989 }
1990
1991 /*
1992 * Create the relay command pipe to wake thread_manage_apps.
1993 * Closed in cleanup().
1994 */
1995 static int create_relay_cmd_pipe(void)
1996 {
1997 int ret;
1998
1999 ret = utils_create_pipe_cloexec(live_relay_cmd_pipe);
2000
2001 return ret;
2002 }
2003
2004 void live_stop_threads(void)
2005 {
2006 int ret;
2007 void *status;
2008
2009 stop_threads();
2010
2011 ret = pthread_join(live_listener_thread, &status);
2012 if (ret != 0) {
2013 PERROR("pthread_join live listener");
2014 goto error; /* join error, exit without cleanup */
2015 }
2016
2017 ret = pthread_join(live_worker_thread, &status);
2018 if (ret != 0) {
2019 PERROR("pthread_join live worker");
2020 goto error; /* join error, exit without cleanup */
2021 }
2022
2023 ret = pthread_join(live_dispatcher_thread, &status);
2024 if (ret != 0) {
2025 PERROR("pthread_join live dispatcher");
2026 goto error; /* join error, exit without cleanup */
2027 }
2028
2029 cleanup();
2030
2031 error:
2032 return;
2033 }
2034
2035 /*
2036 * main
2037 */
2038 int live_start_threads(struct lttng_uri *uri,
2039 struct relay_local_data *relay_ctx, int quit_pipe[2])
2040 {
2041 int ret = 0;
2042 void *status;
2043 int is_root;
2044
2045 assert(uri);
2046 live_uri = uri;
2047
2048 live_thread_quit_pipe[0] = quit_pipe[0];
2049 live_thread_quit_pipe[1] = quit_pipe[1];
2050
2051 /* Check if daemon is UID = 0 */
2052 is_root = !getuid();
2053
2054 if (!is_root) {
2055 if (live_uri->port < 1024) {
2056 ERR("Need to be root to use ports < 1024");
2057 ret = -1;
2058 goto exit;
2059 }
2060 }
2061
2062 /* Setup the thread apps communication pipe. */
2063 if ((ret = create_relay_cmd_pipe()) < 0) {
2064 goto exit;
2065 }
2066
2067 /* Init relay command queue. */
2068 cds_wfq_init(&viewer_cmd_queue.queue);
2069
2070 /* Set up max poll set size */
2071 lttng_poll_set_max_size();
2072
2073 /* Setup the dispatcher thread */
2074 ret = pthread_create(&live_dispatcher_thread, NULL,
2075 thread_dispatcher, (void *) NULL);
2076 if (ret != 0) {
2077 PERROR("pthread_create viewer dispatcher");
2078 goto exit_dispatcher;
2079 }
2080
2081 /* Setup the worker thread */
2082 ret = pthread_create(&live_worker_thread, NULL,
2083 thread_worker, relay_ctx);
2084 if (ret != 0) {
2085 PERROR("pthread_create viewer worker");
2086 goto exit_worker;
2087 }
2088
2089 /* Setup the listener thread */
2090 ret = pthread_create(&live_listener_thread, NULL,
2091 thread_listener, (void *) NULL);
2092 if (ret != 0) {
2093 PERROR("pthread_create viewer listener");
2094 goto exit_listener;
2095 }
2096
2097 ret = 0;
2098 goto end;
2099
2100 exit_listener:
2101 ret = pthread_join(live_listener_thread, &status);
2102 if (ret != 0) {
2103 PERROR("pthread_join live listener");
2104 goto error; /* join error, exit without cleanup */
2105 }
2106
2107 exit_worker:
2108 ret = pthread_join(live_worker_thread, &status);
2109 if (ret != 0) {
2110 PERROR("pthread_join live worker");
2111 goto error; /* join error, exit without cleanup */
2112 }
2113
2114 exit_dispatcher:
2115 ret = pthread_join(live_dispatcher_thread, &status);
2116 if (ret != 0) {
2117 PERROR("pthread_join live dispatcher");
2118 goto error; /* join error, exit without cleanup */
2119 }
2120
2121 exit:
2122 cleanup();
2123
2124 end:
2125 error:
2126 return ret;
2127 }
This page took 0.082599 seconds and 3 git commands to generate.