Skip to content

Weaken memory ordering OneToOnRingBuffer #359

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

Open
pveentjer wants to merge 1 commit into
base: master
Choose a base branch
from
Open

Weaken memory ordering OneToOnRingBuffer #359

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
11 changes: 2 additions & 9 deletions agrona/src/main/java/org/agrona/concurrent/ringbuffer/OneToOneRingBuffer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,8 +20,6 @@
import org.agrona.concurrent.ControlledMessageHandler;
import org.agrona.concurrent.MessageHandler;

import java.lang.invoke.VarHandle;

import static java.lang.Math.max;
import static org.agrona.BitUtil.align;
import static org.agrona.concurrent.ControlledMessageHandler.Action.*;
Expand Down Expand Up @@ -100,8 +98,7 @@ public boolean write(final int msgTypeId, final DirectBuffer srcBuffer, final in
return false;
}

buffer.putIntRelease(lengthOffset(recordIndex), -recordLength);
Copy link
Copy Markdown
Contributor Author

@pveentjer pveentjer Apr 14, 2026 •
edited
Loading

Choose a reason for hiding this comment

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

For the actual claim (so writing the negative record length), only opaque is needed. Surrounding loads/stores do not need to be ordered. The content of the record can't be read (apart from the length) until the commit is done. The commit is performed using a release store and will make sure that all writes to the record are visible when the positive length is read by the reader.

VarHandle.releaseFence();
buffer.putIntOpaque(lengthOffset(recordIndex), -recordLength);

buffer.putBytes(encodedMsgOffset(recordIndex), srcBuffer, offset, length);
buffer.putInt(typeOffset(recordIndex), msgTypeId);
Expand All @@ -127,8 +124,7 @@ public int tryClaim(final int msgTypeId, final int length)
return recordIndex;
}

buffer.putIntRelease(lengthOffset(recordIndex), -recordLength);
Copy link
Copy Markdown
Contributor Author

@pveentjer pveentjer Apr 14, 2026

Choose a reason for hiding this comment

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

Also an opaque write for the record length. We do not care for any ordering guarantees. Only when the record is committed, a release store is needed which will order the stores to the record before it.

VarHandle.releaseFence();
Copy link
Copy Markdown
Contributor Author

@pveentjer pveentjer Apr 14, 2026 •
edited
Loading

Choose a reason for hiding this comment

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

The VarHandle.releaseFence can be dropped. THe only ordering that is needed is the release store of the (positive) record length when committing. This will order all the stores to the record before it.

buffer.putIntOpaque(lengthOffset(recordIndex), -recordLength);
buffer.putInt(typeOffset(recordIndex), msgTypeId);

return encodedMsgOffset(recordIndex);
Expand Down Expand Up @@ -472,9 +468,6 @@ else if (requiredCapacity > toBufferEndLength)
if (0 != padding)
{
buffer.putLong(0, 0L);
buffer.putIntRelease(lengthOffset(recordIndex), -padding);
Copy link
Copy Markdown
Contributor Author

@pveentjer pveentjer Apr 14, 2026

Choose a reason for hiding this comment

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

We don't need to announce the negative length. The reading side will not read the record while the length is not positive. And there won't be any abort in this case.

VarHandle.releaseFence();
Copy link
Copy Markdown
Contributor Author

@pveentjer pveentjer Apr 14, 2026 •
edited
Loading

Choose a reason for hiding this comment

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

The releaseFence is not needed. The release store of the commit (so writing the positive length) will order the write of the message type, before it.


buffer.putInt(typeOffset(recordIndex), PADDING_MSG_TYPE_ID);
buffer.putIntRelease(lengthOffset(recordIndex), padding);
}
Expand Down
8 changes: 1 addition & 7 deletions agrona/src/test/java/org/agrona/concurrent/ringbuffer/OneToOneRingBufferTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -109,7 +109,6 @@ void shouldWriteToEmptyBuffer()
final InOrder inOrder = inOrder(buffer);
inOrder.verify(buffer).putLongRelease(TAIL_COUNTER_INDEX, tail + alignedRecordLength);
inOrder.verify(buffer).putLong((int)tail + alignedRecordLength, 0L);
inOrder.verify(buffer).putIntRelease(lengthOffset((int)tail), -recordLength);
inOrder.verify(buffer).putBytes(encodedMsgOffset((int)tail), srcBuffer, srcIndex, length);
inOrder.verify(buffer).putInt(typeOffset((int)tail), MSG_TYPE_ID);
inOrder.verify(buffer).putIntRelease(lengthOffset((int)tail), recordLength);
Expand Down Expand Up @@ -175,13 +174,11 @@ void shouldInsertPaddingRecordPlusMessageOnBufferWrap()
inOrder.verify(buffer).putLongRelease(TAIL_COUNTER_INDEX, tail + alignedRecordLength + HEADER_LENGTH);

inOrder.verify(buffer).putLong(0, 0L);
inOrder.verify(buffer).putIntRelease(lengthOffset((int)tail), -HEADER_LENGTH);
inOrder.verify(buffer).putInt(typeOffset((int)tail), PADDING_MSG_TYPE_ID);
inOrder.verify(buffer).putIntRelease(lengthOffset((int)tail), HEADER_LENGTH);

inOrder.verify(buffer).putLong(alignedRecordLength, 0L);

inOrder.verify(buffer).putIntRelease(lengthOffset(0), -recordLength);
inOrder.verify(buffer).putBytes(encodedMsgOffset(0), srcBuffer, srcIndex, length);
inOrder.verify(buffer).putInt(typeOffset(0), MSG_TYPE_ID);
inOrder.verify(buffer).putIntRelease(lengthOffset(0), recordLength);
Expand All @@ -208,13 +205,11 @@ void shouldInsertPaddingRecordPlusMessageOnBufferWrapWithHeadEqualToTail()
inOrder.verify(buffer).putLongRelease(TAIL_COUNTER_INDEX, tail + alignedRecordLength + HEADER_LENGTH);

inOrder.verify(buffer).putLong(0, 0L);
inOrder.verify(buffer).putIntRelease(lengthOffset((int)tail), -HEADER_LENGTH);
inOrder.verify(buffer).putInt(typeOffset((int)tail), PADDING_MSG_TYPE_ID);
inOrder.verify(buffer).putIntRelease(lengthOffset((int)tail), HEADER_LENGTH);

inOrder.verify(buffer).putLong(alignedRecordLength, 0L);

inOrder.verify(buffer).putIntRelease(lengthOffset(0), -recordLength);
inOrder.verify(buffer).putBytes(encodedMsgOffset(0), srcBuffer, srcIndex, length);
inOrder.verify(buffer).putInt(typeOffset(0), MSG_TYPE_ID);
inOrder.verify(buffer).putIntRelease(lengthOffset(0), recordLength);
Expand Down Expand Up @@ -437,7 +432,7 @@ void tryClaimReturnsOffsetAtWhichMessageBodyCanBeWritten()
final InOrder inOrder = inOrder(buffer);
inOrder.verify(buffer).putLongRelease(TAIL_COUNTER_INDEX, alignedRecordLength);
inOrder.verify(buffer).putLong(alignedRecordLength, 0L);
inOrder.verify(buffer).putIntRelease(lengthOffset(0), -recordLength);
inOrder.verify(buffer).putIntOpaque(lengthOffset(0), -recordLength);
inOrder.verify(buffer).putInt(typeOffset(0), msgTypeId);
}

Expand Down Expand Up @@ -481,7 +476,6 @@ void tryClaimReturnsInsufficientCapacityIfThereIsNotEnoughSpaceInTheBufferAfterW
inOrder.verify(buffer).putLongRelease(TAIL_COUNTER_INDEX, CAPACITY * 2L);
final int paddingIndex = CAPACITY - 10;
inOrder.verify(buffer).putLong(0, 0L);
inOrder.verify(buffer).putIntRelease(lengthOffset(paddingIndex), -10);
inOrder.verify(buffer).putInt(typeOffset(paddingIndex), PADDING_MSG_TYPE_ID);
inOrder.verify(buffer).putIntRelease(lengthOffset(paddingIndex), 10);
inOrder.verifyNoMoreInteractions();
Expand Down
Loading