Repository: incubator-ratis Updated Branches: refs/heads/master c16cb1c65 -> 2d99e6172
RATIS-127. Move the state machine implementation to a new package. Contributed by Jing Chen Project: http://git-wip-us.apache.org/repos/asf/incubator-ratis/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-ratis/commit/2d99e617 Tree: http://git-wip-us.apache.org/repos/asf/incubator-ratis/tree/2d99e617 Diff: http://git-wip-us.apache.org/repos/asf/incubator-ratis/diff/2d99e617 Branch: refs/heads/master Commit: 2d99e61728402ccf848f4c47a38ac15d6275344d Parents: c16cb1c Author: Tsz-Wo Nicholas Sze <[email protected]> Authored: Fri Jan 26 17:20:59 2018 -0800 Committer: Tsz-Wo Nicholas Sze <[email protected]> Committed: Fri Jan 26 17:20:59 2018 -0800 ---------------------------------------------------------------------- .../arithmetic/ArithmeticStateMachine.java | 3 + .../filestore/FileStoreStateMachine.java | 7 +- .../ratis/server/impl/RaftServerImpl.java | 3 +- .../ratis/statemachine/BaseStateMachine.java | 158 --------------- .../statemachine/FileListSnapshotInfo.java | 64 ------ .../statemachine/SimpleStateMachineStorage.java | 132 ------------ .../statemachine/SingleFileSnapshotInfo.java | 38 ---- .../ratis/statemachine/TransactionContext.java | 183 +++++------------ .../statemachine/impl/BaseStateMachine.java | 162 +++++++++++++++ .../statemachine/impl/FileListSnapshotInfo.java | 65 ++++++ .../impl/SimpleStateMachineStorage.java | 133 ++++++++++++ .../impl/SingleFileSnapshotInfo.java | 38 ++++ .../impl/TransactionContextImpl.java | 200 +++++++++++++++++++ .../java/org/apache/ratis/MiniRaftCluster.java | 2 +- .../ratis/server/storage/TestRaftStorage.java | 2 +- .../SimpleStateMachine4Testing.java | 6 +- .../ratis/statemachine/TestStateMachine.java | 3 +- 17 files changed, 666 insertions(+), 533 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-ratis/blob/2d99e617/ratis-examples/src/main/java/org/apache/ratis/examples/arithmetic/ArithmeticStateMachine.java ---------------------------------------------------------------------- diff --git a/ratis-examples/src/main/java/org/apache/ratis/examples/arithmetic/ArithmeticStateMachine.java b/ratis-examples/src/main/java/org/apache/ratis/examples/arithmetic/ArithmeticStateMachine.java index f7488b5..4eb0756 100644 --- a/ratis-examples/src/main/java/org/apache/ratis/examples/arithmetic/ArithmeticStateMachine.java +++ b/ratis-examples/src/main/java/org/apache/ratis/examples/arithmetic/ArithmeticStateMachine.java @@ -28,6 +28,9 @@ import org.apache.ratis.server.protocol.TermIndex; import org.apache.ratis.server.storage.RaftStorage; import org.apache.ratis.shaded.proto.RaftProtos.LogEntryProto; import org.apache.ratis.statemachine.*; +import org.apache.ratis.statemachine.impl.BaseStateMachine; +import org.apache.ratis.statemachine.impl.SimpleStateMachineStorage; +import org.apache.ratis.statemachine.impl.SingleFileSnapshotInfo; import org.apache.ratis.util.AutoCloseableLock; import org.apache.ratis.util.Preconditions; import org.slf4j.Logger; http://git-wip-us.apache.org/repos/asf/incubator-ratis/blob/2d99e617/ratis-examples/src/main/java/org/apache/ratis/examples/filestore/FileStoreStateMachine.java ---------------------------------------------------------------------- diff --git a/ratis-examples/src/main/java/org/apache/ratis/examples/filestore/FileStoreStateMachine.java b/ratis-examples/src/main/java/org/apache/ratis/examples/filestore/FileStoreStateMachine.java index 538c970..2a09e1b 100644 --- a/ratis-examples/src/main/java/org/apache/ratis/examples/filestore/FileStoreStateMachine.java +++ b/ratis-examples/src/main/java/org/apache/ratis/examples/filestore/FileStoreStateMachine.java @@ -27,10 +27,11 @@ import org.apache.ratis.shaded.com.google.protobuf.InvalidProtocolBufferExceptio import org.apache.ratis.shaded.proto.ExamplesProtos.*; import org.apache.ratis.shaded.proto.RaftProtos.LogEntryProto; import org.apache.ratis.shaded.proto.RaftProtos.SMLogEntryProto; -import org.apache.ratis.statemachine.BaseStateMachine; -import org.apache.ratis.statemachine.SimpleStateMachineStorage; +import org.apache.ratis.statemachine.impl.BaseStateMachine; +import org.apache.ratis.statemachine.impl.SimpleStateMachineStorage; import org.apache.ratis.statemachine.StateMachineStorage; import org.apache.ratis.statemachine.TransactionContext; +import org.apache.ratis.statemachine.impl.TransactionContextImpl; import org.apache.ratis.util.FileUtils; import org.apache.ratis.util.Preconditions; import org.slf4j.Logger; @@ -106,7 +107,7 @@ public class FileStoreStateMachine extends BaseStateMachine { log = SMLogEntryProto.newBuilder().setData(content).build(); } - return new TransactionContext(this, request, log); + return new TransactionContextImpl(this, request, log); } @Override http://git-wip-us.apache.org/repos/asf/incubator-ratis/blob/2d99e617/ratis-server/src/main/java/org/apache/ratis/server/impl/RaftServerImpl.java ---------------------------------------------------------------------- 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 44499be..29b3a7e 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 @@ -32,6 +32,7 @@ import org.apache.ratis.shaded.proto.RaftProtos.*; import org.apache.ratis.statemachine.SnapshotInfo; import org.apache.ratis.statemachine.StateMachine; import org.apache.ratis.statemachine.TransactionContext; +import org.apache.ratis.statemachine.impl.TransactionContextImpl; import org.apache.ratis.util.*; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -990,7 +991,7 @@ public class RaftServerImpl implements RaftServerProtocol, RaftServerAsynchronou // check whether there is a TransactionContext because we are the leader. TransactionContext trx = getTransactionContext(next.getIndex()); if (trx == null) { - trx = new TransactionContext(stateMachine, next); + trx = new TransactionContextImpl(stateMachine, next); } // Let the StateMachine inject logic for committed transactions in sequential order. http://git-wip-us.apache.org/repos/asf/incubator-ratis/blob/2d99e617/ratis-server/src/main/java/org/apache/ratis/statemachine/BaseStateMachine.java ---------------------------------------------------------------------- diff --git a/ratis-server/src/main/java/org/apache/ratis/statemachine/BaseStateMachine.java b/ratis-server/src/main/java/org/apache/ratis/statemachine/BaseStateMachine.java deleted file mode 100644 index dbf19b0..0000000 --- a/ratis-server/src/main/java/org/apache/ratis/statemachine/BaseStateMachine.java +++ /dev/null @@ -1,158 +0,0 @@ -/** - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.apache.ratis.statemachine; - -import java.io.IOException; -import java.util.Collection; -import java.util.concurrent.CompletableFuture; - -import org.apache.ratis.conf.RaftProperties; -import org.apache.ratis.protocol.Message; -import org.apache.ratis.protocol.RaftClientReply; -import org.apache.ratis.protocol.RaftClientRequest; -import org.apache.ratis.protocol.RaftPeerId; -import org.apache.ratis.server.impl.RaftConfiguration; -import org.apache.ratis.server.impl.RaftServerConstants; -import org.apache.ratis.server.storage.RaftStorage; -import org.apache.ratis.shaded.proto.RaftProtos.SMLogEntryProto; -import org.apache.ratis.util.LifeCycle; - -/** - * Base implementation for StateMachines. - */ -public class BaseStateMachine implements StateMachine { - - private volatile RaftPeerId id; - protected RaftProperties properties; - protected RaftStorage storage; - protected RaftConfiguration raftConf; - protected final LifeCycle lifeCycle = new LifeCycle(getClass().getSimpleName()); - - public RaftPeerId getId() { - return id; - } - - @Override - public LifeCycle.State getLifeCycleState() { - return lifeCycle.getCurrentState(); - } - - @Override - public void initialize(RaftPeerId id, RaftProperties properties, - RaftStorage storage) throws IOException { - this.id = id; - lifeCycle.setName(getClass().getSimpleName() + ":" + id); - this.properties = properties; - this.storage = storage; - } - - @Override - public void setRaftConfiguration(RaftConfiguration conf) { - this.raftConf = conf; - } - - @Override - public RaftConfiguration getRaftConfiguration() { - return this.raftConf; - } - - @Override - public SnapshotInfo getLatestSnapshot() { - return getStateMachineStorage().getLatestSnapshot(); - } - - @Override - public void notifyNotLeader(Collection<TransactionContext> pendingEntries) throws IOException { - // do nothing - } - - @Override - public void pause() { - } - - @Override - public void reinitialize(RaftPeerId id, RaftProperties properties, - RaftStorage storage) throws IOException { - } - - @Override - public TransactionContext applyTransactionSerial(TransactionContext trx) { - return trx; - } - - @Override - public CompletableFuture<Message> applyTransaction(TransactionContext trx) { - // return the same message contained in the entry - Message msg = () -> trx.getLogEntry().getSmLogEntry().getData(); - return CompletableFuture.completedFuture(msg); - } - - @Override - public long takeSnapshot() throws IOException { - return RaftServerConstants.INVALID_LOG_INDEX; - } - - @Override - public StateMachineStorage getStateMachineStorage() { - return new StateMachineStorage() { - @Override - public void init(RaftStorage raftStorage) throws IOException { - } - - @Override - public SnapshotInfo getLatestSnapshot() { - return null; - } - - @Override - public void format() throws IOException { - } - }; - } - - @Override - public CompletableFuture<RaftClientReply> query( - RaftClientRequest request) { - return null; - } - - @Override - public TransactionContext startTransaction(RaftClientRequest request) - throws IOException { - return new TransactionContext(this, request, - SMLogEntryProto.newBuilder() - .setData(request.getMessage().getContent()) - .build()); - } - - @Override - public TransactionContext cancelTransaction(TransactionContext trx) throws IOException { - return trx; - } - - @Override - public TransactionContext preAppendTransaction(TransactionContext trx) throws IOException { - return trx; - } - - @Override - public void close() throws IOException { - // do nothing - } -} http://git-wip-us.apache.org/repos/asf/incubator-ratis/blob/2d99e617/ratis-server/src/main/java/org/apache/ratis/statemachine/FileListSnapshotInfo.java ---------------------------------------------------------------------- diff --git a/ratis-server/src/main/java/org/apache/ratis/statemachine/FileListSnapshotInfo.java b/ratis-server/src/main/java/org/apache/ratis/statemachine/FileListSnapshotInfo.java deleted file mode 100644 index 1858603..0000000 --- a/ratis-server/src/main/java/org/apache/ratis/statemachine/FileListSnapshotInfo.java +++ /dev/null @@ -1,64 +0,0 @@ -/** - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -package org.apache.ratis.statemachine; - -import java.util.Collections; -import java.util.List; - -import org.apache.ratis.server.protocol.TermIndex; -import org.apache.ratis.server.storage.FileInfo; - -/** - * Each snapshot has a list of files. - * - * The objects of this class are immutable. - */ -public class FileListSnapshotInfo implements SnapshotInfo { - private final TermIndex termIndex; - private final List<FileInfo> files; - - public FileListSnapshotInfo(List<FileInfo> files, long term, long index) { - this.termIndex = TermIndex.newTermIndex(term, index); - this.files = Collections.unmodifiableList(files); - } - - @Override - public TermIndex getTermIndex() { - return termIndex; - } - - @Override - public long getTerm() { - return termIndex.getTerm(); - } - - @Override - public long getIndex() { - return termIndex.getIndex(); - } - - @Override - public List<FileInfo> getFiles() { - return files; - } - - @Override - public String toString() { - return termIndex + ":" + files; - } -} http://git-wip-us.apache.org/repos/asf/incubator-ratis/blob/2d99e617/ratis-server/src/main/java/org/apache/ratis/statemachine/SimpleStateMachineStorage.java ---------------------------------------------------------------------- diff --git a/ratis-server/src/main/java/org/apache/ratis/statemachine/SimpleStateMachineStorage.java b/ratis-server/src/main/java/org/apache/ratis/statemachine/SimpleStateMachineStorage.java deleted file mode 100644 index 05c5337..0000000 --- a/ratis-server/src/main/java/org/apache/ratis/statemachine/SimpleStateMachineStorage.java +++ /dev/null @@ -1,132 +0,0 @@ -/** - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -package org.apache.ratis.statemachine; - -import org.apache.ratis.io.MD5Hash; -import org.apache.ratis.server.protocol.TermIndex; -import org.apache.ratis.server.storage.FileInfo; -import org.apache.ratis.server.storage.RaftStorage; -import org.apache.ratis.util.AtomicFileOutputStream; -import org.apache.ratis.util.MD5FileUtil; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import java.io.File; -import java.io.IOException; -import java.nio.file.DirectoryStream; -import java.nio.file.Files; -import java.nio.file.Path; -import java.util.regex.Matcher; -import java.util.regex.Pattern; - -/** - * A StateMachineStorage that stores the snapshot in a single file. - */ -public class SimpleStateMachineStorage implements StateMachineStorage { - - private static final Logger LOG = LoggerFactory.getLogger(SimpleStateMachineStorage.class); - - static final String SNAPSHOT_FILE_PREFIX = "snapshot"; - static final String CORRUPT_SNAPSHOT_FILE_SUFFIX = ".corrupt"; - /** snapshot.term_index */ - static final Pattern SNAPSHOT_REGEX = - Pattern.compile(SNAPSHOT_FILE_PREFIX + "\\.(\\d+)_(\\d+)"); - - private RaftStorage raftStorage; - private File smDir = null; - - private volatile SingleFileSnapshotInfo currentSnapshot = null; - - @Override - public void init(RaftStorage raftStorage) throws IOException { - this.raftStorage = raftStorage; - this.smDir = raftStorage.getStorageDir().getStateMachineDir(); - loadLatestSnapshot(); - } - - @Override - public void format() throws IOException { - // TODO - } - - public static TermIndex getTermIndexFromSnapshotFile(File file) { - final String name = file.getName(); - final Matcher m = SNAPSHOT_REGEX.matcher(name); - if (!m.matches()) { - throw new IllegalArgumentException("File \"" + file - + "\" does not match snapshot file name pattern \"" - + SNAPSHOT_REGEX + "\""); - } - final long term = Long.parseLong(m.group(1)); - final long index = Long.parseLong(m.group(2)); - return TermIndex.newTermIndex(term, index); - } - - protected static String getTmpSnapshotFileName(long term, long endIndex) { - return getSnapshotFileName(term, endIndex) + AtomicFileOutputStream.TMP_EXTENSION; - } - - protected static String getCorruptSnapshotFileName(long term, long endIndex) { - return getSnapshotFileName(term, endIndex) + CORRUPT_SNAPSHOT_FILE_SUFFIX; - } - - public File getSnapshotFile(long term, long endIndex) { - return new File(smDir, getSnapshotFileName(term, endIndex)); - } - - protected File getTmpSnapshotFile(long term, long endIndex) { - return new File(smDir, getTmpSnapshotFileName(term, endIndex)); - } - - protected File getCorruptSnapshotFile(long term, long endIndex) { - return new File(smDir, getCorruptSnapshotFileName(term, endIndex)); - } - - public SingleFileSnapshotInfo findLatestSnapshot() throws IOException { - SingleFileSnapshotInfo latest = null; - try (DirectoryStream<Path> stream = - Files.newDirectoryStream(smDir.toPath())) { - for (Path path : stream) { - Matcher matcher = SNAPSHOT_REGEX.matcher(path.getFileName().toString()); - if (matcher.matches()) { - final long endIndex = Long.parseLong(matcher.group(2)); - if (latest == null || endIndex > latest.getIndex()) { - final long term = Long.parseLong(matcher.group(1)); - MD5Hash fileDigest = MD5FileUtil.readStoredMd5ForFile(path.toFile()); - final FileInfo fileInfo = new FileInfo(path, fileDigest); - latest = new SingleFileSnapshotInfo(fileInfo, term, endIndex); - } - } - } - } - return latest; - } - - public void loadLatestSnapshot() throws IOException { - this.currentSnapshot = findLatestSnapshot(); - } - - public static String getSnapshotFileName(long term, long endIndex) { - return SNAPSHOT_FILE_PREFIX + "." + term + "_" + endIndex; - } - - @Override - public SingleFileSnapshotInfo getLatestSnapshot() { - return currentSnapshot; - } -} http://git-wip-us.apache.org/repos/asf/incubator-ratis/blob/2d99e617/ratis-server/src/main/java/org/apache/ratis/statemachine/SingleFileSnapshotInfo.java ---------------------------------------------------------------------- diff --git a/ratis-server/src/main/java/org/apache/ratis/statemachine/SingleFileSnapshotInfo.java b/ratis-server/src/main/java/org/apache/ratis/statemachine/SingleFileSnapshotInfo.java deleted file mode 100644 index 5bca2c9..0000000 --- a/ratis-server/src/main/java/org/apache/ratis/statemachine/SingleFileSnapshotInfo.java +++ /dev/null @@ -1,38 +0,0 @@ -/** - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -package org.apache.ratis.statemachine; - -import java.util.Arrays; - -import org.apache.ratis.server.storage.FileInfo; - -/** - * Each snapshot only has a single file. - * - * The objects of this class are immutable. - */ -public class SingleFileSnapshotInfo extends FileListSnapshotInfo { - public SingleFileSnapshotInfo(FileInfo fileInfo, long term, long endIndex) { - super(Arrays.asList(fileInfo), term, endIndex); - } - - /** @return the file associated with the snapshot. */ - public FileInfo getFile() { - return getFiles().get(0); - } -} http://git-wip-us.apache.org/repos/asf/incubator-ratis/blob/2d99e617/ratis-server/src/main/java/org/apache/ratis/statemachine/TransactionContext.java ---------------------------------------------------------------------- diff --git a/ratis-server/src/main/java/org/apache/ratis/statemachine/TransactionContext.java b/ratis-server/src/main/java/org/apache/ratis/statemachine/TransactionContext.java index df9a4fe..53ef2f8 100644 --- a/ratis-server/src/main/java/org/apache/ratis/statemachine/TransactionContext.java +++ b/ratis-server/src/main/java/org/apache/ratis/statemachine/TransactionContext.java @@ -45,151 +45,74 @@ import org.apache.ratis.util.Preconditions; * The {@link TransactionContext} will be a committed entry coming from * the RAFT log from the leader. */ -public class TransactionContext { +public interface TransactionContext { - /** The {@link StateMachine} that originated the transaction. */ - private final StateMachine stateMachine; - - /** Original request from the client */ - private RaftClientRequest clientRequest; - - /** Exception from the {@link StateMachine} or from the log */ - private Exception exception; + /** + * Returns the original request from the {@link RaftClientRequest} + * @return the original request from the {@link RaftClientRequest} + */ + RaftClientRequest getClientRequest(); - /** Data from the {@link StateMachine} */ - private SMLogEntryProto smLogEntryProto; + /** + * Returns the data from the {@link StateMachine} + * @return the data from the {@link StateMachine} + */ + SMLogEntryProto getSMLogEntry(); /** - * Context specific to the state machine. - * The {@link StateMachine} can use this object to carry state between - * {@link StateMachine#startTransaction(RaftClientRequest)} and - * {@link StateMachine#applyTransaction(TransactionContext)}. + * Returns the exception from the {@link StateMachine} or the log + * @return the exception from the {@link StateMachine} or the log */ - private Object stateMachineContext; + Exception getException(); /** - * Whether to commit the transaction to the RAFT Log. - * In some cases the {@link StateMachine} may want to indicate - * that the transaction should not be committed + * Sets the {@link StateMachine} the {@link TransactionContext} is specific to, the method would + * not create a new transaction context, it updates the {@link StateMachine} it associates with + * @param stateMachineContext state machine context + * @return transaction context specific to the given {@link StateMachine} */ - private boolean shouldCommit = true; + TransactionContext setStateMachineContext(Object stateMachineContext); - /** Committed LogEntry. */ - private LogEntryProto logEntry; + /** + * Returns the {@link StateMachine} the current {@link TransactionContext} specific to + * @return the {@link StateMachine} the current {@link TransactionContext} specific to + */ + Object getStateMachineContext(); - private TransactionContext(StateMachine stateMachine) { - this.stateMachine = stateMachine; - } + /** + * Set the {@link LogEntryProto} the current {@link TransactionContext} specific to. The log + * entry's body case must be {@link LogEntryBodyCase#SMLOGENTRY}. The current + * {@link TransactionContext} log entry must be null, otherwise, a exception will be thrown + * @param logEntry target {@link LogEntryProto} + * @return the current {@link TransactionContext} itself + */ + TransactionContext setLogEntry(LogEntryProto logEntry); - /** The same as this(stateMachine, clientRequest, smLogEntryProto, null). */ - public TransactionContext( - StateMachine stateMachine, RaftClientRequest clientRequest, - SMLogEntryProto smLogEntryProto) { - this(stateMachine, clientRequest, smLogEntryProto, null); - } + /** + * Sets the data from the {@link StateMachine} + * @param smLogEntryProto data from {@link StateMachine} + * @return the current {@link TransactionContext} itself + */ + TransactionContext setSmLogEntryProto(SMLogEntryProto smLogEntryProto); /** - * Construct a {@link TransactionContext} from a client request. - * Used by the state machine to start a transaction - * and send the Log entry representing the transaction data - * to be applied to the raft log. + * Returns the committed log entry + * @return the committed log entry */ - public TransactionContext( - StateMachine stateMachine, RaftClientRequest clientRequest, - SMLogEntryProto smLogEntryProto, Object stateMachineContext) { - this(stateMachine); - this.clientRequest = clientRequest; - this.smLogEntryProto = smLogEntryProto; - this.stateMachineContext = stateMachineContext; - } - - /** The same as this(stateMachine, clientRequest, exception, null). */ - public TransactionContext( - StateMachine stateMachine, RaftClientRequest clientRequest, - Exception exception) { - this(stateMachine, clientRequest, exception, null); - } + LogEntryProto getLogEntry(); /** - * Construct a {@link TransactionContext} from a client request to signal - * an exception so that the RAFT server will fail the request on behalf - * of the {@link StateMachine}. + * Sets whether to commit the transaction to the RAFT log or not + * @param shouldCommit true if the transaction is supposed to be committed to the RAFT log + * @return the current {@link TransactionContext} itself */ - public TransactionContext( - StateMachine stateMachine, RaftClientRequest clientRequest, - Exception exception, Object stateMachineContext) { - this(stateMachine); - this.clientRequest = clientRequest; - this.exception = exception; - this.stateMachineContext = stateMachineContext; - } + TransactionContext setShouldCommit(boolean shouldCommit); /** - * Construct a {@link TransactionContext} from a {@link LogEntryProto}. - * Used by followers for applying committed entries to the state machine. - * @param logEntry the log entry to be applied + * It indicates if the transaction should be committed to the RAFT log + * @return true if it commits the transaction to the RAFT log, otherwise, false */ - public TransactionContext(StateMachine stateMachine, LogEntryProto logEntry) { - this(stateMachine); - setLogEntry(logEntry); - this.smLogEntryProto = logEntry.getSmLogEntry(); - } - - public RaftClientRequest getClientRequest() { - return clientRequest; - } - - public SMLogEntryProto getSMLogEntry() { - return smLogEntryProto; - } - - public Exception getException() { - return exception; - } - - public TransactionContext setStateMachineContext(Object stateMachineContext) { - this.stateMachineContext = stateMachineContext; - return this; - } - - public Object getStateMachineContext() { - return stateMachineContext; - } - - public TransactionContext setLogEntry(LogEntryProto logEntry) { - Objects.requireNonNull(logEntry, "logEntry == null"); - Preconditions.assertTrue(logEntry.getLogEntryBodyCase() == LogEntryBodyCase.SMLOGENTRY, - () -> "LogEntryBodyCase = " + logEntry.getLogEntryBodyCase() - + " != " + LogEntryBodyCase.SMLOGENTRY + ", logEntry=" + logEntry); - Preconditions.assertTrue(this.logEntry == null, "this.logEntry != null"); - this.logEntry = logEntry; - return this; - } - - public TransactionContext setSmLogEntryProto(SMLogEntryProto smLogEntryProto) { - this.smLogEntryProto = smLogEntryProto; - return this; - } - - public LogEntryProto getLogEntry() { - return logEntry; - } - - private TransactionContext setException(IOException ioe) { - assert exception != null; - this.exception = ioe; - return this; - } - - public TransactionContext setShouldCommit(boolean shouldCommit) { - this.shouldCommit = shouldCommit; - return this; - } - - public boolean shouldCommit() { - // TODO: Hook this up in the server to bypass the RAFT Log and send back a response to client - return this.shouldCommit; - } + boolean shouldCommit(); // proxy StateMachine methods. We do not want to expose the SM to the RaftLog @@ -200,18 +123,12 @@ public class TransactionContext { * log append, it is important to do only required operations here. * @return The Transaction context. */ - public TransactionContext preAppendTransaction() throws IOException { - return stateMachine.preAppendTransaction(this); - } + public TransactionContext preAppendTransaction() throws IOException; /** * Called to notify the state machine that the Transaction passed cannot be appended (or synced). * The exception field will indicate whether there was an exception or not. * @return cancelled transaction */ - public TransactionContext cancelTransaction() throws IOException { - // TODO: This is not called from Raft server / log yet. When an IOException happens, we should - // call this to let the SM know that Transaction cannot be synced - return stateMachine.cancelTransaction(this); - } + public TransactionContext cancelTransaction() throws IOException; } http://git-wip-us.apache.org/repos/asf/incubator-ratis/blob/2d99e617/ratis-server/src/main/java/org/apache/ratis/statemachine/impl/BaseStateMachine.java ---------------------------------------------------------------------- diff --git a/ratis-server/src/main/java/org/apache/ratis/statemachine/impl/BaseStateMachine.java b/ratis-server/src/main/java/org/apache/ratis/statemachine/impl/BaseStateMachine.java new file mode 100644 index 0000000..520d965 --- /dev/null +++ b/ratis-server/src/main/java/org/apache/ratis/statemachine/impl/BaseStateMachine.java @@ -0,0 +1,162 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.ratis.statemachine.impl; + +import java.io.IOException; +import java.util.Collection; +import java.util.concurrent.CompletableFuture; + +import org.apache.ratis.conf.RaftProperties; +import org.apache.ratis.protocol.Message; +import org.apache.ratis.protocol.RaftClientReply; +import org.apache.ratis.protocol.RaftClientRequest; +import org.apache.ratis.protocol.RaftPeerId; +import org.apache.ratis.server.impl.RaftConfiguration; +import org.apache.ratis.server.impl.RaftServerConstants; +import org.apache.ratis.server.storage.RaftStorage; +import org.apache.ratis.shaded.proto.RaftProtos.SMLogEntryProto; +import org.apache.ratis.statemachine.SnapshotInfo; +import org.apache.ratis.statemachine.StateMachine; +import org.apache.ratis.statemachine.StateMachineStorage; +import org.apache.ratis.statemachine.TransactionContext; +import org.apache.ratis.util.LifeCycle; + +/** + * Base implementation for StateMachines. + */ +public class BaseStateMachine implements StateMachine { + + private volatile RaftPeerId id; + protected RaftProperties properties; + protected RaftStorage storage; + protected RaftConfiguration raftConf; + protected final LifeCycle lifeCycle = new LifeCycle(getClass().getSimpleName()); + + public RaftPeerId getId() { + return id; + } + + @Override + public LifeCycle.State getLifeCycleState() { + return lifeCycle.getCurrentState(); + } + + @Override + public void initialize(RaftPeerId id, RaftProperties properties, + RaftStorage storage) throws IOException { + this.id = id; + lifeCycle.setName(getClass().getSimpleName() + ":" + id); + this.properties = properties; + this.storage = storage; + } + + @Override + public void setRaftConfiguration(RaftConfiguration conf) { + this.raftConf = conf; + } + + @Override + public RaftConfiguration getRaftConfiguration() { + return this.raftConf; + } + + @Override + public SnapshotInfo getLatestSnapshot() { + return getStateMachineStorage().getLatestSnapshot(); + } + + @Override + public void notifyNotLeader(Collection<TransactionContext> pendingEntries) throws IOException { + // do nothing + } + + @Override + public void pause() { + } + + @Override + public void reinitialize(RaftPeerId id, RaftProperties properties, + RaftStorage storage) throws IOException { + } + + @Override + public TransactionContext applyTransactionSerial(TransactionContext trx) { + return trx; + } + + @Override + public CompletableFuture<Message> applyTransaction(TransactionContext trx) { + // return the same message contained in the entry + Message msg = () -> trx.getLogEntry().getSmLogEntry().getData(); + return CompletableFuture.completedFuture(msg); + } + + @Override + public long takeSnapshot() throws IOException { + return RaftServerConstants.INVALID_LOG_INDEX; + } + + @Override + public StateMachineStorage getStateMachineStorage() { + return new StateMachineStorage() { + @Override + public void init(RaftStorage raftStorage) throws IOException { + } + + @Override + public SnapshotInfo getLatestSnapshot() { + return null; + } + + @Override + public void format() throws IOException { + } + }; + } + + @Override + public CompletableFuture<RaftClientReply> query( + RaftClientRequest request) { + return null; + } + + @Override + public TransactionContext startTransaction(RaftClientRequest request) + throws IOException { + return new TransactionContextImpl(this, request, + SMLogEntryProto.newBuilder() + .setData(request.getMessage().getContent()) + .build()); + } + + @Override + public TransactionContext cancelTransaction(TransactionContext trx) throws IOException { + return trx; + } + + @Override + public TransactionContext preAppendTransaction(TransactionContext trx) throws IOException { + return trx; + } + + @Override + public void close() throws IOException { + // do nothing + } +} http://git-wip-us.apache.org/repos/asf/incubator-ratis/blob/2d99e617/ratis-server/src/main/java/org/apache/ratis/statemachine/impl/FileListSnapshotInfo.java ---------------------------------------------------------------------- diff --git a/ratis-server/src/main/java/org/apache/ratis/statemachine/impl/FileListSnapshotInfo.java b/ratis-server/src/main/java/org/apache/ratis/statemachine/impl/FileListSnapshotInfo.java new file mode 100644 index 0000000..0e21223 --- /dev/null +++ b/ratis-server/src/main/java/org/apache/ratis/statemachine/impl/FileListSnapshotInfo.java @@ -0,0 +1,65 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.ratis.statemachine.impl; + +import java.util.Collections; +import java.util.List; + +import org.apache.ratis.server.protocol.TermIndex; +import org.apache.ratis.server.storage.FileInfo; +import org.apache.ratis.statemachine.SnapshotInfo; + +/** + * Each snapshot has a list of files. + * + * The objects of this class are immutable. + */ +public class FileListSnapshotInfo implements SnapshotInfo { + private final TermIndex termIndex; + private final List<FileInfo> files; + + public FileListSnapshotInfo(List<FileInfo> files, long term, long index) { + this.termIndex = TermIndex.newTermIndex(term, index); + this.files = Collections.unmodifiableList(files); + } + + @Override + public TermIndex getTermIndex() { + return termIndex; + } + + @Override + public long getTerm() { + return termIndex.getTerm(); + } + + @Override + public long getIndex() { + return termIndex.getIndex(); + } + + @Override + public List<FileInfo> getFiles() { + return files; + } + + @Override + public String toString() { + return termIndex + ":" + files; + } +} http://git-wip-us.apache.org/repos/asf/incubator-ratis/blob/2d99e617/ratis-server/src/main/java/org/apache/ratis/statemachine/impl/SimpleStateMachineStorage.java ---------------------------------------------------------------------- diff --git a/ratis-server/src/main/java/org/apache/ratis/statemachine/impl/SimpleStateMachineStorage.java b/ratis-server/src/main/java/org/apache/ratis/statemachine/impl/SimpleStateMachineStorage.java new file mode 100644 index 0000000..78605b4 --- /dev/null +++ b/ratis-server/src/main/java/org/apache/ratis/statemachine/impl/SimpleStateMachineStorage.java @@ -0,0 +1,133 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.ratis.statemachine.impl; + +import org.apache.ratis.io.MD5Hash; +import org.apache.ratis.server.protocol.TermIndex; +import org.apache.ratis.server.storage.FileInfo; +import org.apache.ratis.server.storage.RaftStorage; +import org.apache.ratis.statemachine.StateMachineStorage; +import org.apache.ratis.util.AtomicFileOutputStream; +import org.apache.ratis.util.MD5FileUtil; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.File; +import java.io.IOException; +import java.nio.file.DirectoryStream; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.regex.Matcher; +import java.util.regex.Pattern; + +/** + * A StateMachineStorage that stores the snapshot in a single file. + */ +public class SimpleStateMachineStorage implements StateMachineStorage { + + private static final Logger LOG = LoggerFactory.getLogger(SimpleStateMachineStorage.class); + + static final String SNAPSHOT_FILE_PREFIX = "snapshot"; + static final String CORRUPT_SNAPSHOT_FILE_SUFFIX = ".corrupt"; + /** snapshot.term_index */ + static final Pattern SNAPSHOT_REGEX = + Pattern.compile(SNAPSHOT_FILE_PREFIX + "\\.(\\d+)_(\\d+)"); + + private RaftStorage raftStorage; + private File smDir = null; + + private volatile SingleFileSnapshotInfo currentSnapshot = null; + + @Override + public void init(RaftStorage raftStorage) throws IOException { + this.raftStorage = raftStorage; + this.smDir = raftStorage.getStorageDir().getStateMachineDir(); + loadLatestSnapshot(); + } + + @Override + public void format() throws IOException { + // TODO + } + + public static TermIndex getTermIndexFromSnapshotFile(File file) { + final String name = file.getName(); + final Matcher m = SNAPSHOT_REGEX.matcher(name); + if (!m.matches()) { + throw new IllegalArgumentException("File \"" + file + + "\" does not match snapshot file name pattern \"" + + SNAPSHOT_REGEX + "\""); + } + final long term = Long.parseLong(m.group(1)); + final long index = Long.parseLong(m.group(2)); + return TermIndex.newTermIndex(term, index); + } + + protected static String getTmpSnapshotFileName(long term, long endIndex) { + return getSnapshotFileName(term, endIndex) + AtomicFileOutputStream.TMP_EXTENSION; + } + + protected static String getCorruptSnapshotFileName(long term, long endIndex) { + return getSnapshotFileName(term, endIndex) + CORRUPT_SNAPSHOT_FILE_SUFFIX; + } + + public File getSnapshotFile(long term, long endIndex) { + return new File(smDir, getSnapshotFileName(term, endIndex)); + } + + protected File getTmpSnapshotFile(long term, long endIndex) { + return new File(smDir, getTmpSnapshotFileName(term, endIndex)); + } + + protected File getCorruptSnapshotFile(long term, long endIndex) { + return new File(smDir, getCorruptSnapshotFileName(term, endIndex)); + } + + public SingleFileSnapshotInfo findLatestSnapshot() throws IOException { + SingleFileSnapshotInfo latest = null; + try (DirectoryStream<Path> stream = + Files.newDirectoryStream(smDir.toPath())) { + for (Path path : stream) { + Matcher matcher = SNAPSHOT_REGEX.matcher(path.getFileName().toString()); + if (matcher.matches()) { + final long endIndex = Long.parseLong(matcher.group(2)); + if (latest == null || endIndex > latest.getIndex()) { + final long term = Long.parseLong(matcher.group(1)); + MD5Hash fileDigest = MD5FileUtil.readStoredMd5ForFile(path.toFile()); + final FileInfo fileInfo = new FileInfo(path, fileDigest); + latest = new SingleFileSnapshotInfo(fileInfo, term, endIndex); + } + } + } + } + return latest; + } + + public void loadLatestSnapshot() throws IOException { + this.currentSnapshot = findLatestSnapshot(); + } + + public static String getSnapshotFileName(long term, long endIndex) { + return SNAPSHOT_FILE_PREFIX + "." + term + "_" + endIndex; + } + + @Override + public SingleFileSnapshotInfo getLatestSnapshot() { + return currentSnapshot; + } +} http://git-wip-us.apache.org/repos/asf/incubator-ratis/blob/2d99e617/ratis-server/src/main/java/org/apache/ratis/statemachine/impl/SingleFileSnapshotInfo.java ---------------------------------------------------------------------- diff --git a/ratis-server/src/main/java/org/apache/ratis/statemachine/impl/SingleFileSnapshotInfo.java b/ratis-server/src/main/java/org/apache/ratis/statemachine/impl/SingleFileSnapshotInfo.java new file mode 100644 index 0000000..797db17 --- /dev/null +++ b/ratis-server/src/main/java/org/apache/ratis/statemachine/impl/SingleFileSnapshotInfo.java @@ -0,0 +1,38 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.ratis.statemachine.impl; + +import java.util.Arrays; + +import org.apache.ratis.server.storage.FileInfo; + +/** + * Each snapshot only has a single file. + * + * The objects of this class are immutable. + */ +public class SingleFileSnapshotInfo extends FileListSnapshotInfo { + public SingleFileSnapshotInfo(FileInfo fileInfo, long term, long endIndex) { + super(Arrays.asList(fileInfo), term, endIndex); + } + + /** @return the file associated with the snapshot. */ + public FileInfo getFile() { + return getFiles().get(0); + } +} http://git-wip-us.apache.org/repos/asf/incubator-ratis/blob/2d99e617/ratis-server/src/main/java/org/apache/ratis/statemachine/impl/TransactionContextImpl.java ---------------------------------------------------------------------- diff --git a/ratis-server/src/main/java/org/apache/ratis/statemachine/impl/TransactionContextImpl.java b/ratis-server/src/main/java/org/apache/ratis/statemachine/impl/TransactionContextImpl.java new file mode 100644 index 0000000..38bfb9a --- /dev/null +++ b/ratis-server/src/main/java/org/apache/ratis/statemachine/impl/TransactionContextImpl.java @@ -0,0 +1,200 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.ratis.statemachine.impl; + +import java.io.IOException; +import java.util.Objects; +import org.apache.ratis.protocol.RaftClientRequest; +import org.apache.ratis.shaded.proto.RaftProtos.LogEntryProto; +import org.apache.ratis.shaded.proto.RaftProtos.LogEntryProto.LogEntryBodyCase; +import org.apache.ratis.shaded.proto.RaftProtos.SMLogEntryProto; +import org.apache.ratis.statemachine.StateMachine; +import org.apache.ratis.statemachine.TransactionContext; +import org.apache.ratis.util.Preconditions; + +/** + * Implementation of {@link TransactionContext} + */ +public class TransactionContextImpl implements TransactionContext { + + /** The {@link StateMachine} that originated the transaction. */ + private final StateMachine stateMachine; + + /** Original request from the client */ + private RaftClientRequest clientRequest; + + /** Exception from the {@link StateMachine} or from the log */ + private Exception exception; + + /** Data from the {@link StateMachine} */ + private SMLogEntryProto smLogEntryProto; + + /** + * Context specific to the state machine. + * The {@link StateMachine} can use this object to carry state between + * {@link StateMachine#startTransaction(RaftClientRequest)} and + * {@link StateMachine#applyTransaction(TransactionContext)}. + */ + private Object stateMachineContext; + + /** + * Whether to commit the transaction to the RAFT Log. + * In some cases the {@link StateMachine} may want to indicate + * that the transaction should not be committed + */ + private boolean shouldCommit = true; + + /** Committed LogEntry. */ + private LogEntryProto logEntry; + + private TransactionContextImpl(StateMachine stateMachine) { + this.stateMachine = stateMachine; + } + + /** The same as this(stateMachine, clientRequest, smLogEntryProto, null). */ + public TransactionContextImpl( + StateMachine stateMachine, RaftClientRequest clientRequest, + SMLogEntryProto smLogEntryProto) { + this(stateMachine, clientRequest, smLogEntryProto, null); + } + + /** + * Construct a {@link TransactionContext} from a client request. + * Used by the state machine to start a transaction + * and send the Log entry representing the transaction data + * to be applied to the raft log. + */ + public TransactionContextImpl( + StateMachine stateMachine, RaftClientRequest clientRequest, + SMLogEntryProto smLogEntryProto, Object stateMachineContext) { + this(stateMachine); + this.clientRequest = clientRequest; + this.smLogEntryProto = smLogEntryProto; + this.stateMachineContext = stateMachineContext; + } + + /** The same as this(stateMachine, clientRequest, exception, null). */ + public TransactionContextImpl ( + StateMachine stateMachine, RaftClientRequest clientRequest, + Exception exception) { + this(stateMachine, clientRequest, exception, null); + } + + /** + * Construct a {@link TransactionContext} from a client request to signal + * an exception so that the RAFT server will fail the request on behalf + * of the {@link StateMachine}. + */ + public TransactionContextImpl( + StateMachine stateMachine, RaftClientRequest clientRequest, + Exception exception, Object stateMachineContext) { + this(stateMachine); + this.clientRequest = clientRequest; + this.exception = exception; + this.stateMachineContext = stateMachineContext; + } + + /** + * Construct a {@link TransactionContext} from a {@link LogEntryProto}. + * Used by followers for applying committed entries to the state machine. + * @param logEntry the log entry to be applied + */ + public TransactionContextImpl(StateMachine stateMachine, LogEntryProto logEntry) { + this(stateMachine); + setLogEntry(logEntry); + this.smLogEntryProto = logEntry.getSmLogEntry(); + } + + @Override + public RaftClientRequest getClientRequest() { + return clientRequest; + } + + @Override + public SMLogEntryProto getSMLogEntry() { + return smLogEntryProto; + } + + @Override + public Exception getException() { + return exception; + } + + @Override + public TransactionContext setStateMachineContext(Object stateMachineContext) { + this.stateMachineContext = stateMachineContext; + return this; + } + + @Override + public Object getStateMachineContext() { + return stateMachineContext; + } + + @Override + public TransactionContext setLogEntry(LogEntryProto logEntry) { + Objects.requireNonNull(logEntry, "logEntry == null"); + Preconditions.assertTrue(logEntry.getLogEntryBodyCase() == LogEntryBodyCase.SMLOGENTRY, + () -> "LogEntryBodyCase = " + logEntry.getLogEntryBodyCase() + + " != " + LogEntryBodyCase.SMLOGENTRY + ", logEntry=" + logEntry); + Preconditions.assertTrue(this.logEntry == null, "this.logEntry != null"); + this.logEntry = logEntry; + return this; + } + + @Override + public TransactionContext setSmLogEntryProto(SMLogEntryProto smLogEntryProto) { + this.smLogEntryProto = smLogEntryProto; + return this; + } + + @Override + public LogEntryProto getLogEntry() { + return logEntry; + } + + private TransactionContext setException(IOException ioe) { + assert exception != null; + this.exception = ioe; + return this; + } + + @Override + public TransactionContext setShouldCommit(boolean shouldCommit) { + this.shouldCommit = shouldCommit; + return this; + } + + @Override + public boolean shouldCommit() { + // TODO: Hook this up in the server to bypass the RAFT Log and send back a response to client + return this.shouldCommit; + } + + @Override + public TransactionContext preAppendTransaction() throws IOException { + return stateMachine.preAppendTransaction(this); + } + + @Override + public TransactionContext cancelTransaction() throws IOException { + // TODO: This is not called from Raft server / log yet. When an IOException happens, we should + // call this to let the SM know that Transaction cannot be synced + return stateMachine.cancelTransaction(this); + } +} http://git-wip-us.apache.org/repos/asf/incubator-ratis/blob/2d99e617/ratis-server/src/test/java/org/apache/ratis/MiniRaftCluster.java ---------------------------------------------------------------------- diff --git a/ratis-server/src/test/java/org/apache/ratis/MiniRaftCluster.java b/ratis-server/src/test/java/org/apache/ratis/MiniRaftCluster.java index 26df59c..f4df193 100644 --- a/ratis-server/src/test/java/org/apache/ratis/MiniRaftCluster.java +++ b/ratis-server/src/test/java/org/apache/ratis/MiniRaftCluster.java @@ -28,7 +28,7 @@ import org.apache.ratis.server.impl.RaftServerImpl; import org.apache.ratis.server.impl.RaftServerProxy; import org.apache.ratis.server.storage.MemoryRaftLog; import org.apache.ratis.server.storage.RaftLog; -import org.apache.ratis.statemachine.BaseStateMachine; +import org.apache.ratis.statemachine.impl.BaseStateMachine; import org.apache.ratis.statemachine.StateMachine; import org.apache.ratis.util.*; import org.slf4j.Logger; http://git-wip-us.apache.org/repos/asf/incubator-ratis/blob/2d99e617/ratis-server/src/test/java/org/apache/ratis/server/storage/TestRaftStorage.java ---------------------------------------------------------------------- diff --git a/ratis-server/src/test/java/org/apache/ratis/server/storage/TestRaftStorage.java b/ratis-server/src/test/java/org/apache/ratis/server/storage/TestRaftStorage.java index 6c14123..4a26f8c 100644 --- a/ratis-server/src/test/java/org/apache/ratis/server/storage/TestRaftStorage.java +++ b/ratis-server/src/test/java/org/apache/ratis/server/storage/TestRaftStorage.java @@ -21,7 +21,7 @@ import org.apache.ratis.BaseTest; import org.apache.ratis.server.impl.RaftServerConstants.StartupOption; import org.apache.ratis.server.protocol.TermIndex; import org.apache.ratis.server.storage.RaftStorageDirectory.StorageState; -import org.apache.ratis.statemachine.SimpleStateMachineStorage; +import org.apache.ratis.statemachine.impl.SimpleStateMachineStorage; import org.apache.ratis.util.FileUtils; import org.junit.After; import org.junit.Assert; http://git-wip-us.apache.org/repos/asf/incubator-ratis/blob/2d99e617/ratis-server/src/test/java/org/apache/ratis/statemachine/SimpleStateMachine4Testing.java ---------------------------------------------------------------------- diff --git a/ratis-server/src/test/java/org/apache/ratis/statemachine/SimpleStateMachine4Testing.java b/ratis-server/src/test/java/org/apache/ratis/statemachine/SimpleStateMachine4Testing.java index d041fcf..8d53faf 100644 --- a/ratis-server/src/test/java/org/apache/ratis/statemachine/SimpleStateMachine4Testing.java +++ b/ratis-server/src/test/java/org/apache/ratis/statemachine/SimpleStateMachine4Testing.java @@ -34,6 +34,10 @@ import org.apache.ratis.server.storage.LogOutputStream; import org.apache.ratis.server.storage.RaftStorage; import org.apache.ratis.shaded.proto.RaftProtos.LogEntryProto; import org.apache.ratis.shaded.proto.RaftProtos.SMLogEntryProto; +import org.apache.ratis.statemachine.impl.BaseStateMachine; +import org.apache.ratis.statemachine.impl.SimpleStateMachineStorage; +import org.apache.ratis.statemachine.impl.SingleFileSnapshotInfo; +import org.apache.ratis.statemachine.impl.TransactionContextImpl; import org.apache.ratis.util.Daemon; import org.apache.ratis.util.LifeCycle; import org.apache.ratis.util.MD5FileUtil; @@ -238,7 +242,7 @@ public class SimpleStateMachine4Testing extends BaseStateMachine { Thread.currentThread().interrupt(); } } - return new TransactionContext(this, request, SMLogEntryProto.newBuilder() + return new TransactionContextImpl(this, request, SMLogEntryProto.newBuilder() .setData(request.getMessage().getContent()) .build()); } http://git-wip-us.apache.org/repos/asf/incubator-ratis/blob/2d99e617/ratis-server/src/test/java/org/apache/ratis/statemachine/TestStateMachine.java ---------------------------------------------------------------------- diff --git a/ratis-server/src/test/java/org/apache/ratis/statemachine/TestStateMachine.java b/ratis-server/src/test/java/org/apache/ratis/statemachine/TestStateMachine.java index 73ce69d..611435a 100644 --- a/ratis-server/src/test/java/org/apache/ratis/statemachine/TestStateMachine.java +++ b/ratis-server/src/test/java/org/apache/ratis/statemachine/TestStateMachine.java @@ -31,6 +31,7 @@ import org.apache.ratis.server.impl.RaftServerImpl; import org.apache.ratis.server.impl.RaftServerProxy; import org.apache.ratis.server.simulation.MiniRaftClusterWithSimulatedRpc; import org.apache.ratis.shaded.proto.RaftProtos.SMLogEntryProto; +import org.apache.ratis.statemachine.impl.TransactionContextImpl; import org.apache.ratis.util.LogUtils; import org.junit.*; @@ -109,7 +110,7 @@ public class TestStateMachine extends BaseTest { // only leader will get this call isLeader.set(true); // send the next transaction id as the "context" from SM - return new TransactionContext(this, request, SMLogEntryProto.newBuilder() + return new TransactionContextImpl(this, request, SMLogEntryProto.newBuilder() .setData(request.getMessage().getContent()) .build(), transactions.incrementAndGet()); }
