Skip to content

Commit c715c19

Browse files
mjpt777mjpt777
committed
[Java] Changed publication log rotation algorithm so that if a publisher dies, or takes a long pause, mid rotation then other publishers can complete the operation and make progress. Issue #377.
1 parent 6c43b8d commit c715c19

File tree

8 files changed

+127
-97
lines changed

8 files changed

+127
-97
lines changed

‎aeron-client/src/main/java/io/aeron/ExclusivePublication.java

+19-22
Original file line numberDiff line numberDiff line change
@@ -637,44 +637,41 @@ private long newPosition(final int resultingOffset)
637637

638638
return termBeginPosition + resultingOffset;
639639
}
640-
else
640+
641+
if ((termBeginPosition + termBufferLength) >= maxPossiblePosition)
641642
{
642-
if ((termBeginPosition + termBufferLength) >= maxPossiblePosition)
643-
{
644-
return MAX_POSITION_EXCEEDED;
645-
}
643+
return MAX_POSITION_EXCEEDED;
644+
}
646645

647-
final int nextIndex = nextPartitionIndex(activePartitionIndex);
648-
final int nextTermId = termId + 1;
646+
final int nextIndex = nextPartitionIndex(activePartitionIndex);
647+
final int nextTermId = termId + 1;
649648

650-
activePartitionIndex = nextIndex;
651-
termOffset = 0;
652-
termId = nextTermId;
653-
termBeginPosition = computeTermBeginPosition(nextTermId, positionBitsToShift, initialTermId);
649+
activePartitionIndex = nextIndex;
650+
termOffset = 0;
651+
termId = nextTermId;
652+
termBeginPosition = computeTermBeginPosition(nextTermId, positionBitsToShift, initialTermId);
654653

655-
final int termCount = nextTermId - initialTermId;
654+
final int termCount = nextTermId - initialTermId;
656655

657-
initialiseTailWithTermId(logMetaDataBuffer, nextIndex, nextTermId);
658-
activeTermCountOrdered(logMetaDataBuffer, termCount);
656+
initialiseTailWithTermId(logMetaDataBuffer, nextIndex, nextTermId);
657+
activeTermCountOrdered(logMetaDataBuffer, termCount);
659658

660-
return ADMIN_ACTION;
661-
}
659+
return ADMIN_ACTION;
662660
}
663661

664662
private long backPressureStatus(final long currentPosition, final int messageLength)
665663
{
666-
long status = NOT_CONNECTED;
667-
668664
if ((currentPosition + messageLength) >= maxPossiblePosition)
669665
{
670-
status = MAX_POSITION_EXCEEDED;
666+
return MAX_POSITION_EXCEEDED;
671667
}
672-
else if (LogBufferDescriptor.isConnected(logMetaDataBuffer))
668+
669+
if (LogBufferDescriptor.isConnected(logMetaDataBuffer))
673670
{
674-
status = BACK_PRESSURED;
671+
return BACK_PRESSURED;
675672
}
676673

677-
return status;
674+
return NOT_CONNECTED;
678675
}
679676

680677
private void checkForMaxPayloadLength(final int length)

‎aeron-client/src/main/java/io/aeron/Publication.java

+11-24
Original file line numberDiff line numberDiff line change
@@ -594,47 +594,34 @@ public void removeDestination(final String endpointChannel)
594594
private long newPosition(
595595
final int termCount, final int termOffset, final int termId, final long position, final int resultingOffset)
596596
{
597-
long newPosition = ADMIN_ACTION;
598597
if (resultingOffset > 0)
599598
{
600-
newPosition = (position - termOffset) + resultingOffset;
599+
return (position - termOffset) + resultingOffset;
601600
}
602-
else if ((position + termOffset) > maxPossiblePosition)
601+
602+
if ((position + termOffset) > maxPossiblePosition)
603603
{
604-
newPosition = MAX_POSITION_EXCEEDED;
604+
return MAX_POSITION_EXCEEDED;
605605
}
606-
else if (resultingOffset == TermAppender.TRIPPED)
607-
{
608-
final int nextTermCount = termCount + 1;
609-
final int nextIndex = indexByTermCount(nextTermCount);
610606

611-
initialiseTailWithTermId(logMetaDataBuffer, nextIndex, termId + 1);
607+
rotateLog(logMetaDataBuffer, termCount, termId);
612608

613-
if (!casActiveTermCount(logMetaDataBuffer, termCount, nextTermCount))
614-
{
615-
throw new IllegalStateException(
616-
"CAS failed: expected=" + termCount +
617-
" update=" + nextTermCount + " actual=" + activeTermCount(logMetaDataBuffer));
618-
}
619-
}
620-
621-
return newPosition;
609+
return ADMIN_ACTION;
622610
}
623611

624612
private long backPressureStatus(final long currentPosition, final int messageLength)
625613
{
626-
long status = NOT_CONNECTED;
627-
628614
if ((currentPosition + messageLength) >= maxPossiblePosition)
629615
{
630-
status = MAX_POSITION_EXCEEDED;
616+
return MAX_POSITION_EXCEEDED;
631617
}
632-
else if (LogBufferDescriptor.isConnected(logMetaDataBuffer))
618+
619+
if (LogBufferDescriptor.isConnected(logMetaDataBuffer))
633620
{
634-
status = BACK_PRESSURED;
621+
return BACK_PRESSURED;
635622
}
636623

637-
return status;
624+
return NOT_CONNECTED;
638625
}
639626

640627
private void checkForMaxPayloadLength(final int length)

‎aeron-client/src/main/java/io/aeron/logbuffer/ExclusiveTermAppender.java

+8-8
Original file line numberDiff line numberDiff line change
@@ -48,7 +48,7 @@ public class ExclusiveTermAppender
4848
/**
4949
* The append operation tripped the end of the buffer and needs to rotate.
5050
*/
51-
public static final int TRIPPED = -1;
51+
public static final int FAILED = -1;
5252

5353
private final long tailAddressOffset;
5454
private final byte[] tailBuffer;
@@ -80,7 +80,7 @@ public ExclusiveTermAppender(
8080
* @param header for writing the default header.
8181
* @param length of the message to be written.
8282
* @param bufferClaim to be updated with the claimed region.
83-
* @return the resulting offset of the term after the append on success otherwise {@link #TRIPPED}.
83+
* @return the resulting offset of the term after the append on success otherwise {@link #FAILED}.
8484
*/
8585
public int claim(
8686
final int termId,
@@ -117,7 +117,7 @@ public int claim(
117117
* @param termOffset in the term at which to append.
118118
* @param header for writing the default header.
119119
* @param length of the padding to be written.
120-
* @return the resulting offset of the term after success otherwise {@link #TRIPPED}.
120+
* @return the resulting offset of the term after success otherwise {@link #FAILED}.
121121
*/
122122
public int appendPadding(
123123
final int termId,
@@ -157,7 +157,7 @@ public int appendPadding(
157157
* @param srcOffset at which the message begins.
158158
* @param length of the message in the source buffer.
159159
* @param reservedValueSupplier {@link ReservedValueSupplier} for the frame.
160-
* @return the resulting offset of the term after the append on success otherwise {@link #TRIPPED}.
160+
* @return the resulting offset of the term after the append on success otherwise {@link #FAILED}.
161161
*/
162162
public int appendUnfragmentedMessage(
163163
final int termId,
@@ -206,7 +206,7 @@ public int appendUnfragmentedMessage(
206206
* @param vectors to the buffers.
207207
* @param length of the message as a sum of the vectors.
208208
* @param reservedValueSupplier {@link ReservedValueSupplier} for the frame.
209-
* @return the resulting offset of the term after the append on success otherwise {@link #TRIPPED}.
209+
* @return the resulting offset of the term after the append on success otherwise {@link #FAILED}.
210210
*/
211211
public int appendUnfragmentedMessage(
212212
final int termId,
@@ -263,7 +263,7 @@ public int appendUnfragmentedMessage(
263263
* @param length of the message in the source buffer.
264264
* @param maxPayloadLength that the message will be fragmented into.
265265
* @param reservedValueSupplier {@link ReservedValueSupplier} for the frame.
266-
* @return the resulting offset of the term after the append on success otherwise {@link #TRIPPED}.
266+
* @return the resulting offset of the term after the append on success otherwise {@link #FAILED}.
267267
*/
268268
public int appendFragmentedMessage(
269269
final int termId,
@@ -343,7 +343,7 @@ public int appendFragmentedMessage(
343343
* @param length of the message in the source buffer.
344344
* @param maxPayloadLength that the message will be fragmented into.
345345
* @param reservedValueSupplier {@link ReservedValueSupplier} for the frame.
346-
* @return the resulting offset of the term after the append on success otherwise {@link #TRIPPED}.
346+
* @return the resulting offset of the term after the append on success otherwise {@link #FAILED}.
347347
*/
348348
public int appendFragmentedMessage(
349349
final int termId,
@@ -447,7 +447,7 @@ private int handleEndOfLogCondition(
447447
frameLengthOrdered(termBuffer, offset, paddingLength);
448448
}
449449

450-
return TRIPPED;
450+
return FAILED;
451451
}
452452

453453
private void putRawTailOrdered(final int termId, final int termOffset)

‎aeron-client/src/main/java/io/aeron/logbuffer/LogBufferDescriptor.java

+38-4
Original file line numberDiff line numberDiff line change
@@ -542,7 +542,9 @@ public static void applyDefaultHeader(
542542
}
543543

544544
/**
545-
* Rotate the log and update the default headers for the new term.
545+
* Rotate the log and update the tail counter for the new term.
546+
*
547+
* This method is safe for concurrent use.
546548
*
547549
* @param logMetaDataBuffer for the meta data.
548550
* @param currentTermCount from which to rotate.
@@ -551,10 +553,23 @@ public static void applyDefaultHeader(
551553
public static void rotateLog(
552554
final UnsafeBuffer logMetaDataBuffer, final int currentTermCount, final int currentTermId)
553555
{
556+
final int nextTermId = currentTermId + 1;
554557
final int nextTermCount = currentTermCount + 1;
555558
final int nextIndex = indexByTermCount(nextTermCount);
556-
initialiseTailWithTermId(logMetaDataBuffer, nextIndex, currentTermId + 1);
557-
activeTermCountOrdered(logMetaDataBuffer, nextTermCount);
559+
final int expectedTermId = nextTermId - PARTITION_COUNT;
560+
561+
long rawTail;
562+
do
563+
{
564+
rawTail = rawTail(logMetaDataBuffer, nextIndex);
565+
if (expectedTermId != termId(rawTail))
566+
{
567+
break;
568+
}
569+
}
570+
while (!casRawTail(logMetaDataBuffer, nextIndex, rawTail, packTail(nextTermId, 0)));
571+
572+
casActiveTermCount(logMetaDataBuffer, currentTermCount, nextTermCount);
558573
}
559574

560575
/**
@@ -567,7 +582,7 @@ public static void rotateLog(
567582
public static void initialiseTailWithTermId(
568583
final UnsafeBuffer logMetaData, final int partitionIndex, final int termId)
569584
{
570-
logMetaData.putLongOrdered(TERM_TAIL_COUNTERS_OFFSET + (partitionIndex * SIZE_OF_LONG), packTail(termId, 0));
585+
logMetaData.putLong(TERM_TAIL_COUNTERS_OFFSET + (partitionIndex * SIZE_OF_LONG), packTail(termId, 0));
571586
}
572587

573588
/**
@@ -679,4 +694,23 @@ public static long rawTailVolatile(final UnsafeBuffer logMetaDataBuffer)
679694
final int partitionIndex = indexByTermCount(activeTermCount(logMetaDataBuffer));
680695
return logMetaDataBuffer.getLongVolatile(TERM_TAIL_COUNTERS_OFFSET + (SIZE_OF_LONG * partitionIndex));
681696
}
697+
698+
/**
699+
* Compare and set the raw value of the tail for the given partition.
700+
*
701+
* @param logMetaDataBuffer containing the tail counters.
702+
* @param partitionIndex for the tail counter.
703+
* @param expectedRawTail expected current value.
704+
* @param updateRawTail to be applied.
705+
* @return true if the update was successful otherwise false.
706+
*/
707+
public static boolean casRawTail(
708+
final UnsafeBuffer logMetaDataBuffer,
709+
final int partitionIndex,
710+
final long expectedRawTail,
711+
final long updateRawTail)
712+
{
713+
final int index = TERM_TAIL_COUNTERS_OFFSET + (SIZE_OF_LONG * partitionIndex);
714+
return logMetaDataBuffer.compareAndSetLong(index, expectedRawTail, updateRawTail);
715+
}
682716
}

‎aeron-client/src/main/java/io/aeron/logbuffer/TermAppender.java

+12-29
Original file line numberDiff line numberDiff line change
@@ -50,11 +50,6 @@
5050
*/
5151
public class TermAppender
5252
{
53-
/**
54-
* The append operation tripped the end of the buffer and needs to rotate.
55-
*/
56-
public static final int TRIPPED = -1;
57-
5853
/**
5954
* The append operation failed because it was past the end of the buffer.
6055
*/
@@ -98,8 +93,7 @@ public long rawTailVolatile()
9893
* @param length of the message to be written.
9994
* @param bufferClaim to be updated with the claimed region.
10095
* @param activeTermId used for flow control.
101-
* @return the resulting offset of the term after the append on success otherwise {@link #TRIPPED}
102-
* or {@link #FAILED}.
96+
* @return the resulting offset of the term after the append on success otherwise {@link #FAILED}.
10397
*/
10498
public int claim(
10599
final HeaderWriter header,
@@ -141,8 +135,7 @@ public int claim(
141135
* @param length of the message in the source buffer.
142136
* @param reservedValueSupplier {@link ReservedValueSupplier} for the frame.
143137
* @param activeTermId used for flow control.
144-
* @return the resulting offset of the term after the append on success otherwise {@link #TRIPPED} or
145-
* {@link #FAILED}
138+
* @return the resulting offset of the term after the append on success otherwise {@link #FAILED}
146139
*/
147140
public int appendUnfragmentedMessage(
148141
final HeaderWriter header,
@@ -193,8 +186,7 @@ public int appendUnfragmentedMessage(
193186
* @param length of the message as a sum of the vectors.
194187
* @param reservedValueSupplier {@link ReservedValueSupplier} for the frame.
195188
* @param activeTermId used for flow control.
196-
* @return the resulting offset of the term after the append on success otherwise {@link #TRIPPED} or
197-
* {@link #FAILED}.
189+
* @return the resulting offset of the term after the append on success otherwise {@link #FAILED}.
198190
*/
199191
public int appendUnfragmentedMessage(
200192
final HeaderWriter header,
@@ -253,8 +245,7 @@ public int appendUnfragmentedMessage(
253245
* @param maxPayloadLength that the message will be fragmented into.
254246
* @param reservedValueSupplier {@link ReservedValueSupplier} for the frame.
255247
* @param activeTermId used for flow control.
256-
* @return the resulting offset of the term after the append on success otherwise {@link #TRIPPED}
257-
* or {@link #FAILED}.
248+
* @return the resulting offset of the term after the append on success otherwise {@link #FAILED}.
258249
*/
259250
public int appendFragmentedMessage(
260251
final HeaderWriter header,
@@ -336,8 +327,7 @@ public int appendFragmentedMessage(
336327
* @param maxPayloadLength that the message will be fragmented into.
337328
* @param reservedValueSupplier {@link ReservedValueSupplier} for the frame.
338329
* @param activeTermId used for flow control.
339-
* @return the resulting offset of the term after the append on success otherwise {@link #TRIPPED}
340-
* or {@link #FAILED}.
330+
* @return the resulting offset of the term after the append on success otherwise {@link #FAILED}.
341331
*/
342332
public int appendFragmentedMessage(
343333
final HeaderWriter header,
@@ -443,23 +433,16 @@ private int handleEndOfLogCondition(
443433
final int termLength,
444434
final int termId)
445435
{
446-
int resultingOffset = FAILED;
447-
448-
if (termOffset <= termLength)
436+
if (termOffset < termLength)
449437
{
450-
resultingOffset = TRIPPED;
451-
452-
if (termOffset < termLength)
453-
{
454-
final int offset = (int)termOffset;
455-
final int paddingLength = termLength - offset;
456-
header.write(termBuffer, offset, paddingLength, termId);
457-
frameType(termBuffer, offset, PADDING_FRAME_TYPE);
458-
frameLengthOrdered(termBuffer, offset, paddingLength);
459-
}
438+
final int offset = (int)termOffset;
439+
final int paddingLength = termLength - offset;
440+
header.write(termBuffer, offset, paddingLength, termId);
441+
frameType(termBuffer, offset, PADDING_FRAME_TYPE);
442+
frameLengthOrdered(termBuffer, offset, paddingLength);
460443
}
461444

462-
return resultingOffset;
445+
return FAILED;
463446
}
464447

465448
private long getAndAddRawTail(final int alignedLength)

0 commit comments

Comments
 (0)