Skip to content

Commit

Permalink
[C] Due to termOffset being cast to an int we need to protect against…
Browse files Browse the repository at this point in the history 
… negative value which is a change from existing long behaviour.
  • Loading branch information
@mjpt777
mjpt777 committed Oct 13, 2022
1 parent ac23181 commit f379397
Show file tree
Hide file tree
Showing 3 changed files with 27 additions and 27 deletions.
42 changes: 22 additions & 20 deletions aeron-client/src/main/c/aeron_publication.c
View file
Original file line number Diff line number Diff line change
Expand Up @@ -187,17 +187,17 @@ static int64_t aeron_publication_claim(
const int32_t aligned_frame_length = (int32_t)AERON_ALIGN(frame_length, AERON_LOGBUFFER_FRAME_ALIGNMENT);
const int64_t raw_tail = aeron_publication_get_and_add_raw_tail(
term_tail_counter, (size_t)aligned_frame_length);
const int32_t term_offset = (int32_t)(raw_tail & 0xFFFFFFFF);
const int32_t term_id = aeron_logbuffer_term_id(raw_tail);
const int32_t term_length = (int32_t)term_buffer->length;
const int32_t term_offset = aeron_logbuffer_term_offset(raw_tail, term_length);
const int32_t term_id = aeron_logbuffer_term_id(raw_tail);

int32_t resulting_offset = term_offset + aligned_frame_length;
int64_t position = aeron_logbuffer_compute_position(
term_id, resulting_offset, publication->position_bits_to_shift, publication->initial_term_id);
if (resulting_offset > term_length)
{
return aeron_publication_handle_end_of_log_condition(
publication, term_buffer, (int32_t)term_offset, term_length, term_id, position);
publication, term_buffer, term_offset, term_length, term_id, position);
}
else
{
Expand All @@ -210,7 +210,6 @@ static int64_t aeron_publication_claim(
return position;
}


static int64_t aeron_publication_append_unfragmented_message(
aeron_publication_t *publication,
aeron_mapped_buffer_t *term_buffer,
Expand All @@ -224,9 +223,9 @@ static int64_t aeron_publication_append_unfragmented_message(
const int32_t aligned_frame_length = (int32_t)AERON_ALIGN(frame_length, AERON_LOGBUFFER_FRAME_ALIGNMENT);
const int64_t raw_tail = aeron_publication_get_and_add_raw_tail(
term_tail_counter, (size_t)aligned_frame_length);
const int32_t term_offset = (int32_t)(raw_tail & 0xFFFFFFFF);
const int32_t term_id = aeron_logbuffer_term_id(raw_tail);
const int32_t term_length = (int32_t)term_buffer->length;
const int32_t term_offset = aeron_logbuffer_term_offset(raw_tail, term_length);
const int32_t term_id = aeron_logbuffer_term_id(raw_tail);

int32_t resulting_offset = term_offset + aligned_frame_length;
int64_t position = aeron_logbuffer_compute_position(
Expand Down Expand Up @@ -272,9 +271,9 @@ static int64_t aeron_publication_append_fragmented_message(
const size_t required_length =
(num_max_payloads * (max_payload_length + AERON_DATA_HEADER_LENGTH)) + last_frame_length;
const int64_t raw_tail = aeron_publication_get_and_add_raw_tail(term_tail_counter, required_length);
const int32_t term_offset = (int32_t)(raw_tail & 0xFFFFFFFF);
const int32_t term_id = aeron_logbuffer_term_id(raw_tail);
const int32_t term_length = (int32_t)term_buffer->length;
const int32_t term_offset = aeron_logbuffer_term_offset(raw_tail, term_length);
const int32_t term_id = aeron_logbuffer_term_id(raw_tail);

int32_t resulting_offset = term_offset + (int32_t)required_length;
int64_t position = aeron_logbuffer_compute_position(
Expand All @@ -288,7 +287,7 @@ static int64_t aeron_publication_append_fragmented_message(
{
uint8_t flags = AERON_DATA_HEADER_BEGIN_FLAG;
size_t remaining = length;
int32_t frame_offset = (int32_t)term_offset;
int32_t frame_offset = term_offset;

do
{
Expand Down Expand Up @@ -342,9 +341,9 @@ static int64_t aeron_publication_append_unfragmented_messagev(
const int32_t aligned_frame_length = (int32_t)AERON_ALIGN(frame_length, AERON_LOGBUFFER_FRAME_ALIGNMENT);
const int64_t raw_tail = aeron_publication_get_and_add_raw_tail(
term_tail_counter, (size_t)aligned_frame_length);
const int32_t term_offset = (int32_t)(raw_tail & 0xFFFFFFFF);
const int32_t term_id = aeron_logbuffer_term_id(raw_tail);
const int32_t term_length = (int32_t)term_buffer->length;
const int32_t term_offset = aeron_logbuffer_term_offset(raw_tail, term_length);
const int32_t term_id = aeron_logbuffer_term_id(raw_tail);

int32_t resulting_offset = term_offset + aligned_frame_length;
int64_t position = aeron_logbuffer_compute_position(
Expand Down Expand Up @@ -399,9 +398,9 @@ static int64_t aeron_publication_append_fragmented_messagev(
const size_t required_length =
(num_max_payloads * (max_payload_length + AERON_DATA_HEADER_LENGTH)) + last_frame_length;
const int64_t raw_tail = aeron_publication_get_and_add_raw_tail(term_tail_counter, required_length);
const int32_t term_offset = (int32_t)(raw_tail & 0xFFFFFFFF);
const int32_t term_id = aeron_logbuffer_term_id(raw_tail);
const int32_t term_length = (int32_t)term_buffer->length;
const int32_t term_offset = aeron_logbuffer_term_offset(raw_tail, term_length);
const int32_t term_id = aeron_logbuffer_term_id(raw_tail);

int32_t resulting_offset = term_offset + (int32_t)required_length;
int64_t position = aeron_logbuffer_compute_position(
Expand All @@ -415,7 +414,7 @@ static int64_t aeron_publication_append_fragmented_messagev(
{
uint8_t flags = AERON_DATA_HEADER_BEGIN_FLAG;
size_t remaining = length, i = 0;
int32_t frame_offset = (int32_t)term_offset;
int32_t frame_offset = term_offset;
int32_t current_buffer_offset = 0;

do
Expand Down Expand Up @@ -503,7 +502,8 @@ int64_t aeron_publication_offer(
const size_t index = aeron_logbuffer_index_by_term_count(term_count);
const int64_t raw_tail = aeron_publication_raw_tail_volatile(
&publication->log_meta_data->term_tail_counters[index]);
const int32_t term_offset = (int32_t)(raw_tail & 0xFFFFFFFF);
const int32_t term_length = publication->log_meta_data->term_length;
const int32_t term_offset = aeron_logbuffer_term_offset(raw_tail, term_length);
const int32_t term_id = aeron_logbuffer_term_id(raw_tail);

if (term_count != aeron_logbuffer_compute_term_count(term_id, publication->initial_term_id))
Expand Down Expand Up @@ -592,7 +592,8 @@ int64_t aeron_publication_offerv(
const size_t index = aeron_logbuffer_index_by_term_count(term_count);
const int64_t raw_tail = aeron_publication_raw_tail_volatile(
&publication->log_meta_data->term_tail_counters[index]);
const int32_t term_offset = (int32_t)(raw_tail & 0xFFFFFFFF);
const int32_t term_length = publication->log_meta_data->term_length;
const int32_t term_offset = aeron_logbuffer_term_offset(raw_tail, term_length);
const int32_t term_id = aeron_logbuffer_term_id(raw_tail);

if (term_count != aeron_logbuffer_compute_term_count(term_id, publication->initial_term_id))
Expand Down Expand Up @@ -682,7 +683,8 @@ int64_t aeron_publication_try_claim(aeron_publication_t *publication, size_t len
const size_t index = aeron_logbuffer_index_by_term_count(term_count);
const int64_t raw_tail = aeron_publication_raw_tail_volatile(
&publication->log_meta_data->term_tail_counters[index]);
const int32_t term_offset = (int32_t)(raw_tail & 0xFFFFFFFF);
const int32_t term_length = publication->log_meta_data->term_length;
const int32_t term_offset = aeron_logbuffer_term_offset(raw_tail, term_length);
const int32_t term_id = aeron_logbuffer_term_id(raw_tail);

if (term_count != aeron_logbuffer_compute_term_count(term_id, publication->initial_term_id))
Expand Down Expand Up @@ -796,11 +798,11 @@ int64_t aeron_publication_position(aeron_publication_t *publication)
const size_t index = aeron_logbuffer_index_by_term_count(term_count);
const int64_t raw_tail = aeron_publication_raw_tail_volatile(
&publication->log_meta_data->term_tail_counters[index]);
const int64_t term_length = publication->log_meta_data->term_length;
const int64_t term_offset = term_length < (raw_tail & 0xFFFFFFFF) ? term_length : (raw_tail & 0xFFFFFFFF);
const int32_t term_length = publication->log_meta_data->term_length;
const int32_t term_offset = aeron_logbuffer_term_offset(raw_tail, term_length);
const int32_t term_id = aeron_logbuffer_term_id(raw_tail);
const int64_t position = aeron_logbuffer_compute_position(
term_id, (int32_t)term_offset, publication->position_bits_to_shift, publication->initial_term_id);
term_id, term_offset, publication->position_bits_to_shift, publication->initial_term_id);

return position;
}
Expand Down
10 changes: 4 additions & 6 deletions aeron-client/src/main/c/concurrent/aeron_logbuffer_descriptor.h
View file
Original file line number Diff line number Diff line change
Expand Up @@ -17,8 +17,6 @@
#ifndef AERON_LOGBUFFER_DESCRIPTOR_H
#define AERON_LOGBUFFER_DESCRIPTOR_H

#include <assert.h>
#include <string.h>
#include "protocol/aeron_udp_protocol.h"
#include "util/aeron_bitutil.h"
#include "util/aeron_math.h"
Expand Down Expand Up @@ -80,9 +78,8 @@ inline uint64_t aeron_logbuffer_compute_log_length(uint64_t term_length, uint64_

inline int32_t aeron_logbuffer_term_offset(int64_t raw_tail, int32_t term_length)
{
int32_t offset = (int32_t)(raw_tail & 0xFFFFFFFFL);

return offset < term_length ? offset : term_length;
int64_t offset = raw_tail & 0xFFFFFFFFL;
return offset < term_length ? (int32_t)offset : term_length;
}

inline int32_t aeron_logbuffer_term_id(int64_t raw_tail)
Expand Down Expand Up @@ -188,7 +185,8 @@ inline void aeron_logbuffer_fill_default_header(
uint8_t *log_meta_data_buffer, int32_t session_id, int32_t stream_id, int32_t initial_term_id)
{
aeron_logbuffer_metadata_t *log_meta_data = (aeron_logbuffer_metadata_t *)log_meta_data_buffer;
aeron_data_header_t *data_header = (aeron_data_header_t *)(log_meta_data_buffer + sizeof(aeron_logbuffer_metadata_t));
aeron_data_header_t *data_header =
(aeron_data_header_t *)(log_meta_data_buffer + sizeof(aeron_logbuffer_metadata_t));

log_meta_data->default_frame_header_length = AERON_DATA_HEADER_LENGTH;
data_header->frame_header.frame_length = 0;
Expand Down
2 changes: 1 addition & 1 deletion aeron-client/src/main/c/protocol/aeron_udp_protocol.h
View file
Original file line number Diff line number Diff line change
Expand Up @@ -24,7 +24,7 @@
#pragma pack(4)
typedef struct aeron_frame_header_stct
{
int32_t frame_length;
volatile int32_t frame_length;
int8_t version;
uint8_t flags;
int16_t type;
Expand Down

0 comments on commit f379397

Please sign in to comment.