etcd_worker: batch etcd patch

pkg/orchestrator/etcd_worker.go

@@ -283,14 +283,33 @@ func (worker *EtcdWorker) cloneRawState() map[util.EtcdKey][]byte {
 	return ret
 }
 
+var maxBatchPatchSize = 96
+
 func (worker *EtcdWorker) applyPatchGroups(ctx context.Context, patchGroups [][]DataPatch) ([][]DataPatch, error) {
+	getBatchPatches := func(patchGroups [][]DataPatch) ([]DataPatch, int) {
+		batchPatches := patchGroups[0]
+		patchesNum := 1
+		for _, patches := range patchGroups {
+			if patchesNum == 1 {
+				continue
+			}
+			if (len(batchPatches) + len(patches)) > maxBatchPatchSize {
+				break
+			}
+			batchPatches = append(batchPatches, patches...)
+			patchesNum++
+		}
+
+		return batchPatches, patchesNum
+	}
+
 	for len(patchGroups) > 0 {
-		patches := patchGroups[0]
+		patches, n := getBatchPatches(patchGroups)
 		err := worker.applyPatches(ctx, patches)
 		if err != nil {
 			return patchGroups, err
 		}
-		patchGroups = patchGroups[1:]
+		patchGroups = patchGroups[n:]
 	}
 	return patchGroups, nil
 }
@@ -310,10 +329,12 @@ func (worker *EtcdWorker) applyPatches(ctx context.Context, patches []DataPatch)
 	cmps := make([]clientv3.Cmp, 0, len(changedSet))
 	ops := make([]clientv3.Op, 0, len(changedSet))
 	hasDelete := false
+	entrys := make(map[string]int64)
 	for key := range changedSet {
 		// make sure someone else has not updated the key after the last snapshot
 		var cmp clientv3.Cmp
 		if entry, ok := worker.rawState[key]; ok {
+			entrys[key.String()] = entry.modRevision
 			cmp = clientv3.Compare(clientv3.ModRevision(key.String()), "=", entry.modRevision)
 		} else {
 			// if ok is false, it means that the key of this patch is not exist in a committed state

pkg/orchestrator/interfaces.go

@@ -68,10 +68,10 @@ func (s *SingleDataPatch) Patch(valueMap map[util.EtcdKey][]byte, changedSet map
 	return nil
 }
 
-// MultiDatePatch represents an update to many keys
-type MultiDatePatch func(valueMap map[util.EtcdKey][]byte, changedSet map[util.EtcdKey]struct{}) error
+// MultiDataPatch represents an update to many keys
+type MultiDataPatch func(valueMap map[util.EtcdKey][]byte, changedSet map[util.EtcdKey]struct{}) error
 
 // Patch implements the DataPatch interface
-func (m MultiDatePatch) Patch(valueMap map[util.EtcdKey][]byte, changedSet map[util.EtcdKey]struct{}) error {
+func (m MultiDataPatch) Patch(valueMap map[util.EtcdKey][]byte, changedSet map[util.EtcdKey]struct{}) error {
 	return m(valueMap, changedSet)
 }

pkg/orchestrator/reactor_state.go

@@ -100,6 +100,7 @@ func (s *GlobalReactorState) Update(key util.EtcdKey, value []byte, _ bool) erro
 }
 
 // GetPatches implements the ReactorState interface
+// Every []DataPatch slice in [][]DataPatch slice is the patches of a ChangefeedReactorState
 func (s *GlobalReactorState) GetPatches() [][]DataPatch {
 	pendingPatches := s.pendingPatches
 	for _, changefeedState := range s.Changefeeds {