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