*: use atomic value to store coordinator info (pingcap#1414)

ref pingcap#1328

Author: lidezhu

downstreamadapter/eventcollector/event_collector.go

@@ -16,6 +16,7 @@ package eventcollector
 import (
 	"context"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/pingcap/log"
@@ -112,8 +113,7 @@ type EventCollector struct {
 	ds dynstream.DynamicStream[common.GID, common.DispatcherID, dispatcher.DispatcherEvent, *dispatcherStat, *EventsHandler]
 
 	coordinatorInfo struct {
-		sync.Mutex
-		id node.ID
+		value atomic.Value
 	}
 
 	wg     sync.WaitGroup
@@ -401,15 +401,14 @@ func (c *EventCollector) processDispatcherRequests(ctx context.Context) {
 }
 
 func (c *EventCollector) setCoordinatorInfo(id node.ID) {
-	c.coordinatorInfo.Lock()
-	defer c.coordinatorInfo.Unlock()
-	c.coordinatorInfo.id = id
+	c.coordinatorInfo.value.Store(id)
 }
 
 func (c *EventCollector) getCoordinatorInfo() node.ID {
-	c.coordinatorInfo.Lock()
-	defer c.coordinatorInfo.Unlock()
-	return c.coordinatorInfo.id
+	if v := c.coordinatorInfo.value.Load(); v != nil {
+		return v.(node.ID)
+	}
+	return ""
 }
 
 func (c *EventCollector) processLogCoordinatorRequest(ctx context.Context) {

logservice/eventstore/event_store.go

@@ -163,8 +163,7 @@ type eventStore struct {
 	messageCenter messaging.MessageCenter
 
 	coordinatorInfo struct {
-		sync.Mutex
-		id node.ID
+		value atomic.Value
 	}
 	// The channel is used to gather the subscription info
 	// which need to be uploaded to log coordinator periodically.
@@ -270,15 +269,14 @@ func (p *writeTaskPool) run(ctx context.Context) {
 }
 
 func (e *eventStore) setCoordinatorInfo(id node.ID) {
-	e.coordinatorInfo.Lock()
-	defer e.coordinatorInfo.Unlock()
-	e.coordinatorInfo.id = id
+	e.coordinatorInfo.value.Store(id)
 }
 
 func (e *eventStore) getCoordinatorInfo() node.ID {
-	e.coordinatorInfo.Lock()
-	defer e.coordinatorInfo.Unlock()
-	return e.coordinatorInfo.id
+	if v := e.coordinatorInfo.value.Load(); v != nil {
+		return v.(node.ID)
+	}
+	return ""
 }
 
 func (e *eventStore) Name() string {