You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ratis.apache.org by ji...@apache.org on 2017/01/31 21:16:54 UTC
[09/54] [abbrv] incubator-ratis git commit: Change RaftServerRpc to
extend RaftServerProtocol and Closeable.
Change RaftServerRpc to extend RaftServerProtocol and Closeable.
Project: http://git-wip-us.apache.org/repos/asf/incubator-ratis/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-ratis/commit/673a2827
Tree: http://git-wip-us.apache.org/repos/asf/incubator-ratis/tree/673a2827
Diff: http://git-wip-us.apache.org/repos/asf/incubator-ratis/diff/673a2827
Branch: refs/heads/master
Commit: 673a28278346fb641f7c1e81c20bc37168be33f0
Parents: 56e9b71
Author: Tsz-Wo Nicholas Sze <sz...@hortonworks.com>
Authored: Mon Jan 2 23:59:14 2017 +0800
Committer: Tsz-Wo Nicholas Sze <sz...@hortonworks.com>
Committed: Mon Jan 2 23:59:14 2017 +0800
----------------------------------------------------------------------
.../org/apache/raft/grpc/RaftGRpcService.java | 10 ++++-----
.../raft/hadooprpc/server/HadoopRpcService.java | 8 +++----
.../raft/netty/server/NettyRpcService.java | 8 +++----
.../org/apache/raft/server/RaftServerRpc.java | 23 ++++++++------------
.../apache/raft/server/impl/LeaderElection.java | 2 +-
.../apache/raft/server/impl/LogAppender.java | 4 ++--
.../apache/raft/server/impl/RaftServerImpl.java | 11 ++++------
.../apache/raft/server/impl/ServerState.java | 2 +-
.../server/simulation/SimulatedServerRpc.java | 8 +++----
9 files changed, 34 insertions(+), 42 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-ratis/blob/673a2827/raft-grpc/src/main/java/org/apache/raft/grpc/RaftGRpcService.java
----------------------------------------------------------------------
diff --git a/raft-grpc/src/main/java/org/apache/raft/grpc/RaftGRpcService.java b/raft-grpc/src/main/java/org/apache/raft/grpc/RaftGRpcService.java
index f3d894a..d0c98c3 100644
--- a/raft-grpc/src/main/java/org/apache/raft/grpc/RaftGRpcService.java
+++ b/raft-grpc/src/main/java/org/apache/raft/grpc/RaftGRpcService.java
@@ -90,14 +90,14 @@ public class RaftGRpcService implements RaftServerRpc {
@Override
public void run() {
System.err.println("*** shutting down gRPC server since JVM is shutting down");
- RaftGRpcService.this.shutdown();
+ RaftGRpcService.this.close();
System.err.println("*** server shut down");
}
});
}
@Override
- public void shutdown() {
+ public void close() {
if (server != null) {
server.shutdown();
}
@@ -110,21 +110,21 @@ public class RaftGRpcService implements RaftServerRpc {
}
@Override
- public AppendEntriesReplyProto sendAppendEntries(
+ public AppendEntriesReplyProto appendEntries(
AppendEntriesRequestProto request) throws IOException {
throw new UnsupportedOperationException(
"Blocking AppendEntries call is not supported");
}
@Override
- public InstallSnapshotReplyProto sendInstallSnapshot(
+ public InstallSnapshotReplyProto installSnapshot(
InstallSnapshotRequestProto request) throws IOException {
throw new UnsupportedOperationException(
"Blocking InstallSnapshot call is not supported");
}
@Override
- public RequestVoteReplyProto sendRequestVote(RequestVoteRequestProto request)
+ public RequestVoteReplyProto requestVote(RequestVoteRequestProto request)
throws IOException {
CodeInjectionForTesting.execute(GRPC_SEND_SERVER_REQUEST, selfId,
null, request);
http://git-wip-us.apache.org/repos/asf/incubator-ratis/blob/673a2827/raft-hadoop/src/main/java/org/apache/raft/hadooprpc/server/HadoopRpcService.java
----------------------------------------------------------------------
diff --git a/raft-hadoop/src/main/java/org/apache/raft/hadooprpc/server/HadoopRpcService.java b/raft-hadoop/src/main/java/org/apache/raft/hadooprpc/server/HadoopRpcService.java
index 3330d78..b73deca 100644
--- a/raft-hadoop/src/main/java/org/apache/raft/hadooprpc/server/HadoopRpcService.java
+++ b/raft-hadoop/src/main/java/org/apache/raft/hadooprpc/server/HadoopRpcService.java
@@ -122,12 +122,12 @@ public class HadoopRpcService implements RaftServerRpc {
}
@Override
- public void shutdown() {
+ public void close() {
ipcServer.stop();
}
@Override
- public AppendEntriesReplyProto sendAppendEntries(
+ public AppendEntriesReplyProto appendEntries(
AppendEntriesRequestProto request) throws IOException {
Preconditions.checkArgument(id.equals(request.getServerRequest().getRequestorId()));
CodeInjectionForTesting.execute(SEND_SERVER_REQUEST, id, null, request);
@@ -142,7 +142,7 @@ public class HadoopRpcService implements RaftServerRpc {
}
@Override
- public InstallSnapshotReplyProto sendInstallSnapshot(
+ public InstallSnapshotReplyProto installSnapshot(
InstallSnapshotRequestProto request) throws IOException {
Preconditions.checkArgument(id.equals(request.getServerRequest().getRequestorId()));
CodeInjectionForTesting.execute(SEND_SERVER_REQUEST, id, null, request);
@@ -157,7 +157,7 @@ public class HadoopRpcService implements RaftServerRpc {
}
@Override
- public RequestVoteReplyProto sendRequestVote(
+ public RequestVoteReplyProto requestVote(
RequestVoteRequestProto request) throws IOException {
Preconditions.checkArgument(id.equals(request.getServerRequest().getRequestorId()));
CodeInjectionForTesting.execute(SEND_SERVER_REQUEST, id, null, request);
http://git-wip-us.apache.org/repos/asf/incubator-ratis/blob/673a2827/raft-netty/src/main/java/org/apache/raft/netty/server/NettyRpcService.java
----------------------------------------------------------------------
diff --git a/raft-netty/src/main/java/org/apache/raft/netty/server/NettyRpcService.java b/raft-netty/src/main/java/org/apache/raft/netty/server/NettyRpcService.java
index 19f5979..08e379a 100644
--- a/raft-netty/src/main/java/org/apache/raft/netty/server/NettyRpcService.java
+++ b/raft-netty/src/main/java/org/apache/raft/netty/server/NettyRpcService.java
@@ -112,7 +112,7 @@ public final class NettyRpcService implements RaftServerRpc {
}
@Override
- public void shutdown() {
+ public void close() {
lifeCycle.checkStateAndClose(() -> {
bossGroup.shutdownGracefully();
workerGroup.shutdownGracefully();
@@ -199,7 +199,7 @@ public final class NettyRpcService implements RaftServerRpc {
}
@Override
- public RequestVoteReplyProto sendRequestVote(RequestVoteRequestProto request) throws IOException {
+ public RequestVoteReplyProto requestVote(RequestVoteRequestProto request) throws IOException {
Preconditions.checkArgument(id.equals(request.getServerRequest().getRequestorId()));
CodeInjectionForTesting.execute(SEND_SERVER_REQUEST, id, null, request);
@@ -211,7 +211,7 @@ public final class NettyRpcService implements RaftServerRpc {
}
@Override
- public AppendEntriesReplyProto sendAppendEntries(AppendEntriesRequestProto request) throws IOException {
+ public AppendEntriesReplyProto appendEntries(AppendEntriesRequestProto request) throws IOException {
Preconditions.checkArgument(id.equals(request.getServerRequest().getRequestorId()));
CodeInjectionForTesting.execute(SEND_SERVER_REQUEST, id, null, request);
@@ -223,7 +223,7 @@ public final class NettyRpcService implements RaftServerRpc {
}
@Override
- public InstallSnapshotReplyProto sendInstallSnapshot(InstallSnapshotRequestProto request) throws IOException {
+ public InstallSnapshotReplyProto installSnapshot(InstallSnapshotRequestProto request) throws IOException {
Preconditions.checkArgument(id.equals(request.getServerRequest().getRequestorId()));
CodeInjectionForTesting.execute(SEND_SERVER_REQUEST, id, null, request);
http://git-wip-us.apache.org/repos/asf/incubator-ratis/blob/673a2827/raft-server/src/main/java/org/apache/raft/server/RaftServerRpc.java
----------------------------------------------------------------------
diff --git a/raft-server/src/main/java/org/apache/raft/server/RaftServerRpc.java b/raft-server/src/main/java/org/apache/raft/server/RaftServerRpc.java
index de81ec2..6526bea 100644
--- a/raft-server/src/main/java/org/apache/raft/server/RaftServerRpc.java
+++ b/raft-server/src/main/java/org/apache/raft/server/RaftServerRpc.java
@@ -18,27 +18,22 @@
package org.apache.raft.server;
import org.apache.raft.protocol.RaftPeer;
-import org.apache.raft.shaded.proto.RaftProtos.*;
+import org.apache.raft.server.protocol.RaftServerProtocol;
-import java.io.IOException;
+import java.io.Closeable;
import java.net.InetSocketAddress;
-public interface RaftServerRpc {
+/**
+ * An server-side interface for supporting different RPC implementations
+ * such as Netty, gRPC and Hadoop.
+ */
+public interface RaftServerRpc extends RaftServerProtocol, Closeable {
+ /** Start the RPC service. */
void start();
- void shutdown();
-
+ /** @return the address where this RPC server is listening to. */
InetSocketAddress getInetSocketAddress();
- AppendEntriesReplyProto sendAppendEntries(
- AppendEntriesRequestProto request) throws IOException;
-
- InstallSnapshotReplyProto sendInstallSnapshot(
- InstallSnapshotRequestProto request) throws IOException;
-
- RequestVoteReplyProto sendRequestVote(RequestVoteRequestProto request)
- throws IOException;
-
/** add information of the given peers */
void addPeers(Iterable<RaftPeer> peers);
}
http://git-wip-us.apache.org/repos/asf/incubator-ratis/blob/673a2827/raft-server/src/main/java/org/apache/raft/server/impl/LeaderElection.java
----------------------------------------------------------------------
diff --git a/raft-server/src/main/java/org/apache/raft/server/impl/LeaderElection.java b/raft-server/src/main/java/org/apache/raft/server/impl/LeaderElection.java
index 8552029..39bdb13 100644
--- a/raft-server/src/main/java/org/apache/raft/server/impl/LeaderElection.java
+++ b/raft-server/src/main/java/org/apache/raft/server/impl/LeaderElection.java
@@ -187,7 +187,7 @@ class LeaderElection extends Daemon {
final RequestVoteRequestProto r = server.createRequestVoteRequest(
peer.getId(), electionTerm, lastEntry);
service.submit(
- () -> server.getServerRpc().sendRequestVote(r));
+ () -> server.getServerRpc().requestVote(r));
submitted++;
}
return submitted;
http://git-wip-us.apache.org/repos/asf/incubator-ratis/blob/673a2827/raft-server/src/main/java/org/apache/raft/server/impl/LogAppender.java
----------------------------------------------------------------------
diff --git a/raft-server/src/main/java/org/apache/raft/server/impl/LogAppender.java b/raft-server/src/main/java/org/apache/raft/server/impl/LogAppender.java
index 3b18f13..cf613ca 100644
--- a/raft-server/src/main/java/org/apache/raft/server/impl/LogAppender.java
+++ b/raft-server/src/main/java/org/apache/raft/server/impl/LogAppender.java
@@ -198,7 +198,7 @@ public class LogAppender extends Daemon {
follower.updateLastRpcSendTime();
final AppendEntriesReplyProto r = server.getServerRpc()
- .sendAppendEntries(request);
+ .appendEntries(request);
follower.updateLastRpcResponseTime();
return r;
@@ -328,7 +328,7 @@ public class LogAppender extends Daemon {
for (InstallSnapshotRequestProto request :
new SnapshotRequestIter(snapshot, requestId)) {
follower.updateLastRpcSendTime();
- reply = server.getServerRpc().sendInstallSnapshot(request);
+ reply = server.getServerRpc().installSnapshot(request);
follower.updateLastRpcResponseTime();
if (!reply.getServerReply().getSuccess()) {
http://git-wip-us.apache.org/repos/asf/incubator-ratis/blob/673a2827/raft-server/src/main/java/org/apache/raft/server/impl/RaftServerImpl.java
----------------------------------------------------------------------
diff --git a/raft-server/src/main/java/org/apache/raft/server/impl/RaftServerImpl.java b/raft-server/src/main/java/org/apache/raft/server/impl/RaftServerImpl.java
index 131e002..3026afa 100644
--- a/raft-server/src/main/java/org/apache/raft/server/impl/RaftServerImpl.java
+++ b/raft-server/src/main/java/org/apache/raft/server/impl/RaftServerImpl.java
@@ -132,12 +132,13 @@ public class RaftServerImpl implements RaftServer {
this.state.setInitialConf(conf);
}
+ @Override
public void setServerRpc(RaftServerRpc serverRpc) {
this.serverRpc = serverRpc;
// add peers into rpc service
RaftConfiguration conf = getRaftConf();
if (conf != null) {
- addPeersToRPC(conf.getPeers());
+ serverRpc.addPeers(conf.getPeers());
}
}
@@ -201,7 +202,7 @@ public class RaftServerImpl implements RaftServer {
shutdownElectionDaemon();
shutdownLeaderState();
- serverRpc.shutdown();
+ serverRpc.close();
state.close();
} catch (Exception ignored) {
LOG.warn("Failed to kill " + state.getSelfId(), ignored);
@@ -418,7 +419,7 @@ public class RaftServerImpl implements RaftServer {
}
// add new peers into the rpc service
- addPeersToRPC(Arrays.asList(peersInNewConf));
+ getServerRpc().addPeers(Arrays.asList(peersInNewConf));
// add staging state into the leaderState
pending = leaderState.startSetConfiguration(request);
}
@@ -726,10 +727,6 @@ public class RaftServerImpl implements RaftServer {
}
}
- public void addPeersToRPC(Iterable<RaftPeer> peers) {
- serverRpc.addPeers(peers);
- }
-
synchronized void replyPendingRequest(long logIndex,
CompletableFuture<Message> message) {
if (isLeader() && leaderState != null) { // is leader and is running
http://git-wip-us.apache.org/repos/asf/incubator-ratis/blob/673a2827/raft-server/src/main/java/org/apache/raft/server/impl/ServerState.java
----------------------------------------------------------------------
diff --git a/raft-server/src/main/java/org/apache/raft/server/impl/ServerState.java b/raft-server/src/main/java/org/apache/raft/server/impl/ServerState.java
index c91968c..8611101 100644
--- a/raft-server/src/main/java/org/apache/raft/server/impl/ServerState.java
+++ b/raft-server/src/main/java/org/apache/raft/server/impl/ServerState.java
@@ -285,7 +285,7 @@ public class ServerState implements Closeable {
final RaftConfiguration conf = ServerProtoUtils.toRaftConfiguration(
entry.getIndex(), entry.getConfigurationEntry());
configurationManager.addConfiguration(entry.getIndex(), conf);
- server.addPeersToRPC(conf.getPeers());
+ server.getServerRpc().addPeers(conf.getPeers());
}
}
}
http://git-wip-us.apache.org/repos/asf/incubator-ratis/blob/673a2827/raft-server/src/test/java/org/apache/raft/server/simulation/SimulatedServerRpc.java
----------------------------------------------------------------------
diff --git a/raft-server/src/test/java/org/apache/raft/server/simulation/SimulatedServerRpc.java b/raft-server/src/test/java/org/apache/raft/server/simulation/SimulatedServerRpc.java
index cc3fb35..8a7e752 100644
--- a/raft-server/src/test/java/org/apache/raft/server/simulation/SimulatedServerRpc.java
+++ b/raft-server/src/test/java/org/apache/raft/server/simulation/SimulatedServerRpc.java
@@ -69,7 +69,7 @@ class SimulatedServerRpc implements RaftServerRpc {
}
@Override
- public void shutdown() {
+ public void close() {
try {
interruptAndJoin();
executor.shutdown();
@@ -86,7 +86,7 @@ class SimulatedServerRpc implements RaftServerRpc {
}
@Override
- public AppendEntriesReplyProto sendAppendEntries(AppendEntriesRequestProto request)
+ public AppendEntriesReplyProto appendEntries(AppendEntriesRequestProto request)
throws IOException {
RaftServerReply reply = serverHandler.getRpc()
.sendRequest(new RaftServerRequest(request));
@@ -94,7 +94,7 @@ class SimulatedServerRpc implements RaftServerRpc {
}
@Override
- public InstallSnapshotReplyProto sendInstallSnapshot(InstallSnapshotRequestProto request)
+ public InstallSnapshotReplyProto installSnapshot(InstallSnapshotRequestProto request)
throws IOException {
RaftServerReply reply = serverHandler.getRpc()
.sendRequest(new RaftServerRequest(request));
@@ -102,7 +102,7 @@ class SimulatedServerRpc implements RaftServerRpc {
}
@Override
- public RequestVoteReplyProto sendRequestVote(RequestVoteRequestProto request)
+ public RequestVoteReplyProto requestVote(RequestVoteRequestProto request)
throws IOException {
RaftServerReply reply = serverHandler.getRpc()
.sendRequest(new RaftServerRequest(request));