Merge ba07d974ec into f000da3b35

internal/archiver/archiver_test.go

@@ -376,6 +376,62 @@ func BenchmarkArchiverSaveFileSmall(b *testing.B) {
 	}
 }
 
+func BenchmarkArchiverSaveFileConcurrent(b *testing.B) {
+	const fileSize = 4 * 1024
+	const numFiles = 64
+
+	d := TestDir{}
+	for i := 0; i < numFiles; i++ {
+		d[fmt.Sprintf("file%d", i)] = TestFile{
+			Content: string(rtest.Random(23+i, fileSize)),
+		}
+	}
+
+	b.SetBytes(fileSize * numFiles)
+
+	for i := 0; i < b.N; i++ {
+		b.StopTimer()
+		tempdir, repo := prepareTempdirRepoSrc(b, d)
+
+		arch := New(repo, fs.Track{FS: fs.Local{}}, Options{})
+		arch.Error = func(item string, err error) error {
+			b.Errorf("archiver error for %v: %v", item, err)
+			return err
+		}
+		b.StartTimer()
+
+		err := repo.WithBlobUploader(context.Background(), func(ctx context.Context, uploader restic.BlobSaverWithAsync) error {
+			wg, ctx := errgroup.WithContext(ctx)
+			arch.runWorkers(ctx, wg, uploader)
+
+			var futures []futureNode
+			for j := 0; j < numFiles; j++ {
+				filename := filepath.Join(tempdir, fmt.Sprintf("file%d", j))
+				file, err := arch.FS.OpenFile(filename, fs.O_NOFOLLOW, false)
+				if err != nil {
+					b.Fatal(err)
+				}
+
+				fn := arch.fileSaver.Save(ctx, "/", filename, file, func() {}, func() {}, func(*data.Node, ItemStats) {})
+				futures = append(futures, fn)
+			}
+
+			for _, fn := range futures {
+				fnr := fn.take(ctx)
+				if fnr.err != nil {
+					b.Fatal(fnr.err)
+				}
+			}
+
+			arch.stopWorkers()
+			return wg.Wait()
+		})
+		if err != nil {
+			b.Fatal(err)
+		}
+	}
+}
+
 func BenchmarkArchiverSaveFileLarge(b *testing.B) {
 	const fileSize = 40*1024*1024 + 1287898
 	d := TestDir{"file": TestFile{