Enable ConcurrentStreamParts option and add detailed timing to S3/MinIO benchmarks

Author: sandraromanchenko

pbm/storage/mio/minio.go

@@ -181,8 +181,9 @@ func (m *Minio) Save(name string, data io.Reader, options ...storage.Option) err
 	}
 
 	putOpts := minio.PutObjectOptions{
-		PartSize:   uint64(partSize),
-		NumThreads: uint(max(runtime.NumCPU()/2, 1)),
+		PartSize:              uint64(partSize),
+		NumThreads:            uint(max(runtime.NumCPU()/2, 1)),
+		ConcurrentStreamParts: true,
 	}
 	_, err := m.cl.PutObject(
 		context.Background(),

pbm/storage/mio/minio_test.go

@@ -275,6 +275,7 @@ go test ./pbm/storage/mio -bench=BenchmarkMinioPutObject -run=^$ -v \
 -part-size=100
 */
 func BenchmarkMinioPutObject(b *testing.B) {
+	var uploadTimes []time.Duration
 	numThreds := uint(max(runtime.GOMAXPROCS(0), 1))
 	fsize := *fileSize * 1024 * 1024
 	pSize := *partSize * 1024 * 1024
@@ -306,14 +307,15 @@ func BenchmarkMinioPutObject(b *testing.B) {
 		r := io.LimitReader(infR, fsize)
 
 		fname := time.Now().Format("2006-01-02T15:04:05")
-		b.Logf("uploading file: %s ....", fname)
 
 		putOpts := minio.PutObjectOptions{
-			PartSize:   uint64(pSize),
-			NumThreads: numThreds,
+			PartSize:              uint64(pSize),
+			NumThreads:            numThreds,
+			ConcurrentStreamParts: true,
 		}
 
 		b.StartTimer()
+		startTime := time.Now()
 		_, err = mc.PutObject(
 			context.Background(),
 			bucket,
@@ -322,10 +324,22 @@ func BenchmarkMinioPutObject(b *testing.B) {
 			-1,
 			putOpts,
 		)
+		uploadDuration := time.Since(startTime)
+		uploadTimes = append(uploadTimes, uploadDuration)
+		b.Logf("uploading file: %s, completed in: %v", fname, uploadDuration)
 		if err != nil {
 			b.Fatalf("put object: %v", err)
 		}
 	}
+
+	if len(uploadTimes) > 0 {
+		var totalDuration time.Duration
+		for _, duration := range uploadTimes {
+			totalDuration += duration
+		}
+		averageTime := totalDuration / time.Duration(len(uploadTimes))
+		b.Logf("average upload time: %v", averageTime)
+	}
 }
 
 // BenchmarkMinioStorageSave measures the performance of uploading file on the

pbm/storage/s3/s3_test.go

@@ -264,6 +264,7 @@ go test ./pbm/storage/s3 -bench=BenchmarkS3Upload -run=^$ -v \
 -part-size=100
 */
 func BenchmarkS3Upload(b *testing.B) {
+	var uploadTimes []time.Duration
 	numThreds := max(runtime.GOMAXPROCS(0), 1)
 	fsize := *fileSize * 1024 * 1024
 	pSize := *partSize * 1024 * 1024
@@ -298,7 +299,6 @@ func BenchmarkS3Upload(b *testing.B) {
 		r := io.LimitReader(infR, fsize)
 
 		fname := time.Now().Format("2006-01-02T15:04:05")
-		b.Logf("uploading file: %s ....", fname)
 
 		putInput := &s3.PutObjectInput{
 			Bucket:       aws.String(bucket),
@@ -308,15 +308,28 @@ func BenchmarkS3Upload(b *testing.B) {
 		}
 
 		b.StartTimer()
+		startTime := time.Now()
 		_, err := manager.NewUploader(s3Client, func(u *manager.Uploader) {
 			u.PartSize = pSize
 			u.LeavePartsOnError = true
 			u.Concurrency = numThreds
 		}).Upload(context.Background(), putInput)
+		uploadDuration := time.Since(startTime)
+		uploadTimes = append(uploadTimes, uploadDuration)
+		b.Logf("uploading file: %s, completed in: %v", fname, uploadDuration)
 		if err != nil {
 			b.Fatalf("put object: %v", err)
 		}
 	}
+
+	if len(uploadTimes) > 0 {
+		var totalDuration time.Duration
+		for _, duration := range uploadTimes {
+			totalDuration += duration
+		}
+		averageTime := totalDuration / time.Duration(len(uploadTimes))
+		b.Logf("average upload time: %v", averageTime)
+	}
 }
 
 // BenchmarkS3StorageSave measures the performance of uploading file on the