Skip to content

Commit

Permalink
etcdserver: adjust election ticks on restart
Browse files Browse the repository at this point in the history
Signed-off-by: Gyuho Lee <gyuhox@gmail.com>
  • Loading branch information
gyuho committed Mar 11, 2018
1 parent 7910dcf commit 9f356c8
Show file tree
Hide file tree
Showing 2 changed files with 47 additions and 2 deletions.
12 changes: 10 additions & 2 deletions etcdserver/raft.go
Original file line number Diff line number Diff line change
Expand Up @@ -97,6 +97,7 @@ type raftNode struct {
term uint64
lead uint64

tickMu sync.Mutex
raftNodeConfig

// a chan to send/receive snapshot
Expand Down Expand Up @@ -151,6 +152,13 @@ func newRaftNode(cfg raftNodeConfig) *raftNode {
return r
}

// raft.Node does not have locks in Raft package
func (r *raftNode) tick() {
r.tickMu.Lock()
r.Tick()
r.tickMu.Unlock()
}

// start prepares and starts raftNode in a new goroutine. It is no longer safe
// to modify the fields after it has been started.
func (r *raftNode) start(rh *raftReadyHandler) {
Expand All @@ -163,7 +171,7 @@ func (r *raftNode) start(rh *raftReadyHandler) {
for {
select {
case <-r.ticker.C:
r.Tick()
r.tick()
case rd := <-r.Ready():
if rd.SoftState != nil {
newLeader := rd.SoftState.Lead != raft.None && atomic.LoadUint64(&r.lead) != rd.SoftState.Lead
Expand Down Expand Up @@ -376,7 +384,7 @@ func (r *raftNode) resumeSending() {
// speeding up election process.
func (r *raftNode) advanceTicks(ticks int) {
for i := 0; i < ticks; i++ {
r.Tick()
r.tick()
}
}

Expand Down
37 changes: 37 additions & 0 deletions etcdserver/server.go
Original file line number Diff line number Diff line change
Expand Up @@ -530,12 +530,49 @@ func NewServer(cfg ServerConfig) (srv *EtcdServer, err error) {
return srv, nil
}

func (s *EtcdServer) adjustTicks() {
clusterN := len(s.cluster.Members())

// single-node fresh start, or single-node recovers from snapshot
if clusterN == 1 {
ticks := s.Cfg.ElectionTicks - 1
plog.Infof("%s as single-node; fast-forwarding %d ticks (election ticks %d)", s.ID(), ticks, s.Cfg.ElectionTicks)
s.r.advanceTicks(ticks)
return
}

// retry up to "rafthttp.ConnReadTimeout", which is 5-sec
for i := 0; i < 5; i++ {
select {
case <-time.After(time.Second):
case <-s.stopping:
return
}

peerN := s.r.transport.ActivePeers()
if peerN > 1 {
// multi-node received peer connection reports
// adjust ticks, in case slow leader message receive
ticks := s.Cfg.ElectionTicks - 2
plog.Infof("%s initialzed peer connection; fast-forwarding %d ticks (election ticks %d) with %d active peer(s)", s.ID(), ticks, s.Cfg.ElectionTicks, peerN)
s.r.advanceTicks(ticks)
return
}
}

// 1. all connections failed, or
// 2. no active peers, or
// 3. restarted single-node with no snapshot
// do nothing, because advancing ticks would have no effect
}

// Start performs any initialization of the Server necessary for it to
// begin serving requests. It must be called before Do or Process.
// Start must be non-blocking; any long-running server functionality
// should be implemented in goroutines.
func (s *EtcdServer) Start() {
s.start()
s.goAttach(func() { s.adjustTicks() })
s.goAttach(func() { s.publish(s.Cfg.ReqTimeout()) })
s.goAttach(s.purgeFile)
s.goAttach(func() { monitorFileDescriptor(s.stopping) })
Expand Down

0 comments on commit 9f356c8

Please sign in to comment.