Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Logbuffer metadata extra fields #1700

Merged
merged 25 commits into from
Jan 8, 2025
Merged

Logbuffer metadata extra fields #1700

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
25 commits
Select commit Hold shift + click to select a range
a378c8f
Initial checking extra properties logbuffer metadata
pveentjer Dec 20, 2024
708407e
WIP
pveentjer Dec 20, 2024
a0b6493
Minor cleanup
pveentjer Dec 20, 2024
ce1040f
Further improvements
pveentjer Dec 20, 2024
0c7a358
Minor fixes
pveentjer Dec 20, 2024
2bd9830
Booleans are now encoded as byte instead of int
pveentjer Dec 20, 2024
57ea62d
Removed the extra cacheline of padding after the frame header
pveentjer Dec 20, 2024
6c4e28e
Fixed checkstyle issues
pveentjer Dec 20, 2024
e6039b3
Moved as many fields as possible before the frame header
pveentjer Dec 30, 2024
b5cf57c
Checkstyle fixes
pveentjer Dec 30, 2024
ee98864
Lot of minor improvements
pveentjer Dec 31, 2024
b092c5f
More work on descriptor
pveentjer Dec 31, 2024
8ee1291
More work log buffer descriptor
pveentjer Dec 31, 2024
a94276a
hacks
pveentjer Jan 2, 2025
84c9146
Disabled the fields after the end of the original record; now java_sy…
pveentjer Jan 2, 2025
37e060e
Added the default_header to the struct
pveentjer Jan 2, 2025
7a7fe9a
All working apart from the assert from aeron_logbuffer_metadata_t size
pveentjer Jan 2, 2025
e4625f3
Fixed padding problem
pveentjer Jan 2, 2025
6066761
Update the order of the fields to reflect master
pveentjer Jan 2, 2025
e177c9c
Furher alignment
pveentjer Jan 2, 2025
015496f
Merge fixes
pveentjer Jan 2, 2025
13264fe
Fixes
pveentjer Jan 2, 2025
c7e8a57
Updated javadoc field layout
pveentjer Jan 3, 2025
1fbfb2c
Processed review comments
pveentjer Jan 7, 2025
7742a73
Removed padding from LogBufferDescriptor documentation
pveentjer Jan 7, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
254 changes: 253 additions & 1 deletion aeron-client/src/main/java/io/aeron/logbuffer/LogBufferDescriptor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -135,11 +135,33 @@ public class LogBufferDescriptor
*/
public static final int LOG_DEFAULT_FRAME_HEADER_OFFSET;



/**
* Maximum length of a frame header.
*/
public static final int LOG_DEFAULT_FRAME_HEADER_MAX_LENGTH = CACHE_LINE_LENGTH * 2;


// todo: Add documentation
public static final int LOG_ACTIVE_TERM_ID_OFFSET;
public static final int LOG_TERM_OFFSET_OFFSET;
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

A natural place to store this information would the default frame header, i.e.:

        defaultDataHeader.sessionId(sessionId).streamId(streamId).termId(initialTermId).termOffset(termOffset);
        storeDefaultFrameHeader(logMetaData, defaultDataHeader);

Copy link
Contributor Author

@pveentjer pveentjer Dec 30, 2024 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Ok. I'll set the term offset on the default frame header and will drop it from the LogBufferDescriptor

public static final int LOG_IS_SPARSE_OFFSET;
public static final int LOG_IS_TETHER_OFFSET;
vyazelenko marked this conversation as resolved.
Show resolved Hide resolved
public static final int LOG_IS_REJOIN_OFFSET;
pveentjer marked this conversation as resolved.
Show resolved Hide resolved
public static final int LOG_IS_RELIABLE_OFFSET;
pveentjer marked this conversation as resolved.
Show resolved Hide resolved
public static final int LOG_SOCKET_RCVBUF_LENGTH_OFFSET;
public static final int LOG_SOCKET_SNDBUF_LENGTH_OFFSET;
public static final int LOG_RECEIVER_WINDOW_LENGTH_OFFSET;
public static final int LOG_PUBLICATION_WINDOW_LENGTH_OFFSET;
public static final int LOG_GROUP_OFFSET;
public static final int LOG_UNTETHERED_WINDOW_LIMIT_TIMEOUT_NS_OFFSET;
public static final int LOG_UNTETHERED_RESTING_TIMEOUT_NS_OFFSET;
public static final int LOG_MAX_RESEND_OFFSET;
public static final int LOG_LINGER_TIMEOUT_NS_OFFSET;
public static final int LOG_SIGNAL_EOS_OFFSET;
public static final int LOG_SPIES_SIMULATE_CONNECTION_OFFSET;

/**
* Total length of the log metadata buffer in bytes.
* <pre>
Expand Down Expand Up @@ -217,7 +239,67 @@ public class LogBufferDescriptor
offset += CACHE_LINE_LENGTH;
LOG_DEFAULT_FRAME_HEADER_OFFSET = offset;

LOG_META_DATA_LENGTH = align(offset + LOG_DEFAULT_FRAME_HEADER_MAX_LENGTH, PAGE_MIN_SIZE);
// the new fields will be added with 1 cacheline of padding after the frame header.
offset+= LOG_DEFAULT_FRAME_HEADER_MAX_LENGTH+CACHE_LINE_LENGTH;

LOG_ACTIVE_TERM_ID_OFFSET = offset;
offset += SIZE_OF_INT;

LOG_TERM_OFFSET_OFFSET = offset;
offset += SIZE_OF_INT;

LOG_IS_SPARSE_OFFSET = offset;
offset += SIZE_OF_INT;

LOG_IS_TETHER_OFFSET = offset;
offset += SIZE_OF_INT;

LOG_IS_REJOIN_OFFSET = offset;
offset += SIZE_OF_INT;

LOG_IS_RELIABLE_OFFSET = offset;
offset += SIZE_OF_INT;

LOG_SOCKET_RCVBUF_LENGTH_OFFSET = offset;
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Let's try to use available space after the page_size field (LOG_PAGE_SIZE_OFFSET + SIZE_OF_INT) where we still have 36 bytes before the default header starts.
That should be enough to fit 6 ints or 5 ints + 6 booleans.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Done

offset += SIZE_OF_INT;

LOG_SOCKET_SNDBUF_LENGTH_OFFSET = offset;
offset += SIZE_OF_INT;

LOG_RECEIVER_WINDOW_LENGTH_OFFSET = offset;
offset += SIZE_OF_INT;

LOG_PUBLICATION_WINDOW_LENGTH_OFFSET = offset;
offset += SIZE_OF_INT;

LOG_GROUP_OFFSET = offset;
offset += SIZE_OF_INT;

LOG_MAX_RESEND_OFFSET = offset;
offset += SIZE_OF_INT;

LOG_UNTETHERED_WINDOW_LIMIT_TIMEOUT_NS_OFFSET = offset;
if((LOG_UNTETHERED_WINDOW_LIMIT_TIMEOUT_NS_OFFSET % SIZE_OF_LONG)!=0){
throw new Error();
}
offset += SIZE_OF_LONG;

LOG_UNTETHERED_RESTING_TIMEOUT_NS_OFFSET = offset;
offset += SIZE_OF_LONG;

LOG_LINGER_TIMEOUT_NS_OFFSET = offset;
offset += SIZE_OF_LONG;

LOG_SIGNAL_EOS_OFFSET = offset;
offset += SIZE_OF_INT;

LOG_SPIES_SIMULATE_CONNECTION_OFFSET = offset;
offset += SIZE_OF_INT;


LOG_META_DATA_LENGTH = align(offset, PAGE_MIN_SIZE);
System.out.println("Offset:"+offset);
System.out.println("LOG_META_DATA_LENGTH:"+LOG_META_DATA_LENGTH);
}

/**
Expand Down Expand Up @@ -878,4 +960,174 @@ public static int computeAssembledFrameLength(final int length, final int maxPay

return HEADER_LENGTH + (numMaxPayloads * maxPayloadSize) + remainingPayload;
}

// todo: javadoc
public static int activeTermId(final UnsafeBuffer metadataBuffer) {
return metadataBuffer.getInt(LOG_ACTIVE_TERM_ID_OFFSET);
}

// todo: javadoc
public static void activeTermId(final UnsafeBuffer metadataBuffer, final int value) {
metadataBuffer.putInt(LOG_ACTIVE_TERM_ID_OFFSET, value);
}

// todo: javadoc
public static int termOffset(final UnsafeBuffer metadataBuffer) {
return metadataBuffer.getInt(LOG_TERM_OFFSET_OFFSET);
}

// todo: javadoc
public static void termOffset(final UnsafeBuffer metadataBuffer, final int value) {
metadataBuffer.putInt(LOG_TERM_OFFSET_OFFSET, value);
}

// todo: javadoc
public static boolean isSparse(final UnsafeBuffer metadataBuffer) {
return metadataBuffer.getInt(LOG_IS_SPARSE_OFFSET) == 1;
}

// todo: javadoc
public static void isSparse(final UnsafeBuffer metadataBuffer, final boolean value) {
metadataBuffer.putInt(LOG_IS_SPARSE_OFFSET, value ? 1 : 0);
}

// todo: javadoc
public static boolean isTether(final UnsafeBuffer metadataBuffer) {
return metadataBuffer.getInt(LOG_IS_TETHER_OFFSET) == 1;
}

// todo: javadoc
public static void isTether(final UnsafeBuffer metadataBuffer, final boolean value) {
metadataBuffer.putInt(LOG_IS_TETHER_OFFSET, value ? 1 : 0);
}

// todo: javadoc
public static boolean isRejoin(final UnsafeBuffer metadataBuffer) {
return metadataBuffer.getInt(LOG_IS_REJOIN_OFFSET) == 1;
}

// todo: javadoc
public static void isRejoin(final UnsafeBuffer metadataBuffer, final boolean value) {
metadataBuffer.putInt(LOG_IS_REJOIN_OFFSET, value ? 1 : 0);
}

// todo: javadoc
public static boolean isReliable(final UnsafeBuffer metadataBuffer) {
return metadataBuffer.getInt(LOG_IS_RELIABLE_OFFSET) == 1;
}

// todo: javadoc
public static void isReliable(final UnsafeBuffer metadataBuffer, final boolean value) {
metadataBuffer.putInt(LOG_IS_RELIABLE_OFFSET, value ? 1 : 0);
}

// todo: javadoc
public static int socketRcvbufLength(final UnsafeBuffer metadataBuffer) {
return metadataBuffer.getInt(LOG_SOCKET_RCVBUF_LENGTH_OFFSET);
}

// todo: javadoc
public static void socketRcvbufLength(final UnsafeBuffer metadataBuffer, final int value) {
metadataBuffer.putInt(LOG_SOCKET_RCVBUF_LENGTH_OFFSET, value);
}

// todo: javadoc
public static int socketSndbufLength(final UnsafeBuffer metadataBuffer) {
return metadataBuffer.getInt(LOG_SOCKET_SNDBUF_LENGTH_OFFSET);
}

// todo: javadoc
public static void socketSndbufLength(final UnsafeBuffer metadataBuffer, final int value) {
metadataBuffer.putInt(LOG_SOCKET_SNDBUF_LENGTH_OFFSET, value);
}

// todo: javadoc
public static int receiverWindowLength(final UnsafeBuffer metadataBuffer) {
return metadataBuffer.getInt(LOG_RECEIVER_WINDOW_LENGTH_OFFSET);
}

// todo: javadoc
public static void receiverWindowLength(final UnsafeBuffer metadataBuffer, final int value) {
metadataBuffer.putInt(LOG_RECEIVER_WINDOW_LENGTH_OFFSET, value);
}

// todo: javadoc
public static int publicationWindowLength(final UnsafeBuffer metadataBuffer) {
return metadataBuffer.getInt(LOG_PUBLICATION_WINDOW_LENGTH_OFFSET);
}

// todo: javadoc
public static void publicationWindowLength(final UnsafeBuffer metadataBuffer, final int value) {
metadataBuffer.putInt(LOG_PUBLICATION_WINDOW_LENGTH_OFFSET, value);
}

// todo: javadoc
public static long untetheredWindowLimitTimeoutNs(final UnsafeBuffer metadataBuffer) {
return metadataBuffer.getLong(LOG_UNTETHERED_WINDOW_LIMIT_TIMEOUT_NS_OFFSET);
}

// todo: javadoc
public static void untetheredWindowLimitTimeoutNs(final UnsafeBuffer metadataBuffer, final long value) {
metadataBuffer.putLong(LOG_UNTETHERED_WINDOW_LIMIT_TIMEOUT_NS_OFFSET, value);
}

// todo: javadoc
public static long untetheredRestingTimeoutNs(final UnsafeBuffer metadataBuffer) {
return metadataBuffer.getLong(LOG_UNTETHERED_RESTING_TIMEOUT_NS_OFFSET);
}

// todo: javadoc
public static void untetheredRestingTimeoutNs(final UnsafeBuffer metadataBuffer, final long value) {
metadataBuffer.putLong(LOG_UNTETHERED_RESTING_TIMEOUT_NS_OFFSET, value);
}

// todo: javadoc
public static boolean group(final UnsafeBuffer metadataBuffer) {
return metadataBuffer.getInt(LOG_GROUP_OFFSET) == 1;
}

// todo: javadoc
public static void group(final UnsafeBuffer metadataBuffer, final boolean value) {
metadataBuffer.putInt(LOG_GROUP_OFFSET, value ? 1 : 0);
}

// todo: javadoc
public static int maxResend(final UnsafeBuffer metadataBuffer) {
return metadataBuffer.getInt(LOG_MAX_RESEND_OFFSET);
}

// todo: javadoc
public static void maxResend(final UnsafeBuffer metadataBuffer, final int value) {
metadataBuffer.putInt(LOG_MAX_RESEND_OFFSET, value);
}

// todo: javadoc
public static long lingerTimeoutNs(final UnsafeBuffer metadataBuffer) {
return metadataBuffer.getLong(LOG_LINGER_TIMEOUT_NS_OFFSET);
}

// todo: javadoc
public static void lingerTimeoutNs(final UnsafeBuffer metadataBuffer, final long value) {
metadataBuffer.putLong(LOG_LINGER_TIMEOUT_NS_OFFSET, value);
}

// todo: javadoc
public static boolean signalEos(final UnsafeBuffer metadataBuffer) {
return metadataBuffer.getInt(LOG_SIGNAL_EOS_OFFSET) == 1;
}

// todo: javadoc
public static void signalEos(final UnsafeBuffer metadataBuffer, final boolean value) {
metadataBuffer.putInt(LOG_SIGNAL_EOS_OFFSET,value ? 1 : 0);
}

// todo: javadoc
public static boolean spiesSimulateConnection(final UnsafeBuffer metadataBuffer) {
return metadataBuffer.getInt(LOG_SPIES_SIMULATE_CONNECTION_OFFSET) == 1;
}

// todo: javadoc
public static void spiesSimulateConnection(final UnsafeBuffer metadataBuffer, final boolean value) {
metadataBuffer.putInt(LOG_SPIES_SIMULATE_CONNECTION_OFFSET, value ? 1 : 0);
}
}
Loading
Loading