s3: move "delete-directory-if-empty" to read time

move "delete-directory-if-empty" to read time instead of entry deletion time

the listing speed for a s3 bucket folder will slow down if it has many sub folders

related to 0d345ac97d

fix https://github.com/chrislusf/seaweedfs/issues/1647

fix https://github.com/chrislusf/seaweedfs/issues/1670

weed/filer/filer_delete_entry.go

@@ -3,8 +3,6 @@ package filer
 import (
 	"context"
 	"fmt"
-	"strings"
-
 	"github.com/chrislusf/seaweedfs/weed/glog"
 	"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
 	"github.com/chrislusf/seaweedfs/weed/pb/master_pb"
@@ -60,11 +58,6 @@ func (f *Filer) DeleteEntryMetaAndData(ctx context.Context, p util.FullPath, isR
 		collectionName := entry.Name()
 		f.doDeleteCollection(collectionName)
 		f.deleteBucket(collectionName)
-	} else {
-		parent, _ := p.DirAndName()
-		if err := f.removeEmptyParentFolder(ctx, util.FullPath(parent)); err != nil {
-			glog.Errorf("clean up empty folders for %s : %v", p, err)
-		}
 	}
 
 	return nil
@@ -159,25 +152,3 @@ func (f *Filer) maybeDeleteHardLinks(hardLinkIds []HardLinkId) {
 		}
 	}
 }
-
-func (f *Filer) removeEmptyParentFolder(ctx context.Context, dir util.FullPath) error {
-	if !strings.HasPrefix(string(dir), f.DirBucketsPath) {
-		return nil
-	}
-	parent, _ := dir.DirAndName()
-	if parent == f.DirBucketsPath {
-		// should not delete bucket itself
-		return nil
-	}
-	entries, err := f.ListDirectoryEntries(ctx, dir, "", false, 1, "")
-	if err != nil {
-		return err
-	}
-	if len(entries) > 0 {
-		return nil
-	}
-	if err := f.Store.DeleteEntry(ctx, dir); err != nil {
-		return err
-	}
-	return f.removeEmptyParentFolder(ctx, util.FullPath(parent))
-}

weed/s3api/s3api_objects_list_handlers.go

@@ -246,8 +246,8 @@ func (s3a *S3ApiServer) doListFilerEntries(client filer_pb.SeaweedFilerClient, d
 		if entry.IsDirectory {
 			// println("ListEntries", dir, "dir:", entry.Name)
 			if entry.Name != ".uploads" { // FIXME no need to apply to all directories. this extra also affects maxKeys
-				eachEntryFn(dir, entry)
 				if delimiter != "/" {
+					eachEntryFn(dir, entry)
 					// println("doListFilerEntries2 dir", dir+"/"+entry.Name, "maxKeys", maxKeys-counter)
 					subCounter, subIsTruncated, subNextMarker, subErr := s3a.doListFilerEntries(client, dir+"/"+entry.Name, "", maxKeys-counter, "", delimiter, eachEntryFn)
 					if subErr != nil {
@@ -262,7 +262,18 @@ func (s3a *S3ApiServer) doListFilerEntries(client filer_pb.SeaweedFilerClient, d
 						return
 					}
 				} else {
-					counter++
+					var isEmpty bool
+					if isEmpty, err = s3a.isDirectoryAllEmpty(client, dir+"/"+entry.Name); err != nil {
+						return
+					}
+					if isEmpty {
+						if err = doDeleteEntry(client, dir, entry.Name, true, true); err != nil {
+							return
+						}
+					} else {
+						eachEntryFn(dir, entry)
+						counter++
+					}
 				}
 			}
 		} else {
@@ -299,3 +310,15 @@ func getListObjectsV1Args(values url.Values) (prefix, marker, delimiter string,
 	}
 	return
 }
+
+func (s3a *S3ApiServer) isDirectoryAllEmpty(filerClient filer_pb.SeaweedFilerClient, dir string) (isEmpty bool, err error) {
+	// println("+ isDirectoryAllEmpty", dir)
+	subCounter, _, _, subErr := s3a.doListFilerEntries(filerClient, dir, "", 32, "", "/", func(dir string, entry *filer_pb.Entry) {
+	})
+	if subErr != nil {
+		err = fmt.Errorf("isDirectoryAllEmpty: %v", subErr)
+	}
+	isEmpty = subCounter <= 0
+	// println("- isDirectoryAllEmpty", dir, isEmpty)
+	return
+}