Merge remote-tracking branch 'origin/master'

real-logic · Dec 9, 2020 · d9ed706 · d9ed706
2 parents f91a67b + 412eb91
commit d9ed706
Show file tree

Hide file tree

Showing 2 changed files with 15 additions and 2 deletions.
diff --git a/agrona/src/main/java/org/agrona/concurrent/ringbuffer/OneToOneRingBuffer.java b/agrona/src/main/java/org/agrona/concurrent/ringbuffer/OneToOneRingBuffer.java
@@ -16,6 +16,7 @@
 package org.agrona.concurrent.ringbuffer;
 
 import org.agrona.DirectBuffer;
+import org.agrona.UnsafeAccess;
 import org.agrona.concurrent.*;
 
 import static org.agrona.BitUtil.align;
@@ -90,6 +91,9 @@ public boolean write(final int msgTypeId, final DirectBuffer srcBuffer, final in
             return false;
         }
 
+        buffer.putIntOrdered(lengthOffset(recordIndex), -recordLength);
+        UnsafeAccess.UNSAFE.storeFence();
+
         buffer.putBytes(encodedMsgOffset(recordIndex), srcBuffer, offset, length);
         buffer.putInt(typeOffset(recordIndex), msgTypeId);
         buffer.putIntOrdered(lengthOffset(recordIndex), recordLength);
@@ -114,8 +118,9 @@ public int tryClaim(final int msgTypeId, final int length)
             return recordIndex;
         }
 
-        buffer.putInt(typeOffset(recordIndex), msgTypeId);
         buffer.putIntOrdered(lengthOffset(recordIndex), -recordLength);
+        UnsafeAccess.UNSAFE.storeFence();
+        buffer.putInt(typeOffset(recordIndex), msgTypeId);
 
         return encodedMsgOffset(recordIndex);
     }
@@ -447,6 +452,9 @@ private int claimCapacity(final AtomicBuffer buffer, final int recordLength)
         if (0 != padding)
         {
             buffer.putLong(0, 0L);
+            buffer.putIntOrdered(lengthOffset(recordIndex), -padding);
+            UnsafeAccess.UNSAFE.storeFence();
+
             buffer.putInt(typeOffset(recordIndex), PADDING_MSG_TYPE_ID);
             buffer.putIntOrdered(lengthOffset(recordIndex), padding);
             recordIndex = 0;

diff --git a/agrona/src/test/java/org/agrona/concurrent/ringbuffer/OneToOneRingBufferTest.java b/agrona/src/test/java/org/agrona/concurrent/ringbuffer/OneToOneRingBufferTest.java
@@ -75,6 +75,7 @@ public void shouldWriteToEmptyBuffer()
         final InOrder inOrder = inOrder(buffer);
         inOrder.verify(buffer).putLongOrdered(TAIL_COUNTER_INDEX, tail + alignedRecordLength);
         inOrder.verify(buffer).putLong((int)tail + alignedRecordLength, 0L);
+        inOrder.verify(buffer).putIntOrdered(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).putIntOrdered(lengthOffset((int)tail), recordLength);
@@ -140,11 +141,13 @@ public void shouldInsertPaddingRecordPlusMessageOnBufferWrap()
         inOrder.verify(buffer).putLongOrdered(TAIL_COUNTER_INDEX, tail + alignedRecordLength + HEADER_LENGTH);
 
         inOrder.verify(buffer).putLong(0, 0L);
+        inOrder.verify(buffer).putIntOrdered(lengthOffset((int)tail), -HEADER_LENGTH);
         inOrder.verify(buffer).putInt(typeOffset((int)tail), PADDING_MSG_TYPE_ID);
         inOrder.verify(buffer).putIntOrdered(lengthOffset((int)tail), HEADER_LENGTH);
 
         inOrder.verify(buffer).putLong(alignedRecordLength, 0L);
 
+        inOrder.verify(buffer).putIntOrdered(lengthOffset(0), -recordLength);
         inOrder.verify(buffer).putBytes(encodedMsgOffset(0), srcBuffer, srcIndex, length);
         inOrder.verify(buffer).putInt(typeOffset(0), MSG_TYPE_ID);
         inOrder.verify(buffer).putIntOrdered(lengthOffset(0), recordLength);
@@ -171,11 +174,13 @@ public void shouldInsertPaddingRecordPlusMessageOnBufferWrapWithHeadEqualToTail(
         inOrder.verify(buffer).putLongOrdered(TAIL_COUNTER_INDEX, tail + alignedRecordLength + HEADER_LENGTH);
 
         inOrder.verify(buffer).putLong(0, 0L);
+        inOrder.verify(buffer).putIntOrdered(lengthOffset((int)tail), -HEADER_LENGTH);
         inOrder.verify(buffer).putInt(typeOffset((int)tail), PADDING_MSG_TYPE_ID);
         inOrder.verify(buffer).putIntOrdered(lengthOffset((int)tail), HEADER_LENGTH);
 
         inOrder.verify(buffer).putLong(alignedRecordLength, 0L);
 
+        inOrder.verify(buffer).putIntOrdered(lengthOffset(0), -recordLength);
         inOrder.verify(buffer).putBytes(encodedMsgOffset(0), srcBuffer, srcIndex, length);
         inOrder.verify(buffer).putInt(typeOffset(0), MSG_TYPE_ID);
         inOrder.verify(buffer).putIntOrdered(lengthOffset(0), recordLength);
@@ -398,8 +403,8 @@ public void tryClaimReturnsOffsetAtWhichMessageBodyCanBeWritten()
         final InOrder inOrder = inOrder(buffer);
         inOrder.verify(buffer).putLongOrdered(TAIL_COUNTER_INDEX, alignedRecordLength);
         inOrder.verify(buffer).putLong(alignedRecordLength, 0L);
-        inOrder.verify(buffer).putInt(typeOffset(0), msgTypeId);
         inOrder.verify(buffer).putIntOrdered(lengthOffset(0), -recordLength);
+        inOrder.verify(buffer).putInt(typeOffset(0), msgTypeId);
     }
 
     @Test