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