Merge branch 'master' of https://github.com/chrislusf/raft

seaweedfs · Jun 23, 2022 · 3936262 · 3936262
2 parents 0b95ac8 + 2be29d9
commit 3936262
Show file tree

Hide file tree

Showing 2 changed files with 13 additions and 11 deletions.
diff --git a/grpc_transporter.go b/grpc_transporter.go
@@ -19,13 +19,13 @@ var (
 // An GrpcTransporter is a default transport layer used to communicate between
 // multiple servers.
 type GrpcTransporter struct {
- grpcDialOption grpc.DialOption
+ grpcDialOptions []grpc.DialOption
 }
 
 // Creates a new HTTP transporter with the given path prefix.
-func NewGrpcTransporter(grpcDialOption grpc.DialOption) *GrpcTransporter {
+func NewGrpcTransporter(grpcDialOptions ...grpc.DialOption) *GrpcTransporter {
  t := &GrpcTransporter{
- grpcDialOption: grpcDialOption,
+ grpcDialOptions: grpcDialOptions,
  }
  return t
 }
@@ -49,7 +49,7 @@ func NewGrpcServer(server Server) *GrpcServer {
 // Sends an AppendEntries RPC to a peer.
 func (t *GrpcTransporter) SendAppendEntriesRequest(server Server, peer *Peer, req *AppendEntriesRequest) (ret *AppendEntriesResponse) {
 
- err := withRaftServerClient(peer.ConnectionString, t.grpcDialOption, func(client protobuf.RaftClient) error {
+ err := withRaftServerClient(peer.ConnectionString, t.grpcDialOptions, func(client protobuf.RaftClient) error {
  ctx, cancel := context2.WithTimeout(context2.Background(), time.Duration(5*time.Second))
  defer cancel()
 
@@ -83,7 +83,7 @@ func (t *GrpcTransporter) SendAppendEntriesRequest(server Server, peer *Peer, re
 // Sends a RequestVote RPC to a peer.
 func (t *GrpcTransporter) SendVoteRequest(server Server, peer *Peer, req *RequestVoteRequest) (ret *RequestVoteResponse) {
 
- err := withRaftServerClient(peer.ConnectionString, t.grpcDialOption, func(client protobuf.RaftClient) error {
+ err := withRaftServerClient(peer.ConnectionString, t.grpcDialOptions, func(client protobuf.RaftClient) error {
  ctx, cancel := context2.WithTimeout(context2.Background(), time.Duration(5*time.Second))
  defer cancel()
 
@@ -117,7 +117,7 @@ func (t *GrpcTransporter) SendVoteRequest(server Server, peer *Peer, req *Reques
 // Sends a SnapshotRequest RPC to a peer.
 func (t *GrpcTransporter) SendSnapshotRequest(server Server, peer *Peer, req *SnapshotRequest) (ret *SnapshotResponse) {
 
- err := withRaftServerClient(peer.ConnectionString, t.grpcDialOption, func(client protobuf.RaftClient) error {
+ err := withRaftServerClient(peer.ConnectionString, t.grpcDialOptions, func(client protobuf.RaftClient) error {
  ctx, cancel := context2.WithTimeout(context2.Background(), time.Duration(5*time.Second))
  defer cancel()
 
@@ -149,7 +149,7 @@ func (t *GrpcTransporter) SendSnapshotRequest(server Server, peer *Peer, req *Sn
 // Sends a SnapshotRequest RPC to a peer.
 func (t *GrpcTransporter) SendSnapshotRecoveryRequest(server Server, peer *Peer, req *SnapshotRecoveryRequest) (ret *SnapshotRecoveryResponse) {
 
- err := withRaftServerClient(peer.ConnectionString, t.grpcDialOption, func(client protobuf.RaftClient) error {
+ err := withRaftServerClient(peer.ConnectionString, t.grpcDialOptions, func(client protobuf.RaftClient) error {
  ctx, cancel := context2.WithTimeout(context2.Background(), time.Duration(5*time.Second))
  defer cancel()
 
@@ -285,12 +285,12 @@ func (t *GrpcServer) OnSendSnapshotRecoveryRequest(ctx context2.Context, pbReq *
  }, nil
 }
 
-func withRaftServerClient(raftServer string, grpcDialOption grpc.DialOption, fn func(protobuf.RaftClient) error) error {
+func withRaftServerClient(raftServer string, grpcDialOptions []grpc.DialOption, fn func(protobuf.RaftClient) error) error {
 
  return withCachedGrpcClient(func(grpcConnection *grpc.ClientConn) error {
  client := protobuf.NewRaftClient(grpcConnection)
  return fn(client)
- }, raftServer, grpcDialOption)
+ }, raftServer, grpcDialOptions...)
 
 }
 

diff --git a/server.go b/server.go
@@ -661,7 +661,7 @@ func (s *server) sendAsync(value interface{}) {
 }
 
 func (s *server) checkQuorumActive(timeout time.Duration) bool {
- s.debugln("check.quorum.active")
+ s.debugln("check.quorum.active with quorum size: ", s.QuorumSize(), " member count: ", s.MemberCount())
  act := 1
  now := time.Now()
  for _, peer := range s.peers {
@@ -1267,7 +1267,9 @@ func (s *server) TakeSnapshot() error {
  // Attach snapshot to pending snapshot and save it to disk.
  s.pendingSnapshot.Peers = peers
  s.pendingSnapshot.State = state
- s.saveSnapshot()
+ if err := s.saveSnapshot(); err != nil {
+ return err
+ }
 
  // We keep some log entries after the snapshot.
  // We do not want to send the whole snapshot to the slightly slow machines