#include "lttng-relayd.h"
#include "lttng-viewer.h"
#include "utils.h"
+#include "health-relayd.h"
static struct lttng_uri *live_uri;
static
int notify_thread_pipe(int wpipe)
{
- int ret;
+ ssize_t ret;
- do {
- ret = write(wpipe, "!", 1);
- } while (ret < 0 && errno == EINTR);
- if (ret < 0 || ret != 1) {
+ ret = lttng_write(wpipe, "!", 1);
+ if (ret < 1) {
PERROR("write poll pipe");
}
- return ret;
+ return (int) ret;
}
/*
DBG("[thread] Relay live listener started");
+ health_register(health_relayd, HEALTH_RELAYD_TYPE_LIVE_LISTENER);
+
+ health_code_update();
+
live_control_sock = init_socket(live_uri);
if (!live_control_sock) {
goto error_sock_control;
}
while (1) {
+ health_code_update();
+
DBG("Listener accepting live viewers connections");
restart:
+ health_poll_entry();
ret = lttng_poll_wait(&events, -1);
+ health_poll_exit();
if (ret < 0) {
/*
* Restart interrupted system call.
DBG("Relay new viewer connection received");
for (i = 0; i < nb_fd; i++) {
+ health_code_update();
+
/* Fetch once the poll data */
revents = LTTNG_POLL_GETEV(&events, i);
pollfd = LTTNG_POLL_GETFD(&events, i);
lttcomm_destroy_sock(live_control_sock);
error_sock_control:
if (err) {
+ health_error();
DBG("Live viewer listener thread exited with error");
}
+ health_unregister(health_relayd);
DBG("Live viewer listener thread cleanup complete");
stop_threads();
return NULL;
static
void *thread_dispatcher(void *data)
{
- int ret;
+ int err = -1;
+ ssize_t ret;
struct cds_wfq_node *node;
struct relay_command *relay_cmd = NULL;
DBG("[thread] Live viewer relay dispatcher started");
+ health_register(health_relayd, HEALTH_RELAYD_TYPE_LIVE_DISPATCHER);
+
+ health_code_update();
+
while (!CMM_LOAD_SHARED(live_dispatch_thread_exit)) {
+ health_code_update();
+
/* Atomically prepare the queue futex */
futex_nto1_prepare(&viewer_cmd_queue.futex);
do {
+ health_code_update();
+
/* Dequeue commands */
node = cds_wfq_dequeue_blocking(&viewer_cmd_queue.queue);
if (node == NULL) {
* so we can be assured that the data will be read at some point in
* time or wait to the end of the world :)
*/
- do {
- ret = write(live_relay_cmd_pipe[1], relay_cmd,
- sizeof(*relay_cmd));
- } while (ret < 0 && errno == EINTR);
+ ret = lttng_write(live_relay_cmd_pipe[1], relay_cmd,
+ sizeof(*relay_cmd));
free(relay_cmd);
- if (ret < 0 || ret != sizeof(struct relay_command)) {
+ if (ret < sizeof(struct relay_command)) {
PERROR("write cmd pipe");
goto error;
}
} while (node != NULL);
/* Futex wait on queue. Blocking call on futex() */
+ health_poll_entry();
futex_nto1_wait(&viewer_cmd_queue.futex);
+ health_poll_exit();
}
+ /* Normal exit, no error */
+ err = 0;
+
error:
+ if (err) {
+ health_error();
+ ERR("Health error occurred in %s", __func__);
+ }
+ health_unregister(health_relayd);
DBG("Live viewer dispatch thread dying");
stop_threads();
return NULL;
cmd->version_check_done = 1;
+ health_code_update();
+
/* Get version from the other side. */
ret = cmd->sock->ops->recvmsg(cmd->sock, &msg, sizeof(msg), 0);
if (ret < 0 || ret != sizeof(msg)) {
goto end;
}
+ health_code_update();
+
reply.major = RELAYD_VERSION_COMM_MAJOR;
reply.minor = RELAYD_VERSION_COMM_MINOR;
if (cmd->type == RELAY_VIEWER_COMMAND) {
reply.viewer_session_id = htobe64(++last_relay_viewer_session_id);
}
+
+ health_code_update();
+
ret = cmd->sock->ops->sendmsg(cmd->sock, &reply,
sizeof(struct lttng_viewer_connect), 0);
if (ret < 0) {
ERR("Relay sending version");
}
+ health_code_update();
+
DBG("Version check done using protocol %u.%u", cmd->major, cmd->minor);
ret = 0;
cds_lfht_count_nodes(sessions_ht->ht, &approx_before, &count, &approx_after);
session_list.sessions_count = htobe32(count);
+ health_code_update();
+
ret = cmd->sock->ops->sendmsg(cmd->sock, &session_list,
sizeof(session_list), 0);
if (ret < 0) {
goto end_unlock;
}
+ health_code_update();
+
cds_lfht_for_each_entry(sessions_ht->ht, &iter.iter, node, node) {
+ health_code_update();
+
node = lttng_ht_iter_get_node_ulong(&iter);
if (!node) {
goto end_unlock;
send_session.id = htobe64(session->id);
send_session.live_timer = htobe32(session->live_timer);
send_session.clients = htobe32(session->viewer_attached);
+ send_session.streams = htobe32(session->stream_count);
+
+ health_code_update();
ret = cmd->sock->ops->sendmsg(cmd->sock, &send_session,
sizeof(send_session), 0);
goto end_unlock;
}
}
+ health_code_update();
+
rcu_read_unlock();
ret = 0;
goto end;
return ret;
}
+/*
+ * Open index file using a given viewer stream.
+ *
+ * Return 0 on success or else a negative value.
+ */
+static int open_index(struct relay_viewer_stream *stream)
+{
+ int ret;
+ char fullpath[PATH_MAX];
+ struct lttng_packet_index_file_hdr hdr;
+
+ if (stream->tracefile_size > 0) {
+ /* For now we don't support on-disk ring buffer. */
+ ret = -1;
+ goto end;
+ }
+
+ ret = snprintf(fullpath, sizeof(fullpath), "%s/" DEFAULT_INDEX_DIR "/%s"
+ DEFAULT_INDEX_FILE_SUFFIX, stream->path_name,
+ stream->channel_name);
+ if (ret < 0) {
+ PERROR("snprintf index path");
+ goto error;
+ }
+
+ DBG("Opening index file %s in read only", fullpath);
+ ret = open(fullpath, O_RDONLY);
+ if (ret < 0) {
+ if (errno == ENOENT) {
+ ret = -ENOENT;
+ goto error;
+ } else {
+ PERROR("opening index in read-only");
+ }
+ goto error;
+ }
+ stream->index_read_fd = ret;
+ DBG("Opening index file %s in read only, (fd: %d)", fullpath, ret);
+
+ ret = lttng_read(stream->index_read_fd, &hdr, sizeof(hdr));
+ if (ret < sizeof(hdr)) {
+ PERROR("Reading index header");
+ goto error;
+ }
+ if (strncmp(hdr.magic, INDEX_MAGIC, sizeof(hdr.magic)) != 0) {
+ ERR("Invalid header magic");
+ ret = -1;
+ goto error;
+ }
+ if (be32toh(hdr.index_major) != INDEX_MAJOR ||
+ be32toh(hdr.index_minor) != INDEX_MINOR) {
+ ERR("Invalid header version");
+ ret = -1;
+ goto error;
+ }
+ ret = 0;
+
+error:
+end:
+ return ret;
+}
+
/*
* Allocate and init a new viewer_stream.
*
* Returns 0 on success or a negative value on error.
*/
static
-int init_viewer_stream(struct relay_stream *stream)
+int init_viewer_stream(struct relay_stream *stream, int seek_last)
{
int ret;
struct relay_viewer_stream *viewer_stream;
viewer_stream->tracefile_count = stream->tracefile_count;
viewer_stream->metadata_flag = stream->metadata_flag;
+ if (seek_last && viewer_stream->total_index_received > 0) {
+ ret = open_index(viewer_stream);
+ if (ret < 0) {
+ goto error;
+ }
+ ret = lseek(viewer_stream->index_read_fd,
+ viewer_stream->total_index_received *
+ sizeof(struct lttng_packet_index),
+ SEEK_CUR);
+ if (ret < 0) {
+ goto error;
+ }
+ viewer_stream->last_sent_index =
+ viewer_stream->total_index_received;
+ }
+
/*
* This is to avoid a race between the initialization of this object and
* the close of the given stream. If the stream is unable to find this
struct lttng_ht_node_u64 *node64;
struct lttng_ht_iter iter;
struct relay_session *session;
+ int seek_last = 0;
assert(cmd);
assert(sessions_ht);
goto end_no_session;
}
+ health_code_update();
+
ret = cmd->sock->ops->recvmsg(cmd->sock, &request, sizeof(request), 0);
if (ret < 0 || ret != sizeof(request)) {
if (ret == 0) {
goto error;
}
+ health_code_update();
+
rcu_read_lock();
lttng_ht_lookup(sessions_ht,
(void *)((unsigned long) be64toh(request.session_id)), &iter);
/* Default behaviour. */
break;
case VIEWER_SEEK_LAST:
- /* TODO */
+ seek_last = 1;
break;
default:
ERR("Wrong seek parameter");
cds_lfht_for_each_entry(relay_streams_ht->ht, &iter.iter, node, node) {
struct relay_viewer_stream *vstream;
+ health_code_update();
+
node = lttng_ht_iter_get_node_ulong(&iter);
if (!node) {
continue;
vstream = live_find_viewer_stream_by_id(stream->stream_handle);
if (!vstream) {
- ret = init_viewer_stream(stream);
+ ret = init_viewer_stream(stream, seek_last);
if (ret < 0) {
goto end_unlock;
}
}
send_reply:
+ health_code_update();
ret = cmd->sock->ops->sendmsg(cmd->sock, &response, sizeof(response), 0);
if (ret < 0) {
ERR("Relay sending viewer attach response");
goto end_unlock;
}
+ health_code_update();
/*
* Unknown or busy session, just return gracefully, the viewer knows what
/* We should only be there if we have a session to attach to. */
assert(session);
cds_lfht_for_each_entry(viewer_streams_ht->ht, &iter.iter, node, node) {
+ health_code_update();
+
node64 = lttng_ht_iter_get_node_u64(&iter);
if (!node64) {
continue;
return ret;
}
-/*
- * Open index file using a given viewer stream.
- *
- * Return 0 on success or else a negative value.
- */
-static int open_index(struct relay_viewer_stream *stream)
-{
- int ret;
- char fullpath[PATH_MAX];
- struct lttng_packet_index_file_hdr hdr;
-
- if (stream->tracefile_size > 0) {
- /* For now we don't support on-disk ring buffer. */
- ret = -1;
- goto end;
- } else {
- ret = snprintf(fullpath, sizeof(fullpath), "%s/" DEFAULT_INDEX_DIR
- "/%s" DEFAULT_INDEX_FILE_SUFFIX,
- stream->path_name, stream->channel_name);
- if (ret < 0) {
- PERROR("snprintf index path");
- goto error;
- }
- }
-
- DBG("Opening index file %s in read only", fullpath);
- ret = open(fullpath, O_RDONLY);
- if (ret < 0) {
- if (errno == ENOENT) {
- ret = ENOENT;
- goto error;
- } else {
- PERROR("opening index in read-only");
- }
- goto error;
- }
- stream->index_read_fd = ret;
- DBG("Opening index file %s in read only, (fd: %d)", fullpath, ret);
-
- do {
- ret = read(stream->index_read_fd, &hdr, sizeof(hdr));
- } while (ret < 0 && errno == EINTR);
- if (ret < 0) {
- PERROR("Reading index header");
- goto error;
- }
- if (strncmp(hdr.magic, INDEX_MAGIC, sizeof(hdr.magic)) != 0) {
- ERR("Invalid header magic");
- ret = -1;
- goto error;
- }
- if (be32toh(hdr.index_major) != INDEX_MAJOR ||
- be32toh(hdr.index_minor) != INDEX_MINOR) {
- ERR("Invalid header version");
- ret = -1;
- goto error;
- }
- ret = 0;
-
-error:
-end:
- return ret;
-}
-
/*
* Get viewer stream from stream id.
*
goto end_no_session;
}
+ health_code_update();
ret = cmd->sock->ops->recvmsg(cmd->sock, &request_index,
sizeof(request_index), 0);
if (ret < 0 || ret != sizeof(request_index)) {
ERR("Relay didn't receive the whole packet");
goto end;
}
+ health_code_update();
rcu_read_lock();
vstream = live_find_viewer_stream_by_id(be64toh(request_index.stream_id));
/* First time, we open the index file */
if (vstream->index_read_fd < 0) {
ret = open_index(vstream);
- if (ret == ENOENT) {
+ if (ret == -ENOENT) {
/*
* The index is created only when the first data packet arrives, it
* might not be ready at the beginning of the session
viewer_index.flags |= LTTNG_VIEWER_FLAG_NEW_METADATA;
}
- do {
- ret = read(vstream->index_read_fd, &packet_index,
- sizeof(packet_index));
- } while (ret < 0 && errno == EINTR);
+ ret = lttng_read(vstream->index_read_fd, &packet_index,
+ sizeof(packet_index));
if (ret < sizeof(packet_index)) {
PERROR("Relay reading index file");
viewer_index.status = htobe32(VIEWER_INDEX_ERR);
send_reply:
viewer_index.flags = htobe32(viewer_index.flags);
+ health_code_update();
ret = cmd->sock->ops->sendmsg(cmd->sock, &viewer_index,
sizeof(viewer_index), 0);
if (ret < 0) {
ERR("Relay index to viewer");
goto end_unlock;
}
+ health_code_update();
DBG("Index %" PRIu64 "for stream %" PRIu64 "sent",
vstream->last_sent_index, vstream->stream_handle);
goto end;
}
+ health_code_update();
ret = cmd->sock->ops->recvmsg(cmd->sock, &get_packet_info,
sizeof(get_packet_info), 0);
if (ret < 0 || ret != sizeof(get_packet_info)) {
ERR("Relay didn't receive the whole packet");
goto end;
}
+ health_code_update();
+
+ /* From this point on, the error label can be reached. */
+ memset(&reply, 0, sizeof(reply));
rcu_read_lock();
stream = live_find_viewer_stream_by_id(be64toh(get_packet_info.stream_id));
stream->read_fd = ret;
}
- memset(&reply, 0, sizeof(reply));
-
if (!stream->ctf_trace->metadata_received ||
stream->ctf_trace->metadata_received >
stream->ctf_trace->metadata_sent) {
reply.status = htobe32(VIEWER_GET_PACKET_ERR);
reply.flags |= LTTNG_VIEWER_FLAG_NEW_METADATA;
-
goto send_reply;
}
PERROR("lseek");
goto error;
}
- read_len = read(stream->read_fd, data, len);
- if (read_len < (ssize_t) len) {
+ read_len = lttng_read(stream->read_fd, data, len);
+ if (read_len < len) {
PERROR("Relay reading trace file, fd: %d, offset: %" PRIu64,
stream->read_fd, be64toh(get_packet_info.offset));
goto error;
send_reply:
reply.flags = htobe32(reply.flags);
+
+ health_code_update();
ret = cmd->sock->ops->sendmsg(cmd->sock, &reply, sizeof(reply), 0);
if (ret < 0) {
ERR("Relay data header to viewer");
goto end_unlock;
}
+ health_code_update();
if (send_data) {
+ health_code_update();
ret = cmd->sock->ops->sendmsg(cmd->sock, data, len, 0);
if (ret < 0) {
ERR("Relay send data to viewer");
goto end_unlock;
}
+ health_code_update();
}
DBG("Sent %u bytes for stream %" PRIu64, len,
goto end;
}
+ health_code_update();
ret = cmd->sock->ops->recvmsg(cmd->sock, &request,
sizeof(request), 0);
if (ret < 0 || ret != sizeof(request)) {
ERR("Relay didn't receive the whole packet");
goto end;
}
+ health_code_update();
rcu_read_lock();
stream = live_find_viewer_stream_by_id(be64toh(request.stream_id));
goto error;
}
- read_len = read(stream->read_fd, data, len);
- if (read_len < (ssize_t) len) {
+ read_len = lttng_read(stream->read_fd, data, len);
+ if (read_len < len) {
PERROR("Relay reading metadata file");
goto error;
}
reply.status = htobe32(VIEWER_METADATA_ERR);
send_reply:
+ health_code_update();
ret = cmd->sock->ops->sendmsg(cmd->sock, &reply, sizeof(reply), 0);
if (ret < 0) {
ERR("Relay data header to viewer");
goto end_unlock;
}
+ health_code_update();
if (len > 0) {
ret = cmd->sock->ops->sendmsg(cmd->sock, data, len, 0);
goto error;
}
- do {
- ret = read(fd, relay_connection, sizeof(*relay_connection));
- } while (ret < 0 && errno == EINTR);
- if (ret < 0 || ret < sizeof(*relay_connection)) {
+ ret = lttng_read(fd, relay_connection, sizeof(*relay_connection));
+ if (ret < sizeof(*relay_connection)) {
PERROR("read relay cmd pipe");
goto error_read;
}
rcu_read_lock();
cds_lfht_for_each_entry(viewer_streams_ht->ht, &iter.iter, node, node) {
+ health_code_update();
+
node = lttng_ht_iter_get_node_u64(&iter);
if (!node) {
continue;
rcu_register_thread();
+ health_register(health_relayd, HEALTH_RELAYD_TYPE_LIVE_WORKER);
+
/* table of connections indexed on socket */
relay_connections_ht = lttng_ht_new(0, LTTNG_HT_TYPE_ULONG);
if (!relay_connections_ht) {
while (1) {
int i;
+ health_code_update();
+
/* Infinite blocking call, waiting for transmission */
DBG3("Relayd live viewer worker thread polling...");
+ health_poll_entry();
ret = lttng_poll_wait(&events, -1);
+ health_poll_exit();
if (ret < 0) {
/*
* Restart interrupted system call.
uint32_t revents = LTTNG_POLL_GETEV(&events, i);
int pollfd = LTTNG_POLL_GETFD(&events, i);
+ health_code_update();
+
/* Thread quit pipe has been closed. Killing thread. */
ret = check_thread_quit_pipe(pollfd, revents);
if (ret) {
/* connection closed */
if (ret <= 0) {
cleanup_poll_connection(&events, pollfd);
- del_connection( relay_connections_ht, &iter,
+ del_connection(relay_connections_ht, &iter,
relay_connection);
DBG("Viewer control connection closed with %d",
pollfd);
/* empty the hash table and free the memory */
rcu_read_lock();
cds_lfht_for_each_entry(relay_connections_ht->ht, &iter.iter, node, node) {
+ health_code_update();
+
node = lttng_ht_iter_get_node_ulong(&iter);
if (!node) {
continue;
DBG("Viewer worker thread exited with error");
}
DBG("Viewer worker thread cleanup complete");
+ if (err) {
+ health_error();
+ ERR("Health error occurred in %s", __func__);
+ }
+ health_unregister(health_relayd);
stop_threads();
rcu_unregister_thread();
return NULL;
return ret;
}
-void live_stop_threads()
+void live_stop_threads(void)
{
int ret;
void *status;
projects / lttng-tools.git / blobdiff