{miniogw, testsuite}: run deletion within DeleteObjects concurrently

This changes DeleteObjects from executing deletion sequentially. How
many deletions to run concurrently is controlled with a new
configuration parameter (default=100).

Updates storj/storj-private#252

Change-Id: Ie12ea00c2d0e806be9b7c1d0769c21ddf60b0c97

miniogw/config.go

@@ -25,4 +25,5 @@ type S3CompatibilityConfig struct {
 	FullyCompatibleListing       bool  `help:"make ListObjects(V2) fully S3-compatible (specifically: always return lexicographically ordered results) but slow" default:"false"`
 	DisableCopyObject            bool  `help:"return 501 (Not Implemented) for CopyObject calls" default:"false"`
 	MinPartSize                  int64 `help:"minimum part size for multipart uploads" default:"5242880"` // 5 MiB
+	DeleteObjectsConcurrency     int   `help:"how many objects to delete in parallel with DeleteObjects" default:"100"`
 }

miniogw/gateway.go

@@ -20,6 +20,7 @@ import (
 	"github.com/zeebo/errs"
 
 	"storj.io/common/memory"
+	"storj.io/common/sync2"
 	minio "storj.io/minio/cmd"
 	"storj.io/minio/cmd/config/storageclass"
 	xhttp "storj.io/minio/cmd/http"
@@ -998,14 +999,23 @@ func (layer *gatewayLayer) DeleteObject(ctx context.Context, bucket, objectPath
 func (layer *gatewayLayer) DeleteObjects(ctx context.Context, bucket string, objects []minio.ObjectToDelete, opts minio.ObjectOptions) ([]minio.DeletedObject, []error) {
 	// TODO: implement multiple object deletion in libuplink API
 	deleted, errs := make([]minio.DeletedObject, len(objects)), make([]error, len(objects))
+
+	limiter := sync2.NewLimiter(layer.compatibilityConfig.DeleteObjectsConcurrency)
+
 	for i, object := range objects {
-		_, deleteErr := layer.DeleteObject(ctx, bucket, object.ObjectName, opts)
-		if deleteErr != nil && !errors.As(deleteErr, &minio.ObjectNotFound{}) {
-			errs[i] = ConvertError(deleteErr, bucket, object.ObjectName)
-			continue
-		}
-		deleted[i].ObjectName = object.ObjectName
+		i, object := i, object
+		limiter.Go(ctx, func() {
+			_, err := layer.DeleteObject(ctx, bucket, object.ObjectName, opts)
+			if err != nil && !errors.As(err, &minio.ObjectNotFound{}) {
+				errs[i] = ConvertError(err, bucket, object.ObjectName)
+				return
+			}
+			deleted[i].ObjectName = object.ObjectName
+		})
 	}
+
+	limiter.Wait()
+
 	return deleted, errs
 }
 

testsuite/miniogw/gateway_test.go

@@ -4022,6 +4022,7 @@ func runTestWithPathCipher(t *testing.T, pathCipher storj.CipherSuite, test func
 			MaxKeysLimit:                 maxKeysLimit,
 			MaxKeysExhaustiveLimit:       100000,
 			MaxUploadsLimit:              maxUploadsLimit,
+			DeleteObjectsConcurrency:     100,
 		}
 
 		layer, err := miniogw.NewStorjGateway(s3Compatibility).NewGatewayLayer(auth.Credentials{})