Fix(multipart-put): add parts on multipart complete (#403)

Warh40k · web-flow · commit fc415c83d0ff · 2025-06-20T11:22:47.000+03:00
fix(multipart-put): add parts on multipart complete

Signed-off-by: Warh40k &lt;nicita.zinkevitch@yandex.ru&gt;
diff --git a/pkg/bench/multipart_put.go b/pkg/bench/multipart_put.go
@@ -50,7 +50,7 @@ func (g *MultipartPut) Start(ctx context.Context, wait chan struct{}) error {
 					continue
 				}
 
-				err = g.uploadParts(ctx, thread, objectName, uploadID)
+				parts, err := g.uploadParts(ctx, thread, objectName, uploadID)
 				if errors.Is(err, context.Canceled) {
 					return nil
 				}
@@ -59,7 +59,7 @@ func (g *MultipartPut) Start(ctx context.Context, wait chan struct{}) error {
 					continue
 				}
 
-				err = g.completeMultipartUpload(ctx, objectName, uploadID)
+				err = g.completeMultipartUpload(ctx, objectName, uploadID, parts)
 				if err != nil {
 					g.Error("complete multipart upload")
 				}
@@ -89,13 +89,15 @@ func (g *MultipartPut) createMultupartUpload(ctx context.Context, objectName str
 	return c.NewMultipartUpload(nonTerm, g.Bucket, objectName, g.PutOpts)
 }
 
-func (g *MultipartPut) uploadParts(ctx context.Context, thread uint16, objectName, uploadID string) error {
+func (g *MultipartPut) uploadParts(ctx context.Context, thread uint16, objectName, uploadID string) ([]minio.CompletePart, error) {
 	partIdxCh := make(chan int, g.PartsNumber)
 	for i := 0; i < g.PartsNumber; i++ {
 		partIdxCh <- i + 1
 	}
 	close(partIdxCh)
 
+	parts := make([]minio.CompletePart, g.PartsNumber)
+
 	eg, ctx := errgroup.WithContext(ctx)
 
 	// Non-terminating context.
@@ -157,6 +159,7 @@ func (g *MultipartPut) uploadParts(ctx context.Context, thread uint16, objectNam
 					}
 					g.Error(err)
 				}
+				parts[res.PartNumber-1] = minio.CompletePart{PartNumber: res.PartNumber, ETag: res.ETag}
 
 				g.Collector.Receiver() <- op
 			}
@@ -165,16 +168,24 @@ func (g *MultipartPut) uploadParts(ctx context.Context, thread uint16, objectNam
 		})
 	}
 
-	return eg.Wait()
+	err := eg.Wait()
+
+	return parts, err
 }
 
-func (g *MultipartPut) completeMultipartUpload(_ context.Context, objectName, uploadID string) error {
+func (g *MultipartPut) completeMultipartUpload(ctx context.Context, objectName, uploadID string, parts []minio.CompletePart) error {
+	select {
+	case <-ctx.Done():
+		return nil
+	default:
+	}
+
 	// Non-terminating context.
 	nonTerm := context.Background()
 
 	cl, done := g.Client()
 	c := minio.Core{Client: cl}
 	defer done()
-	_, err := c.CompleteMultipartUpload(nonTerm, g.Bucket, objectName, uploadID, nil, g.PutOpts)
+	_, err := c.CompleteMultipartUpload(nonTerm, g.Bucket, objectName, uploadID, parts, g.PutOpts)
 	return err
 }

Original file line number	Diff line number	Diff line change
`@@ -50,7 +50,7 @@ func (g *MultipartPut) Start(ctx context.Context, wait chan struct{}) error {`
`50`	`50`	`continue`
`51`	`51`	`}`
`52`	`52`
`53`		`- err = g.uploadParts(ctx, thread, objectName, uploadID)`
	`53`	`+ parts, err := g.uploadParts(ctx, thread, objectName, uploadID)`
`54`	`54`	`if errors.Is(err, context.Canceled) {`
`55`	`55`	`return nil`
`56`	`56`	`}`
`@@ -59,7 +59,7 @@ func (g *MultipartPut) Start(ctx context.Context, wait chan struct{}) error {`
`59`	`59`	`continue`
`60`	`60`	`}`
`61`	`61`
`62`		`- err = g.completeMultipartUpload(ctx, objectName, uploadID)`
	`62`	`+ err = g.completeMultipartUpload(ctx, objectName, uploadID, parts)`
`63`	`63`	`if err != nil {`
`64`	`64`	`g.Error("complete multipart upload")`
`65`	`65`	`}`
`@@ -89,13 +89,15 @@ func (g *MultipartPut) createMultupartUpload(ctx context.Context, objectName str`
`89`	`89`	`return c.NewMultipartUpload(nonTerm, g.Bucket, objectName, g.PutOpts)`
`90`	`90`	`}`
`91`	`91`
`92`		`-func (g *MultipartPut) uploadParts(ctx context.Context, thread uint16, objectName, uploadID string) error {`
	`92`	`+func (g *MultipartPut) uploadParts(ctx context.Context, thread uint16, objectName, uploadID string) ([]minio.CompletePart, error) {`
`93`	`93`	`partIdxCh := make(chan int, g.PartsNumber)`
`94`	`94`	`for i := 0; i < g.PartsNumber; i++ {`
`95`	`95`	`partIdxCh <- i + 1`
`96`	`96`	`}`
`97`	`97`	`close(partIdxCh)`
`98`	`98`
	`99`	`+ parts := make([]minio.CompletePart, g.PartsNumber)`
	`100`	`+`
`99`	`101`	`eg, ctx := errgroup.WithContext(ctx)`
`100`	`102`
`101`	`103`	`// Non-terminating context.`
`@@ -157,6 +159,7 @@ func (g *MultipartPut) uploadParts(ctx context.Context, thread uint16, objectNam`
`157`	`159`	`}`
`158`	`160`	`g.Error(err)`
`159`	`161`	`}`
	`162`	`+ parts[res.PartNumber-1] = minio.CompletePart{PartNumber: res.PartNumber, ETag: res.ETag}`
`160`	`163`
`161`	`164`	`g.Collector.Receiver() <- op`
`162`	`165`	`}`
`@@ -165,16 +168,24 @@ func (g *MultipartPut) uploadParts(ctx context.Context, thread uint16, objectNam`
`165`	`168`	`})`
`166`	`169`	`}`
`167`	`170`
`168`		`- return eg.Wait()`
	`171`	`+ err := eg.Wait()`
	`172`	`+`
	`173`	`+ return parts, err`
`169`	`174`	`}`
`170`	`175`
`171`		`-func (g *MultipartPut) completeMultipartUpload(_ context.Context, objectName, uploadID string) error {`
	`176`	`+func (g *MultipartPut) completeMultipartUpload(ctx context.Context, objectName, uploadID string, parts []minio.CompletePart) error {`
	`177`	`+ select {`
	`178`	`+ case <-ctx.Done():`
	`179`	`+ return nil`
	`180`	`+ default:`
	`181`	`+ }`
	`182`	`+`
`172`	`183`	`// Non-terminating context.`
`173`	`184`	`nonTerm := context.Background()`
`174`	`185`
`175`	`186`	`cl, done := g.Client()`
`176`	`187`	`c := minio.Core{Client: cl}`
`177`	`188`	`defer done()`
`178`		`- _, err := c.CompleteMultipartUpload(nonTerm, g.Bucket, objectName, uploadID, nil, g.PutOpts)`
	`189`	`+ _, err := c.CompleteMultipartUpload(nonTerm, g.Bucket, objectName, uploadID, parts, g.PutOpts)`
`179`	`190`	`return err`
`180`	`191`	`}`