From 015dd3a147cc6f1ca0bfc02ddcd74453fc08ebf6 Mon Sep 17 00:00:00 2001
From: Chris Lu <chris.lu@gmail.com>
Date: Sun, 10 May 2020 23:25:39 -0700
Subject: [PATCH] batch file id deletion

---
 weed/filer2/filer_deletion.go | 30 +++++++++++++++++++++---------
 1 file changed, 21 insertions(+), 9 deletions(-)

diff --git a/weed/filer2/filer_deletion.go b/weed/filer2/filer_deletion.go
index e463e9e6a..a6b229771 100644
--- a/weed/filer2/filer_deletion.go
+++ b/weed/filer2/filer_deletion.go
@@ -34,19 +34,31 @@ func (f *Filer) loopProcessingDeletion() {
 
 	lookupFunc := LookupByMasterClientFn(f.MasterClient)
 
+	DeletionBatchSize := 100000 // roughly 20 bytes cost per file id.
+
 	var deletionCount int
 	for {
 		deletionCount = 0
 		f.fileIdDeletionQueue.Consume(func(fileIds []string) {
-			deletionCount = len(fileIds)
-			deleteResults, err := operation.DeleteFilesWithLookupVolumeId(f.GrpcDialOption, fileIds, lookupFunc)
-			if err != nil {
-				glog.V(0).Infof("deleting fileIds len=%d error: %v", deletionCount, err)
-			} else {
-				glog.V(1).Infof("deleting fileIds len=%d", deletionCount)
-			}
-			if len(deleteResults) != deletionCount {
-				glog.V(0).Infof("delete %d fileIds actual %d", deletionCount, len(deleteResults))
+			for len(fileIds) > 0 {
+				var toDeleteFileIds []string
+				if len(fileIds) > DeletionBatchSize {
+					toDeleteFileIds = fileIds[:DeletionBatchSize]
+					fileIds = fileIds[DeletionBatchSize:]
+				} else {
+					toDeleteFileIds = fileIds
+					fileIds = fileIds[:0]
+				}
+				deletionCount = len(toDeleteFileIds)
+				deleteResults, err := operation.DeleteFilesWithLookupVolumeId(f.GrpcDialOption, toDeleteFileIds, lookupFunc)
+				if err != nil {
+					glog.V(0).Infof("deleting fileIds len=%d error: %v", deletionCount, err)
+				} else {
+					glog.V(1).Infof("deleting fileIds len=%d", deletionCount)
+				}
+				if len(deleteResults) != deletionCount {
+					glog.V(0).Infof("delete %d fileIds actual %d", deletionCount, len(deleteResults))
+				}
 			}
 		})