}
/*
- * 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;
}
/* Stopping all threads */
DBG("Terminating all live threads");
- ret = notify_thread_pipe(live_conn_pipe[1]);
+ ret = notify_thread_pipe(thread_quit_pipe[1]);
if (ret < 0) {
ERR("write error on thread quit pipe");
}
}
/* 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;
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 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);
+}
+
/*
* Send the viewer the list of current sessions.
*/
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();
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;
}
+ send_streams = 1;
+ response.status = htobe32(LTTNG_VIEWER_NEW_STREAMS_OK);
+
if (!send_streams) {
goto send_reply;
}
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;
}
} 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)) {
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;
}
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) {
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);
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);
}
}
-/*
- * 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;