Method Run

dgraph/cmd/zero/raft.go:964–1132

()

const tickDur = 100 * time.Millisecond

func (n *node) Run() {
	// lastLead is for detecting leadership changes
	//
	// etcd has a similar mechanism for tracking leader changes, with their
	// raftReadyHandler.getLead() function that returns the previous leader
	lastLead := uint64(math.MaxUint64)

	var leader bool
	ticker := time.Tick(tickDur)

	// snapshot can cause select loop to block while deleting entries, so run
	// it in goroutine
	readStateCh := make(chan raft.ReadState, 100)
	closer := z.NewCloser(0)
	defer func() {
		closer.SignalAndWait()
		n.closer.Done()
		glog.Infof("Zero Node.Run finished.")
	}()

	closer.AddRunning(1)
	go n.snapshotPeriodically(closer)

	closer.AddRunning(1)
	go n.updateZeroMembershipPeriodically(closer)

	closer.AddRunning(1)
	go n.checkQuorum(closer)

	closer.AddRunning(1)
	go n.RunReadIndexLoop(closer, readStateCh)

	if !x.WorkerConfig.HardSync {
		closer.AddRunning(1)
		go x.StoreSync(n.Store, closer)
	}
	// We only stop runReadIndexLoop after the for loop below has finished interacting with it.
	// That way we know sending to readStateCh will not deadlock.

	var timer x.Timer
	for {
		select {
		case <-n.closer.HasBeenClosed():
			n.Raft().Stop()
			return
		case <-ticker:
			n.Raft().Tick()
		case rd := <-n.Raft().Ready():
			timer.Start()
			_, span := otel.Tracer("").Start(n.ctx, "Zero.RunLoop")
			for _, rs := range rd.ReadStates {
				// No need to use select-case-default on pushing to readStateCh. It is typically
				// empty.
				readStateCh <- rs
			}
			span.AddEvent(fmt.Sprintf("Pushed %d readstates", len(rd.ReadStates)))

			if rd.SoftState != nil {