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