X-Git-Url: https://git.liburcu.org/?a=blobdiff_plain;f=src%2Fbin%2Flttng-relayd%2Fmain.c;h=d3f9fe202a6c73f7a9ba2a7f7281fccb0a1fa7b7;hb=2c93dc32c897a1bf6535a84817a6e5430b51189b;hp=bdc27475eeaf3c89552ca3cd507ff8b5235c96c8;hpb=3a9e5d1671c6096fed41d7f6dc50cc28479ee55c;p=lttng-tools.git diff --git a/src/bin/lttng-relayd/main.c b/src/bin/lttng-relayd/main.c index bdc27475e..d3f9fe202 100644 --- a/src/bin/lttng-relayd/main.c +++ b/src/bin/lttng-relayd/main.c @@ -1750,24 +1750,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; @@ -2505,6 +2487,125 @@ end_no_session: return ret; } +static ssize_t relay_unpack_rotate_streams_header( + const struct lttng_buffer_view *payload, + struct lttcomm_relayd_rotate_streams *_rotate_streams) +{ + struct lttcomm_relayd_rotate_streams rotate_streams; + /* + * Set to the smallest version (packed) of `lttcomm_relayd_rotate_streams`. + * This is the smallest version of this structure, but it can be larger; + * this variable is updated once the proper size of the structure is known. + * + * See comment at the declaration of this structure for more information. + */ + ssize_t header_len = sizeof(struct lttcomm_relayd_rotate_streams_packed); + size_t expected_payload_size_no_padding, + expected_payload_size_3_bytes_padding, + expected_payload_size_7_bytes_padding; + + if (payload->size < header_len) { + ERR("Unexpected payload size in \"relay_rotate_session_stream\": expected >= %zu bytes, got %zu bytes", + header_len, payload->size); + goto error; + } + + /* + * Some versions incorrectly omitted the LTTNG_PACKED annotation on the + * `new_chunk_id` optional field of struct lttcomm_relayd_rotate_streams. + * + * We start by "unpacking" `stream_count` to figure out the padding length + * emited by our peer. + */ + memcpy(&rotate_streams.stream_count, payload->data, + sizeof(rotate_streams.stream_count)); + rotate_streams = (typeof(rotate_streams)) { + .stream_count = be32toh(rotate_streams.stream_count), + }; + + /* + * Payload size expected given the possible padding lengths in + * `struct lttcomm_relayd_rotate_streams`. + */ + expected_payload_size_no_padding = (rotate_streams.stream_count * + sizeof(*rotate_streams.rotation_positions)) + + sizeof(struct lttcomm_relayd_rotate_streams_packed); + expected_payload_size_3_bytes_padding = (rotate_streams.stream_count * + sizeof(*rotate_streams.rotation_positions)) + + sizeof(struct lttcomm_relayd_rotate_streams_3_bytes_padding); + expected_payload_size_7_bytes_padding = (rotate_streams.stream_count * + sizeof(*rotate_streams.rotation_positions)) + + sizeof(struct lttcomm_relayd_rotate_streams_7_bytes_padding); + + if (payload->size == expected_payload_size_no_padding) { + struct lttcomm_relayd_rotate_streams_packed packed_rotate_streams; + + /* + * This handles cases where someone might build with + * -fpack-struct or any other toolchain that wouldn't produce + * padding to align `value`. + */ + DBG("Received `struct lttcomm_relayd_rotate_streams` with no padding"); + + header_len = sizeof(packed_rotate_streams); + memcpy(&packed_rotate_streams, payload->data, header_len); + + /* Unpack the packed structure to the natively-packed version. */ + *_rotate_streams = (typeof(*_rotate_streams)) { + .stream_count = be32toh(packed_rotate_streams.stream_count), + .new_chunk_id = (typeof(_rotate_streams->new_chunk_id)) { + .is_set = !!packed_rotate_streams.new_chunk_id.is_set, + .value = be64toh(packed_rotate_streams.new_chunk_id.value), + } + }; + } else if (payload->size == expected_payload_size_3_bytes_padding) { + struct lttcomm_relayd_rotate_streams_3_bytes_padding padded_rotate_streams; + + DBG("Received `struct lttcomm_relayd_rotate_streams` with 3 bytes of padding (4-byte aligned peer)"); + + header_len = sizeof(padded_rotate_streams); + memcpy(&padded_rotate_streams, payload->data, header_len); + + /* Unpack the 3-byte padded structure to the natively-packed version. */ + *_rotate_streams = (typeof(*_rotate_streams)) { + .stream_count = be32toh(padded_rotate_streams.stream_count), + .new_chunk_id = (typeof(_rotate_streams->new_chunk_id)) { + .is_set = !!padded_rotate_streams.new_chunk_id.is_set, + .value = be64toh(padded_rotate_streams.new_chunk_id.value), + } + }; + } else if (payload->size == expected_payload_size_7_bytes_padding) { + struct lttcomm_relayd_rotate_streams_7_bytes_padding padded_rotate_streams; + + DBG("Received `struct lttcomm_relayd_rotate_streams` with 7 bytes of padding (8-byte aligned peer)"); + + header_len = sizeof(padded_rotate_streams); + memcpy(&padded_rotate_streams, payload->data, header_len); + + /* Unpack the 7-byte padded structure to the natively-packed version. */ + *_rotate_streams = (typeof(*_rotate_streams)) { + .stream_count = be32toh(padded_rotate_streams.stream_count), + .new_chunk_id = (typeof(_rotate_streams->new_chunk_id)) { + .is_set = !!padded_rotate_streams.new_chunk_id.is_set, + .value = be64toh(padded_rotate_streams.new_chunk_id.value), + } + }; + + header_len = sizeof(padded_rotate_streams); + } else { + ERR("Unexpected payload size in \"relay_rotate_session_stream\": expected %zu, %zu or %zu bytes, got %zu bytes", + expected_payload_size_no_padding, + expected_payload_size_3_bytes_padding, + expected_payload_size_7_bytes_padding, + payload->size); + goto error; + } + + return header_len; +error: + return -1; +} + /* * relay_rotate_session_stream: rotate a stream to a new tracefile for the * session rotation feature (not the tracefile rotation feature). @@ -2522,11 +2623,11 @@ static int relay_rotate_session_streams( struct lttcomm_relayd_rotate_streams rotate_streams; struct lttcomm_relayd_generic_reply reply = {}; struct relay_stream *stream = NULL; - const size_t header_len = sizeof(struct lttcomm_relayd_rotate_streams); struct lttng_trace_chunk *next_trace_chunk = NULL; struct lttng_buffer_view stream_positions; char chunk_id_buf[MAX_INT_DEC_LEN(uint64_t)]; const char *chunk_id_str = "none"; + ssize_t header_len; if (!session || !conn->version_check_done) { ERR("Trying to rotate a stream before version check"); @@ -2540,24 +2641,12 @@ static int relay_rotate_session_streams( goto end_no_reply; } - if (payload->size < header_len) { - ERR("Unexpected payload size in \"relay_rotate_session_stream\": expected >= %zu bytes, got %zu bytes", - header_len, payload->size); + header_len = relay_unpack_rotate_streams_header(payload, &rotate_streams); + if (header_len < 0) { ret = -1; goto end_no_reply; } - memcpy(&rotate_streams, payload->data, header_len); - - /* Convert header to host endianness. */ - rotate_streams = (typeof(rotate_streams)) { - .stream_count = be32toh(rotate_streams.stream_count), - .new_chunk_id = (typeof(rotate_streams.new_chunk_id)) { - .is_set = !!rotate_streams.new_chunk_id.is_set, - .value = be64toh(rotate_streams.new_chunk_id.value), - } - }; - if (rotate_streams.new_chunk_id.is_set) { /* * Retrieve the trace chunk the stream must transition to. As @@ -2566,7 +2655,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]; @@ -2595,7 +2687,7 @@ static int relay_rotate_session_streams( chunk_id_str); stream_positions = lttng_buffer_view_from_view(payload, - sizeof(rotate_streams), -1); + header_len, -1); if (!stream_positions.data || stream_positions.size < (rotate_streams.stream_count * @@ -2654,8 +2746,6 @@ end_no_reply: return ret; } - - /* * relay_create_trace_chunk: create a new trace chunk */ @@ -2699,6 +2789,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, @@ -2710,7 +2802,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 = ""; @@ -2789,7 +2880,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]; @@ -2804,7 +2897,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 @@ -2813,7 +2905,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; @@ -2822,9 +2914,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, @@ -2899,7 +2990,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];