golang源码分析：etcd（13）

func NewServer(cfg config.ServerConfig) (srv *EtcdServer, err error) {
      srv.lessor = lease.NewLessor(srv.Logger(), srv.be, srv.cluster, lease.LessorConfig{
    MinLeaseTTL:                int64(math.Ceil(minTTL.Seconds())),
    CheckpointInterval:         cfg.LeaseCheckpointInterval,
    CheckpointPersist:          cfg.LeaseCheckpointPersist,
    ExpiredLeasesRetryInterval: srv.Cfg.ReqTimeout(),
  })

type TxnDelete interface {
  DeleteRange(key, end []byte) (n, rev int64)
  End()
}

type Lessor interface {
  // SetRangeDeleter lets the lessor create TxnDeletes to the store.
  // Lessor deletes the items in the revoked or expired lease by creating
  // new TxnDeletes.
  SetRangeDeleter(rd RangeDeleter)


  SetCheckpointer(cp Checkpointer)


  // Grant grants a lease that expires at least after TTL seconds.
  Grant(id LeaseID, ttl int64) (*Lease, error)
  // Revoke revokes a lease with given ID. The item attached to the
  // given lease will be removed. If the ID does not exist, an error
  // will be returned.
  Revoke(id LeaseID) error


  // Checkpoint applies the remainingTTL of a lease. The remainingTTL is used in Promote to set
  // the expiry of leases to less than the full TTL when possible.
  Checkpoint(id LeaseID, remainingTTL int64) error


  // Attach attaches given leaseItem to the lease with given LeaseID.
  // If the lease does not exist, an error will be returned.
  Attach(id LeaseID, items []LeaseItem) error


  // GetLease returns LeaseID for given item.
  // If no lease found, NoLease value will be returned.
  GetLease(item LeaseItem) LeaseID


  // Detach detaches given leaseItem from the lease with given LeaseID.
  // If the lease does not exist, an error will be returned.
  Detach(id LeaseID, items []LeaseItem) error


  // Promote promotes the lessor to be the primary lessor. Primary lessor manages
  // the expiration and renew of leases.
  // Newly promoted lessor renew the TTL of all lease to extend + previous TTL.
  Promote(extend time.Duration)


  // Demote demotes the lessor from being the primary lessor.
  Demote()


  // Renew renews a lease with given ID. It returns the renewed TTL. If the ID does not exist,
  // an error will be returned.
  Renew(id LeaseID) (int64, error)


  // Lookup gives the lease at a given lease id, if any
  Lookup(id LeaseID) *Lease


  // Leases lists all leases.
  Leases() []*Lease


  // ExpiredLeasesC returns a chan that is used to receive expired leases.
  ExpiredLeasesC() <-chan []*Lease


  // Recover recovers the lessor state from the given backend and RangeDeleter.
  Recover(b backend.Backend, rd RangeDeleter)


  // Stop stops the lessor for managing leases. The behavior of calling Stop multiple
  // times is undefined.
  Stop()
}

type lessor struct {
  mu sync.RWMutex


  // demotec is set when the lessor is the primary.
  // demotec will be closed if the lessor is demoted.
  demotec chan struct{}


  leaseMap             map[LeaseID]*Lease
  leaseExpiredNotifier *LeaseExpiredNotifier
  leaseCheckpointHeap  LeaseQueue
  itemMap              map[LeaseItem]LeaseID


  // When a lease expires, the lessor will delete the
  // leased range (or key) by the RangeDeleter.
  rd RangeDeleter


  // When a lease's deadline should be persisted to preserve the remaining TTL across leader
  // elections and restarts, the lessor will checkpoint the lease by the Checkpointer.
  cp Checkpointer


  // backend to persist leases. We only persist lease ID and expiry for now.
  // The leased items can be recovered by iterating all the keys in kv.
  b backend.Backend


  // minLeaseTTL is the minimum lease TTL that can be granted for a lease. Any
  // requests for shorter TTLs are extended to the minimum TTL.
  minLeaseTTL int64


  expiredC chan []*Lease
  // stopC is a channel whose closure indicates that the lessor should be stopped.
  stopC chan struct{}
  // doneC is a channel whose closure indicates that the lessor is stopped.
  doneC chan struct{}


  lg *zap.Logger


  // Wait duration between lease checkpoints.
  checkpointInterval time.Duration
  // the interval to check if the expired lease is revoked
  expiredLeaseRetryInterval time.Duration
  // whether lessor should always persist remaining TTL (always enabled in v3.6).
  checkpointPersist bool
  // cluster is used to adapt lessor logic based on cluster version
  cluster cluster
}  

type LessorConfig struct {
  MinLeaseTTL                int64
  CheckpointInterval         time.Duration
  ExpiredLeasesRetryInterval time.Duration
  CheckpointPersist          bool
}

func NewLessor(lg *zap.Logger, b backend.Backend, cluster cluster, cfg LessorConfig) Lessor {
  return newLessor(lg, b, cluster, cfg)
}

func newLessor(lg *zap.Logger, b backend.Backend, cluster cluster, cfg LessorConfig) *lessor 

      l := &lessor{
        b:                         b,
      l.initAndRecover()
      go l.runLoop()

func (le *lessor) runLoop() {
      for {
    le.revokeExpiredLeases()
    le.checkpointScheduledLeases()

func (le *lessor) revokeExpiredLeases() {
        if le.isPrimary() {
    ls = le.findExpiredLeases(revokeLimit)
  }
        if len(ls) != 0 {
    select {
    case <-le.stopC:
      return
    case le.expiredC <- ls:

func (le *lessor) checkpointScheduledLeases() {
          if le.isPrimary() {
      cps = le.findDueScheduledCheckpoints(maxLeaseCheckpointBatchSize)
    }
      le.cp(context.Background(), &pb.LeaseCheckpointRequest{Checkpoints: cps})

func (le *lessor) ExpiredLeasesC() <-chan []*Lease {
  return le.expiredC
}

func (le *lessor) Grant(id LeaseID, ttl int64) (*Lease, error) {
      l := &Lease{
      if _, ok := le.leaseMap[id]; ok {
        if le.isPrimary() {
    l.refresh(0)
  } else {
    l.forever()
  }
        le.leaseMap[id] = l
  l.persistTo(le.b)    

func (l *Lease) persistTo(b backend.Backend) {

func (le *lessor) Revoke(id LeaseID) error {
      delete(le.leaseMap, id)
      txn := le.rd()
      keys := l.Keys()
  sort.StringSlice(keys).Sort()
  for _, key := range keys {
    txn.DeleteRange([]byte(key), nil)
  }

func (le *lessor) Renew(id LeaseID) (int64, error) {
      l := le.leaseMap[id]
        l.refresh(0)
  item := &LeaseWithTime{id: l.ID, time: l.expiry}
  le.leaseExpiredNotifier.RegisterOrUpdate(item)

func (le *lessor) Promote(extend time.Duration) {

type FakeLessor struct{}

func (fl *FakeLessor) Grant(id LeaseID, ttl int64) (*Lease, error) { return nil, nil }

type FakeTxnDelete struct {
  backend.BatchTx
}

type Lease struct {
  ID           LeaseID
  ttl          int64 // time to live of the lease in seconds
  remainingTTL int64 // remaining time to live in seconds, if zero valued it is considered unset and the full ttl should be used
  // expiryMu protects concurrent accesses to expiry
  expiryMu sync.RWMutex
  // expiry is time when lease should expire. no expiration when expiry.IsZero() is true
  expiry time.Time


  // mu protects concurrent accesses to itemSet
  mu      sync.RWMutex
  itemSet map[LeaseItem]struct{}
  revokec chan struct{}
}

func (l *Lease) persistTo(b backend.Backend) {
      tx := b.BatchTx()
      tx.LockInsideApply()
      schema.MustUnsafePutLease(tx, &lpb)

type LeaseItem struct {
  Key string
}

LeasePrefix         = "/leases"
LeaseInternalPrefix = "/leases/internal"

type leaseHandler struct {
  l      lease.Lessor
  waitch func() <-chan struct{}
}

func (h *leaseHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
      switch r.URL.Path {
        case LeasePrefix:
          ttl, rerr := h.l.Renew(lease.LeaseID(lreq.ID))
        case LeaseInternalPrefix:
          l := h.l.Lookup(lease.LeaseID(lreq.LeaseTimeToLiveRequest.ID))

func RenewHTTP(ctx context.Context, id lease.LeaseID, url string, rt http.RoundTripper) (int64, error) {
      req, err := http.NewRequest(http.MethodPost, url, bytes.NewReader(lreq))    

func TimeToLiveHTTP(ctx context.Context, id lease.LeaseID, keys bool, url string, rt http.RoundTripper) (*leasepb.LeaseInternalResponse, error) {
      req, err := http.NewRequest(http.MethodPost, url, bytes.NewReader(lreq))

message Lease {
  int64 ID = 1;
  int64 TTL = 2;
  int64 RemainingTTL = 3;
}

message LeaseInternalRequest {
  etcdserverpb.LeaseTimeToLiveRequest LeaseTimeToLiveRequest = 1;
}

message LeaseInternalResponse {
  etcdserverpb.LeaseTimeToLiveResponse LeaseTimeToLiveResponse = 1;
}

type LeaseWithTime struct {
  id    LeaseID
  time  time.Time
  index int
}

type LeaseQueue []*LeaseWithTime

func (pq *LeaseQueue) Push(x interface{}) {

type LeaseExpiredNotifier struct {
  m     map[LeaseID]*LeaseWithTime
  queue LeaseQueue
}

func (mq *LeaseExpiredNotifier) RegisterOrUpdate(item *LeaseWithTime) {
        if old, ok := mq.m[item.id]; ok {
    old.time = item.time
    heap.Fix(&mq.queue, old.index)
  } else {
    heap.Push(&mq.queue, item)
    mq.m[item.id] = item
  }

leaseGranted = prometheus.NewCounter(prometheus.CounterOpts{
    Namespace: "etcd_debugging",
    Subsystem: "lease",
    Name:      "granted_total",
    Help:      "The total number of granted leases.",
  })