chore(core): use Tick() instead of NewTicker() (#9548)

**Description**

- Replaces all uses of `time.NewTicker()` with `time.Tick()`. Since Go
1.23 the garbage collector can automatically clean up unreferenced
tickers. Therefore, we can just use `Tick()` instead of `NewTicker()`
and remove all `defer ticker.Stop()`.
- Removes all calls to stop a timer. Similar to tickers, since Go 1.23
timers can also be garbage collected automatically without calling
`timer.Stop()`.
- Remove two ticker member variables that were only used in a single
function.
- Changes the function `PollTillPassOrTimeout()` to use the `timeout`
argument

Author: xqqp

conn/node.go

@@ -150,10 +150,9 @@ func (n *Node) ReportRaftComms() {
 	if !glog.V(3) {
 		return
 	}
-	ticker := time.NewTicker(time.Second)
-	defer ticker.Stop()
+	ticker := time.Tick(time.Second)
 
-	for range ticker.C {
+	for range ticker {
 		out := atomic.SwapInt64(&n.heartbeatsOut, 0)
 		in := atomic.SwapInt64(&n.heartbeatsIn, 0)
 		glog.Infof("RaftComm: [%#x] Heartbeats out: %d, in: %d", n.Id, out, in)
@@ -401,11 +400,10 @@ func (n *Node) streamMessages(to uint64, s *stream) {
 	// Let's set the deadline to 10s because if we increase it, then it takes longer to recover from
 	// a partition and get a new leader.
 	deadline := time.Now().Add(10 * time.Second)
-	ticker := time.NewTicker(time.Second)
-	defer ticker.Stop()
+	ticker := time.Tick(time.Second)
 
 	var logged int
-	for range ticker.C { // Don't do this in an busy-wait loop, use a ticker.
+	for range ticker { // Don't do this in a busy-wait loop, use a ticker.
 		// doSendMessage would block doing a stream. So, time.Now().After is
 		// only there to avoid a busy-wait.
 		if err := n.doSendMessage(to, s.msgCh); err != nil {
@@ -460,11 +458,8 @@ func (n *Node) doSendMessage(to uint64, msgCh chan []byte) error {
 
 	ctx = mc.Context()
 
-	fastTick := time.NewTicker(5 * time.Second)
-	defer fastTick.Stop()
-
-	ticker := time.NewTicker(3 * time.Minute)
-	defer ticker.Stop()
+	fastTick := time.Tick(5 * time.Second)
+	ticker := time.Tick(3 * time.Minute)
 
 	for {
 		select {
@@ -495,7 +490,7 @@ func (n *Node) doSendMessage(to uint64, msgCh chan []byte) error {
 				// RAFT would automatically retry.
 				return err
 			}
-		case <-fastTick.C:
+		case <-fastTick:
 			// We use this ticker, because during network partitions, mc.Send is
 			// unable to actually send packets, and also does not complain about
 			// them. We could have potentially used the separately tracked
@@ -512,7 +507,7 @@ func (n *Node) doSendMessage(to uint64, msgCh chan []byte) error {
 				pool.SetUnhealthy()
 				return errors.Wrapf(err, "while calling IsPeer %x", to)
 			}
-		case <-ticker.C:
+		case <-ticker:
 			if lastPackets == packets {
 				span.AddEvent(fmt.Sprintf("No activity for a while [Packets == %d]. Closing connection.", packets))
 				return mc.CloseSend()

conn/node_test.go

@@ -22,12 +22,11 @@ import (
 )
 
 func (n *Node) run(wg *sync.WaitGroup) {
-	ticker := time.NewTicker(20 * time.Millisecond)
-	defer ticker.Stop()
+	ticker := time.Tick(20 * time.Millisecond)
 
 	for {
 		select {
-		case <-ticker.C:
+		case <-ticker:
 			n.Raft().Tick()
 		case rd := <-n.Raft().Ready():
 			n.SaveToStorage(&rd.HardState, rd.Entries, &rd.Snapshot)

conn/raft_server.go

@@ -266,9 +266,6 @@ func (w *RaftServer) RaftMessage(server pb.Raft_RaftMessageServer) error {
 // Heartbeat rpc call is used to check connection with other workers after worker
 // tcp server for this instance starts.
 func (w *RaftServer) Heartbeat(_ *api.Payload, stream pb.Raft_HeartbeatServer) error {
-	ticker := time.NewTicker(echoDuration)
-	defer ticker.Stop()
-
 	node := w.GetNode()
 	if node == nil {
 		return ErrNoNode
@@ -287,13 +284,14 @@ func (w *RaftServer) Heartbeat(_ *api.Payload, stream pb.Raft_HeartbeatServer) e
 	}
 
 	ctx := stream.Context()
+	ticker := time.Tick(echoDuration)
 
 	for {
 		info.Uptime = int64(time.Since(node.StartTime) / time.Second)
 		select {
 		case <-ctx.Done():
 			return ctx.Err()
-		case <-ticker.C:
+		case <-ticker:
 			if err := stream.Send(&info); err != nil {
 				return err
 			}

contrib/jepsen/main.go

@@ -237,18 +237,15 @@ func jepsenServe() error {
 		// lein run serve runs indefinitely, so there's no need to wait for the
 		// command to finish.
 		_ = cmd.Start()
-		ticker := time.NewTicker(time.Second)
-		defer ticker.Stop()
-
+		ticker := time.Tick(time.Second)
 		timeout := time.NewTimer(5 * time.Minute)
-		defer timeout.Stop()
 		for {
 			select {
 			case <-timeout.C:
 				wg.Done()
 				errCh <- errors.New("lein run serve couldn't run after 5 minutes")
 				return
-			case <-ticker.C:
+			case <-ticker:
 				if err := checkServing(); err == nil {
 					wg.Done()
 					errCh <- nil

dgraph/cmd/bulk/progress.go

@@ -56,13 +56,12 @@ func (p *progress) setPhase(ph phase) {
 }
 
 func (p *progress) report() {
-	t := time.NewTicker(time.Second)
-	defer t.Stop()
+	t := time.Tick(time.Second)
 
 	z.StatsPrint() // Just print once.
 	for {
 		select {
-		case <-t.C:
+		case <-t:
 			p.reportOnce()
 		case <-p.shutdown:
 			p.shutdown <- struct{}{}

dgraph/cmd/bulk/reduce.go

@@ -478,8 +478,7 @@ func (r *reducer) reduce(partitionKeys [][]byte, mapItrs []*mapIterator, ci *cou
 		writerCh <- req
 	}
 
-	ticker := time.NewTicker(time.Minute)
-	defer ticker.Stop()
+	ticker := time.Tick(time.Minute)
 
 	buffers := make(chan *z.Buffer, 3)
 
@@ -502,7 +501,7 @@ func (r *reducer) reduce(partitionKeys [][]byte, mapItrs []*mapIterator, ci *cou
 
 			hd.Update(int64(cbuf.LenNoPadding()))
 			select {
-			case <-ticker.C:
+			case <-ticker:
 				fmt.Printf("Histogram of buffer sizes: %s\n", hd.String())
 			default:
 			}

dgraph/cmd/increment/increment_test.go

@@ -203,7 +203,6 @@ func TestBestEffortOnly(t *testing.T) {
 	}()
 
 	timer := time.NewTimer(15 * time.Second)
-	defer timer.Stop()
 
 	select {
 	case <-timer.C:

dgraph/cmd/live/batch.go

@@ -54,7 +54,6 @@ type loader struct {
 
 	dc         *dgo.Dgraph
 	alloc      *xidmap.XidMap
-	ticker     *time.Ticker
 	db         *badger.DB
 	requestsWg sync.WaitGroup
 	// If we retry a request, we add one to retryRequestsWg.
@@ -414,8 +413,7 @@ func (l *loader) makeRequests() {
 		buffer = buffer[:i]
 	}
 
-	t := time.NewTicker(5 * time.Second)
-	defer t.Stop()
+	t := time.Tick(5 * time.Second)
 
 loop:
 	for {
@@ -431,7 +429,7 @@ loop:
 				buffer = append(buffer, req)
 			}
 
-		case <-t.C:
+		case <-t:
 			for {
 				drain()
 				if len(buffer) < l.opts.bufferSize {
@@ -462,11 +460,11 @@ loop:
 
 func (l *loader) printCounters() {
 	period := 5 * time.Second
-	l.ticker = time.NewTicker(period)
+	ticker := time.Tick(period)
 	start := time.Now()
 
 	var last Counter
-	for range l.ticker.C {
+	for range ticker {
 		counter := l.Counter()
 		rate := float64(counter.Nquads-last.Nquads) / period.Seconds()
 		elapsed := time.Since(start).Round(time.Second)

dgraph/cmd/zero/oracle.go

@@ -184,8 +184,7 @@ func (o *Oracle) removeSubscriber(id int) {
 // and sends the delta object to each subscriber's channel
 func (o *Oracle) sendDeltasToSubscribers() {
 	delta := &pb.OracleDelta{}
-	ticker := time.NewTicker(time.Second)
-	defer ticker.Stop()
+	ticker := time.Tick(time.Second)
 
 	// waitFor calculates the maximum value of delta.MaxAssigned and all the CommitTs of delta.Txns
 	waitFor := func() uint64 {
@@ -201,7 +200,7 @@ func (o *Oracle) sendDeltasToSubscribers() {
 		var update *pb.OracleDelta
 		select {
 		case update = <-o.updates:
-		case <-ticker.C:
+		case <-ticker:
 			wait := waitFor()
 			if wait == 0 || o.doneUntil.DoneUntil() < wait {
 				goto get_update

dgraph/cmd/zero/raft.go

@@ -690,12 +690,11 @@ func (n *node) initAndStartNode() error {
 
 func (n *node) updateZeroMembershipPeriodically(closer *z.Closer) {
 	defer closer.Done()
-	ticker := time.NewTicker(10 * time.Second)
-	defer ticker.Stop()
+	ticker := time.Tick(10 * time.Second)
 
 	for {
 		select {
-		case <-ticker.C:
+		case <-ticker:
 			n.server.updateZeroLeader()
 		case <-closer.HasBeenClosed():
 			return
@@ -705,8 +704,7 @@ func (n *node) updateZeroMembershipPeriodically(closer *z.Closer) {
 
 func (n *node) checkQuorum(closer *z.Closer) {
 	defer closer.Done()
-	ticker := time.NewTicker(time.Second)
-	defer ticker.Stop()
+	ticker := time.Tick(time.Second)
 
 	quorum := func() {
 		// Make this timeout 1.5x the timeout on RunReadIndexLoop.
@@ -733,7 +731,7 @@ func (n *node) checkQuorum(closer *z.Closer) {
 
 	for {
 		select {
-		case <-ticker.C:
+		case <-ticker:
 			// Only the leader needs to check for the quorum. The quorum is
 			// used by a leader to identify if it is behind a network partition.
 			if n.amLeader() {
@@ -747,12 +745,11 @@ func (n *node) checkQuorum(closer *z.Closer) {
 
 func (n *node) snapshotPeriodically(closer *z.Closer) {
 	defer closer.Done()
-	ticker := time.NewTicker(time.Minute)
-	defer ticker.Stop()
+	ticker := time.Tick(time.Minute)
 
 	for {
 		select {
-		case <-ticker.C:
+		case <-ticker:
 			if err := n.calculateAndProposeSnapshot(); err != nil {
 				glog.Errorf("While calculateAndProposeSnapshot: %v", err)
 			}
@@ -873,8 +870,7 @@ func (n *node) Run() {
 	lastLead := uint64(math.MaxUint64)
 
 	var leader bool
-	ticker := time.NewTicker(tickDur)
-	defer ticker.Stop()
+	ticker := time.Tick(tickDur)
 
 	// snapshot can cause select loop to block while deleting entries, so run
 	// it in goroutine
@@ -911,7 +907,7 @@ func (n *node) Run() {
 		case <-n.closer.HasBeenClosed():
 			n.Raft().Stop()
 			return
-		case <-ticker.C:
+		case <-ticker:
 			n.Raft().Tick()
 		case rd := <-n.Raft().Ready():
 			timer.Start()