X-Git-Url: https://git.lttng.org/?p=lttng-tools.git;a=blobdiff_plain;f=src%2Fbin%2Flttng-relayd%2Fmain.cpp;h=942bfa4d5ab3abded6b6031e0498042e5e1dfbb9;hp=adbf3f3b4c18be0693770903599934588c0c37e5;hb=c5c79321eb1937f3d208210365c512f4a186ec2a;hpb=ce9dbd47eea9fe023691518ef46b58c03b89c236 diff --git a/src/bin/lttng-relayd/main.cpp b/src/bin/lttng-relayd/main.cpp index adbf3f3b4..942bfa4d5 100644 --- a/src/bin/lttng-relayd/main.cpp +++ b/src/bin/lttng-relayd/main.cpp @@ -1758,24 +1758,6 @@ static int relay_close_stream(const struct lttcomm_relayd_hdr *recv_hdr, * request. */ try_stream_close(stream); - if (stream->is_metadata) { - struct relay_viewer_stream *vstream; - - vstream = viewer_stream_get_by_id(stream->stream_handle); - if (vstream) { - if (stream->no_new_metadata_notified) { - /* - * Since all the metadata has been sent to the - * viewer and that we have a request to close - * its stream, we can safely teardown the - * corresponding metadata viewer stream. - */ - viewer_stream_put(vstream); - } - /* Put local reference. */ - viewer_stream_put(vstream); - } - } stream_put(stream); ret = 0; @@ -2681,7 +2663,10 @@ static int relay_rotate_session_streams( */ next_trace_chunk = sessiond_trace_chunk_registry_get_chunk( sessiond_trace_chunk_registry, - session->sessiond_uuid, session->id, + session->sessiond_uuid, + conn->session->id_sessiond.is_set ? + conn->session->id_sessiond.value : + conn->session->id, rotate_streams.new_chunk_id.value); if (!next_trace_chunk) { char uuid_str[LTTNG_UUID_STR_LEN]; @@ -2812,6 +2797,8 @@ static int relay_create_trace_chunk(const struct lttcomm_relayd_hdr *recv_hdr, msg->creation_timestamp = be64toh(msg->creation_timestamp); msg->override_name_length = be32toh(msg->override_name_length); + pthread_mutex_lock(&conn->session->lock); + session->ongoing_rotation = true; if (session->current_trace_chunk && !lttng_trace_chunk_get_name_overridden(session->current_trace_chunk)) { chunk_status = lttng_trace_chunk_rename_path(session->current_trace_chunk, @@ -2823,7 +2810,6 @@ static int relay_create_trace_chunk(const struct lttcomm_relayd_hdr *recv_hdr, goto end; } } - session->ongoing_rotation = true; if (!session->current_trace_chunk) { if (!session->has_rotated) { new_path = ""; @@ -2902,7 +2888,9 @@ static int relay_create_trace_chunk(const struct lttcomm_relayd_hdr *recv_hdr, published_chunk = sessiond_trace_chunk_registry_publish_chunk( sessiond_trace_chunk_registry, conn->session->sessiond_uuid, - conn->session->id, + conn->session->id_sessiond.is_set ? + conn->session->id_sessiond.value : + conn->session->id, chunk); if (!published_chunk) { char uuid_str[LTTNG_UUID_STR_LEN]; @@ -2917,7 +2905,6 @@ static int relay_create_trace_chunk(const struct lttcomm_relayd_hdr *recv_hdr, goto end; } - pthread_mutex_lock(&conn->session->lock); if (conn->session->pending_closure_trace_chunk) { /* * Invalid; this means a second create_trace_chunk command was @@ -2926,7 +2913,7 @@ static int relay_create_trace_chunk(const struct lttcomm_relayd_hdr *recv_hdr, ERR("Invalid trace chunk close command received; a trace chunk is already waiting for a trace chunk close command"); reply_code = LTTNG_ERR_INVALID_PROTOCOL; ret = -1; - goto end_unlock_session; + goto end; } conn->session->pending_closure_trace_chunk = conn->session->current_trace_chunk; @@ -2935,9 +2922,8 @@ static int relay_create_trace_chunk(const struct lttcomm_relayd_hdr *recv_hdr, if (!conn->session->pending_closure_trace_chunk) { session->ongoing_rotation = false; } -end_unlock_session: - pthread_mutex_unlock(&conn->session->lock); end: + pthread_mutex_unlock(&conn->session->lock); reply.ret_code = htobe32((uint32_t) reply_code); send_ret = conn->sock->ops->sendmsg(conn->sock, &reply, @@ -3010,7 +2996,9 @@ static int relay_close_trace_chunk(const struct lttcomm_relayd_hdr *recv_hdr, chunk = sessiond_trace_chunk_registry_get_chunk( sessiond_trace_chunk_registry, conn->session->sessiond_uuid, - conn->session->id, + conn->session->id_sessiond.is_set ? + conn->session->id_sessiond.value : + conn->session->id, chunk_id); if (!chunk) { char uuid_str[LTTNG_UUID_STR_LEN]; @@ -3320,86 +3308,68 @@ end_no_reply: return ret; } -#define DBG_CMD(cmd_name, conn) \ - DBG3("Processing \"%s\" command for socket %i", cmd_name, conn->sock->fd); - static int relay_process_control_command(struct relay_connection *conn, const struct lttcomm_relayd_hdr *header, const struct lttng_buffer_view *payload) { int ret = 0; + DBG3("Processing \"%s\" command for socket %i", + lttcomm_relayd_command_str((lttcomm_relayd_command) header->cmd), + conn->sock->fd); switch (header->cmd) { case RELAYD_CREATE_SESSION: - DBG_CMD("RELAYD_CREATE_SESSION", conn); ret = relay_create_session(header, conn, payload); break; case RELAYD_ADD_STREAM: - DBG_CMD("RELAYD_ADD_STREAM", conn); ret = relay_add_stream(header, conn, payload); break; case RELAYD_START_DATA: - DBG_CMD("RELAYD_START_DATA", conn); ret = relay_start(header, conn, payload); break; case RELAYD_SEND_METADATA: - DBG_CMD("RELAYD_SEND_METADATA", conn); ret = relay_recv_metadata(header, conn, payload); break; case RELAYD_VERSION: - DBG_CMD("RELAYD_VERSION", conn); ret = relay_send_version(header, conn, payload); break; case RELAYD_CLOSE_STREAM: - DBG_CMD("RELAYD_CLOSE_STREAM", conn); ret = relay_close_stream(header, conn, payload); break; case RELAYD_DATA_PENDING: - DBG_CMD("RELAYD_DATA_PENDING", conn); ret = relay_data_pending(header, conn, payload); break; case RELAYD_QUIESCENT_CONTROL: - DBG_CMD("RELAYD_QUIESCENT_CONTROL", conn); ret = relay_quiescent_control(header, conn, payload); break; case RELAYD_BEGIN_DATA_PENDING: - DBG_CMD("RELAYD_BEGIN_DATA_PENDING", conn); ret = relay_begin_data_pending(header, conn, payload); break; case RELAYD_END_DATA_PENDING: - DBG_CMD("RELAYD_END_DATA_PENDING", conn); ret = relay_end_data_pending(header, conn, payload); break; case RELAYD_SEND_INDEX: - DBG_CMD("RELAYD_SEND_INDEX", conn); ret = relay_recv_index(header, conn, payload); break; case RELAYD_STREAMS_SENT: - DBG_CMD("RELAYD_STREAMS_SENT", conn); ret = relay_streams_sent(header, conn, payload); break; case RELAYD_RESET_METADATA: - DBG_CMD("RELAYD_RESET_METADATA", conn); ret = relay_reset_metadata(header, conn, payload); break; case RELAYD_ROTATE_STREAMS: - DBG_CMD("RELAYD_ROTATE_STREAMS", conn); ret = relay_rotate_session_streams(header, conn, payload); break; case RELAYD_CREATE_TRACE_CHUNK: - DBG_CMD("RELAYD_CREATE_TRACE_CHUNK", conn); ret = relay_create_trace_chunk(header, conn, payload); break; case RELAYD_CLOSE_TRACE_CHUNK: - DBG_CMD("RELAYD_CLOSE_TRACE_CHUNK", conn); ret = relay_close_trace_chunk(header, conn, payload); break; case RELAYD_TRACE_CHUNK_EXISTS: - DBG_CMD("RELAYD_TRACE_CHUNK_EXISTS", conn); ret = relay_trace_chunk_exists(header, conn, payload); break; case RELAYD_GET_CONFIGURATION: - DBG_CMD("RELAYD_GET_CONFIGURATION", conn); ret = relay_get_configuration(header, conn, payload); break; case RELAYD_UPDATE_SYNC_INFO: @@ -3547,9 +3517,9 @@ static enum relay_connection_status relay_process_control_receive_header( memcpy(&conn->protocol.ctrl.state.receive_payload.header, &header, sizeof(header)); - DBG("Done receiving control command header: fd = %i, cmd = %" PRIu32 ", cmd_version = %" PRIu32 ", payload size = %" PRIu64 " bytes", - conn->sock->fd, header.cmd, header.cmd_version, - header.data_size); + DBG("Done receiving control command header: fd = %i, cmd = %s, cmd_version = %" PRIu32 ", payload size = %" PRIu64 " bytes", + conn->sock->fd, lttcomm_relayd_command_str((enum lttcomm_relayd_command) header.cmd), + header.cmd_version, header.data_size); if (header.data_size > DEFAULT_NETWORK_RELAYD_CTRL_MAX_PAYLOAD_SIZE) { ERR("Command header indicates a payload (%" PRIu64 " bytes) that exceeds the maximal payload size allowed on a control connection.",