2 * Copyright (C) 2013 - Julien Desfossez <jdesfossez@efficios.com>
3 * David Goulet <dgoulet@efficios.com>
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.
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
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.
29 #include <sys/mount.h>
30 #include <sys/resource.h>
31 #include <sys/socket.h>
33 #include <sys/types.h>
36 #include <urcu/futex.h>
37 #include <urcu/uatomic.h>
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>
56 #include "lttng-relayd.h"
57 #include "lttng-viewer.h"
59 #include "health-relayd.h"
61 static struct lttng_uri
*live_uri
;
64 * This pipe is used to inform the worker thread that a command is queued and
65 * ready to be processed.
67 static int live_relay_cmd_pipe
[2] = { -1, -1 };
69 /* Shared between threads */
70 static int live_dispatch_thread_exit
;
72 static pthread_t live_listener_thread
;
73 static pthread_t live_dispatcher_thread
;
74 static pthread_t live_worker_thread
;
77 * Relay command queue.
79 * The live_thread_listener and live_thread_dispatcher communicate with this
82 static struct relay_cmd_queue viewer_cmd_queue
;
84 static uint64_t last_relay_viewer_session_id
;
98 * Write to writable pipe used to notify a thread.
101 int notify_thread_pipe(int wpipe
)
105 ret
= lttng_write(wpipe
, "!", 1);
107 PERROR("write poll pipe");
114 * Stop all threads by closing the thread quit pipe.
117 void stop_threads(void)
121 /* Stopping all threads */
122 DBG("Terminating all live threads");
123 ret
= notify_thread_pipe(thread_quit_pipe
[1]);
125 ERR("write error on thread quit pipe");
128 /* Dispatch thread */
129 CMM_STORE_SHARED(live_dispatch_thread_exit
, 1);
130 futex_nto1_wake(&viewer_cmd_queue
.futex
);
134 * Create a poll set with O_CLOEXEC and add the thread quit pipe to the set.
137 int create_thread_poll_set(struct lttng_poll_event
*events
, int size
)
141 if (events
== NULL
|| size
== 0) {
146 ret
= lttng_poll_create(events
, size
, LTTNG_CLOEXEC
);
152 ret
= lttng_poll_add(events
, thread_quit_pipe
[0], LPOLLIN
| LPOLLERR
);
164 * Check if the thread quit pipe was triggered.
166 * Return 1 if it was triggered else 0;
169 int check_thread_quit_pipe(int fd
, uint32_t events
)
171 if (fd
== thread_quit_pipe
[0] && (events
& LPOLLIN
)) {
179 * Create and init socket from uri.
182 struct lttcomm_sock
*init_socket(struct lttng_uri
*uri
)
185 struct lttcomm_sock
*sock
= NULL
;
187 sock
= lttcomm_alloc_sock_from_uri(uri
);
189 ERR("Allocating socket");
193 ret
= lttcomm_create_sock(sock
);
197 DBG("Listening on sock %d for live", sock
->fd
);
199 ret
= sock
->ops
->bind(sock
);
204 ret
= sock
->ops
->listen(sock
, -1);
214 lttcomm_destroy_sock(sock
);
220 * This thread manages the listening for new connections on the network
223 void *thread_listener(void *data
)
225 int i
, ret
, pollfd
, err
= -1;
227 uint32_t revents
, nb_fd
;
228 struct lttng_poll_event events
;
229 struct lttcomm_sock
*live_control_sock
;
231 DBG("[thread] Relay live listener started");
233 health_register(health_relayd
, HEALTH_RELAYD_TYPE_LIVE_LISTENER
);
235 health_code_update();
237 live_control_sock
= init_socket(live_uri
);
238 if (!live_control_sock
) {
239 goto error_sock_control
;
242 /* Pass 2 as size here for the thread quit pipe and control sockets. */
243 ret
= create_thread_poll_set(&events
, 2);
245 goto error_create_poll
;
248 /* Add the control socket */
249 ret
= lttng_poll_add(&events
, live_control_sock
->fd
, LPOLLIN
| LPOLLRDHUP
);
254 lttng_relay_notify_ready();
257 health_code_update();
259 DBG("Listener accepting live viewers connections");
263 ret
= lttng_poll_wait(&events
, -1);
267 * Restart interrupted system call.
269 if (errno
== EINTR
) {
276 DBG("Relay new viewer connection received");
277 for (i
= 0; i
< nb_fd
; i
++) {
278 health_code_update();
280 /* Fetch once the poll data */
281 revents
= LTTNG_POLL_GETEV(&events
, i
);
282 pollfd
= LTTNG_POLL_GETFD(&events
, i
);
284 /* Thread quit pipe has been closed. Killing thread. */
285 ret
= check_thread_quit_pipe(pollfd
, revents
);
291 if (revents
& (LPOLLERR
| LPOLLHUP
| LPOLLRDHUP
)) {
292 ERR("socket poll error");
294 } else if (revents
& LPOLLIN
) {
296 * Get allocated in this thread, enqueued to a global queue,
297 * dequeued and freed in the worker thread.
299 struct relay_command
*relay_cmd
;
300 struct lttcomm_sock
*newsock
;
302 relay_cmd
= zmalloc(sizeof(*relay_cmd
));
304 PERROR("relay command zmalloc");
308 assert(pollfd
== live_control_sock
->fd
);
309 newsock
= live_control_sock
->ops
->accept(live_control_sock
);
311 PERROR("accepting control sock");
315 DBG("Relay viewer connection accepted socket %d", newsock
->fd
);
316 ret
= setsockopt(newsock
->fd
, SOL_SOCKET
, SO_REUSEADDR
, &val
,
319 PERROR("setsockopt inet");
320 lttcomm_destroy_sock(newsock
);
324 relay_cmd
->sock
= newsock
;
327 * Lock free enqueue the request.
329 cds_wfq_enqueue(&viewer_cmd_queue
.queue
, &relay_cmd
->node
);
332 * Wake the dispatch queue futex. Implicit memory
333 * barrier with the exchange in cds_wfq_enqueue.
335 futex_nto1_wake(&viewer_cmd_queue
.futex
);
343 lttng_poll_clean(&events
);
345 if (live_control_sock
->fd
>= 0) {
346 ret
= live_control_sock
->ops
->close(live_control_sock
);
351 lttcomm_destroy_sock(live_control_sock
);
355 DBG("Live viewer listener thread exited with error");
357 health_unregister(health_relayd
);
358 DBG("Live viewer listener thread cleanup complete");
364 * This thread manages the dispatching of the requests to worker threads
367 void *thread_dispatcher(void *data
)
371 struct cds_wfq_node
*node
;
372 struct relay_command
*relay_cmd
= NULL
;
374 DBG("[thread] Live viewer relay dispatcher started");
376 health_register(health_relayd
, HEALTH_RELAYD_TYPE_LIVE_DISPATCHER
);
378 health_code_update();
380 while (!CMM_LOAD_SHARED(live_dispatch_thread_exit
)) {
381 health_code_update();
383 /* Atomically prepare the queue futex */
384 futex_nto1_prepare(&viewer_cmd_queue
.futex
);
387 health_code_update();
389 /* Dequeue commands */
390 node
= cds_wfq_dequeue_blocking(&viewer_cmd_queue
.queue
);
392 DBG("Woken up but nothing in the live-viewer "
393 "relay command queue");
394 /* Continue thread execution */
398 relay_cmd
= caa_container_of(node
, struct relay_command
, node
);
399 DBG("Dispatching viewer request waiting on sock %d",
400 relay_cmd
->sock
->fd
);
403 * Inform worker thread of the new request. This call is blocking
404 * so we can be assured that the data will be read at some point in
405 * time or wait to the end of the world :)
407 ret
= lttng_write(live_relay_cmd_pipe
[1], relay_cmd
,
410 if (ret
< sizeof(struct relay_command
)) {
411 PERROR("write cmd pipe");
414 } while (node
!= NULL
);
416 /* Futex wait on queue. Blocking call on futex() */
418 futex_nto1_wait(&viewer_cmd_queue
.futex
);
422 /* Normal exit, no error */
428 ERR("Health error occurred in %s", __func__
);
430 health_unregister(health_relayd
);
431 DBG("Live viewer dispatch thread dying");
437 * Establish connection with the viewer and check the versions.
439 * Return 0 on success or else negative value.
442 int viewer_connect(struct relay_command
*cmd
)
445 struct lttng_viewer_connect reply
, msg
;
449 cmd
->version_check_done
= 1;
451 health_code_update();
453 /* Get version from the other side. */
454 ret
= cmd
->sock
->ops
->recvmsg(cmd
->sock
, &msg
, sizeof(msg
), 0);
455 if (ret
< 0 || ret
!= sizeof(msg
)) {
457 /* Orderly shutdown. Not necessary to print an error. */
458 DBG("Socket %d did an orderly shutdown", cmd
->sock
->fd
);
460 ERR("Relay failed to receive the version values.");
466 health_code_update();
468 reply
.major
= RELAYD_VERSION_COMM_MAJOR
;
469 reply
.minor
= RELAYD_VERSION_COMM_MINOR
;
471 /* Major versions must be the same */
472 if (reply
.major
!= be32toh(msg
.major
)) {
473 DBG("Incompatible major versions (%u vs %u)", reply
.major
,
479 cmd
->major
= reply
.major
;
480 /* We adapt to the lowest compatible version */
481 if (reply
.minor
<= be32toh(msg
.minor
)) {
482 cmd
->minor
= reply
.minor
;
484 cmd
->minor
= be32toh(msg
.minor
);
487 if (be32toh(msg
.type
) == VIEWER_CLIENT_COMMAND
) {
488 cmd
->type
= RELAY_VIEWER_COMMAND
;
489 } else if (be32toh(msg
.type
) == VIEWER_CLIENT_NOTIFICATION
) {
490 cmd
->type
= RELAY_VIEWER_NOTIFICATION
;
492 ERR("Unknown connection type : %u", be32toh(msg
.type
));
497 reply
.major
= htobe32(reply
.major
);
498 reply
.minor
= htobe32(reply
.minor
);
499 if (cmd
->type
== RELAY_VIEWER_COMMAND
) {
500 reply
.viewer_session_id
= htobe64(++last_relay_viewer_session_id
);
503 health_code_update();
505 ret
= cmd
->sock
->ops
->sendmsg(cmd
->sock
, &reply
,
506 sizeof(struct lttng_viewer_connect
), 0);
508 ERR("Relay sending version");
511 health_code_update();
513 DBG("Version check done using protocol %u.%u", cmd
->major
, cmd
->minor
);
521 * Send the viewer the list of current sessions.
523 * Return 0 on success or else a negative value.
526 int viewer_list_sessions(struct relay_command
*cmd
,
527 struct lttng_ht
*sessions_ht
)
530 struct lttng_viewer_list_sessions session_list
;
532 long approx_before
, approx_after
;
533 struct lttng_ht_node_ulong
*node
;
534 struct lttng_ht_iter iter
;
535 struct lttng_viewer_session send_session
;
536 struct relay_session
*session
;
538 DBG("List sessions received");
540 if (cmd
->version_check_done
== 0) {
541 ERR("Trying to list sessions before version check");
547 cds_lfht_count_nodes(sessions_ht
->ht
, &approx_before
, &count
, &approx_after
);
548 session_list
.sessions_count
= htobe32(count
);
550 health_code_update();
552 ret
= cmd
->sock
->ops
->sendmsg(cmd
->sock
, &session_list
,
553 sizeof(session_list
), 0);
555 ERR("Relay sending sessions list");
559 health_code_update();
561 cds_lfht_for_each_entry(sessions_ht
->ht
, &iter
.iter
, node
, node
) {
562 health_code_update();
564 node
= lttng_ht_iter_get_node_ulong(&iter
);
568 session
= caa_container_of(node
, struct relay_session
, session_n
);
570 strncpy(send_session
.session_name
, session
->session_name
,
571 sizeof(send_session
.session_name
));
572 strncpy(send_session
.hostname
, session
->hostname
,
573 sizeof(send_session
.hostname
));
574 send_session
.id
= htobe64(session
->id
);
575 send_session
.live_timer
= htobe32(session
->live_timer
);
576 send_session
.clients
= htobe32(session
->viewer_attached
);
577 send_session
.streams
= htobe32(session
->stream_count
);
579 health_code_update();
581 ret
= cmd
->sock
->ops
->sendmsg(cmd
->sock
, &send_session
,
582 sizeof(send_session
), 0);
584 ERR("Relay sending session info");
588 health_code_update();
603 * Open index file using a given viewer stream.
605 * Return 0 on success or else a negative value.
607 static int open_index(struct relay_viewer_stream
*stream
)
610 char fullpath
[PATH_MAX
];
611 struct ctf_packet_index_file_hdr hdr
;
613 if (stream
->tracefile_count
> 0) {
614 ret
= snprintf(fullpath
, sizeof(fullpath
), "%s/" DEFAULT_INDEX_DIR
"/%s_%"
615 PRIu64 DEFAULT_INDEX_FILE_SUFFIX
, stream
->path_name
,
616 stream
->channel_name
, stream
->tracefile_count_current
);
618 ret
= snprintf(fullpath
, sizeof(fullpath
), "%s/" DEFAULT_INDEX_DIR
"/%s"
619 DEFAULT_INDEX_FILE_SUFFIX
, stream
->path_name
,
620 stream
->channel_name
);
623 PERROR("snprintf index path");
627 DBG("Opening index file %s in read only", fullpath
);
628 ret
= open(fullpath
, O_RDONLY
);
630 if (errno
== ENOENT
) {
634 PERROR("opening index in read-only");
638 stream
->index_read_fd
= ret
;
639 DBG("Opening index file %s in read only, (fd: %d)", fullpath
, ret
);
641 ret
= lttng_read(stream
->index_read_fd
, &hdr
, sizeof(hdr
));
642 if (ret
< sizeof(hdr
)) {
643 PERROR("Reading index header");
646 if (be32toh(hdr
.magic
) != CTF_INDEX_MAGIC
) {
647 ERR("Invalid header magic");
651 if (be32toh(hdr
.index_major
) != CTF_INDEX_MAJOR
||
652 be32toh(hdr
.index_minor
) != CTF_INDEX_MINOR
) {
653 ERR("Invalid header version");
664 * Allocate and init a new viewer_stream.
666 * Copies the values from the stream passed in parameter and insert the new
667 * stream in the viewer_streams_ht.
669 * MUST be called with rcu_read_lock held.
671 * Returns 0 on success or a negative value on error.
674 int init_viewer_stream(struct relay_stream
*stream
, int seek_last
)
677 struct relay_viewer_stream
*viewer_stream
;
681 viewer_stream
= zmalloc(sizeof(*viewer_stream
));
682 if (!viewer_stream
) {
683 PERROR("relay viewer stream zmalloc");
687 viewer_stream
->session_id
= stream
->session
->id
;
688 viewer_stream
->stream_handle
= stream
->stream_handle
;
689 viewer_stream
->path_name
= strndup(stream
->path_name
,
690 LTTNG_VIEWER_PATH_MAX
);
691 viewer_stream
->channel_name
= strndup(stream
->channel_name
,
692 LTTNG_VIEWER_NAME_MAX
);
693 viewer_stream
->tracefile_count
= stream
->tracefile_count
;
694 viewer_stream
->metadata_flag
= stream
->metadata_flag
;
695 viewer_stream
->tracefile_count_last
= -1ULL;
697 viewer_stream
->tracefile_count_current
=
698 stream
->tracefile_count_current
;
700 viewer_stream
->tracefile_count_current
=
701 stream
->oldest_tracefile_id
;
704 viewer_stream
->ctf_trace
= stream
->ctf_trace
;
705 if (viewer_stream
->metadata_flag
) {
706 viewer_stream
->ctf_trace
->viewer_metadata_stream
=
709 uatomic_inc(&viewer_stream
->ctf_trace
->refcount
);
711 lttng_ht_node_init_u64(&viewer_stream
->stream_n
, stream
->stream_handle
);
712 lttng_ht_add_unique_u64(viewer_streams_ht
, &viewer_stream
->stream_n
);
714 viewer_stream
->index_read_fd
= -1;
715 viewer_stream
->read_fd
= -1;
718 * This is to avoid a race between the initialization of this object and
719 * the close of the given stream. If the stream is unable to find this
720 * viewer stream when closing, this copy will at least take the latest
722 * We also need that for the seek_last.
724 viewer_stream
->total_index_received
= stream
->total_index_received
;
727 * If we never received an index for the current stream, delay
728 * the opening of the index, otherwise open it right now.
730 if (viewer_stream
->tracefile_count_current
==
731 stream
->tracefile_count_current
&&
732 viewer_stream
->total_index_received
== 0) {
733 viewer_stream
->index_read_fd
= -1;
735 ret
= open_index(viewer_stream
);
741 if (seek_last
&& viewer_stream
->index_read_fd
> 0) {
742 ret
= lseek(viewer_stream
->index_read_fd
,
743 viewer_stream
->total_index_received
*
744 sizeof(struct ctf_packet_index
),
749 viewer_stream
->last_sent_index
=
750 viewer_stream
->total_index_received
;
760 * Rotate a stream to the next tracefile.
762 * Returns 0 on success, 1 on EOF, a negative value on error.
765 int rotate_viewer_stream(struct relay_viewer_stream
*viewer_stream
,
766 struct relay_stream
*stream
)
769 uint64_t tracefile_id
;
771 assert(viewer_stream
);
773 tracefile_id
= (viewer_stream
->tracefile_count_current
+ 1) %
774 viewer_stream
->tracefile_count
;
776 * Detect the last tracefile to open.
778 if (viewer_stream
->tracefile_count_last
!= -1ULL &&
779 viewer_stream
->tracefile_count_last
==
780 viewer_stream
->tracefile_count_current
) {
786 pthread_mutex_lock(&stream
->viewer_stream_rotation_lock
);
789 * The writer and the reader are not working in the same
790 * tracefile, we can read up to EOF, we don't care about the
791 * total_index_received.
793 if (!stream
|| (stream
->tracefile_count_current
!= tracefile_id
)) {
794 viewer_stream
->close_write_flag
= 1;
797 * We are opening a file that is still open in write, make
798 * sure we limit our reading to the number of indexes
801 viewer_stream
->close_write_flag
= 0;
803 viewer_stream
->total_index_received
=
804 stream
->total_index_received
;
807 viewer_stream
->tracefile_count_current
= tracefile_id
;
809 ret
= close(viewer_stream
->index_read_fd
);
811 PERROR("close index file %d",
812 viewer_stream
->index_read_fd
);
814 viewer_stream
->index_read_fd
= -1;
815 ret
= close(viewer_stream
->read_fd
);
817 PERROR("close tracefile %d",
818 viewer_stream
->read_fd
);
820 viewer_stream
->read_fd
= -1;
822 pthread_mutex_lock(&viewer_stream
->overwrite_lock
);
823 viewer_stream
->abort_flag
= 0;
824 pthread_mutex_unlock(&viewer_stream
->overwrite_lock
);
826 viewer_stream
->index_read_fd
= -1;
827 viewer_stream
->read_fd
= -1;
830 pthread_mutex_unlock(&stream
->viewer_stream_rotation_lock
);
832 ret
= open_index(viewer_stream
);
845 * Send the viewer the list of current sessions.
848 int viewer_attach_session(struct relay_command
*cmd
,
849 struct lttng_ht
*sessions_ht
)
851 int ret
, send_streams
= 0;
852 uint32_t nb_streams
= 0, nb_streams_ready
= 0;
853 struct lttng_viewer_attach_session_request request
;
854 struct lttng_viewer_attach_session_response response
;
855 struct lttng_viewer_stream send_stream
;
856 struct relay_stream
*stream
;
857 struct relay_viewer_stream
*viewer_stream
;
858 struct lttng_ht_node_ulong
*node
;
859 struct lttng_ht_node_u64
*node64
;
860 struct lttng_ht_iter iter
;
861 struct relay_session
*session
;
867 DBG("Attach session received");
869 if (cmd
->version_check_done
== 0) {
870 ERR("Trying to attach session before version check");
875 health_code_update();
877 ret
= cmd
->sock
->ops
->recvmsg(cmd
->sock
, &request
, sizeof(request
), 0);
878 if (ret
< 0 || ret
!= sizeof(request
)) {
880 /* Orderly shutdown. Not necessary to print an error. */
881 DBG("Socket %d did an orderly shutdown", cmd
->sock
->fd
);
883 ERR("Relay failed to receive the attach parameters.");
889 health_code_update();
892 lttng_ht_lookup(sessions_ht
,
893 (void *)((unsigned long) be64toh(request
.session_id
)), &iter
);
894 node
= lttng_ht_iter_get_node_ulong(&iter
);
896 DBG("Relay session %" PRIu64
" not found",
897 be64toh(request
.session_id
));
898 response
.status
= htobe32(VIEWER_ATTACH_UNK
);
902 session
= caa_container_of(node
, struct relay_session
, session_n
);
903 if (cmd
->session_id
== session
->id
) {
904 /* Same viewer already attached, just send the stream list. */
906 response
.status
= htobe32(VIEWER_ATTACH_OK
);
907 } else if (session
->viewer_attached
!= 0) {
908 DBG("Already a viewer attached");
909 response
.status
= htobe32(VIEWER_ATTACH_ALREADY
);
911 } else if (session
->live_timer
== 0) {
912 DBG("Not live session");
913 response
.status
= htobe32(VIEWER_ATTACH_NOT_LIVE
);
916 session
->viewer_attached
++;
918 response
.status
= htobe32(VIEWER_ATTACH_OK
);
919 cmd
->session_id
= session
->id
;
920 cmd
->session
= session
;
923 switch (be32toh(request
.seek
)) {
924 case VIEWER_SEEK_BEGINNING
:
925 /* Default behaviour. */
927 case VIEWER_SEEK_LAST
:
931 ERR("Wrong seek parameter");
932 response
.status
= htobe32(VIEWER_ATTACH_SEEK_ERR
);
938 /* We should only be there if we have a session to attach to. */
942 * Fill the viewer_streams_ht to count the number of streams
943 * ready to be sent and avoid concurrency issues on the
944 * relay_streams_ht and don't rely on a total session stream count.
946 cds_lfht_for_each_entry(relay_streams_ht
->ht
, &iter
.iter
, node
, node
) {
947 struct relay_viewer_stream
*vstream
;
949 health_code_update();
951 node
= lttng_ht_iter_get_node_ulong(&iter
);
955 stream
= caa_container_of(node
, struct relay_stream
, stream_n
);
956 if (stream
->session
!= cmd
->session
) {
962 * Don't send streams with no ctf_trace, they are not
965 if (!stream
->ctf_trace
|| !stream
->viewer_ready
) {
970 vstream
= live_find_viewer_stream_by_id(stream
->stream_handle
);
972 ret
= init_viewer_stream(stream
, seek_last
);
979 /* We must have the same amount of existing stream and ready stream. */
980 if (nb_streams
!= nb_streams_ready
) {
983 response
.streams_count
= htobe32(nb_streams
);
987 health_code_update();
988 ret
= cmd
->sock
->ops
->sendmsg(cmd
->sock
, &response
, sizeof(response
), 0);
990 ERR("Relay sending viewer attach response");
993 health_code_update();
996 * Unknown or empty session, just return gracefully, the viewer knows what
999 if (!send_streams
|| !nb_streams
) {
1004 /* We should only be there if we have a session to attach to. */
1006 cds_lfht_for_each_entry(viewer_streams_ht
->ht
, &iter
.iter
, node
, node
) {
1007 health_code_update();
1009 node64
= lttng_ht_iter_get_node_u64(&iter
);
1013 viewer_stream
= caa_container_of(node64
, struct relay_viewer_stream
,
1015 if (viewer_stream
->session_id
!= cmd
->session
->id
) {
1019 send_stream
.id
= htobe64(viewer_stream
->stream_handle
);
1020 send_stream
.ctf_trace_id
= htobe64(viewer_stream
->ctf_trace
->id
);
1021 send_stream
.metadata_flag
= htobe32(viewer_stream
->metadata_flag
);
1022 strncpy(send_stream
.path_name
, viewer_stream
->path_name
,
1023 sizeof(send_stream
.path_name
));
1024 strncpy(send_stream
.channel_name
, viewer_stream
->channel_name
,
1025 sizeof(send_stream
.channel_name
));
1027 ret
= cmd
->sock
->ops
->sendmsg(cmd
->sock
, &send_stream
,
1028 sizeof(send_stream
), 0);
1030 ERR("Relay sending stream %" PRIu64
, viewer_stream
->stream_handle
);
1033 DBG("Sent stream %" PRIu64
" to viewer", viewer_stream
->stream_handle
);
1045 * Get viewer stream from stream id.
1047 * RCU read side lock MUST be acquired.
1049 struct relay_viewer_stream
*live_find_viewer_stream_by_id(uint64_t stream_id
)
1051 struct lttng_ht_node_u64
*node
;
1052 struct lttng_ht_iter iter
;
1053 struct relay_viewer_stream
*stream
= NULL
;
1055 lttng_ht_lookup(viewer_streams_ht
, &stream_id
, &iter
);
1056 node
= lttng_ht_iter_get_node_u64(&iter
);
1058 DBG("Relay viewer stream %" PRIu64
" not found", stream_id
);
1061 stream
= caa_container_of(node
, struct relay_viewer_stream
, stream_n
);
1068 void deferred_free_viewer_stream(struct rcu_head
*head
)
1070 struct relay_viewer_stream
*stream
=
1071 caa_container_of(head
, struct relay_viewer_stream
, rcu_node
);
1073 free(stream
->path_name
);
1074 free(stream
->channel_name
);
1079 void delete_viewer_stream(struct relay_viewer_stream
*vstream
)
1082 struct lttng_ht_iter iter
;
1084 iter
.iter
.node
= &vstream
->stream_n
.node
;
1085 delret
= lttng_ht_del(viewer_streams_ht
, &iter
);
1090 void destroy_viewer_stream(struct relay_viewer_stream
*vstream
)
1092 unsigned long ret_ref
;
1096 ret_ref
= uatomic_add_return(&vstream
->ctf_trace
->refcount
, -1);
1097 assert(ret_ref
>= 0);
1099 if (vstream
->read_fd
>= 0) {
1100 ret
= close(vstream
->read_fd
);
1102 PERROR("close read_fd");
1105 if (vstream
->index_read_fd
>= 0) {
1106 ret
= close(vstream
->index_read_fd
);
1108 PERROR("close index_read_fd");
1113 * If the only stream left in the HT is the metadata stream,
1114 * we need to remove it because we won't detect a EOF for this
1117 if (ret_ref
== 1 && vstream
->ctf_trace
->viewer_metadata_stream
) {
1118 delete_viewer_stream(vstream
->ctf_trace
->viewer_metadata_stream
);
1119 destroy_viewer_stream(vstream
->ctf_trace
->viewer_metadata_stream
);
1120 vstream
->ctf_trace
->metadata_stream
= NULL
;
1121 DBG("Freeing ctf_trace %" PRIu64
, vstream
->ctf_trace
->id
);
1123 * The streaming-side is already closed and we can't receive a new
1124 * stream concurrently at this point (since the session is being
1125 * destroyed), so when we detect the refcount equals 0, we are the
1126 * only owners of the ctf_trace and we can free it ourself.
1128 free(vstream
->ctf_trace
);
1131 call_rcu(&vstream
->rcu_node
, deferred_free_viewer_stream
);
1135 * Send the next index for a stream.
1137 * Return 0 on success or else a negative value.
1140 int viewer_get_next_index(struct relay_command
*cmd
,
1141 struct lttng_ht
*sessions_ht
)
1144 struct lttng_viewer_get_next_index request_index
;
1145 struct lttng_viewer_index viewer_index
;
1146 struct ctf_packet_index packet_index
;
1147 struct relay_viewer_stream
*vstream
;
1148 struct relay_stream
*rstream
;
1151 assert(sessions_ht
);
1153 DBG("Viewer get next index");
1155 if (cmd
->version_check_done
== 0) {
1156 ERR("Trying to request index before version check");
1158 goto end_no_session
;
1161 health_code_update();
1162 ret
= cmd
->sock
->ops
->recvmsg(cmd
->sock
, &request_index
,
1163 sizeof(request_index
), 0);
1164 if (ret
< 0 || ret
!= sizeof(request_index
)) {
1166 ERR("Relay didn't receive the whole packet");
1169 health_code_update();
1172 vstream
= live_find_viewer_stream_by_id(be64toh(request_index
.stream_id
));
1178 memset(&viewer_index
, 0, sizeof(viewer_index
));
1181 * The viewer should not ask for index on metadata stream.
1183 if (vstream
->metadata_flag
) {
1184 viewer_index
.status
= htobe32(VIEWER_INDEX_HUP
);
1188 /* First time, we open the index file */
1189 if (vstream
->index_read_fd
< 0) {
1190 ret
= open_index(vstream
);
1191 if (ret
== -ENOENT
) {
1193 * The index is created only when the first data packet arrives, it
1194 * might not be ready at the beginning of the session
1196 viewer_index
.status
= htobe32(VIEWER_INDEX_RETRY
);
1198 } else if (ret
< 0) {
1199 viewer_index
.status
= htobe32(VIEWER_INDEX_ERR
);
1204 rstream
= relay_stream_find_by_id(vstream
->stream_handle
);
1206 if (vstream
->abort_flag
) {
1207 /* Rotate on abort (overwrite). */
1208 DBG("Viewer rotate because of overwrite");
1209 ret
= rotate_viewer_stream(vstream
, rstream
);
1212 } else if (ret
== 1) {
1213 viewer_index
.status
= htobe32(VIEWER_INDEX_HUP
);
1214 delete_viewer_stream(vstream
);
1215 destroy_viewer_stream(vstream
);
1219 pthread_mutex_lock(&rstream
->viewer_stream_rotation_lock
);
1220 if (rstream
->tracefile_count_current
== vstream
->tracefile_count_current
) {
1221 if (rstream
->beacon_ts_end
!= -1ULL &&
1222 vstream
->last_sent_index
== rstream
->total_index_received
) {
1223 viewer_index
.status
= htobe32(VIEWER_INDEX_INACTIVE
);
1224 viewer_index
.timestamp_end
= htobe64(rstream
->beacon_ts_end
);
1225 pthread_mutex_unlock(&rstream
->viewer_stream_rotation_lock
);
1228 * Reader and writer are working in the same tracefile, so we care
1229 * about the number of index received and sent. Otherwise, we read
1232 } else if (rstream
->total_index_received
<= vstream
->last_sent_index
1233 && !vstream
->close_write_flag
) {
1234 pthread_mutex_unlock(&rstream
->viewer_stream_rotation_lock
);
1235 /* No new index to send, retry later. */
1236 viewer_index
.status
= htobe32(VIEWER_INDEX_RETRY
);
1240 pthread_mutex_unlock(&rstream
->viewer_stream_rotation_lock
);
1241 } else if (!rstream
&& vstream
->close_write_flag
&&
1242 vstream
->total_index_received
== vstream
->last_sent_index
) {
1243 /* Last index sent and current tracefile closed in write */
1244 viewer_index
.status
= htobe32(VIEWER_INDEX_HUP
);
1245 delete_viewer_stream(vstream
);
1246 destroy_viewer_stream(vstream
);
1249 vstream
->close_write_flag
= 1;
1252 if (!vstream
->ctf_trace
->metadata_received
||
1253 vstream
->ctf_trace
->metadata_received
>
1254 vstream
->ctf_trace
->metadata_sent
) {
1255 viewer_index
.flags
|= LTTNG_VIEWER_FLAG_NEW_METADATA
;
1258 pthread_mutex_lock(&vstream
->overwrite_lock
);
1259 if (vstream
->abort_flag
) {
1261 * The file is being overwritten by the writer, we cannot
1264 viewer_index
.status
= htobe32(VIEWER_INDEX_RETRY
);
1265 pthread_mutex_unlock(&vstream
->overwrite_lock
);
1266 ret
= rotate_viewer_stream(vstream
, rstream
);
1269 } else if (ret
== 1) {
1270 viewer_index
.status
= htobe32(VIEWER_INDEX_HUP
);
1271 delete_viewer_stream(vstream
);
1272 destroy_viewer_stream(vstream
);
1277 ret
= lttng_read(vstream
->index_read_fd
, &packet_index
,
1278 sizeof(packet_index
));
1279 pthread_mutex_unlock(&vstream
->overwrite_lock
);
1280 if (ret
< sizeof(packet_index
)) {
1282 * The tracefile is closed in write, so we read up to EOF.
1284 if (vstream
->close_write_flag
== 1) {
1285 viewer_index
.status
= htobe32(VIEWER_INDEX_RETRY
);
1286 /* Rotate on normal EOF */
1287 ret
= rotate_viewer_stream(vstream
, rstream
);
1290 } else if (ret
== 1) {
1291 viewer_index
.status
= htobe32(VIEWER_INDEX_HUP
);
1292 delete_viewer_stream(vstream
);
1293 destroy_viewer_stream(vstream
);
1297 PERROR("Relay reading index file %d",
1298 vstream
->index_read_fd
);
1299 viewer_index
.status
= htobe32(VIEWER_INDEX_ERR
);
1303 viewer_index
.status
= htobe32(VIEWER_INDEX_OK
);
1304 vstream
->last_sent_index
++;
1308 * Indexes are stored in big endian, no need to switch before sending.
1310 viewer_index
.offset
= packet_index
.offset
;
1311 viewer_index
.packet_size
= packet_index
.packet_size
;
1312 viewer_index
.content_size
= packet_index
.content_size
;
1313 viewer_index
.timestamp_begin
= packet_index
.timestamp_begin
;
1314 viewer_index
.timestamp_end
= packet_index
.timestamp_end
;
1315 viewer_index
.events_discarded
= packet_index
.events_discarded
;
1316 viewer_index
.stream_id
= packet_index
.stream_id
;
1319 viewer_index
.flags
= htobe32(viewer_index
.flags
);
1320 health_code_update();
1321 ret
= cmd
->sock
->ops
->sendmsg(cmd
->sock
, &viewer_index
,
1322 sizeof(viewer_index
), 0);
1324 ERR("Relay index to viewer");
1327 health_code_update();
1329 DBG("Index %" PRIu64
"for stream %" PRIu64
"sent",
1330 vstream
->last_sent_index
, vstream
->stream_handle
);
1341 * Send the next index for a stream
1343 * Return 0 on success or else a negative value.
1346 int viewer_get_packet(struct relay_command
*cmd
)
1348 int ret
, send_data
= 0;
1352 struct lttng_viewer_get_packet get_packet_info
;
1353 struct lttng_viewer_trace_packet reply
;
1354 struct relay_viewer_stream
*stream
;
1358 DBG2("Relay get data packet");
1360 if (cmd
->version_check_done
== 0) {
1361 ERR("Trying to get packet before version check");
1366 health_code_update();
1367 ret
= cmd
->sock
->ops
->recvmsg(cmd
->sock
, &get_packet_info
,
1368 sizeof(get_packet_info
), 0);
1369 if (ret
< 0 || ret
!= sizeof(get_packet_info
)) {
1371 ERR("Relay didn't receive the whole packet");
1374 health_code_update();
1376 /* From this point on, the error label can be reached. */
1377 memset(&reply
, 0, sizeof(reply
));
1380 stream
= live_find_viewer_stream_by_id(be64toh(get_packet_info
.stream_id
));
1384 assert(stream
->ctf_trace
);
1387 * First time we read this stream, we need open the tracefile, we should
1388 * only arrive here if an index has already been sent to the viewer, so the
1389 * tracefile must exist, if it does not it is a fatal error.
1391 if (stream
->read_fd
< 0) {
1392 char fullpath
[PATH_MAX
];
1394 if (stream
->tracefile_count
> 0) {
1395 ret
= snprintf(fullpath
, PATH_MAX
, "%s/%s_%" PRIu64
, stream
->path_name
,
1396 stream
->channel_name
,
1397 stream
->tracefile_count_current
);
1399 ret
= snprintf(fullpath
, PATH_MAX
, "%s/%s", stream
->path_name
,
1400 stream
->channel_name
);
1405 ret
= open(fullpath
, O_RDONLY
);
1407 PERROR("Relay opening trace file");
1410 stream
->read_fd
= ret
;
1413 if (!stream
->ctf_trace
->metadata_received
||
1414 stream
->ctf_trace
->metadata_received
>
1415 stream
->ctf_trace
->metadata_sent
) {
1416 reply
.status
= htobe32(VIEWER_GET_PACKET_ERR
);
1417 reply
.flags
|= LTTNG_VIEWER_FLAG_NEW_METADATA
;
1421 len
= be32toh(get_packet_info
.len
);
1422 data
= zmalloc(len
);
1424 PERROR("relay data zmalloc");
1428 ret
= lseek(stream
->read_fd
, be64toh(get_packet_info
.offset
), SEEK_SET
);
1431 * If the read fd was closed by the streaming side, the
1432 * abort_flag will be set to 1, otherwise it is an error.
1434 if (stream
->abort_flag
== 0) {
1438 reply
.status
= htobe32(VIEWER_GET_PACKET_EOF
);
1441 read_len
= lttng_read(stream
->read_fd
, data
, len
);
1442 if (read_len
< len
) {
1444 * If the read fd was closed by the streaming side, the
1445 * abort_flag will be set to 1, otherwise it is an error.
1447 if (stream
->abort_flag
== 0) {
1448 PERROR("Relay reading trace file, fd: %d, offset: %" PRIu64
,
1450 be64toh(get_packet_info
.offset
));
1453 reply
.status
= htobe32(VIEWER_GET_PACKET_EOF
);
1457 reply
.status
= htobe32(VIEWER_GET_PACKET_OK
);
1458 reply
.len
= htobe32(len
);
1463 reply
.status
= htobe32(VIEWER_GET_PACKET_ERR
);
1466 reply
.flags
= htobe32(reply
.flags
);
1468 health_code_update();
1469 ret
= cmd
->sock
->ops
->sendmsg(cmd
->sock
, &reply
, sizeof(reply
), 0);
1471 ERR("Relay data header to viewer");
1474 health_code_update();
1477 health_code_update();
1478 ret
= cmd
->sock
->ops
->sendmsg(cmd
->sock
, data
, len
, 0);
1480 ERR("Relay send data to viewer");
1483 health_code_update();
1486 DBG("Sent %u bytes for stream %" PRIu64
, len
,
1487 be64toh(get_packet_info
.stream_id
));
1498 * Send the session's metadata
1500 * Return 0 on success else a negative value.
1503 int viewer_get_metadata(struct relay_command
*cmd
)
1509 struct lttng_viewer_get_metadata request
;
1510 struct lttng_viewer_metadata_packet reply
;
1511 struct relay_viewer_stream
*stream
;
1515 DBG("Relay get metadata");
1517 if (cmd
->version_check_done
== 0) {
1518 ERR("Trying to get metadata before version check");
1523 health_code_update();
1524 ret
= cmd
->sock
->ops
->recvmsg(cmd
->sock
, &request
,
1525 sizeof(request
), 0);
1526 if (ret
< 0 || ret
!= sizeof(request
)) {
1528 ERR("Relay didn't receive the whole packet");
1531 health_code_update();
1534 stream
= live_find_viewer_stream_by_id(be64toh(request
.stream_id
));
1535 if (!stream
|| !stream
->metadata_flag
) {
1536 ERR("Invalid metadata stream");
1539 assert(stream
->ctf_trace
);
1540 assert(stream
->ctf_trace
->metadata_sent
<=
1541 stream
->ctf_trace
->metadata_received
);
1543 len
= stream
->ctf_trace
->metadata_received
-
1544 stream
->ctf_trace
->metadata_sent
;
1546 reply
.status
= htobe32(VIEWER_NO_NEW_METADATA
);
1550 /* first time, we open the metadata file */
1551 if (stream
->read_fd
< 0) {
1552 char fullpath
[PATH_MAX
];
1554 ret
= snprintf(fullpath
, PATH_MAX
, "%s/%s", stream
->path_name
,
1555 stream
->channel_name
);
1559 ret
= open(fullpath
, O_RDONLY
);
1561 PERROR("Relay opening metadata file");
1564 stream
->read_fd
= ret
;
1567 reply
.len
= htobe64(len
);
1568 data
= zmalloc(len
);
1570 PERROR("viewer metadata zmalloc");
1574 read_len
= lttng_read(stream
->read_fd
, data
, len
);
1575 if (read_len
< len
) {
1576 PERROR("Relay reading metadata file");
1579 stream
->ctf_trace
->metadata_sent
+= read_len
;
1580 reply
.status
= htobe32(VIEWER_METADATA_OK
);
1584 reply
.status
= htobe32(VIEWER_METADATA_ERR
);
1587 health_code_update();
1588 ret
= cmd
->sock
->ops
->sendmsg(cmd
->sock
, &reply
, sizeof(reply
), 0);
1590 ERR("Relay data header to viewer");
1593 health_code_update();
1596 ret
= cmd
->sock
->ops
->sendmsg(cmd
->sock
, data
, len
, 0);
1598 ERR("Relay send data to viewer");
1603 DBG("Sent %" PRIu64
" bytes of metadata for stream %" PRIu64
, len
,
1604 be64toh(request
.stream_id
));
1606 DBG("Metadata sent");
1616 * live_relay_unknown_command: send -1 if received unknown command
1619 void live_relay_unknown_command(struct relay_command
*cmd
)
1621 struct lttcomm_relayd_generic_reply reply
;
1624 reply
.ret_code
= htobe32(LTTNG_ERR_UNK
);
1625 ret
= cmd
->sock
->ops
->sendmsg(cmd
->sock
, &reply
,
1626 sizeof(struct lttcomm_relayd_generic_reply
), 0);
1628 ERR("Relay sending unknown command");
1633 * Process the commands received on the control socket
1636 int process_control(struct lttng_viewer_cmd
*recv_hdr
,
1637 struct relay_command
*cmd
, struct lttng_ht
*sessions_ht
)
1641 switch (be32toh(recv_hdr
->cmd
)) {
1642 case VIEWER_CONNECT
:
1643 ret
= viewer_connect(cmd
);
1645 case VIEWER_LIST_SESSIONS
:
1646 ret
= viewer_list_sessions(cmd
, sessions_ht
);
1648 case VIEWER_ATTACH_SESSION
:
1649 ret
= viewer_attach_session(cmd
, sessions_ht
);
1651 case VIEWER_GET_NEXT_INDEX
:
1652 ret
= viewer_get_next_index(cmd
, sessions_ht
);
1654 case VIEWER_GET_PACKET
:
1655 ret
= viewer_get_packet(cmd
);
1657 case VIEWER_GET_METADATA
:
1658 ret
= viewer_get_metadata(cmd
);
1661 ERR("Received unknown viewer command (%u)", be32toh(recv_hdr
->cmd
));
1662 live_relay_unknown_command(cmd
);
1672 void cleanup_poll_connection(struct lttng_poll_event
*events
, int pollfd
)
1678 lttng_poll_del(events
, pollfd
);
1680 ret
= close(pollfd
);
1682 ERR("Closing pollfd %d", pollfd
);
1687 * Create and add connection to the given hash table.
1689 * Return poll add value or else -1 on error.
1692 int add_connection(int fd
, struct lttng_poll_event
*events
,
1693 struct lttng_ht
*relay_connections_ht
)
1696 struct relay_command
*relay_connection
;
1699 assert(relay_connections_ht
);
1701 relay_connection
= zmalloc(sizeof(struct relay_command
));
1702 if (relay_connection
== NULL
) {
1703 PERROR("Relay command zmalloc");
1707 ret
= lttng_read(fd
, relay_connection
, sizeof(*relay_connection
));
1708 if (ret
< sizeof(*relay_connection
)) {
1709 PERROR("read relay cmd pipe");
1713 lttng_ht_node_init_ulong(&relay_connection
->sock_n
,
1714 (unsigned long) relay_connection
->sock
->fd
);
1716 lttng_ht_add_unique_ulong(relay_connections_ht
,
1717 &relay_connection
->sock_n
);
1720 return lttng_poll_add(events
, relay_connection
->sock
->fd
,
1721 LPOLLIN
| LPOLLRDHUP
);
1724 free(relay_connection
);
1730 void deferred_free_connection(struct rcu_head
*head
)
1732 struct relay_command
*relay_connection
=
1733 caa_container_of(head
, struct relay_command
, rcu_node
);
1735 if (relay_connection
->session
&&
1736 relay_connection
->session
->viewer_attached
> 0) {
1737 relay_connection
->session
->viewer_attached
--;
1739 lttcomm_destroy_sock(relay_connection
->sock
);
1740 free(relay_connection
);
1744 * Delete all streams for a specific session ID.
1747 void viewer_del_streams(uint64_t session_id
)
1749 struct relay_viewer_stream
*stream
;
1750 struct lttng_ht_iter iter
;
1753 cds_lfht_for_each_entry(viewer_streams_ht
->ht
, &iter
.iter
, stream
,
1755 health_code_update();
1757 if (stream
->session_id
!= session_id
) {
1761 delete_viewer_stream(stream
);
1762 assert(stream
->ctf_trace
);
1764 if (stream
->metadata_flag
) {
1766 * The metadata viewer stream is destroyed once the refcount on the
1767 * ctf trace goes to 0 in the destroy stream function thus there is
1768 * no explicit call to that function here.
1770 stream
->ctf_trace
->metadata_sent
= 0;
1771 stream
->ctf_trace
->viewer_metadata_stream
= NULL
;
1773 destroy_viewer_stream(stream
);
1780 * Delete and free a connection.
1782 * RCU read side lock MUST be acquired.
1785 void del_connection(struct lttng_ht
*relay_connections_ht
,
1786 struct lttng_ht_iter
*iter
, struct relay_command
*relay_connection
)
1790 assert(relay_connections_ht
);
1792 assert(relay_connection
);
1794 DBG("Cleaning connection of session ID %" PRIu64
,
1795 relay_connection
->session_id
);
1797 ret
= lttng_ht_del(relay_connections_ht
, iter
);
1800 viewer_del_streams(relay_connection
->session_id
);
1802 call_rcu(&relay_connection
->rcu_node
, deferred_free_connection
);
1806 * This thread does the actual work
1809 void *thread_worker(void *data
)
1813 struct relay_command
*relay_connection
;
1814 struct lttng_poll_event events
;
1815 struct lttng_ht
*relay_connections_ht
;
1816 struct lttng_ht_node_ulong
*node
;
1817 struct lttng_ht_iter iter
;
1818 struct lttng_viewer_cmd recv_hdr
;
1819 struct relay_local_data
*relay_ctx
= (struct relay_local_data
*) data
;
1820 struct lttng_ht
*sessions_ht
= relay_ctx
->sessions_ht
;
1822 DBG("[thread] Live viewer relay worker started");
1824 rcu_register_thread();
1826 health_register(health_relayd
, HEALTH_RELAYD_TYPE_LIVE_WORKER
);
1828 /* table of connections indexed on socket */
1829 relay_connections_ht
= lttng_ht_new(0, LTTNG_HT_TYPE_ULONG
);
1830 if (!relay_connections_ht
) {
1831 goto relay_connections_ht_error
;
1834 ret
= create_thread_poll_set(&events
, 2);
1836 goto error_poll_create
;
1839 ret
= lttng_poll_add(&events
, live_relay_cmd_pipe
[0], LPOLLIN
| LPOLLRDHUP
);
1848 health_code_update();
1850 /* Infinite blocking call, waiting for transmission */
1851 DBG3("Relayd live viewer worker thread polling...");
1852 health_poll_entry();
1853 ret
= lttng_poll_wait(&events
, -1);
1857 * Restart interrupted system call.
1859 if (errno
== EINTR
) {
1868 * Process control. The control connection is prioritised so we don't
1869 * starve it with high throughput tracing data on the data
1872 for (i
= 0; i
< nb_fd
; i
++) {
1873 /* Fetch once the poll data */
1874 uint32_t revents
= LTTNG_POLL_GETEV(&events
, i
);
1875 int pollfd
= LTTNG_POLL_GETFD(&events
, i
);
1877 health_code_update();
1879 /* Thread quit pipe has been closed. Killing thread. */
1880 ret
= check_thread_quit_pipe(pollfd
, revents
);
1886 /* Inspect the relay cmd pipe for new connection */
1887 if (pollfd
== live_relay_cmd_pipe
[0]) {
1888 if (revents
& (LPOLLERR
| LPOLLHUP
| LPOLLRDHUP
)) {
1889 ERR("Relay live pipe error");
1891 } else if (revents
& LPOLLIN
) {
1892 DBG("Relay live viewer command received");
1893 ret
= add_connection(live_relay_cmd_pipe
[0],
1894 &events
, relay_connections_ht
);
1899 } else if (revents
) {
1901 lttng_ht_lookup(relay_connections_ht
,
1902 (void *)((unsigned long) pollfd
), &iter
);
1903 node
= lttng_ht_iter_get_node_ulong(&iter
);
1905 DBG2("Relay viewer sock %d not found", pollfd
);
1909 relay_connection
= caa_container_of(node
, struct relay_command
,
1912 if (revents
& (LPOLLERR
)) {
1913 cleanup_poll_connection(&events
, pollfd
);
1914 del_connection(relay_connections_ht
, &iter
,
1916 } else if (revents
& (LPOLLHUP
| LPOLLRDHUP
)) {
1917 DBG("Viewer socket %d hung up", pollfd
);
1918 cleanup_poll_connection(&events
, pollfd
);
1919 del_connection(relay_connections_ht
, &iter
,
1921 } else if (revents
& LPOLLIN
) {
1922 ret
= relay_connection
->sock
->ops
->recvmsg(
1923 relay_connection
->sock
, &recv_hdr
,
1924 sizeof(struct lttng_viewer_cmd
),
1926 /* connection closed */
1928 cleanup_poll_connection(&events
, pollfd
);
1929 del_connection(relay_connections_ht
, &iter
,
1931 DBG("Viewer control connection closed with %d",
1934 if (relay_connection
->session
) {
1935 DBG2("Relay viewer worker receiving data for "
1936 "session: %" PRIu64
,
1937 relay_connection
->session
->id
);
1939 ret
= process_control(&recv_hdr
, relay_connection
,
1942 /* Clear the session on error. */
1943 cleanup_poll_connection(&events
, pollfd
);
1944 del_connection(relay_connections_ht
, &iter
,
1946 DBG("Viewer connection closed with %d", pollfd
);
1957 lttng_poll_clean(&events
);
1959 /* empty the hash table and free the memory */
1961 cds_lfht_for_each_entry(relay_connections_ht
->ht
, &iter
.iter
, node
, node
) {
1962 health_code_update();
1964 node
= lttng_ht_iter_get_node_ulong(&iter
);
1969 relay_connection
= caa_container_of(node
, struct relay_command
,
1971 del_connection(relay_connections_ht
, &iter
, relay_connection
);
1975 lttng_ht_destroy(relay_connections_ht
);
1976 relay_connections_ht_error
:
1977 /* Close relay cmd pipes */
1978 utils_close_pipe(live_relay_cmd_pipe
);
1980 DBG("Viewer worker thread exited with error");
1982 DBG("Viewer worker thread cleanup complete");
1985 ERR("Health error occurred in %s", __func__
);
1987 health_unregister(health_relayd
);
1989 rcu_unregister_thread();
1994 * Create the relay command pipe to wake thread_manage_apps.
1995 * Closed in cleanup().
1997 static int create_relay_cmd_pipe(void)
2001 ret
= utils_create_pipe_cloexec(live_relay_cmd_pipe
);
2006 void live_stop_threads(void)
2013 ret
= pthread_join(live_listener_thread
, &status
);
2015 PERROR("pthread_join live listener");
2016 goto error
; /* join error, exit without cleanup */
2019 ret
= pthread_join(live_worker_thread
, &status
);
2021 PERROR("pthread_join live worker");
2022 goto error
; /* join error, exit without cleanup */
2025 ret
= pthread_join(live_dispatcher_thread
, &status
);
2027 PERROR("pthread_join live dispatcher");
2028 goto error
; /* join error, exit without cleanup */
2040 int live_start_threads(struct lttng_uri
*uri
,
2041 struct relay_local_data
*relay_ctx
)
2050 /* Check if daemon is UID = 0 */
2051 is_root
= !getuid();
2054 if (live_uri
->port
< 1024) {
2055 ERR("Need to be root to use ports < 1024");
2061 /* Setup the thread apps communication pipe. */
2062 if ((ret
= create_relay_cmd_pipe()) < 0) {
2066 /* Init relay command queue. */
2067 cds_wfq_init(&viewer_cmd_queue
.queue
);
2069 /* Set up max poll set size */
2070 lttng_poll_set_max_size();
2072 /* Setup the dispatcher thread */
2073 ret
= pthread_create(&live_dispatcher_thread
, NULL
,
2074 thread_dispatcher
, (void *) NULL
);
2076 PERROR("pthread_create viewer dispatcher");
2077 goto exit_dispatcher
;
2080 /* Setup the worker thread */
2081 ret
= pthread_create(&live_worker_thread
, NULL
,
2082 thread_worker
, relay_ctx
);
2084 PERROR("pthread_create viewer worker");
2088 /* Setup the listener thread */
2089 ret
= pthread_create(&live_listener_thread
, NULL
,
2090 thread_listener
, (void *) NULL
);
2092 PERROR("pthread_create viewer listener");
2100 ret
= pthread_join(live_listener_thread
, &status
);
2102 PERROR("pthread_join live listener");
2103 goto error
; /* join error, exit without cleanup */
2107 ret
= pthread_join(live_worker_thread
, &status
);
2109 PERROR("pthread_join live worker");
2110 goto error
; /* join error, exit without cleanup */
2114 ret
= pthread_join(live_dispatcher_thread
, &status
);
2116 PERROR("pthread_join live dispatcher");
2117 goto error
; /* join error, exit without cleanup */