diff --git a/ratis-grpc/src/main/java/org/apache/ratis/grpc/server/GrpcServerProtocolService.java b/ratis-grpc/src/main/java/org/apache/ratis/grpc/server/GrpcServerProtocolService.java index 766e14321a..dde01c39a3 100644 --- a/ratis-grpc/src/main/java/org/apache/ratis/grpc/server/GrpcServerProtocolService.java +++ b/ratis-grpc/src/main/java/org/apache/ratis/grpc/server/GrpcServerProtocolService.java @@ -30,6 +30,7 @@ import org.apache.ratis.proto.RaftProtos.*; import org.apache.ratis.proto.grpc.RaftServerProtocolServiceGrpc.RaftServerProtocolServiceImplBase; import org.apache.ratis.util.ProtoUtils; +import org.apache.ratis.util.ReferenceCountedObject; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -226,7 +227,7 @@ public StreamObserver appendEntries( RaftServerProtocol.Op.APPEND_ENTRIES, responseObserver) { @Override CompletableFuture process(AppendEntriesRequestProto request) throws IOException { - return server.appendEntriesAsync(request); + return server.appendEntriesAsync(ReferenceCountedObject.wrap(request)); } @Override diff --git a/ratis-server-api/src/main/java/org/apache/ratis/server/protocol/RaftServerAsynchronousProtocol.java b/ratis-server-api/src/main/java/org/apache/ratis/server/protocol/RaftServerAsynchronousProtocol.java index 8a904069ba..1244e72545 100644 --- a/ratis-server-api/src/main/java/org/apache/ratis/server/protocol/RaftServerAsynchronousProtocol.java +++ b/ratis-server-api/src/main/java/org/apache/ratis/server/protocol/RaftServerAsynchronousProtocol.java @@ -22,14 +22,41 @@ import org.apache.ratis.proto.RaftProtos.ReadIndexReplyProto; import org.apache.ratis.proto.RaftProtos.AppendEntriesReplyProto; import org.apache.ratis.proto.RaftProtos.AppendEntriesRequestProto; +import org.apache.ratis.util.ReferenceCountedObject; import java.io.IOException; import java.util.concurrent.CompletableFuture; public interface RaftServerAsynchronousProtocol { - CompletableFuture appendEntriesAsync(AppendEntriesRequestProto request) - throws IOException; + /** + * It is recommended to override {@link #appendEntriesAsync(ReferenceCountedObject)} instead. + * Then, it does not have to override this method. + */ + default CompletableFuture appendEntriesAsync(AppendEntriesRequestProto request) + throws IOException { + throw new UnsupportedOperationException(); + } + + /** + * A referenced counted request is submitted from a client for processing. + * Implementations of this method should retain the request, process it and then release it. + * The request may be retained even after the future returned by this method has completed. + * + * @return a future of the reply + * @see ReferenceCountedObject + */ + default CompletableFuture appendEntriesAsync( + ReferenceCountedObject requestRef) throws IOException { + // Default implementation for backward compatibility. + try { + return appendEntriesAsync(requestRef.retain()) + .whenComplete((r, e) -> requestRef.release()); + } catch (Exception e) { + requestRef.release(); + throw e; + } + } CompletableFuture readIndexAsync(ReadIndexRequestProto request) throws IOException; diff --git a/ratis-server-api/src/main/java/org/apache/ratis/server/raftlog/RaftLogSequentialOps.java b/ratis-server-api/src/main/java/org/apache/ratis/server/raftlog/RaftLogSequentialOps.java index 7b9f42b6bd..e4523cd935 100644 --- a/ratis-server-api/src/main/java/org/apache/ratis/server/raftlog/RaftLogSequentialOps.java +++ b/ratis-server-api/src/main/java/org/apache/ratis/server/raftlog/RaftLogSequentialOps.java @@ -22,8 +22,10 @@ import org.apache.ratis.server.RaftConfiguration; import org.apache.ratis.statemachine.TransactionContext; import org.apache.ratis.util.Preconditions; +import org.apache.ratis.util.ReferenceCountedObject; import org.apache.ratis.util.StringUtils; import org.apache.ratis.util.function.CheckedSupplier; +import org.apache.ratis.util.function.UncheckedAutoCloseableSupplier; import java.util.Arrays; import java.util.List; @@ -132,21 +134,36 @@ default CompletableFuture appendEntry(LogEntryProto entry, TransactionCont /** * The same as append(Arrays.asList(entries)). * - * @deprecated use {@link #append(List)} + * @deprecated use {@link #append(ReferenceCountedObject)}. */ @Deprecated default List> append(LogEntryProto... entries) { return append(Arrays.asList(entries)); } + /** + * @deprecated use {@link #append(ReferenceCountedObject)}. + */ + @Deprecated + default List> append(List entries) { + throw new UnsupportedOperationException(); + } + /** * Append asynchronously all the given log entries. * Used by the followers. * * If an existing entry conflicts with a new one (same index but different terms), * delete the existing entry and all entries that follow it (ยง5.3). + * + * A reference counter is also submitted. + * For each entry, implementations of this method should retain the counter, process it and then release. */ - List> append(List entries); + default List> append(ReferenceCountedObject> entriesRef) { + try(UncheckedAutoCloseableSupplier> entries = entriesRef.retainAndReleaseOnClose()) { + return append(entries.get()); + } + } /** * Truncate asynchronously the log entries till the given index (inclusively). diff --git a/ratis-server/src/main/java/org/apache/ratis/server/impl/RaftServerImpl.java b/ratis-server/src/main/java/org/apache/ratis/server/impl/RaftServerImpl.java index de6491364c..8ad835474c 100644 --- a/ratis-server/src/main/java/org/apache/ratis/server/impl/RaftServerImpl.java +++ b/ratis-server/src/main/java/org/apache/ratis/server/impl/RaftServerImpl.java @@ -1460,27 +1460,30 @@ private void validateEntries(long expectedTerm, TermIndex previous, public AppendEntriesReplyProto appendEntries(AppendEntriesRequestProto r) throws IOException { try { - return appendEntriesAsync(r).join(); + return appendEntriesAsync(ReferenceCountedObject.wrap(r)).join(); } catch (CompletionException e) { throw IOUtils.asIOException(JavaUtils.unwrapCompletionException(e)); } } @Override - public CompletableFuture appendEntriesAsync(AppendEntriesRequestProto r) - throws IOException { + public CompletableFuture appendEntriesAsync( + ReferenceCountedObject requestRef) throws IOException { + final AppendEntriesRequestProto r = requestRef.retain(); final RaftRpcRequestProto request = r.getServerRequest(); final List entries = r.getEntriesList(); final TermIndex previous = r.hasPreviousLog()? TermIndex.valueOf(r.getPreviousLog()) : null; final RaftPeerId requestorId = RaftPeerId.valueOf(request.getRequestorId()); - preAppendEntriesAsync(requestorId, ProtoUtils.toRaftGroupId(request.getRaftGroupId()), r.getLeaderTerm(), - previous, r.getLeaderCommit(), r.getInitializing(), entries); try { + preAppendEntriesAsync(requestorId, ProtoUtils.toRaftGroupId(request.getRaftGroupId()), r.getLeaderTerm(), + previous, r.getLeaderCommit(), r.getInitializing(), entries); return appendEntriesAsync(requestorId, r.getLeaderTerm(), previous, r.getLeaderCommit(), - request.getCallId(), r.getInitializing(), r.getCommitInfosList(), entries); + request.getCallId(), r.getInitializing(), r.getCommitInfosList(), entries, requestRef) + .whenComplete((reply, e) -> requestRef.release()); } catch(Exception t) { LOG.error("{}: Failed appendEntriesAsync {}", getMemberId(), r, t); + requestRef.release(); throw t; } } @@ -1554,7 +1557,8 @@ ExecutorService getServerExecutor() { @SuppressWarnings("checkstyle:parameternumber") private CompletableFuture appendEntriesAsync( RaftPeerId leaderId, long leaderTerm, TermIndex previous, long leaderCommit, long callId, boolean initializing, - List commitInfos, List entries) throws IOException { + List commitInfos, List entries, + ReferenceCountedObject requestRef) throws IOException { final boolean isHeartbeat = entries.isEmpty(); logAppendEntries(isHeartbeat, () -> getMemberId() + ": receive appendEntries(" + leaderId + ", " + leaderTerm + ", " @@ -1612,8 +1616,9 @@ leaderId, getMemberId(), currentTerm, followerCommit, state.getNextIndex(), state.updateConfiguration(entries); } + final List> futures = entries.isEmpty() ? Collections.emptyList() - : state.getLog().append(entries); + : state.getLog().append(requestRef.delegate(entries)); commitInfos.forEach(commitInfoCache::update); CodeInjectionForTesting.execute(LOG_SYNC, getId(), null); diff --git a/ratis-server/src/main/java/org/apache/ratis/server/impl/RaftServerProxy.java b/ratis-server/src/main/java/org/apache/ratis/server/impl/RaftServerProxy.java index cb7918e51d..589c7eeb14 100644 --- a/ratis-server/src/main/java/org/apache/ratis/server/impl/RaftServerProxy.java +++ b/ratis-server/src/main/java/org/apache/ratis/server/impl/RaftServerProxy.java @@ -645,10 +645,16 @@ public StartLeaderElectionReplyProto startLeaderElection(StartLeaderElectionRequ } @Override - public CompletableFuture appendEntriesAsync(AppendEntriesRequestProto request) { - final RaftGroupId groupId = ProtoUtils.toRaftGroupId(request.getServerRequest().getRaftGroupId()); - return getImplFuture(groupId) - .thenCompose(impl -> impl.executeSubmitServerRequestAsync(() -> impl.appendEntriesAsync(request))); + public CompletableFuture appendEntriesAsync( + ReferenceCountedObject requestRef) { + AppendEntriesRequestProto request = requestRef.retain(); + try { + final RaftGroupId groupId = ProtoUtils.toRaftGroupId(request.getServerRequest().getRaftGroupId()); + return getImplFuture(groupId) + .thenCompose(impl -> impl.executeSubmitServerRequestAsync(() -> impl.appendEntriesAsync(requestRef))); + } finally { + requestRef.release(); + } } @Override diff --git a/ratis-server/src/main/java/org/apache/ratis/server/raftlog/RaftLogBase.java b/ratis-server/src/main/java/org/apache/ratis/server/raftlog/RaftLogBase.java index 708b499cd3..0b6fa15b6f 100644 --- a/ratis-server/src/main/java/org/apache/ratis/server/raftlog/RaftLogBase.java +++ b/ratis-server/src/main/java/org/apache/ratis/server/raftlog/RaftLogBase.java @@ -31,7 +31,9 @@ import org.apache.ratis.util.JavaUtils; import org.apache.ratis.util.OpenCloseState; import org.apache.ratis.util.Preconditions; +import org.apache.ratis.util.ReferenceCountedObject; import org.apache.ratis.util.TimeDuration; +import org.apache.ratis.util.function.UncheckedAutoCloseableSupplier; import java.io.IOException; import java.util.List; @@ -354,11 +356,19 @@ public final CompletableFuture appendEntry(LogEntryProto entry, Transactio protected abstract CompletableFuture appendEntryImpl(LogEntryProto entry, TransactionContext context); @Override - public final List> append(List entries) { + public final List> append(ReferenceCountedObject> entries) { return runner.runSequentially(() -> appendImpl(entries)); } - protected abstract List> appendImpl(List entries); + protected List> appendImpl(List entries) { + throw new UnsupportedOperationException(); + } + + protected List> appendImpl(ReferenceCountedObject> entriesRef) { + try(UncheckedAutoCloseableSupplier> entries = entriesRef.retainAndReleaseOnClose()) { + return appendImpl(entries.get()); + } + } @Override public String toString() { diff --git a/ratis-server/src/main/java/org/apache/ratis/server/raftlog/memory/MemoryRaftLog.java b/ratis-server/src/main/java/org/apache/ratis/server/raftlog/memory/MemoryRaftLog.java index ebb1e27d77..7435bb178d 100644 --- a/ratis-server/src/main/java/org/apache/ratis/server/raftlog/memory/MemoryRaftLog.java +++ b/ratis-server/src/main/java/org/apache/ratis/server/raftlog/memory/MemoryRaftLog.java @@ -28,6 +28,7 @@ import org.apache.ratis.statemachine.TransactionContext; import org.apache.ratis.util.AutoCloseableLock; import org.apache.ratis.util.Preconditions; +import org.apache.ratis.util.ReferenceCountedObject; import java.io.IOException; import java.util.ArrayList; @@ -35,6 +36,7 @@ import java.util.List; import java.util.concurrent.CompletableFuture; import java.util.concurrent.atomic.AtomicReference; +import java.util.function.Function; import java.util.function.LongSupplier; /** @@ -42,10 +44,10 @@ */ public class MemoryRaftLog extends RaftLogBase { static class EntryList { - private final List entries = new ArrayList<>(); + private final List> entries = new ArrayList<>(); LogEntryProto get(int i) { - return i >= 0 && i < entries.size() ? entries.get(i) : null; + return i >= 0 && i < entries.size() ? entries.get(i).get() : null; } TermIndex getTermIndex(int i) { @@ -62,18 +64,25 @@ int size() { void truncate(int index) { if (entries.size() > index) { - entries.subList(index, entries.size()).clear(); + clear(index, entries.size()); } } void purge(int index) { if (entries.size() > index) { - entries.subList(0, index).clear(); + clear(0, index); } } - void add(LogEntryProto entry) { - entries.add(entry); + void clear(int from, int to) { + List> subList = entries.subList(from, to); + subList.forEach(ReferenceCountedObject::release); + subList.clear(); + } + + void add(ReferenceCountedObject entryRef) { + entryRef.retain(); + entries.add(entryRef); } } @@ -170,7 +179,9 @@ protected CompletableFuture appendEntryImpl(LogEntryProto entry, Transacti checkLogState(); try(AutoCloseableLock writeLock = writeLock()) { validateLogEntry(entry); - entries.add(entry); + Function> wrap = context != null ? + context::wrap : ReferenceCountedObject::wrap; + entries.add(wrap.apply(entry)); } return CompletableFuture.completedFuture(entry.getIndex()); } @@ -181,12 +192,14 @@ public long getStartIndex() { } @Override - public List> appendImpl(List logEntryProtos) { + public List> appendImpl(ReferenceCountedObject> entriesRef) { checkLogState(); + final List logEntryProtos = entriesRef.retain(); if (logEntryProtos == null || logEntryProtos.isEmpty()) { + entriesRef.release(); return Collections.emptyList(); } - try(AutoCloseableLock writeLock = writeLock()) { + try (AutoCloseableLock writeLock = writeLock()) { // Before truncating the entries, we first need to check if some // entries are duplicated. If the leader sends entry 6, entry 7, then // entry 6 again, without this check the follower may truncate entry 7 @@ -214,10 +227,12 @@ public List> appendImpl(List logEntryProt } for (int i = index; i < logEntryProtos.size(); i++) { LogEntryProto logEntryProto = logEntryProtos.get(i); - this.entries.add(logEntryProto); + entries.add(entriesRef.delegate(logEntryProto)); futures.add(CompletableFuture.completedFuture(logEntryProto.getIndex())); } return futures; + } finally { + entriesRef.release(); } } diff --git a/ratis-server/src/main/java/org/apache/ratis/server/raftlog/segmented/SegmentedRaftLog.java b/ratis-server/src/main/java/org/apache/ratis/server/raftlog/segmented/SegmentedRaftLog.java index 1cfb5933f8..b4e9589658 100644 --- a/ratis-server/src/main/java/org/apache/ratis/server/raftlog/segmented/SegmentedRaftLog.java +++ b/ratis-server/src/main/java/org/apache/ratis/server/raftlog/segmented/SegmentedRaftLog.java @@ -36,6 +36,7 @@ import org.apache.ratis.proto.RaftProtos.LogEntryProto; import org.apache.ratis.statemachine.StateMachine; import org.apache.ratis.statemachine.TransactionContext; +import org.apache.ratis.statemachine.impl.TransactionContextImpl; import org.apache.ratis.thirdparty.com.google.protobuf.ByteString; import org.apache.ratis.util.AutoCloseableLock; import org.apache.ratis.util.AwaitToRun; @@ -454,12 +455,14 @@ private boolean isSegmentFull(LogSegment segment, LogEntryProto entry) { } @Override - public List> appendImpl(List entries) { + protected List> appendImpl(ReferenceCountedObject> entriesRef) { checkLogState(); + final List entries = entriesRef.retain(); if (entries == null || entries.isEmpty()) { + entriesRef.release(); return Collections.emptyList(); } - try(AutoCloseableLock writeLock = writeLock()) { + try (AutoCloseableLock writeLock = writeLock()) { final TruncateIndices ti = cache.computeTruncateIndices(server::notifyTruncatedLogEntry, entries); final long truncateIndex = ti.getTruncateIndex(); final int index = ti.getArrayIndex(); @@ -474,9 +477,15 @@ public List> appendImpl(List entries) { } for (int i = index; i < entries.size(); i++) { final LogEntryProto entry = entries.get(i); - futures.add(appendEntry(entry, server.getTransactionContext(entry, true))); + TransactionContextImpl transactionContext = (TransactionContextImpl) server.getTransactionContext(entry, true); + if (transactionContext != null) { + transactionContext.setDelegatedRef(entriesRef); + } + futures.add(appendEntry(entry, transactionContext)); } return futures; + } finally { + entriesRef.release(); } } diff --git a/ratis-test/src/test/java/org/apache/ratis/server/raftlog/memory/MemoryRaftLogTest.java b/ratis-test/src/test/java/org/apache/ratis/server/raftlog/memory/MemoryRaftLogTest.java index 5d8d090a38..503de34536 100644 --- a/ratis-test/src/test/java/org/apache/ratis/server/raftlog/memory/MemoryRaftLogTest.java +++ b/ratis-test/src/test/java/org/apache/ratis/server/raftlog/memory/MemoryRaftLogTest.java @@ -33,6 +33,7 @@ import org.apache.ratis.server.raftlog.RaftLog; import org.apache.ratis.statemachine.impl.SimpleStateMachine4Testing; import org.apache.ratis.statemachine.StateMachine; +import org.apache.ratis.util.ReferenceCountedObject; import org.apache.ratis.util.Slf4jUtils; import org.junit.Test; import org.slf4j.event.Level; @@ -56,11 +57,11 @@ public void testEntryDoNotPerformTruncation() throws Exception { List entries1 = new ArrayList<>(); entries1.add(LogEntryProto.newBuilder().setIndex(0).setTerm(0).build()); entries1.add(LogEntryProto.newBuilder().setIndex(1).setTerm(0).build()); - raftLog.append(entries1).forEach(CompletableFuture::join); + raftLog.append(ReferenceCountedObject.wrap(entries1)).forEach(CompletableFuture::join); List entries2 = new ArrayList<>(); entries2.add(LogEntryProto.newBuilder().setIndex(0).setTerm(0).build()); - raftLog.append(entries2).forEach(CompletableFuture::join); + raftLog.append(ReferenceCountedObject.wrap(entries2)).forEach(CompletableFuture::join); final LogEntryHeader[] termIndices = raftLog.getEntries(0, 10); assertEquals(2, termIndices.length); @@ -84,11 +85,11 @@ public void testEntryPerformTruncation() throws Exception { List entries1 = new ArrayList<>(); entries1.add(LogEntryProto.newBuilder().setIndex(0).setTerm(0).build()); entries1.add(LogEntryProto.newBuilder().setIndex(1).setTerm(0).build()); - raftLog.append(entries1).forEach(CompletableFuture::join); + raftLog.append(ReferenceCountedObject.wrap(entries1)).forEach(CompletableFuture::join); List entries2 = new ArrayList<>(); entries2.add(LogEntryProto.newBuilder().setIndex(0).setTerm(2).build()); - raftLog.append(entries2).forEach(CompletableFuture::join); + raftLog.append(ReferenceCountedObject.wrap(entries2)).forEach(CompletableFuture::join); final LogEntryHeader[] termIndices = raftLog.getEntries(0, 10); assertEquals(1, termIndices.length); diff --git a/ratis-test/src/test/java/org/apache/ratis/server/raftlog/segmented/TestCacheEviction.java b/ratis-test/src/test/java/org/apache/ratis/server/raftlog/segmented/TestCacheEviction.java index 996f7ef527..675df51d6b 100644 --- a/ratis-test/src/test/java/org/apache/ratis/server/raftlog/segmented/TestCacheEviction.java +++ b/ratis-test/src/test/java/org/apache/ratis/server/raftlog/segmented/TestCacheEviction.java @@ -38,6 +38,7 @@ import org.apache.ratis.statemachine.impl.SimpleStateMachine4Testing; import org.apache.ratis.statemachine.StateMachine; import org.apache.ratis.util.JavaUtils; +import org.apache.ratis.util.ReferenceCountedObject; import org.apache.ratis.util.SizeInBytes; import org.junit.Assert; import org.junit.Test; @@ -174,7 +175,7 @@ public void testEvictionInSegmentedLog() throws Exception { raftLog.open(RaftLog.INVALID_LOG_INDEX, null); List slist = TestSegmentedRaftLog.prepareRanges(0, maxCachedNum, 7, 0); List entries = generateEntries(slist); - raftLog.append(entries).forEach(CompletableFuture::join); + raftLog.append(ReferenceCountedObject.wrap(entries)).forEach(CompletableFuture::join); // check the current cached segment number: the last segment is still open Assert.assertEquals(maxCachedNum - 1, @@ -184,7 +185,7 @@ public void testEvictionInSegmentedLog() throws Exception { Mockito.when(info.getFollowerNextIndices()).thenReturn(new long[]{21, 40, 40}); slist = TestSegmentedRaftLog.prepareRanges(maxCachedNum, maxCachedNum + 2, 7, 7 * maxCachedNum); entries = generateEntries(slist); - raftLog.append(entries).forEach(CompletableFuture::join); + raftLog.append(ReferenceCountedObject.wrap(entries)).forEach(CompletableFuture::join); // check the cached segment number again. since the slowest follower is on // index 21, the eviction should happen and evict 3 segments diff --git a/ratis-test/src/test/java/org/apache/ratis/server/raftlog/segmented/TestSegmentedRaftLog.java b/ratis-test/src/test/java/org/apache/ratis/server/raftlog/segmented/TestSegmentedRaftLog.java index 38fa45e6fa..3d5d5f87d0 100644 --- a/ratis-test/src/test/java/org/apache/ratis/server/raftlog/segmented/TestSegmentedRaftLog.java +++ b/ratis-test/src/test/java/org/apache/ratis/server/raftlog/segmented/TestSegmentedRaftLog.java @@ -43,6 +43,7 @@ import org.apache.ratis.statemachine.StateMachine; import org.apache.ratis.statemachine.impl.BaseStateMachine; import org.apache.ratis.util.LifeCycle; +import org.apache.ratis.util.ReferenceCountedObject; import org.apache.ratis.util.Slf4jUtils; import org.apache.ratis.util.FileUtils; import org.apache.ratis.util.JavaUtils; @@ -551,7 +552,7 @@ public void testAppendEntriesWithInconsistency() throws Exception { LOG.info("newEntries[0] = {}", newEntries.get(0)); final int last = newEntries.size() - 1; LOG.info("newEntries[{}] = {}", last, newEntries.get(last)); - raftLog.append(newEntries).forEach(CompletableFuture::join); + raftLog.append(ReferenceCountedObject.wrap(newEntries)).forEach(CompletableFuture::join); checkFailedEntries(entries, 650, retryCache); checkEntries(raftLog, entries, 0, 650); @@ -710,8 +711,9 @@ public void testAsyncFlushPerf1() throws Exception { long start = System.nanoTime(); for (int i = 0; i < entries.size(); i += 5) { // call append API - futures.add(raftLog.append(Arrays.asList( - entries.get(i), entries.get(i + 1), entries.get(i + 2), entries.get(i + 3), entries.get(i + 4)))); + List entries1 = Arrays.asList( + entries.get(i), entries.get(i + 1), entries.get(i + 2), entries.get(i + 3), entries.get(i + 4)); + futures.add(raftLog.append(ReferenceCountedObject.wrap(entries1))); } for (List> futureList: futures) { futureList.forEach(CompletableFuture::join);