fix the race condition between goroutine and channel on the same leases to be revoked
@@ -842,19 +842,37 @@ func (s *EtcdServer) run() {
f := func(context.Context) { s.applyAll(&ep, &ap) }
sched.Schedule(f)
case leases := <-expiredLeaseC:
+ s.revokeExpiredLeases(leases)
+ case err := <-s.errorc:
+ lg.Warn("server error", zap.Error(err))
+ lg.Warn("data-dir used by this member must be removed")
+ return
+ case <-getSyncC():
+ if s.v2store.HasTTLKeys() {
+ s.sync(s.Cfg.ReqTimeout())
+ }
+ case <-s.stop:
+ return
+ }
+ }
+}
+
+func (s *EtcdServer) revokeExpiredLeases(leases []*lease.Lease) {
s.GoAttach(func() {
+ lg := s.Logger()
// Increases throughput of expired leases deletion process through parallelization
c := make(chan struct{}, maxPendingRevokes)
- for _, lease := range leases {
+ for _, curLease := range leases {
select {
case c <- struct{}{}:
case <-s.stopping:
return
}
- lid := lease.ID
+
+ f := func(lid int64) {
s.GoAttach(func() {
ctx := s.authStore.WithRoot(s.ctx)
- _, lerr := s.LeaseRevoke(ctx, &pb.LeaseRevokeRequest{ID: int64(lid)})
+ _, lerr := s.LeaseRevoke(ctx, &pb.LeaseRevokeRequest{ID: lid})
if lerr == nil {
leaseExpired.Inc()
} else {
@@ -868,19 +886,10 @@ func (s *EtcdServer) run() {
<-c
})
}
- })
- case err := <-s.errorc:
- lg.Warn("server error", zap.Error(err))
- lg.Warn("data-dir used by this member must be removed")
- return
- case <-getSyncC():
- if s.v2store.HasTTLKeys() {
- s.sync(s.Cfg.ReqTimeout())
- }
- case <-s.stop:
- return
- }
+
+ f(int64(curLease.ID))
}
+ })
}
// Cleanup removes allocated objects by EtcdServer.NewServer in
