Refactor the kv client.

Author: 3AceShowHand

api/http.go

@@ -14,13 +14,14 @@
 package api
 
 import (
+	"net/http/pprof"
+
 	"github.com/gin-gonic/gin"
 	v1 "github.com/pingcap/ticdc/api/v1"
 	v2 "github.com/pingcap/ticdc/api/v2"
 	"github.com/pingcap/ticdc/pkg/server"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/promhttp"
-	"net/http/pprof"
 )
 
 // RegisterRoutes create a router for OpenAPI

cmd/cdc/server/server.go

@@ -102,7 +102,7 @@ func (o *options) run(cmd *cobra.Command) error {
 
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
-	
+
 	util.LogHTTPProxies()
 	svr, err := server.New(o.serverConfig, o.pdEndpoints)
 	if err != nil {

logservice/logpuller/region_request_worker.go

@@ -76,35 +76,26 @@ func newRegionRequestWorker(
 		workerID:   workerIDGen.Add(1),
 		client:     client,
 		store:      store,
-		requestsCh: make(chan regionInfo, 1024), // 256 is an arbitrary number.
+		requestsCh: make(chan regionInfo, 256), // 256 is an arbitrary number.
 
 		requestHeader: &cdcpb.Header{ClusterId: client.clusterID, TicdcVersion: version.ReleaseSemver()},
 	}
 	worker.requestedRegions.subscriptions = make(map[SubscriptionID]regionFeedStates)
 
-	waitForPreFetching := func() {
-		if worker.preFetchForConnecting != nil {
-			log.Panic("preFetchForConnecting should be nil",
-				zap.Uint64("workerID", worker.workerID),
-				zap.Uint64("storeID", store.storeID),
-				zap.String("addr", store.storeAddr))
-		}
-		for regionRequest := range worker.requestsCh {
-			if !regionRequest.isStopped() {
-				worker.preFetchForConnecting = &regionRequest
-				return
-			}
-		}
-	}
-
 	g.Go(func() error {
 		for {
+			// fetch for the first region request, so that can establish the grpc stream.
+			for regionRequest := range worker.requestsCh {
+				if !regionRequest.isStopped() {
+					worker.preFetchForConnecting = &regionRequest
+					break
+				}
+			}
 			select {
 			case <-ctx.Done():
 				return ctx.Err()
 			default:
 			}
-			waitForPreFetching()
 			var regionErr error
 			if err := version.CheckStoreVersion(ctx, worker.client.pd, worker.store.storeID); err != nil {
 				if errors.Cause(err) == context.Canceled {
@@ -129,21 +120,15 @@ func newRegionRequestWorker(
 			for subID, m := range worker.clearRegionStates() {
 				for _, state := range m {
 					state.markStopped(regionErr)
-					regionEvent := regionEvent{
+					event := regionEvent{
 						state:  state,
 						worker: worker,
 					}
-					worker.client.pushRegionEventToDS(subID, regionEvent)
+					worker.client.pushRegionEventToDS(subID, event)
 				}
 			}
 			// The store may fail forever, so we need try to re-schedule all pending regions.
-			for _, region := range worker.clearPendingRegions() {
-				if region.isStopped() {
-					// It means it's a special task for stopping the table.
-					continue
-				}
-				client.onRegionFail(newRegionErrorInfo(region, regionErr))
-			}
+			worker.rescheduleRegions(regionErr)
 			if err := util.Hang(ctx, time.Second); err != nil {
 				return err
 			}
@@ -199,7 +184,7 @@ func (s *regionRequestWorker) run(ctx context.Context) (canceled bool) {
 		timer := time.After(10 * time.Second)
 		g.Go(func() error {
 			<-timer
-			err := errors.New("inject force reconnect")
+			err = errors.New("inject force reconnect")
 			log.Info("inject force reconnect", zap.Error(err))
 			return err
 		})
@@ -341,19 +326,11 @@ func (s *regionRequestWorker) processRegionSendTask(
 		return nil
 	}
 
-	fetchMoreReq := func() (regionInfo, error) {
-		for {
-			var region regionInfo
-			select {
-			case <-ctx.Done():
-				return region, ctx.Err()
-			case region = <-s.requestsCh:
-				return region, nil
-			}
-		}
-	}
-
-	region := *s.preFetchForConnecting
+	var (
+		err    error
+		region regionInfo
+	)
+	region = *s.preFetchForConnecting
 	s.preFetchForConnecting = nil
 	for {
 		// TODO: can region be nil?
@@ -367,23 +344,17 @@ func (s *regionRequestWorker) processRegionSendTask(
 
 		// It means it's a special task for stopping the table.
 		if region.isStopped() {
-			req := cdcpb.ChangeDataRequest{
-				Header:    s.requestHeader,
-				RequestId: uint64(subID),
-				Request: &cdcpb.ChangeDataRequest_Deregister_{
-					Deregister: &cdcpb.ChangeDataRequest_Deregister{},
-				},
-			}
-			if err := doSend(req); err != nil {
+			req := s.newDeregisterRegionRequest(subID)
+			if err = doSend(req); err != nil {
 				return err
 			}
 			for _, state := range s.takeRegionStates(subID) {
 				state.markStopped(&requestCancelledErr{})
-				regionEvent := regionEvent{
+				event := regionEvent{
 					state:  state,
 					worker: s,
 				}
-				s.client.pushRegionEventToDS(subID, regionEvent)
+				s.client.pushRegionEventToDS(subID, event)
 			}
 		} else if region.subscribedSpan.stopped.Load() {
 			// It can be skipped directly because there must be no pending states from
@@ -395,19 +366,30 @@ func (s *regionRequestWorker) processRegionSendTask(
 			state.start()
 			s.addRegionState(subID, region.verID.GetID(), state)
 
-			if err := doSend(s.createRegionRequest(region)); err != nil {
+			if err = doSend(s.newRegisterRegionRequest(region)); err != nil {
 				return err
 			}
 		}
 
-		var err error
-		if region, err = fetchMoreReq(); err != nil {
-			return err
+		select {
+		case <-ctx.Done():
+			return errors.Trace(ctx.Err())
+		case region = <-s.requestsCh:
 		}
 	}
 }
 
-func (s *regionRequestWorker) createRegionRequest(region regionInfo) cdcpb.ChangeDataRequest {
+func (s *regionRequestWorker) newDeregisterRegionRequest(subscriptionID SubscriptionID) cdcpb.ChangeDataRequest {
+	return cdcpb.ChangeDataRequest{
+		Header:    s.requestHeader,
+		RequestId: uint64(subscriptionID),
+		Request: &cdcpb.ChangeDataRequest_Deregister_{
+			Deregister: &cdcpb.ChangeDataRequest_Deregister{},
+		},
+	}
+}
+
+func (s *regionRequestWorker) newRegisterRegionRequest(region regionInfo) cdcpb.ChangeDataRequest {
 	return cdcpb.ChangeDataRequest{
 		Header:       s.requestHeader,
 		RegionId:     region.verID.GetID(),
@@ -471,16 +453,20 @@ func (s *regionRequestWorker) clearRegionStates() map[SubscriptionID]regionFeedS
 	return subscriptions
 }
 
-func (s *regionRequestWorker) clearPendingRegions() []regionInfo {
-	regions := make([]regionInfo, 0, len(s.requestsCh))
+func (s *regionRequestWorker) rescheduleRegions(regionErr error) {
 	if s.preFetchForConnecting != nil {
-		region := *s.preFetchForConnecting
-		s.preFetchForConnecting = nil
-		regions = append(regions, region)
+		if !s.preFetchForConnecting.isStopped() {
+			region := *s.preFetchForConnecting
+			s.preFetchForConnecting = nil
+			s.client.onRegionFail(newRegionErrorInfo(region, regionErr))
+		}
 	}
-	// TODO: do we need to start with i := 0(i := len(regions)) if s.preFetchForConnecting is nil?
-	for i := 1; i < cap(regions); i++ {
-		regions = append(regions, <-s.requestsCh)
+
+	for region := range s.requestsCh {
+		if region.isStopped() {
+			// It means it's a special task for stopping the table.
+			continue
+		}
+		s.client.onRegionFail(newRegionErrorInfo(region, regionErr))
 	}
-	return regions
 }

logservice/logpuller/subscription_client.go

@@ -472,7 +472,7 @@ type requestedStore struct {
 	storeID   uint64
 	storeAddr string
 	// Use to select a worker to send request.
-	nextWorker     atomic.Uint32
+	nextWorker     int
 	requestWorkers []*regionRequestWorker
 }
 
@@ -483,9 +483,15 @@ func (rs *requestedStore) broadcastRegionRequest(region regionInfo) {
 }
 
 func (rs *requestedStore) sendRegionRequest(region regionInfo) {
-	index := rs.nextWorker.Add(1) % uint32(len(rs.requestWorkers))
-	worker := rs.requestWorkers[index]
-	worker.sendRegionRequest(region)
+	index := rs.nextWorker % len(rs.requestWorkers)
+	rs.requestWorkers[index].sendRegionRequest(region)
+	rs.nextWorker++
+}
+
+func (rs *requestedStore) stopWorkers() {
+	for _, w := range rs.requestWorkers {
+		close(w.requestsCh)
+	}
 }
 
 // handleRegions receives regionInfo from regionCh and attch rpcCtx to them,
@@ -508,12 +514,7 @@ func (s *SubscriptionClient) handleRegions(ctx context.Context, eg *errgroup.Gro
 
 	defer func() {
 		for _, rs := range stores {
-			for _, w := range rs.requestWorkers {
-				close(w.requestsCh)
-				for range w.requestsCh {
-					// TODO: do we need handle it?
-				}
-			}
+			rs.stopWorkers()
 		}
 	}()