* 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
*/
-#define _GNU_SOURCE
#define _LGPL_SOURCE
#include <getopt.h>
#include <grp.h>
#include <urcu/rculist.h>
#include <unistd.h>
#include <fcntl.h>
-#include <config.h>
#include <lttng/lttng.h>
#include <common/common.h>
continue;
}
/*
- * stream published is protected by the session
- * lock.
+ * stream published is protected by the session lock.
*/
if (!stream->published) {
goto next;
}
- /*
- * Stream has no data, don't consider it yet.
- */
- if (stream->is_metadata) {
- if (!stream->metadata_received) {
- goto next;
- }
- } else {
- if (stream->prev_seq == -1ULL) {
- goto next;
- }
- }
vstream = viewer_stream_get_by_id(stream->stream_handle);
if (!vstream) {
vstream = viewer_stream_create(stream, seek_t);
goto exit;
}
- if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
- ERR("socket poll error");
- goto error;
- } else if (revents & LPOLLIN) {
+ if (revents & LPOLLIN) {
/*
* A new connection is requested, therefore a
* viewer connection is allocated in this
* exchange in cds_wfcq_enqueue.
*/
futex_nto1_wake(&viewer_conn_queue.futex);
+ } else if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
+ ERR("socket poll error");
+ goto error;
+ } else {
+ ERR("Unexpected poll events %u for sock %d", revents, pollfd);
+ goto error;
}
}
}
buf_count = new_buf_count;
}
send_session = &send_session_buf[count];
- strncpy(send_session->session_name, session->session_name,
- sizeof(send_session->session_name));
- strncpy(send_session->hostname, session->hostname,
- sizeof(send_session->hostname));
+ if (lttng_strncpy(send_session->session_name,
+ session->session_name,
+ sizeof(send_session->session_name))) {
+ ret = -1;
+ rcu_read_unlock();
+ goto end_free;
+ }
+ if (lttng_strncpy(send_session->hostname, session->hostname,
+ sizeof(send_session->hostname))) {
+ ret = -1;
+ rcu_read_unlock();
+ goto end_free;
+ }
send_session->id = htobe64(session->id);
send_session->live_timer = htobe32(session->live_timer);
if (session->viewer_attached) {
return ret;
}
+/*
+ * Detach a viewer session.
+ *
+ * Return 0 on success or else a negative value.
+ */
+static
+int viewer_detach_session(struct relay_connection *conn)
+{
+ int ret;
+ struct lttng_viewer_detach_session_response response;
+ struct lttng_viewer_detach_session_request request;
+ struct relay_session *session = NULL;
+ uint64_t viewer_session_to_close;
+
+ DBG("Viewer detach session received");
+
+ assert(conn);
+
+ health_code_update();
+
+ /* Receive the request from the connected client. */
+ ret = recv_request(conn->sock, &request, sizeof(request));
+ if (ret < 0) {
+ goto end;
+ }
+ viewer_session_to_close = be64toh(request.session_id);
+
+ if (!conn->viewer_session) {
+ DBG("Client trying to detach before creating a live viewer session");
+ response.status = htobe32(LTTNG_VIEWER_DETACH_SESSION_ERR);
+ goto send_reply;
+ }
+
+ health_code_update();
+
+ memset(&response, 0, sizeof(response));
+ DBG("Detaching from session ID %" PRIu64, viewer_session_to_close);
+
+ session = session_get_by_id(be64toh(request.session_id));
+ if (!session) {
+ DBG("Relay session %" PRIu64 " not found",
+ be64toh(request.session_id));
+ response.status = htobe32(LTTNG_VIEWER_DETACH_SESSION_UNK);
+ goto send_reply;
+ }
+
+ ret = viewer_session_is_attached(conn->viewer_session, session);
+ if (ret != 1) {
+ DBG("Not attached to this session");
+ response.status = htobe32(LTTNG_VIEWER_DETACH_SESSION_ERR);
+ goto send_reply_put;
+ }
+
+ viewer_session_close_one_session(conn->viewer_session, session);
+ response.status = htobe32(LTTNG_VIEWER_DETACH_SESSION_OK);
+ DBG("Session %" PRIu64 " detached.", viewer_session_to_close);
+
+send_reply_put:
+ session_put(session);
+
+send_reply:
+ health_code_update();
+ ret = send_response(conn->sock, &response, sizeof(response));
+ if (ret < 0) {
+ goto end;
+ }
+ health_code_update();
+ ret = 0;
+
+end:
+ return ret;
+}
/*
* live_relay_unknown_command: send -1 if received unknown command
case LTTNG_VIEWER_CREATE_SESSION:
ret = viewer_create_session(conn);
break;
+ case LTTNG_VIEWER_DETACH_SESSION:
+ ret = viewer_detach_session(conn);
+ break;
default:
ERR("Received unknown viewer command (%u)",
be32toh(recv_hdr->cmd));
/* Inspect the relay conn pipe for new connection. */
if (pollfd == live_conn_pipe[0]) {
- if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
- ERR("Relay live pipe error");
- goto error;
- } else if (revents & LPOLLIN) {
+ if (revents & LPOLLIN) {
struct relay_connection *conn;
ret = lttng_read(live_conn_pipe[0],
LPOLLIN | LPOLLRDHUP);
connection_ht_add(viewer_connections_ht, conn);
DBG("Connection socket %d added to poll", conn->sock->fd);
+ } else if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
+ ERR("Relay live pipe error");
+ goto error;
+ } else {
+ ERR("Unexpected poll events %u for sock %d", revents, pollfd);
+ goto error;
}
} else {
/* Connection activity. */
continue;
}
- if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
- cleanup_connection_pollfd(&events, pollfd);
- /* Put "create" ownership reference. */
- connection_put(conn);
- } else if (revents & LPOLLIN) {
+ if (revents & LPOLLIN) {
ret = conn->sock->ops->recvmsg(conn->sock, &recv_hdr,
sizeof(recv_hdr), 0);
if (ret <= 0) {
DBG("Viewer connection closed with %d", pollfd);
}
}
+ } else if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
+ cleanup_connection_pollfd(&events, pollfd);
+ /* Put "create" ownership reference. */
+ connection_put(conn);
+ } else {
+ ERR("Unexpected poll events %u for sock %d", revents, pollfd);
+ connection_put(conn);
+ goto error;
}
/* Put local "get_by_sock" reference. */
connection_put(conn);
projects / lttng-tools.git / blobdiff