*/
#define _GNU_SOURCE
+#define _LGPL_SOURCE
#include <getopt.h>
#include <grp.h>
#include <limits.h>
#include <common/common.h>
#include <common/compat/poll.h>
#include <common/compat/socket.h>
+#include <common/compat/endian.h>
#include <common/defaults.h>
#include <common/futex.h>
#include <common/index/index.h>
* Cleanup the daemon
*/
static
-void cleanup(void)
+void cleanup_relayd_live(void)
{
DBG("Cleaning up");
}
/*
- * Atomically check if new streams got added in the session since the last
- * check and reset the flag to 0.
+ * Atomically check if new streams got added in one of the sessions attached
+ * and reset the flag to 0.
*
* Returns 1 if new streams got added, 0 if nothing changed, a negative value
* on error.
*/
static
-int check_new_streams(uint64_t session_id, struct lttng_ht *sessions_ht)
+int check_new_streams(struct relay_connection *conn)
{
- int ret;
- unsigned long current_val;
struct relay_session *session;
+ unsigned long current_val;
+ int ret = 0;
- assert(sessions_ht);
-
- session = session_find_by_id(sessions_ht, session_id);
- if (!session) {
- DBG("Relay session %" PRIu64 " not found", session_id);
- ret = -1;
- goto error;
+ if (!conn->viewer_session) {
+ goto end;
+ }
+ cds_list_for_each_entry(session,
+ &conn->viewer_session->sessions_head,
+ viewer_session_list) {
+ current_val = uatomic_cmpxchg(&session->new_streams, 1, 0);
+ ret = current_val;
+ if (ret == 1) {
+ goto end;
+ }
}
- current_val = uatomic_cmpxchg(&session->new_streams, 1, 0);
- ret = current_val;
-
-error:
+end:
return ret;
}
return ret;
}
-/*
- * Write to writable pipe used to notify a thread.
- */
-static
-int notify_thread_pipe(int wpipe)
-{
- ssize_t ret;
-
- ret = lttng_write(wpipe, "!", 1);
- if (ret < 1) {
- PERROR("write poll pipe");
- }
-
- return (int) ret;
-}
-
-/*
- * Stop all threads by closing the thread quit pipe.
- */
-static
-void stop_threads(void)
+int relayd_live_stop(void)
{
- int ret;
-
- /* Stopping all threads */
- DBG("Terminating all live threads");
- ret = notify_thread_pipe(live_conn_pipe[1]);
- if (ret < 0) {
- ERR("write error on thread quit pipe");
- }
-
- /* Dispatch thread */
+ /* Stop dispatch thread */
CMM_STORE_SHARED(live_dispatch_thread_exit, 1);
futex_nto1_wake(&viewer_conn_queue.futex);
+ return 0;
}
/*
}
/* Add quit pipe */
- ret = lttng_poll_add(events, live_conn_pipe[0], LPOLLIN | LPOLLERR);
+ ret = lttng_poll_add(events, thread_quit_pipe[0], LPOLLIN | LPOLLERR);
if (ret < 0) {
goto error;
}
* Return 1 if it was triggered else 0;
*/
static
-int check_live_conn_pipe(int fd, uint32_t events)
+int check_thread_quit_pipe(int fd, uint32_t events)
{
- if (fd == live_conn_pipe[0] && (events & LPOLLIN)) {
+ if (fd == thread_quit_pipe[0] && (events & LPOLLIN)) {
return 1;
}
pollfd = LTTNG_POLL_GETFD(&events, i);
/* Thread quit pipe has been closed. Killing thread. */
- ret = check_live_conn_pipe(pollfd, revents);
+ ret = check_thread_quit_pipe(pollfd, revents);
if (ret) {
err = 0;
goto exit;
new_conn->sock = newsock;
/* Enqueue request for the dispatcher thread. */
- cds_wfq_enqueue(&viewer_conn_queue.queue, &new_conn->qnode);
+ cds_wfcq_enqueue(&viewer_conn_queue.head, &viewer_conn_queue.tail,
+ &new_conn->qnode);
/*
* Wake the dispatch queue futex. Implicit memory barrier with
- * the exchange in cds_wfq_enqueue.
+ * the exchange in cds_wfcq_enqueue.
*/
futex_nto1_wake(&viewer_conn_queue.futex);
}
}
health_unregister(health_relayd);
DBG("Live viewer listener thread cleanup complete");
- stop_threads();
+ if (lttng_relay_stop_threads()) {
+ ERR("Error stopping threads");
+ }
return NULL;
}
{
int err = -1;
ssize_t ret;
- struct cds_wfq_node *node;
+ struct cds_wfcq_node *node;
struct relay_connection *conn = NULL;
DBG("[thread] Live viewer relay dispatcher started");
health_code_update();
/* Dequeue commands */
- node = cds_wfq_dequeue_blocking(&viewer_conn_queue.queue);
+ node = cds_wfcq_dequeue_blocking(&viewer_conn_queue.head,
+ &viewer_conn_queue.tail);
if (node == NULL) {
DBG("Woken up but nothing in the live-viewer "
"relay command queue");
}
health_unregister(health_relayd);
DBG("Live viewer dispatch thread dying");
- stop_threads();
+ if (lttng_relay_stop_threads()) {
+ ERR("Error stopping threads");
+ }
return NULL;
}
health_code_update();
+ memset(&reply, 0, sizeof(reply));
reply.major = RELAYD_VERSION_COMM_MAJOR;
reply.minor = RELAYD_VERSION_COMM_MINOR;
}
health_code_update();
- rcu_read_unlock();
ret = 0;
- goto end;
-
end_unlock:
rcu_read_unlock();
+ return ret;
+}
+
+/*
+ * Check if a connection is attached to a session.
+ * Return 1 if attached, 0 if not attached, a negative value on error.
+ */
+static
+int session_attached(struct relay_connection *conn, uint64_t session_id)
+{
+ struct relay_session *session;
+ int found = 0;
+
+ if (!conn->viewer_session) {
+ goto end;
+ }
+ cds_list_for_each_entry(session,
+ &conn->viewer_session->sessions_head,
+ viewer_session_list) {
+ if (session->id == session_id) {
+ found = 1;
+ goto end;
+ }
+ }
end:
- return ret;
+ return found;
+}
+
+/*
+ * Delete all streams for a specific session ID.
+ */
+static void destroy_viewer_streams_by_session(struct relay_session *session)
+{
+ struct relay_viewer_stream *stream;
+ struct lttng_ht_iter iter;
+
+ assert(session);
+
+ rcu_read_lock();
+ cds_lfht_for_each_entry(viewer_streams_ht->ht, &iter.iter, stream,
+ stream_n.node) {
+ struct ctf_trace *ctf_trace;
+
+ health_code_update();
+ if (stream->session_id != session->id) {
+ continue;
+ }
+
+ ctf_trace = ctf_trace_find_by_path(session->ctf_traces_ht,
+ stream->path_name);
+ assert(ctf_trace);
+
+ viewer_stream_delete(stream);
+
+ if (stream->metadata_flag) {
+ ctf_trace->metadata_sent = 0;
+ ctf_trace->viewer_metadata_stream = NULL;
+ }
+
+ viewer_stream_destroy(ctf_trace, stream);
+ }
+ rcu_read_unlock();
+}
+
+static void try_destroy_streams(struct relay_session *session)
+{
+ struct ctf_trace *ctf_trace;
+ struct lttng_ht_iter iter;
+
+ assert(session);
+
+ cds_lfht_for_each_entry(session->ctf_traces_ht->ht, &iter.iter, ctf_trace,
+ node.node) {
+ /* Attempt to destroy the ctf trace of that session. */
+ ctf_trace_try_destroy(session, ctf_trace);
+ }
+}
+
+/*
+ * Cleanup a session.
+ */
+static void cleanup_session(struct relay_connection *conn,
+ struct relay_session *session)
+{
+ /*
+ * Very important that this is done before destroying the session so we
+ * can put back every viewer stream reference from the ctf_trace.
+ */
+ destroy_viewer_streams_by_session(session);
+ try_destroy_streams(session);
+ cds_list_del(&session->viewer_session_list);
+ session_viewer_try_destroy(conn->sessions_ht, session);
}
/*
struct lttng_viewer_new_streams_request request;
struct lttng_viewer_new_streams_response response;
struct relay_session *session;
+ uint64_t session_id;
assert(conn);
if (ret < 0) {
goto error;
}
+ session_id = be64toh(request.session_id);
health_code_update();
+ memset(&response, 0, sizeof(response));
+
rcu_read_lock();
- session = session_find_by_id(conn->sessions_ht,
- be64toh(request.session_id));
+ session = session_find_by_id(conn->sessions_ht, session_id);
if (!session) {
- DBG("Relay session %" PRIu64 " not found",
- be64toh(request.session_id));
+ DBG("Relay session %" PRIu64 " not found", session_id);
response.status = htobe32(LTTNG_VIEWER_NEW_STREAMS_ERR);
goto send_reply;
}
- if (conn->session_id == session->id) {
- /* We confirmed the viewer is asking for the same session. */
- send_streams = 1;
- response.status = htobe32(LTTNG_VIEWER_NEW_STREAMS_OK);
- } else {
+ if (!session_attached(conn, session_id)) {
send_streams = 0;
response.status = htobe32(LTTNG_VIEWER_NEW_STREAMS_ERR);
goto send_reply;
}
- if (!send_streams) {
- goto send_reply;
- }
+ send_streams = 1;
+ response.status = htobe32(LTTNG_VIEWER_NEW_STREAMS_OK);
ret = make_viewer_streams(session, LTTNG_VIEWER_SEEK_LAST, NULL, &nb_unsent,
&nb_created);
if (nb_streams == 0 && session->close_flag) {
send_streams = 0;
response.status = htobe32(LTTNG_VIEWER_NEW_STREAMS_HUP);
+ /*
+ * Remove the session from the attached list of the connection
+ * and try to destroy it.
+ */
+ cds_list_del(&session->viewer_session_list);
+ cleanup_session(conn, session);
goto send_reply;
}
health_code_update();
+ memset(&response, 0, sizeof(response));
+
if (!conn->viewer_session) {
DBG("Client trying to attach before creating a live viewer session");
response.status = htobe32(LTTNG_VIEWER_ATTACH_NO_SESSION);
} else {
send_streams = 1;
response.status = htobe32(LTTNG_VIEWER_ATTACH_OK);
- conn->session_id = session->id;
- conn->session = session;
+ cds_list_add(&session->viewer_session_list,
+ &conn->viewer_session->sessions_head);
}
switch (be32toh(request.seek)) {
goto send_reply;
}
- if (!send_streams) {
- goto send_reply;
- }
-
ret = make_viewer_streams(session, seek_type, &nb_streams, NULL, NULL);
if (ret < 0) {
goto end_unlock;
return ret;
}
+/*
+ * Open the index file if needed for the given vstream.
+ *
+ * If an index file is successfully opened, the index_read_fd of the stream is
+ * set with it.
+ *
+ * Return 0 on success, a negative value on error (-ENOENT if not ready yet).
+ */
+static int try_open_index(struct relay_viewer_stream *vstream,
+ struct relay_stream *rstream)
+{
+ int ret = 0;
+
+ assert(vstream);
+ assert(rstream);
+
+ if (vstream->index_read_fd >= 0) {
+ goto end;
+ }
+
+ /*
+ * First time, we open the index file and at least one index is ready. The
+ * race between the read and write of the total_index_received is
+ * acceptable here since the client will be notified to simply come back
+ * and get the next index.
+ */
+ if (rstream->total_index_received <= 0) {
+ ret = -ENOENT;
+ goto end;
+ }
+ ret = index_open(vstream->path_name, vstream->channel_name,
+ vstream->tracefile_count, vstream->tracefile_count_current);
+ if (ret >= 0) {
+ vstream->index_read_fd = ret;
+ ret = 0;
+ goto end;
+ }
+
+end:
+ return ret;
+}
+
+/*
+ * Check the status of the index for the given stream. This function updates
+ * the index structure if needed and can destroy the vstream also for the HUP
+ * situation.
+ *
+ * Return 0 means that we can proceed with the index. A value of 1 means that
+ * the index has been updated and is ready to be send to the client. A negative
+ * value indicates an error that can't be handled.
+ */
+static int check_index_status(struct relay_viewer_stream *vstream,
+ struct relay_stream *rstream, struct ctf_trace *trace,
+ struct lttng_viewer_index *index)
+{
+ int ret;
+
+ assert(vstream);
+ assert(rstream);
+ assert(index);
+ assert(trace);
+
+ if (!rstream->close_flag) {
+ /* Rotate on abort (overwrite). */
+ if (vstream->abort_flag) {
+ DBG("Viewer stream %" PRIu64 " rotate because of overwrite",
+ vstream->stream_handle);
+ ret = viewer_stream_rotate(vstream, rstream);
+ if (ret < 0) {
+ goto error;
+ } else if (ret == 1) {
+ /* EOF */
+ index->status = htobe32(LTTNG_VIEWER_INDEX_HUP);
+ goto hup;
+ }
+ /* ret == 0 means successful so we continue. */
+ }
+
+ /* Check if we are in the same trace file at this point. */
+ if (rstream->tracefile_count_current == vstream->tracefile_count_current) {
+ if (rstream->beacon_ts_end != -1ULL &&
+ vstream->last_sent_index == rstream->total_index_received) {
+ /*
+ * We've received a synchronization beacon and the last index
+ * available has been sent, the index for now is inactive.
+ */
+ index->status = htobe32(LTTNG_VIEWER_INDEX_INACTIVE);
+ index->timestamp_end = htobe64(rstream->beacon_ts_end);
+ index->stream_id = htobe64(rstream->ctf_stream_id);
+ goto index_ready;
+ } else if (rstream->total_index_received <= vstream->last_sent_index
+ && !vstream->close_write_flag) {
+ /*
+ * Reader and writer are working in the same tracefile, so we care
+ * about the number of index received and sent. Otherwise, we read
+ * up to EOF.
+ */
+ index->status = htobe32(LTTNG_VIEWER_INDEX_RETRY);
+ goto index_ready;
+ }
+ }
+ /* Nothing to do with the index, continue with it. */
+ ret = 0;
+ } else if (rstream->close_flag && vstream->close_write_flag &&
+ vstream->total_index_received == vstream->last_sent_index) {
+ /* Last index sent and current tracefile closed in write */
+ index->status = htobe32(LTTNG_VIEWER_INDEX_HUP);
+ goto hup;
+ } else {
+ vstream->close_write_flag = 1;
+ ret = 0;
+ }
+
+error:
+ return ret;
+
+hup:
+ viewer_stream_delete(vstream);
+ viewer_stream_destroy(trace, vstream);
+index_ready:
+ return 1;
+}
+
/*
* Send the next index for a stream.
*
int viewer_get_next_index(struct relay_connection *conn)
{
int ret;
+ ssize_t read_ret;
struct lttng_viewer_get_next_index request_index;
struct lttng_viewer_index viewer_index;
struct ctf_packet_index packet_index;
health_code_update();
rcu_read_lock();
- session = session_find_by_id(conn->sessions_ht, conn->session_id);
- if (!session) {
+ vstream = viewer_stream_find_by_id(be64toh(request_index.stream_id));
+ if (!vstream) {
ret = -1;
goto end_unlock;
}
- vstream = viewer_stream_find_by_id(be64toh(request_index.stream_id));
- if (!vstream) {
+ session = session_find_by_id(conn->sessions_ht, vstream->session_id);
+ if (!session) {
ret = -1;
goto end_unlock;
}
goto send_reply;
}
- /* First time, we open the index file */
- if (vstream->index_read_fd < 0) {
- ret = index_open(vstream->path_name, vstream->channel_name,
- vstream->tracefile_count, vstream->tracefile_count_current);
+ rstream = stream_find_by_id(relay_streams_ht, vstream->stream_handle);
+ assert(rstream);
+
+ /* Try to open an index if one is needed for that stream. */
+ ret = try_open_index(vstream, rstream);
+ if (ret < 0) {
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.status = htobe32(LTTNG_VIEWER_INDEX_RETRY);
- goto send_reply;
- } else if (ret < 0) {
+ } else {
+ /* Unhandled error. */
viewer_index.status = htobe32(LTTNG_VIEWER_INDEX_ERR);
- goto send_reply;
}
- vstream->index_read_fd = ret;
+ goto send_reply;
}
- rstream = stream_find_by_id(relay_streams_ht, vstream->stream_handle);
- assert(rstream);
-
- if (!rstream->close_flag) {
- if (vstream->abort_flag) {
- /* Rotate on abort (overwrite). */
- DBG("Viewer rotate because of overwrite");
- ret = viewer_stream_rotate(vstream, rstream);
- if (ret < 0) {
- goto end_unlock;
- } else if (ret == 1) {
- viewer_index.status = htobe32(LTTNG_VIEWER_INDEX_HUP);
- viewer_stream_delete(vstream);
- viewer_stream_destroy(ctf_trace, vstream);
- goto send_reply;
- }
- /* ret == 0 means successful so we continue. */
- }
-
- pthread_mutex_lock(&rstream->viewer_stream_rotation_lock);
- if (rstream->tracefile_count_current == vstream->tracefile_count_current) {
- if (rstream->beacon_ts_end != -1ULL &&
- vstream->last_sent_index == rstream->total_index_received) {
- viewer_index.status = htobe32(LTTNG_VIEWER_INDEX_INACTIVE);
- viewer_index.timestamp_end = htobe64(rstream->beacon_ts_end);
- pthread_mutex_unlock(&rstream->viewer_stream_rotation_lock);
- goto send_reply;
- } else if (rstream->total_index_received <= vstream->last_sent_index
- && !vstream->close_write_flag) {
- /*
- * Reader and writer are working in the same tracefile, so we care
- * about the number of index received and sent. Otherwise, we read
- * up to EOF.
- */
- pthread_mutex_unlock(&rstream->viewer_stream_rotation_lock);
- /* No new index to send, retry later. */
- viewer_index.status = htobe32(LTTNG_VIEWER_INDEX_RETRY);
- goto send_reply;
- }
- }
- pthread_mutex_unlock(&rstream->viewer_stream_rotation_lock);
- } else if (rstream->close_flag && vstream->close_write_flag &&
- vstream->total_index_received == vstream->last_sent_index) {
- /* Last index sent and current tracefile closed in write */
- viewer_index.status = htobe32(LTTNG_VIEWER_INDEX_HUP);
- viewer_stream_delete(vstream);
- viewer_stream_destroy(ctf_trace, vstream);
+ pthread_mutex_lock(&rstream->viewer_stream_rotation_lock);
+ ret = check_index_status(vstream, rstream, ctf_trace, &viewer_index);
+ pthread_mutex_unlock(&rstream->viewer_stream_rotation_lock);
+ if (ret < 0) {
+ goto end_unlock;
+ } else if (ret == 1) {
+ /*
+ * This means the viewer index data structure has been populated by the
+ * check call thus we now send back the reply to the client.
+ */
goto send_reply;
- } else {
- vstream->close_write_flag = 1;
}
+ /* At this point, ret MUST be 0 thus we continue with the get. */
+ assert(!ret);
if (!ctf_trace->metadata_received ||
ctf_trace->metadata_received > ctf_trace->metadata_sent) {
viewer_index.flags |= LTTNG_VIEWER_FLAG_NEW_METADATA;
}
- ret = check_new_streams(vstream->session_id, conn->sessions_ht);
+ ret = check_new_streams(conn);
if (ret < 0) {
goto end_unlock;
} else if (ret == 1) {
viewer_index.flags |= LTTNG_VIEWER_FLAG_NEW_STREAM;
}
+ pthread_mutex_lock(&rstream->viewer_stream_rotation_lock);
pthread_mutex_lock(&vstream->overwrite_lock);
if (vstream->abort_flag) {
- /*
- * The file is being overwritten by the writer, we cannot * use it.
- */
- viewer_index.status = htobe32(LTTNG_VIEWER_INDEX_RETRY);
+ /* The file is being overwritten by the writer, we cannot use it. */
pthread_mutex_unlock(&vstream->overwrite_lock);
ret = viewer_stream_rotate(vstream, rstream);
+ pthread_mutex_unlock(&rstream->viewer_stream_rotation_lock);
if (ret < 0) {
goto end_unlock;
} else if (ret == 1) {
viewer_index.status = htobe32(LTTNG_VIEWER_INDEX_HUP);
viewer_stream_delete(vstream);
viewer_stream_destroy(ctf_trace, vstream);
- goto send_reply;
+ } else {
+ viewer_index.status = htobe32(LTTNG_VIEWER_INDEX_RETRY);
}
goto send_reply;
}
- ret = lttng_read(vstream->index_read_fd, &packet_index,
+ read_ret = lttng_read(vstream->index_read_fd, &packet_index,
sizeof(packet_index));
pthread_mutex_unlock(&vstream->overwrite_lock);
- if (ret < sizeof(packet_index)) {
- /*
- * The tracefile is closed in write, so we read up to EOF.
- */
- if (vstream->close_write_flag == 1) {
- viewer_index.status = htobe32(LTTNG_VIEWER_INDEX_RETRY);
- /* Rotate on normal EOF */
+ pthread_mutex_unlock(&rstream->viewer_stream_rotation_lock);
+ if (read_ret < 0) {
+ viewer_index.status = htobe32(LTTNG_VIEWER_INDEX_HUP);
+ viewer_stream_delete(vstream);
+ viewer_stream_destroy(ctf_trace, vstream);
+ goto send_reply;
+ } else if (read_ret < sizeof(packet_index)) {
+ pthread_mutex_lock(&rstream->viewer_stream_rotation_lock);
+ if (vstream->close_write_flag) {
ret = viewer_stream_rotate(vstream, rstream);
if (ret < 0) {
+ pthread_mutex_unlock(&rstream->viewer_stream_rotation_lock);
goto end_unlock;
} else if (ret == 1) {
viewer_index.status = htobe32(LTTNG_VIEWER_INDEX_HUP);
viewer_stream_delete(vstream);
viewer_stream_destroy(ctf_trace, vstream);
- goto send_reply;
+ } else {
+ viewer_index.status = htobe32(LTTNG_VIEWER_INDEX_RETRY);
}
} else {
- PERROR("Relay reading index file %d", vstream->index_read_fd);
+ ERR("Relay reading index file %d", vstream->index_read_fd);
viewer_index.status = htobe32(LTTNG_VIEWER_INDEX_ERR);
}
+ pthread_mutex_unlock(&rstream->viewer_stream_rotation_lock);
goto send_reply;
} else {
viewer_index.status = htobe32(LTTNG_VIEWER_INDEX_OK);
struct lttng_viewer_get_packet get_packet_info;
struct lttng_viewer_trace_packet reply;
struct relay_viewer_stream *stream;
+ struct relay_session *session;
struct ctf_trace *ctf_trace;
assert(conn);
goto error;
}
- ctf_trace = ctf_trace_find_by_path(conn->session->ctf_traces_ht,
+ session = session_find_by_id(conn->sessions_ht, stream->session_id);
+ if (!session) {
+ ret = -1;
+ goto error;
+ }
+
+ ctf_trace = ctf_trace_find_by_path(session->ctf_traces_ht,
stream->path_name);
assert(ctf_trace);
goto send_reply;
}
- ret = check_new_streams(stream->session_id, conn->sessions_ht);
+ ret = check_new_streams(conn);
if (ret < 0) {
goto end_unlock;
} else if (ret == 1) {
struct lttng_viewer_metadata_packet reply;
struct relay_viewer_stream *stream;
struct ctf_trace *ctf_trace;
+ struct relay_session *session;
assert(conn);
}
health_code_update();
+ memset(&reply, 0, sizeof(reply));
+
rcu_read_lock();
stream = viewer_stream_find_by_id(be64toh(request.stream_id));
if (!stream || !stream->metadata_flag) {
goto error;
}
- ctf_trace = ctf_trace_find_by_path(conn->session->ctf_traces_ht,
+ session = session_find_by_id(conn->sessions_ht, stream->session_id);
+ if (!session) {
+ ret = -1;
+ goto error;
+ }
+
+ ctf_trace = ctf_trace_find_by_path(session->ctf_traces_ht,
stream->path_name);
assert(ctf_trace);
assert(ctf_trace->metadata_sent <= ctf_trace->metadata_received);
DBG("Viewer create session received");
+ memset(&resp, 0, sizeof(resp));
resp.status = htobe32(LTTNG_VIEWER_CREATE_SESSION_OK);
- conn->viewer_session = zmalloc(sizeof(conn->viewer_session));
+ conn->viewer_session = zmalloc(sizeof(*conn->viewer_session));
if (!conn->viewer_session) {
ERR("Allocation viewer session");
resp.status = htobe32(LTTNG_VIEWER_CREATE_SESSION_ERR);
{
struct lttcomm_relayd_generic_reply reply;
+ memset(&reply, 0, sizeof(reply));
reply.ret_code = htobe32(LTTNG_ERR_UNK);
(void) send_response(conn->sock, &reply, sizeof(reply));
}
}
}
-/*
- * Delete all streams for a specific session ID.
- */
-static void destroy_viewer_streams_by_session(struct relay_session *session)
-{
- struct relay_viewer_stream *stream;
- struct lttng_ht_iter iter;
-
- assert(session);
-
- rcu_read_lock();
- cds_lfht_for_each_entry(viewer_streams_ht->ht, &iter.iter, stream,
- stream_n.node) {
- struct ctf_trace *ctf_trace;
-
- health_code_update();
- if (stream->session_id != session->id) {
- continue;
- }
-
- ctf_trace = ctf_trace_find_by_path(session->ctf_traces_ht,
- stream->path_name);
- assert(ctf_trace);
-
- viewer_stream_delete(stream);
-
- if (stream->metadata_flag) {
- ctf_trace->metadata_sent = 0;
- ctf_trace->viewer_metadata_stream = NULL;
- }
-
- viewer_stream_destroy(ctf_trace, stream);
- }
- rcu_read_unlock();
-}
-
-static void try_destroy_streams(struct relay_session *session)
-{
- struct ctf_trace *ctf_trace;
- struct lttng_ht_iter iter;
-
- assert(session);
-
- cds_lfht_for_each_entry(session->ctf_traces_ht->ht, &iter.iter, ctf_trace,
- node.node) {
- /* Attempt to destroy the ctf trace of that session. */
- ctf_trace_try_destroy(session, ctf_trace);
- }
-}
-
/*
* Delete and destroy a connection.
*
static void destroy_connection(struct lttng_ht *relay_connections_ht,
struct relay_connection *conn)
{
- struct relay_session *session;
+ struct relay_session *session, *tmp_session;
assert(relay_connections_ht);
assert(conn);
- DBG("Cleaning connection of session ID %" PRIu64, conn->session_id);
-
connection_delete(relay_connections_ht, conn);
+ if (!conn->viewer_session) {
+ goto end;
+ }
+
rcu_read_lock();
- session = session_find_by_id(conn->sessions_ht, conn->session_id);
- if (session) {
- /*
- * Very important that this is done before destroying the session so we
- * can put back every viewer stream reference from the ctf_trace.
- */
- destroy_viewer_streams_by_session(session);
- try_destroy_streams(session);
- session_viewer_try_destroy(conn->sessions_ht, session);
+ cds_list_for_each_entry_safe(session, tmp_session,
+ &conn->viewer_session->sessions_head,
+ viewer_session_list) {
+ DBG("Cleaning connection of session ID %" PRIu64, session->id);
+ cleanup_session(conn, session);
}
rcu_read_unlock();
+end:
connection_destroy(conn);
}
health_code_update();
/* Thread quit pipe has been closed. Killing thread. */
- ret = check_live_conn_pipe(pollfd, revents);
+ ret = check_thread_quit_pipe(pollfd, revents);
if (ret) {
err = 0;
goto exit;
ERR("Health error occurred in %s", __func__);
}
health_unregister(health_relayd);
- stop_threads();
+ if (lttng_relay_stop_threads()) {
+ ERR("Error stopping threads");
+ }
rcu_unregister_thread();
return NULL;
}
*/
static int create_conn_pipe(void)
{
- int ret;
-
- ret = utils_create_pipe_cloexec(live_conn_pipe);
-
- return ret;
+ return utils_create_pipe_cloexec(live_conn_pipe);
}
-void live_stop_threads(void)
+int relayd_live_join(void)
{
- int ret;
+ int ret, retval = 0;
void *status;
- stop_threads();
-
ret = pthread_join(live_listener_thread, &status);
- if (ret != 0) {
+ if (ret) {
+ errno = ret;
PERROR("pthread_join live listener");
- goto error; /* join error, exit without cleanup */
+ retval = -1;
}
ret = pthread_join(live_worker_thread, &status);
- if (ret != 0) {
+ if (ret) {
+ errno = ret;
PERROR("pthread_join live worker");
- goto error; /* join error, exit without cleanup */
+ retval = -1;
}
ret = pthread_join(live_dispatcher_thread, &status);
- if (ret != 0) {
+ if (ret) {
+ errno = ret;
PERROR("pthread_join live dispatcher");
- goto error; /* join error, exit without cleanup */
+ retval = -1;
}
- cleanup();
+ cleanup_relayd_live();
-error:
- return;
+ return retval;
}
/*
* main
*/
-int live_start_threads(struct lttng_uri *uri,
+int relayd_live_create(struct lttng_uri *uri,
struct relay_local_data *relay_ctx)
{
- int ret = 0;
+ int ret = 0, retval = 0;
void *status;
int is_root;
- assert(uri);
+ if (!uri) {
+ retval = -1;
+ goto exit_init_data;
+ }
live_uri = uri;
/* Check if daemon is UID = 0 */
if (!is_root) {
if (live_uri->port < 1024) {
ERR("Need to be root to use ports < 1024");
- ret = -1;
- goto exit;
+ retval = -1;
+ goto exit_init_data;
}
}
/* Setup the thread apps communication pipe. */
- if ((ret = create_conn_pipe()) < 0) {
- goto exit;
+ if (create_conn_pipe()) {
+ retval = -1;
+ goto exit_init_data;
}
/* Init relay command queue. */
- cds_wfq_init(&viewer_conn_queue.queue);
+ cds_wfcq_init(&viewer_conn_queue.head, &viewer_conn_queue.tail);
/* Set up max poll set size */
lttng_poll_set_max_size();
/* Setup the dispatcher thread */
ret = pthread_create(&live_dispatcher_thread, NULL,
thread_dispatcher, (void *) NULL);
- if (ret != 0) {
+ if (ret) {
+ errno = ret;
PERROR("pthread_create viewer dispatcher");
- goto exit_dispatcher;
+ retval = -1;
+ goto exit_dispatcher_thread;
}
/* Setup the worker thread */
ret = pthread_create(&live_worker_thread, NULL,
thread_worker, relay_ctx);
- if (ret != 0) {
+ if (ret) {
+ errno = ret;
PERROR("pthread_create viewer worker");
- goto exit_worker;
+ retval = -1;
+ goto exit_worker_thread;
}
/* Setup the listener thread */
ret = pthread_create(&live_listener_thread, NULL,
thread_listener, (void *) NULL);
- if (ret != 0) {
+ if (ret) {
+ errno = ret;
PERROR("pthread_create viewer listener");
- goto exit_listener;
+ retval = -1;
+ goto exit_listener_thread;
}
- ret = 0;
- goto end;
+ /*
+ * All OK, started all threads.
+ */
+ return retval;
+
-exit_listener:
ret = pthread_join(live_listener_thread, &status);
- if (ret != 0) {
+ if (ret) {
+ errno = ret;
PERROR("pthread_join live listener");
- goto error; /* join error, exit without cleanup */
+ retval = -1;
}
+exit_listener_thread:
-exit_worker:
ret = pthread_join(live_worker_thread, &status);
- if (ret != 0) {
+ if (ret) {
+ errno = ret;
PERROR("pthread_join live worker");
- goto error; /* join error, exit without cleanup */
+ retval = -1;
}
+exit_worker_thread:
-exit_dispatcher:
ret = pthread_join(live_dispatcher_thread, &status);
- if (ret != 0) {
+ if (ret) {
+ errno = ret;
PERROR("pthread_join live dispatcher");
- goto error; /* join error, exit without cleanup */
+ retval = -1;
}
+exit_dispatcher_thread:
-exit:
- cleanup();
+exit_init_data:
+ cleanup_relayd_live();
-end:
-error:
- return ret;
+ return retval;
}
projects / lttng-tools.git / blobdiff