optimize memory allocate (#555)

* update

* revert

logservice/eventstore/event_store.go

@@ -299,7 +299,7 @@ func (e *eventStore) uploadStatePeriodically(ctx context.Context) error {
 				Subscriptions: make(map[int64]*logservicepb.SubscriptionStates),
 			}
 			for tableID, dispatcherIDs := range e.dispatcherMeta.tableToDispatchers {
-				subStates := make([]*logservicepb.SubscriptionState, 0)
+				subStates := make([]*logservicepb.SubscriptionState, 0, len(dispatcherIDs))
 				subIDs := make(map[logpuller.SubscriptionID]bool)
 				for dispatcherID := range dispatcherIDs {
 					dispatcherStat := e.dispatcherMeta.dispatcherStats[dispatcherID]

logservice/logpuller/region_request_worker.go

@@ -67,6 +67,8 @@ type regionRequestWorker struct {
 
 	// used to indicate whether there are new pending ts events for every processores in dispatchResolvedTs
 	boolCache []bool
+	// used to avoid repeated hash calculation in dispatchResolvedTs
+	slotCache map[uint64]int
 }
 
 func newRegionRequestWorker(
@@ -83,6 +85,7 @@ func newRegionRequestWorker(
 		requestsCh: make(chan regionInfo, 256), // 256 is an arbitrary number.
 
 		boolCache: make([]bool, len(client.changeEventProcessors)),
+		slotCache: make(map[uint64]int),
 	}
 	worker.requestedRegions.subscriptions = make(map[SubscriptionID]regionFeedStates)
 	worker.tsBatches.events = make([]statefulEvent, len(client.changeEventProcessors))
@@ -432,13 +435,18 @@ func (s *regionRequestWorker) dispatchResolvedTs(resolvedTs *cdcpb.ResolvedTs) e
 	for i := range s.boolCache {
 		s.boolCache[i] = false
 	}
+	regionsLen := len(resolvedTs.Regions)/len(s.client.changeEventProcessors) + 1 // an average number
 	s.client.metrics.batchResolvedSize.Observe(float64(len(resolvedTs.Regions)))
-	for _, regionID := range resolvedTs.Regions {
-		slot := hashRegionID(regionID, len(s.client.changeEventProcessors))
+	for i, regionID := range resolvedTs.Regions {
+		// We suppose that the length of changeEventProcessors is constant, so we can cache the hashes of the region
+		if _, exist := s.slotCache[regionID]; !exist {
+			s.slotCache[regionID] = hashRegionID(regionID, len(s.client.changeEventProcessors))
+		}
+		slot := s.slotCache[regionID]
 		if !s.boolCache[slot] {
 			s.tsBatches.events[slot].resolvedTsBatches = append(s.tsBatches.events[slot].resolvedTsBatches, resolvedTsBatch{
 				ts:      resolvedTs.Ts,
-				regions: make([]*regionFeedState, 0, len(resolvedTs.Regions)),
+				regions: make([]*regionFeedState, 0, min(len(resolvedTs.Regions)-i, regionsLen)),
 			})
 			s.boolCache[slot] = true
 		}