projects / lttng-tools.git / blob
? search:
summary | shortlog | log | commit | commitdiff | tree
blame | history | raw | HEAD
Tests: Add test_utils_expand_path to unit tests
[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 = 0;
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 lttng_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 (strncmp(hdr.magic, INDEX_MAGIC, sizeof(hdr.magic)) != 0) {
653 ERR("Invalid header magic");
654 ret = -1;
655 goto error;
656 }
657 if (be32toh(hdr.index_major) != INDEX_MAJOR ||
658 be32toh(hdr.index_minor) != 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 if (seek_last) {
702 viewer_stream->tracefile_count_current =
703 stream->tracefile_count_current;
704 } else {
705 viewer_stream->tracefile_count_current =
706 stream->oldest_tracefile_id;
707 }
708
709 /*
710 * The deletion of this ctf_trace object is only done in a call RCU of the
711 * relay stream making it valid as long as we have the read side lock.
712 */
713 viewer_stream->ctf_trace = stream->ctf_trace;
714 uatomic_inc(&viewer_stream->ctf_trace->refcount);
715
716 lttng_ht_node_init_u64(&viewer_stream->stream_n, stream->stream_handle);
717 lttng_ht_add_unique_u64(viewer_streams_ht, &viewer_stream->stream_n);
718
719 viewer_stream->index_read_fd = -1;
720 viewer_stream->read_fd = -1;
721
722 /*
723 * This is to avoid a race between the initialization of this object and
724 * the close of the given stream. If the stream is unable to find this
725 * viewer stream when closing, this copy will at least take the latest
726 * value.
727 * We also need that for the seek_last.
728 */
729 viewer_stream->total_index_received = stream->total_index_received;
730
731 /*
732 * If we never received an index for the current stream, delay
733 * the opening of the index, otherwise open it right now.
734 */
735 if (viewer_stream->tracefile_count_current ==
736 stream->tracefile_count_current &&
737 viewer_stream->total_index_received == 0) {
738 viewer_stream->index_read_fd = -1;
739 } else {
740 ret = open_index(viewer_stream);
741 if (ret < 0) {
742 goto error;
743 }
744 }
745
746 if (seek_last && viewer_stream->index_read_fd > 0) {
747 ret = lseek(viewer_stream->index_read_fd,
748 viewer_stream->total_index_received *
749 sizeof(struct lttng_packet_index),
750 SEEK_CUR);
751 if (ret < 0) {
752 goto error;
753 }
754 viewer_stream->last_sent_index =
755 viewer_stream->total_index_received;
756 }
757
758 ret = 0;
759
760 error:
761 return ret;
762 }
763
764 /*
765 * Rotate a stream to the next tracefile.
766 *
767 * Returns 0 on success, a negative value on error.
768 */
769 static
770 int rotate_viewer_stream(struct relay_viewer_stream *viewer_stream,
771 struct relay_stream *stream)
772 {
773 int ret;
774 uint64_t tracefile_id;
775
776 assert(viewer_stream);
777
778 tracefile_id = (viewer_stream->tracefile_count_current + 1) %
779 viewer_stream->tracefile_count;
780
781 if (stream) {
782 pthread_mutex_lock(&stream->viewer_stream_rotation_lock);
783 }
784 /*
785 * The writer and the reader are not working in the same
786 * tracefile, we can read up to EOF, we don't care about the
787 * total_index_received.
788 */
789 if (!stream || (stream->tracefile_count_current != tracefile_id)) {
790 viewer_stream->close_write_flag = 1;
791 } else {
792 /*
793 * We are opening a file that is still open in write, make
794 * sure we limit our reading to the number of indexes
795 * received.
796 */
797 viewer_stream->close_write_flag = 0;
798 if (stream) {
799 viewer_stream->total_index_received =
800 stream->total_index_received;
801 }
802 }
803 viewer_stream->tracefile_count_current = tracefile_id;
804 pthread_mutex_unlock(&stream->viewer_stream_rotation_lock);
805
806 if (viewer_stream->abort_flag == 0) {
807 ret = close(viewer_stream->index_read_fd);
808 if (ret < 0) {
809 PERROR("close index file");
810 }
811 ret = close(viewer_stream->read_fd);
812 if (ret < 0) {
813 PERROR("close tracefile");
814 }
815 } else {
816 viewer_stream->abort_flag = 0;
817 }
818
819 viewer_stream->read_fd = -1;
820
821 ret = open_index(viewer_stream);
822 if (ret < 0) {
823 goto error;
824 }
825
826 ret = 0;
827
828 error:
829 return ret;
830 }
831
832 /*
833 * Send the viewer the list of current sessions.
834 */
835 static
836 int viewer_attach_session(struct relay_command *cmd,
837 struct lttng_ht *sessions_ht)
838 {
839 int ret, send_streams = 0, nb_streams = 0;
840 struct lttng_viewer_attach_session_request request;
841 struct lttng_viewer_attach_session_response response;
842 struct lttng_viewer_stream send_stream;
843 struct relay_stream *stream;
844 struct relay_viewer_stream *viewer_stream;
845 struct lttng_ht_node_ulong *node;
846 struct lttng_ht_node_u64 *node64;
847 struct lttng_ht_iter iter;
848 struct relay_session *session;
849 int seek_last = 0;
850
851 assert(cmd);
852 assert(sessions_ht);
853
854 DBG("Attach session received");
855
856 if (cmd->version_check_done == 0) {
857 ERR("Trying to attach session before version check");
858 ret = -1;
859 goto end_no_session;
860 }
861
862 health_code_update();
863
864 ret = cmd->sock->ops->recvmsg(cmd->sock, &request, sizeof(request), 0);
865 if (ret < 0 || ret != sizeof(request)) {
866 if (ret == 0) {
867 /* Orderly shutdown. Not necessary to print an error. */
868 DBG("Socket %d did an orderly shutdown", cmd->sock->fd);
869 } else {
870 ERR("Relay failed to receive the attach parameters.");
871 }
872 ret = -1;
873 goto error;
874 }
875
876 health_code_update();
877
878 rcu_read_lock();
879 lttng_ht_lookup(sessions_ht,
880 (void *)((unsigned long) be64toh(request.session_id)), &iter);
881 node = lttng_ht_iter_get_node_ulong(&iter);
882 if (node == NULL) {
883 DBG("Relay session %" PRIu64 " not found",
884 be64toh(request.session_id));
885 response.status = htobe32(VIEWER_ATTACH_UNK);
886 goto send_reply;
887 }
888
889 session = caa_container_of(node, struct relay_session, session_n);
890 if (cmd->session_id == session->id) {
891 /* Same viewer already attached, just send the stream list. */
892 send_streams = 1;
893 response.status = htobe32(VIEWER_ATTACH_OK);
894 } else if (session->viewer_attached != 0) {
895 DBG("Already a viewer attached");
896 response.status = htobe32(VIEWER_ATTACH_ALREADY);
897 goto send_reply;
898 } else if (session->live_timer == 0) {
899 DBG("Not live session");
900 response.status = htobe32(VIEWER_ATTACH_NOT_LIVE);
901 goto send_reply;
902 } else {
903 session->viewer_attached++;
904 send_streams = 1;
905 response.status = htobe32(VIEWER_ATTACH_OK);
906 cmd->session_id = session->id;
907 cmd->session = session;
908 }
909
910 switch (be32toh(request.seek)) {
911 case VIEWER_SEEK_BEGINNING:
912 /* Default behaviour. */
913 break;
914 case VIEWER_SEEK_LAST:
915 seek_last = 1;
916 break;
917 default:
918 ERR("Wrong seek parameter");
919 response.status = htobe32(VIEWER_ATTACH_SEEK_ERR);
920 send_streams = 0;
921 goto send_reply;
922 }
923
924 if (send_streams) {
925 /* We should only be there if we have a session to attach to. */
926 assert(session);
927
928 /*
929 * Fill the viewer_streams_ht to count the number of streams
930 * ready to be sent and avoid concurrency issues on the
931 * relay_streams_ht and don't rely on a total session stream count.
932 */
933 cds_lfht_for_each_entry(relay_streams_ht->ht, &iter.iter, node, node) {
934 struct relay_viewer_stream *vstream;
935
936 health_code_update();
937
938 node = lttng_ht_iter_get_node_ulong(&iter);
939 if (!node) {
940 continue;
941 }
942 stream = caa_container_of(node, struct relay_stream, stream_n);
943 if (stream->session != cmd->session) {
944 continue;
945 }
946
947 /*
948 * Don't send streams with no ctf_trace, they are not ready to be
949 * read.
950 */
951 if (!stream->ctf_trace) {
952 continue;
953 }
954
955 vstream = live_find_viewer_stream_by_id(stream->stream_handle);
956 if (!vstream) {
957 ret = init_viewer_stream(stream, seek_last);
958 if (ret < 0) {
959 goto end_unlock;
960 }
961 }
962 nb_streams++;
963 }
964 response.streams_count = htobe32(nb_streams);
965 }
966
967 send_reply:
968 health_code_update();
969 ret = cmd->sock->ops->sendmsg(cmd->sock, &response, sizeof(response), 0);
970 if (ret < 0) {
971 ERR("Relay sending viewer attach response");
972 goto end_unlock;
973 }
974 health_code_update();
975
976 /*
977 * Unknown or busy session, just return gracefully, the viewer knows what
978 * is happening.
979 */
980 if (!send_streams) {
981 ret = 0;
982 goto end_unlock;
983 }
984
985 /* We should only be there if we have a session to attach to. */
986 assert(session);
987 cds_lfht_for_each_entry(viewer_streams_ht->ht, &iter.iter, node, node) {
988 health_code_update();
989
990 node64 = lttng_ht_iter_get_node_u64(&iter);
991 if (!node64) {
992 continue;
993 }
994 viewer_stream = caa_container_of(node64, struct relay_viewer_stream,
995 stream_n);
996 if (viewer_stream->session_id != cmd->session->id) {
997 continue;
998 }
999
1000 send_stream.id = htobe64(viewer_stream->stream_handle);
1001 send_stream.ctf_trace_id = htobe64(viewer_stream->ctf_trace->id);
1002 send_stream.metadata_flag = htobe32(viewer_stream->metadata_flag);
1003 strncpy(send_stream.path_name, viewer_stream->path_name,
1004 sizeof(send_stream.path_name));
1005 strncpy(send_stream.channel_name, viewer_stream->channel_name,
1006 sizeof(send_stream.channel_name));
1007
1008 ret = cmd->sock->ops->sendmsg(cmd->sock, &send_stream,
1009 sizeof(send_stream), 0);
1010 if (ret < 0) {
1011 ERR("Relay sending stream %" PRIu64, viewer_stream->stream_handle);
1012 goto end_unlock;
1013 }
1014 DBG("Sent stream %" PRIu64 " to viewer", viewer_stream->stream_handle);
1015 }
1016 ret = 0;
1017
1018 end_unlock:
1019 rcu_read_unlock();
1020 end_no_session:
1021 error:
1022 return ret;
1023 }
1024
1025 /*
1026 * Get viewer stream from stream id.
1027 *
1028 * RCU read side lock MUST be acquired.
1029 */
1030 struct relay_viewer_stream *live_find_viewer_stream_by_id(uint64_t stream_id)
1031 {
1032 struct lttng_ht_node_u64 *node;
1033 struct lttng_ht_iter iter;
1034 struct relay_viewer_stream *stream = NULL;
1035
1036 lttng_ht_lookup(viewer_streams_ht, &stream_id, &iter);
1037 node = lttng_ht_iter_get_node_u64(&iter);
1038 if (node == NULL) {
1039 DBG("Relay viewer stream %" PRIu64 " not found", stream_id);
1040 goto end;
1041 }
1042 stream = caa_container_of(node, struct relay_viewer_stream, stream_n);
1043
1044 end:
1045 return stream;
1046 }
1047
1048 /*
1049 * Send the next index for a stream.
1050 *
1051 * Return 0 on success or else a negative value.
1052 */
1053 static
1054 int viewer_get_next_index(struct relay_command *cmd,
1055 struct lttng_ht *sessions_ht)
1056 {
1057 int ret;
1058 struct lttng_viewer_get_next_index request_index;
1059 struct lttng_viewer_index viewer_index;
1060 struct lttng_packet_index packet_index;
1061 struct relay_viewer_stream *vstream;
1062 struct relay_stream *rstream;
1063
1064 assert(cmd);
1065 assert(sessions_ht);
1066
1067 DBG("Viewer get next index");
1068
1069 if (cmd->version_check_done == 0) {
1070 ERR("Trying to request index before version check");
1071 ret = -1;
1072 goto end_no_session;
1073 }
1074
1075 health_code_update();
1076 ret = cmd->sock->ops->recvmsg(cmd->sock, &request_index,
1077 sizeof(request_index), 0);
1078 if (ret < 0 || ret != sizeof(request_index)) {
1079 ret = -1;
1080 ERR("Relay didn't receive the whole packet");
1081 goto end;
1082 }
1083 health_code_update();
1084
1085 rcu_read_lock();
1086 vstream = live_find_viewer_stream_by_id(be64toh(request_index.stream_id));
1087 if (!vstream) {
1088 ret = -1;
1089 goto end_unlock;
1090 }
1091
1092 memset(&viewer_index, 0, sizeof(viewer_index));
1093
1094 /*
1095 * The viewer should not ask for index on metadata stream.
1096 */
1097 if (vstream->metadata_flag) {
1098 viewer_index.status = htobe32(VIEWER_INDEX_HUP);
1099 goto send_reply;
1100 }
1101
1102 /* First time, we open the index file */
1103 if (vstream->index_read_fd < 0) {
1104 ret = open_index(vstream);
1105 if (ret == -ENOENT) {
1106 /*
1107 * The index is created only when the first data packet arrives, it
1108 * might not be ready at the beginning of the session
1109 */
1110 viewer_index.status = htobe32(VIEWER_INDEX_RETRY);
1111 goto send_reply;
1112 } else if (ret < 0) {
1113 viewer_index.status = htobe32(VIEWER_INDEX_ERR);
1114 goto send_reply;
1115 }
1116 }
1117
1118 rstream = relay_stream_find_by_id(vstream->stream_handle);
1119 if (rstream) {
1120 if (vstream->abort_flag) {
1121 /* Rotate on abort (overwrite). */
1122 DBG("Viewer rotate because of overwrite");
1123 ret = rotate_viewer_stream(vstream, rstream);
1124 if (ret < 0) {
1125 goto end_unlock;
1126 }
1127 }
1128 if (rstream->beacon_ts_end != -1ULL &&
1129 vstream->last_sent_index == rstream->total_index_received) {
1130 viewer_index.status = htobe32(VIEWER_INDEX_INACTIVE);
1131 viewer_index.timestamp_end = htobe64(rstream->beacon_ts_end);
1132 goto send_reply;
1133 }
1134 /*
1135 * Reader and writer are working in the same tracefile, so we care
1136 * about the number of index received and sent. Otherwise, we read
1137 * up to EOF.
1138 */
1139 pthread_mutex_lock(&rstream->viewer_stream_rotation_lock);
1140 if (rstream->tracefile_count_current == vstream->tracefile_count_current
1141 && rstream->total_index_received <= vstream->last_sent_index
1142 && !vstream->close_write_flag) {
1143 pthread_mutex_unlock(&rstream->viewer_stream_rotation_lock);
1144 /* No new index to send, retry later. */
1145 viewer_index.status = htobe32(VIEWER_INDEX_RETRY);
1146 goto send_reply;
1147 }
1148 pthread_mutex_unlock(&rstream->viewer_stream_rotation_lock);
1149 } else if (!rstream && vstream->close_write_flag &&
1150 vstream->total_index_received == vstream->last_sent_index) {
1151 /* Last index sent and current tracefile closed in write */
1152 viewer_index.status = htobe32(VIEWER_INDEX_HUP);
1153 goto send_reply;
1154 } else {
1155 vstream->close_write_flag = 1;
1156 }
1157
1158 if (!vstream->ctf_trace->metadata_received ||
1159 vstream->ctf_trace->metadata_received >
1160 vstream->ctf_trace->metadata_sent) {
1161 viewer_index.flags |= LTTNG_VIEWER_FLAG_NEW_METADATA;
1162 }
1163
1164 ret = lttng_read(vstream->index_read_fd, &packet_index,
1165 sizeof(packet_index));
1166 if (ret < sizeof(packet_index)) {
1167 /*
1168 * The tracefile is closed in write, so we read up to EOF.
1169 */
1170 if (vstream->close_write_flag == 1) {
1171 viewer_index.status = htobe32(VIEWER_INDEX_RETRY);
1172 /* Rotate on normal EOF */
1173 ret = rotate_viewer_stream(vstream, rstream);
1174 if (ret < 0) {
1175 goto end_unlock;
1176 }
1177 } else {
1178 /*
1179 * If the read fd was closed by the streaming side, the
1180 * abort_flag will be set to 1, otherwise it is an error.
1181 */
1182 if (vstream->abort_flag != 1) {
1183 PERROR("Relay reading index file");
1184 viewer_index.status = htobe32(VIEWER_INDEX_ERR);
1185 goto send_reply;
1186 } else {
1187 viewer_index.status = htobe32(VIEWER_INDEX_HUP);
1188 }
1189 }
1190 goto send_reply;
1191 } else {
1192 viewer_index.status = htobe32(VIEWER_INDEX_OK);
1193 vstream->last_sent_index++;
1194 }
1195
1196 /*
1197 * Indexes are stored in big endian, no need to switch before sending.
1198 */
1199 viewer_index.offset = packet_index.offset;
1200 viewer_index.packet_size = packet_index.packet_size;
1201 viewer_index.content_size = packet_index.content_size;
1202 viewer_index.timestamp_begin = packet_index.timestamp_begin;
1203 viewer_index.timestamp_end = packet_index.timestamp_end;
1204 viewer_index.events_discarded = packet_index.events_discarded;
1205 viewer_index.stream_id = packet_index.stream_id;
1206
1207 send_reply:
1208 viewer_index.flags = htobe32(viewer_index.flags);
1209 health_code_update();
1210 ret = cmd->sock->ops->sendmsg(cmd->sock, &viewer_index,
1211 sizeof(viewer_index), 0);
1212 if (ret < 0) {
1213 ERR("Relay index to viewer");
1214 goto end_unlock;
1215 }
1216 health_code_update();
1217
1218 DBG("Index %" PRIu64 "for stream %" PRIu64 "sent",
1219 vstream->last_sent_index, vstream->stream_handle);
1220
1221 end_unlock:
1222 rcu_read_unlock();
1223
1224 end_no_session:
1225 end:
1226 return ret;
1227 }
1228
1229 /*
1230 * Send the next index for a stream
1231 *
1232 * Return 0 on success or else a negative value.
1233 */
1234 static
1235 int viewer_get_packet(struct relay_command *cmd)
1236 {
1237 int ret, send_data = 0;
1238 char *data = NULL;
1239 uint32_t len = 0;
1240 ssize_t read_len;
1241 struct lttng_viewer_get_packet get_packet_info;
1242 struct lttng_viewer_trace_packet reply;
1243 struct relay_viewer_stream *stream;
1244
1245 assert(cmd);
1246
1247 DBG2("Relay get data packet");
1248
1249 if (cmd->version_check_done == 0) {
1250 ERR("Trying to get packet before version check");
1251 ret = -1;
1252 goto end;
1253 }
1254
1255 health_code_update();
1256 ret = cmd->sock->ops->recvmsg(cmd->sock, &get_packet_info,
1257 sizeof(get_packet_info), 0);
1258 if (ret < 0 || ret != sizeof(get_packet_info)) {
1259 ret = -1;
1260 ERR("Relay didn't receive the whole packet");
1261 goto end;
1262 }
1263 health_code_update();
1264
1265 /* From this point on, the error label can be reached. */
1266 memset(&reply, 0, sizeof(reply));
1267
1268 rcu_read_lock();
1269 stream = live_find_viewer_stream_by_id(be64toh(get_packet_info.stream_id));
1270 if (!stream) {
1271 goto error;
1272 }
1273 assert(stream->ctf_trace);
1274
1275 /*
1276 * First time we read this stream, we need open the tracefile, we should
1277 * only arrive here if an index has already been sent to the viewer, so the
1278 * tracefile must exist, if it does not it is a fatal error.
1279 */
1280 if (stream->read_fd < 0) {
1281 char fullpath[PATH_MAX];
1282
1283 if (stream->tracefile_count > 0) {
1284 ret = snprintf(fullpath, PATH_MAX, "%s/%s_%" PRIu64, stream->path_name,
1285 stream->channel_name,
1286 stream->tracefile_count_current);
1287 } else {
1288 ret = snprintf(fullpath, PATH_MAX, "%s/%s", stream->path_name,
1289 stream->channel_name);
1290 }
1291 if (ret < 0) {
1292 goto error;
1293 }
1294 ret = open(fullpath, O_RDONLY);
1295 if (ret < 0) {
1296 PERROR("Relay opening trace file");
1297 goto error;
1298 }
1299 stream->read_fd = ret;
1300 }
1301
1302 if (!stream->ctf_trace->metadata_received ||
1303 stream->ctf_trace->metadata_received >
1304 stream->ctf_trace->metadata_sent) {
1305 reply.status = htobe32(VIEWER_GET_PACKET_ERR);
1306 reply.flags |= LTTNG_VIEWER_FLAG_NEW_METADATA;
1307 goto send_reply;
1308 }
1309
1310 len = be32toh(get_packet_info.len);
1311 data = zmalloc(len);
1312 if (!data) {
1313 PERROR("relay data zmalloc");
1314 goto error;
1315 }
1316
1317 ret = lseek(stream->read_fd, be64toh(get_packet_info.offset), SEEK_SET);
1318 if (ret < 0) {
1319 /*
1320 * If the read fd was closed by the streaming side, the
1321 * abort_flag will be set to 1, otherwise it is an error.
1322 */
1323 if (stream->abort_flag == 0) {
1324 PERROR("lseek");
1325 goto error;
1326 }
1327 reply.status = htobe32(VIEWER_GET_PACKET_EOF);
1328 goto send_reply;
1329 }
1330 read_len = lttng_read(stream->read_fd, data, len);
1331 if (read_len < len) {
1332 /*
1333 * If the read fd was closed by the streaming side, the
1334 * abort_flag will be set to 1, otherwise it is an error.
1335 */
1336 if (stream->abort_flag == 0) {
1337 PERROR("Relay reading trace file, fd: %d, offset: %" PRIu64,
1338 stream->read_fd,
1339 be64toh(get_packet_info.offset));
1340 goto error;
1341 } else {
1342 reply.status = htobe32(VIEWER_GET_PACKET_EOF);
1343 goto send_reply;
1344 }
1345 }
1346 reply.status = htobe32(VIEWER_GET_PACKET_OK);
1347 reply.len = htobe32(len);
1348 send_data = 1;
1349 goto send_reply;
1350
1351 error:
1352 reply.status = htobe32(VIEWER_GET_PACKET_ERR);
1353
1354 send_reply:
1355 reply.flags = htobe32(reply.flags);
1356
1357 health_code_update();
1358 ret = cmd->sock->ops->sendmsg(cmd->sock, &reply, sizeof(reply), 0);
1359 if (ret < 0) {
1360 ERR("Relay data header to viewer");
1361 goto end_unlock;
1362 }
1363 health_code_update();
1364
1365 if (send_data) {
1366 health_code_update();
1367 ret = cmd->sock->ops->sendmsg(cmd->sock, data, len, 0);
1368 if (ret < 0) {
1369 ERR("Relay send data to viewer");
1370 goto end_unlock;
1371 }
1372 health_code_update();
1373 }
1374
1375 DBG("Sent %u bytes for stream %" PRIu64, len,
1376 be64toh(get_packet_info.stream_id));
1377
1378 end_unlock:
1379 free(data);
1380 rcu_read_unlock();
1381
1382 end:
1383 return ret;
1384 }
1385
1386 /*
1387 * Send the session's metadata
1388 *
1389 * Return 0 on success else a negative value.
1390 */
1391 static
1392 int viewer_get_metadata(struct relay_command *cmd)
1393 {
1394 int ret = 0;
1395 ssize_t read_len;
1396 uint64_t len = 0;
1397 char *data = NULL;
1398 struct lttng_viewer_get_metadata request;
1399 struct lttng_viewer_metadata_packet reply;
1400 struct relay_viewer_stream *stream;
1401
1402 assert(cmd);
1403
1404 DBG("Relay get metadata");
1405
1406 if (cmd->version_check_done == 0) {
1407 ERR("Trying to get metadata before version check");
1408 ret = -1;
1409 goto end;
1410 }
1411
1412 health_code_update();
1413 ret = cmd->sock->ops->recvmsg(cmd->sock, &request,
1414 sizeof(request), 0);
1415 if (ret < 0 || ret != sizeof(request)) {
1416 ret = -1;
1417 ERR("Relay didn't receive the whole packet");
1418 goto end;
1419 }
1420 health_code_update();
1421
1422 rcu_read_lock();
1423 stream = live_find_viewer_stream_by_id(be64toh(request.stream_id));
1424 if (!stream || !stream->metadata_flag) {
1425 ERR("Invalid metadata stream");
1426 goto error;
1427 }
1428 assert(stream->ctf_trace);
1429 assert(stream->ctf_trace->metadata_sent <=
1430 stream->ctf_trace->metadata_received);
1431
1432 len = stream->ctf_trace->metadata_received -
1433 stream->ctf_trace->metadata_sent;
1434 if (len == 0) {
1435 reply.status = htobe32(VIEWER_NO_NEW_METADATA);
1436 goto send_reply;
1437 }
1438
1439 /* first time, we open the metadata file */
1440 if (stream->read_fd < 0) {
1441 char fullpath[PATH_MAX];
1442
1443 ret = snprintf(fullpath, PATH_MAX, "%s/%s", stream->path_name,
1444 stream->channel_name);
1445 if (ret < 0) {
1446 goto error;
1447 }
1448 ret = open(fullpath, O_RDONLY);
1449 if (ret < 0) {
1450 PERROR("Relay opening metadata file");
1451 goto error;
1452 }
1453 stream->read_fd = ret;
1454 }
1455
1456 reply.len = htobe64(len);
1457 data = zmalloc(len);
1458 if (!data) {
1459 PERROR("viewer metadata zmalloc");
1460 goto error;
1461 }
1462
1463 read_len = lttng_read(stream->read_fd, data, len);
1464 if (read_len < len) {
1465 PERROR("Relay reading metadata file");
1466 goto error;
1467 }
1468 stream->ctf_trace->metadata_sent += read_len;
1469 reply.status = htobe32(VIEWER_METADATA_OK);
1470 goto send_reply;
1471
1472 error:
1473 reply.status = htobe32(VIEWER_METADATA_ERR);
1474
1475 send_reply:
1476 health_code_update();
1477 ret = cmd->sock->ops->sendmsg(cmd->sock, &reply, sizeof(reply), 0);
1478 if (ret < 0) {
1479 ERR("Relay data header to viewer");
1480 goto end_unlock;
1481 }
1482 health_code_update();
1483
1484 if (len > 0) {
1485 ret = cmd->sock->ops->sendmsg(cmd->sock, data, len, 0);
1486 if (ret < 0) {
1487 ERR("Relay send data to viewer");
1488 goto end_unlock;
1489 }
1490 }
1491
1492 DBG("Sent %" PRIu64 " bytes of metadata for stream %" PRIu64, len,
1493 be64toh(request.stream_id));
1494
1495 DBG("Metadata sent");
1496
1497 end_unlock:
1498 free(data);
1499 rcu_read_unlock();
1500 end:
1501 return ret;
1502 }
1503
1504 /*
1505 * live_relay_unknown_command: send -1 if received unknown command
1506 */
1507 static
1508 void live_relay_unknown_command(struct relay_command *cmd)
1509 {
1510 struct lttcomm_relayd_generic_reply reply;
1511 int ret;
1512
1513 reply.ret_code = htobe32(LTTNG_ERR_UNK);
1514 ret = cmd->sock->ops->sendmsg(cmd->sock, &reply,
1515 sizeof(struct lttcomm_relayd_generic_reply), 0);
1516 if (ret < 0) {
1517 ERR("Relay sending unknown command");
1518 }
1519 }
1520
1521 /*
1522 * Process the commands received on the control socket
1523 */
1524 static
1525 int process_control(struct lttng_viewer_cmd *recv_hdr,
1526 struct relay_command *cmd, struct lttng_ht *sessions_ht)
1527 {
1528 int ret = 0;
1529
1530 switch (be32toh(recv_hdr->cmd)) {
1531 case VIEWER_CONNECT:
1532 ret = viewer_connect(cmd);
1533 break;
1534 case VIEWER_LIST_SESSIONS:
1535 ret = viewer_list_sessions(cmd, sessions_ht);
1536 break;
1537 case VIEWER_ATTACH_SESSION:
1538 ret = viewer_attach_session(cmd, sessions_ht);
1539 break;
1540 case VIEWER_GET_NEXT_INDEX:
1541 ret = viewer_get_next_index(cmd, sessions_ht);
1542 break;
1543 case VIEWER_GET_PACKET:
1544 ret = viewer_get_packet(cmd);
1545 break;
1546 case VIEWER_GET_METADATA:
1547 ret = viewer_get_metadata(cmd);
1548 break;
1549 default:
1550 ERR("Received unknown viewer command (%u)", be32toh(recv_hdr->cmd));
1551 live_relay_unknown_command(cmd);
1552 ret = -1;
1553 goto end;
1554 }
1555
1556 end:
1557 return ret;
1558 }
1559
1560 static
1561 void cleanup_poll_connection(struct lttng_poll_event *events, int pollfd)
1562 {
1563 int ret;
1564
1565 assert(events);
1566
1567 lttng_poll_del(events, pollfd);
1568
1569 ret = close(pollfd);
1570 if (ret < 0) {
1571 ERR("Closing pollfd %d", pollfd);
1572 }
1573 }
1574
1575 /*
1576 * Create and add connection to the given hash table.
1577 *
1578 * Return poll add value or else -1 on error.
1579 */
1580 static
1581 int add_connection(int fd, struct lttng_poll_event *events,
1582 struct lttng_ht *relay_connections_ht)
1583 {
1584 int ret;
1585 struct relay_command *relay_connection;
1586
1587 assert(events);
1588 assert(relay_connections_ht);
1589
1590 relay_connection = zmalloc(sizeof(struct relay_command));
1591 if (relay_connection == NULL) {
1592 PERROR("Relay command zmalloc");
1593 goto error;
1594 }
1595
1596 ret = lttng_read(fd, relay_connection, sizeof(*relay_connection));
1597 if (ret < sizeof(*relay_connection)) {
1598 PERROR("read relay cmd pipe");
1599 goto error_read;
1600 }
1601
1602 lttng_ht_node_init_ulong(&relay_connection->sock_n,
1603 (unsigned long) relay_connection->sock->fd);
1604 rcu_read_lock();
1605 lttng_ht_add_unique_ulong(relay_connections_ht,
1606 &relay_connection->sock_n);
1607 rcu_read_unlock();
1608
1609 return lttng_poll_add(events, relay_connection->sock->fd,
1610 LPOLLIN | LPOLLRDHUP);
1611
1612 error_read:
1613 free(relay_connection);
1614 error:
1615 return -1;
1616 }
1617
1618 static
1619 void deferred_free_connection(struct rcu_head *head)
1620 {
1621 struct relay_command *relay_connection =
1622 caa_container_of(head, struct relay_command, rcu_node);
1623
1624 if (relay_connection->session &&
1625 relay_connection->session->viewer_attached > 0) {
1626 relay_connection->session->viewer_attached--;
1627 }
1628 lttcomm_destroy_sock(relay_connection->sock);
1629 free(relay_connection);
1630 }
1631
1632 static
1633 void deferred_free_viewer_stream(struct rcu_head *head)
1634 {
1635 struct relay_viewer_stream *stream =
1636 caa_container_of(head, struct relay_viewer_stream, rcu_node);
1637
1638 if (stream->ctf_trace) {
1639 uatomic_dec(&stream->ctf_trace->refcount);
1640 assert(uatomic_read(&stream->ctf_trace->refcount) >= 0);
1641 if (uatomic_read(&stream->ctf_trace->refcount) == 0) {
1642 DBG("Freeing ctf_trace %" PRIu64, stream->ctf_trace->id);
1643 free(stream->ctf_trace);
1644 }
1645 }
1646
1647 free(stream->path_name);
1648 free(stream->channel_name);
1649 free(stream);
1650 }
1651
1652 static
1653 void viewer_del_streams(uint64_t session_id)
1654 {
1655 int ret;
1656 struct relay_viewer_stream *stream;
1657 struct lttng_ht_node_u64 *node;
1658 struct lttng_ht_iter iter;
1659
1660 rcu_read_lock();
1661 cds_lfht_for_each_entry(viewer_streams_ht->ht, &iter.iter, node, node) {
1662 health_code_update();
1663
1664 node = lttng_ht_iter_get_node_u64(&iter);
1665 if (!node) {
1666 continue;
1667 }
1668
1669 stream = caa_container_of(node, struct relay_viewer_stream, stream_n);
1670 if (stream->session_id != session_id) {
1671 continue;
1672 }
1673
1674 if (stream->read_fd >= 0) {
1675 ret = close(stream->read_fd);
1676 if (ret < 0) {
1677 PERROR("close read_fd");
1678 }
1679 }
1680 if (stream->index_read_fd >= 0) {
1681 ret = close(stream->index_read_fd);
1682 if (ret < 0) {
1683 PERROR("close index_read_fd");
1684 }
1685 }
1686 if (stream->metadata_flag && stream->ctf_trace) {
1687 stream->ctf_trace->metadata_sent = 0;
1688 }
1689 ret = lttng_ht_del(viewer_streams_ht, &iter);
1690 assert(!ret);
1691 call_rcu(&stream->rcu_node, deferred_free_viewer_stream);
1692 }
1693 rcu_read_unlock();
1694 }
1695
1696 /*
1697 * Delete and free a connection.
1698 *
1699 * RCU read side lock MUST be acquired.
1700 */
1701 static
1702 void del_connection(struct lttng_ht *relay_connections_ht,
1703 struct lttng_ht_iter *iter, struct relay_command *relay_connection)
1704 {
1705 int ret;
1706
1707 assert(relay_connections_ht);
1708 assert(iter);
1709 assert(relay_connection);
1710
1711 ret = lttng_ht_del(relay_connections_ht, iter);
1712 assert(!ret);
1713
1714 viewer_del_streams(relay_connection->session_id);
1715
1716 call_rcu(&relay_connection->rcu_node, deferred_free_connection);
1717 }
1718
1719 /*
1720 * This thread does the actual work
1721 */
1722 static
1723 void *thread_worker(void *data)
1724 {
1725 int ret, err = -1;
1726 uint32_t nb_fd;
1727 struct relay_command *relay_connection;
1728 struct lttng_poll_event events;
1729 struct lttng_ht *relay_connections_ht;
1730 struct lttng_ht_node_ulong *node;
1731 struct lttng_ht_iter iter;
1732 struct lttng_viewer_cmd recv_hdr;
1733 struct relay_local_data *relay_ctx = (struct relay_local_data *) data;
1734 struct lttng_ht *sessions_ht = relay_ctx->sessions_ht;
1735
1736 DBG("[thread] Live viewer relay worker started");
1737
1738 rcu_register_thread();
1739
1740 health_register(health_relayd, HEALTH_RELAYD_TYPE_LIVE_WORKER);
1741
1742 /* table of connections indexed on socket */
1743 relay_connections_ht = lttng_ht_new(0, LTTNG_HT_TYPE_ULONG);
1744 if (!relay_connections_ht) {
1745 goto relay_connections_ht_error;
1746 }
1747
1748 ret = create_thread_poll_set(&events, 2);
1749 if (ret < 0) {
1750 goto error_poll_create;
1751 }
1752
1753 ret = lttng_poll_add(&events, live_relay_cmd_pipe[0], LPOLLIN | LPOLLRDHUP);
1754 if (ret < 0) {
1755 goto error;
1756 }
1757
1758 restart:
1759 while (1) {
1760 int i;
1761
1762 health_code_update();
1763
1764 /* Infinite blocking call, waiting for transmission */
1765 DBG3("Relayd live viewer worker thread polling...");
1766 health_poll_entry();
1767 ret = lttng_poll_wait(&events, -1);
1768 health_poll_exit();
1769 if (ret < 0) {
1770 /*
1771 * Restart interrupted system call.
1772 */
1773 if (errno == EINTR) {
1774 goto restart;
1775 }
1776 goto error;
1777 }
1778
1779 nb_fd = ret;
1780
1781 /*
1782 * Process control. The control connection is prioritised so we don't
1783 * starve it with high throughput tracing data on the data
1784 * connection.
1785 */
1786 for (i = 0; i < nb_fd; i++) {
1787 /* Fetch once the poll data */
1788 uint32_t revents = LTTNG_POLL_GETEV(&events, i);
1789 int pollfd = LTTNG_POLL_GETFD(&events, i);
1790
1791 health_code_update();
1792
1793 /* Thread quit pipe has been closed. Killing thread. */
1794 ret = check_thread_quit_pipe(pollfd, revents);
1795 if (ret) {
1796 err = 0;
1797 goto exit;
1798 }
1799
1800 /* Inspect the relay cmd pipe for new connection */
1801 if (pollfd == live_relay_cmd_pipe[0]) {
1802 if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
1803 ERR("Relay live pipe error");
1804 goto error;
1805 } else if (revents & LPOLLIN) {
1806 DBG("Relay live viewer command received");
1807 ret = add_connection(live_relay_cmd_pipe[0],
1808 &events, relay_connections_ht);
1809 if (ret < 0) {
1810 goto error;
1811 }
1812 }
1813 } else if (revents) {
1814 rcu_read_lock();
1815 lttng_ht_lookup(relay_connections_ht,
1816 (void *)((unsigned long) pollfd), &iter);
1817 node = lttng_ht_iter_get_node_ulong(&iter);
1818 if (node == NULL) {
1819 DBG2("Relay viewer sock %d not found", pollfd);
1820 rcu_read_unlock();
1821 goto error;
1822 }
1823 relay_connection = caa_container_of(node, struct relay_command,
1824 sock_n);
1825
1826 if (revents & (LPOLLERR)) {
1827 cleanup_poll_connection(&events, pollfd);
1828 del_connection(relay_connections_ht, &iter,
1829 relay_connection);
1830 } else if (revents & (LPOLLHUP | LPOLLRDHUP)) {
1831 DBG("Viewer socket %d hung up", pollfd);
1832 cleanup_poll_connection(&events, pollfd);
1833 del_connection(relay_connections_ht, &iter,
1834 relay_connection);
1835 } else if (revents & LPOLLIN) {
1836 ret = relay_connection->sock->ops->recvmsg(
1837 relay_connection->sock, &recv_hdr,
1838 sizeof(struct lttng_viewer_cmd),
1839 0);
1840 /* connection closed */
1841 if (ret <= 0) {
1842 cleanup_poll_connection(&events, pollfd);
1843 del_connection(relay_connections_ht, &iter,
1844 relay_connection);
1845 DBG("Viewer control connection closed with %d",
1846 pollfd);
1847 } else {
1848 if (relay_connection->session) {
1849 DBG2("Relay viewer worker receiving data for "
1850 "session: %" PRIu64,
1851 relay_connection->session->id);
1852 }
1853 ret = process_control(&recv_hdr, relay_connection,
1854 sessions_ht);
1855 if (ret < 0) {
1856 /* Clear the session on error. */
1857 cleanup_poll_connection(&events, pollfd);
1858 del_connection(relay_connections_ht, &iter,
1859 relay_connection);
1860 DBG("Viewer connection closed with %d", pollfd);
1861 }
1862 }
1863 }
1864 rcu_read_unlock();
1865 }
1866 }
1867 }
1868
1869 exit:
1870 error:
1871 lttng_poll_clean(&events);
1872
1873 /* empty the hash table and free the memory */
1874 rcu_read_lock();
1875 cds_lfht_for_each_entry(relay_connections_ht->ht, &iter.iter, node, node) {
1876 health_code_update();
1877
1878 node = lttng_ht_iter_get_node_ulong(&iter);
1879 if (!node) {
1880 continue;
1881 }
1882
1883 relay_connection = caa_container_of(node, struct relay_command,
1884 sock_n);
1885 del_connection(relay_connections_ht, &iter, relay_connection);
1886 }
1887 rcu_read_unlock();
1888 error_poll_create:
1889 lttng_ht_destroy(relay_connections_ht);
1890 relay_connections_ht_error:
1891 /* Close relay cmd pipes */
1892 utils_close_pipe(live_relay_cmd_pipe);
1893 if (err) {
1894 DBG("Viewer worker thread exited with error");
1895 }
1896 DBG("Viewer worker thread cleanup complete");
1897 if (err) {
1898 health_error();
1899 ERR("Health error occurred in %s", __func__);
1900 }
1901 health_unregister(health_relayd);
1902 stop_threads();
1903 rcu_unregister_thread();
1904 return NULL;
1905 }
1906
1907 /*
1908 * Create the relay command pipe to wake thread_manage_apps.
1909 * Closed in cleanup().
1910 */
1911 static int create_relay_cmd_pipe(void)
1912 {
1913 int ret;
1914
1915 ret = utils_create_pipe_cloexec(live_relay_cmd_pipe);
1916
1917 return ret;
1918 }
1919
1920 void live_stop_threads(void)
1921 {
1922 int ret;
1923 void *status;
1924
1925 stop_threads();
1926
1927 ret = pthread_join(live_listener_thread, &status);
1928 if (ret != 0) {
1929 PERROR("pthread_join live listener");
1930 goto error; /* join error, exit without cleanup */
1931 }
1932
1933 ret = pthread_join(live_worker_thread, &status);
1934 if (ret != 0) {
1935 PERROR("pthread_join live worker");
1936 goto error; /* join error, exit without cleanup */
1937 }
1938
1939 ret = pthread_join(live_dispatcher_thread, &status);
1940 if (ret != 0) {
1941 PERROR("pthread_join live dispatcher");
1942 goto error; /* join error, exit without cleanup */
1943 }
1944
1945 cleanup();
1946
1947 error:
1948 return;
1949 }
1950
1951 /*
1952 * main
1953 */
1954 int live_start_threads(struct lttng_uri *uri,
1955 struct relay_local_data *relay_ctx, int quit_pipe[2])
1956 {
1957 int ret = 0;
1958 void *status;
1959 int is_root;
1960
1961 assert(uri);
1962 live_uri = uri;
1963
1964 live_thread_quit_pipe[0] = quit_pipe[0];
1965 live_thread_quit_pipe[1] = quit_pipe[1];
1966
1967 /* Check if daemon is UID = 0 */
1968 is_root = !getuid();
1969
1970 if (!is_root) {
1971 if (live_uri->port < 1024) {
1972 ERR("Need to be root to use ports < 1024");
1973 ret = -1;
1974 goto exit;
1975 }
1976 }
1977
1978 /* Setup the thread apps communication pipe. */
1979 if ((ret = create_relay_cmd_pipe()) < 0) {
1980 goto exit;
1981 }
1982
1983 /* Init relay command queue. */
1984 cds_wfq_init(&viewer_cmd_queue.queue);
1985
1986 /* Set up max poll set size */
1987 lttng_poll_set_max_size();
1988
1989 /* Setup the dispatcher thread */
1990 ret = pthread_create(&live_dispatcher_thread, NULL,
1991 thread_dispatcher, (void *) NULL);
1992 if (ret != 0) {
1993 PERROR("pthread_create viewer dispatcher");
1994 goto exit_dispatcher;
1995 }
1996
1997 /* Setup the worker thread */
1998 ret = pthread_create(&live_worker_thread, NULL,
1999 thread_worker, relay_ctx);
2000 if (ret != 0) {
2001 PERROR("pthread_create viewer worker");
2002 goto exit_worker;
2003 }
2004
2005 /* Setup the listener thread */
2006 ret = pthread_create(&live_listener_thread, NULL,
2007 thread_listener, (void *) NULL);
2008 if (ret != 0) {
2009 PERROR("pthread_create viewer listener");
2010 goto exit_listener;
2011 }
2012
2013 ret = 0;
2014 goto end;
2015
2016 exit_listener:
2017 ret = pthread_join(live_listener_thread, &status);
2018 if (ret != 0) {
2019 PERROR("pthread_join live listener");
2020 goto error; /* join error, exit without cleanup */
2021 }
2022
2023 exit_worker:
2024 ret = pthread_join(live_worker_thread, &status);
2025 if (ret != 0) {
2026 PERROR("pthread_join live worker");
2027 goto error; /* join error, exit without cleanup */
2028 }
2029
2030 exit_dispatcher:
2031 ret = pthread_join(live_dispatcher_thread, &status);
2032 if (ret != 0) {
2033 PERROR("pthread_join live dispatcher");
2034 goto error; /* join error, exit without cleanup */
2035 }
2036
2037 exit:
2038 cleanup();
2039
2040 end:
2041 error:
2042 return ret;
2043 }
LTTng tools - Deliverables
This page took 0.101204 seconds and 5 git commands to generate.