fileservice: fix object storage semaphore (#18589)

fix object storage semaphore

Approved by: @sukki37

Author: reusee

pkg/fileservice/io.go

@@ -70,3 +70,11 @@ var ioBufferPool = NewPool(
 	nil,
 	nil,
 )
+
+type readerFunc func([]byte) (int, error)
+
+var _ io.Reader = readerFunc(nil)
+
+func (r readerFunc) Read(p []byte) (n int, err error) {
+	return r(p)
+}

pkg/fileservice/object_storage_semaphore.go

@@ -17,6 +17,7 @@ package fileservice
 import (
 	"context"
 	"io"
+	"sync"
 	"time"
 )
 
@@ -70,14 +71,23 @@ func (o *objectStorageSemaphore) Read(ctx context.Context, key string, min *int6
 		o.release()
 		return nil, err
 	}
-	released := false
+
+	release := sync.OnceFunc(func() {
+		o.release()
+	})
+
 	return &readCloser{
-		r: r,
-		closeFunc: func() error {
-			if !released {
-				o.release()
-				released = true
+		r: readerFunc(func(buf []byte) (n int, err error) {
+			n, err = r.Read(buf)
+			if err != nil {
+				// release if error
+				release()
 			}
+			return
+		}),
+		closeFunc: func() error {
+			// release when close
+			release()
 			return r.Close()
 		},
 	}, nil

pkg/fileservice/s3_fs.go

@@ -113,7 +113,7 @@ func NewS3FS(
 	// limit number of concurrent operations
 	concurrency := args.Concurrency
 	if concurrency == 0 {
-		concurrency = 100
+		concurrency = 1024
 	}
 	fs.storage = newObjectStorageSemaphore(
 		fs.storage,