feat(storage): enable parallel writes by using per-repo locking

pkg/api/routes.go

@@ -957,8 +957,8 @@ func (rh *RouteHandler) CheckBlob(response http.ResponseWriter, request *http.Re
 	} else {
 		var lockLatency time.Time
 
-		imgStore.RLock(&lockLatency)
-		defer imgStore.RUnlock(&lockLatency)
+		imgStore.RLockRepo(name, &lockLatency)
+		defer imgStore.RUnlockRepo(name, &lockLatency)
 
 		ok, blen, _, err = imgStore.StatBlob(name, digest)
 	}

pkg/extensions/sync/destination.go

@@ -136,9 +136,9 @@ func (registry *DestinationRegistry) CommitImage(imageReference types.ImageRefer
 		}
 
 		for _, manifest := range indexManifest.Manifests {
-			tempImageStore.RLock(&lockLatency)
+			tempImageStore.RLockRepo(repo, &lockLatency)
 			manifestBuf, err := tempImageStore.GetBlobContent(repo, manifest.Digest)
-			tempImageStore.RUnlock(&lockLatency)
+			tempImageStore.RUnlockRepo(repo, &lockLatency)
 
 			if err != nil {
 				registry.log.Error().Str("errorType", common.TypeOf(err)).

pkg/meta/parse.go

@@ -109,8 +109,8 @@ func ParseRepo(repo string, metaDB mTypes.MetaDB, storeController stypes.StoreCo
 
 	var lockLatency time.Time
 
-	imageStore.RLock(&lockLatency)
-	defer imageStore.RUnlock(&lockLatency)
+	imageStore.RLockRepo(repo, &lockLatency)
+	defer imageStore.RUnlockRepo(repo, &lockLatency)
 
 	indexBlob, err := imageStore.GetIndexContent(repo)
 	if err != nil {
@@ -223,8 +223,8 @@ func getCosignSignatureLayersInfo(
 
 	var lockLatency time.Time
 
-	imageStore.RLock(&lockLatency)
-	defer imageStore.RUnlock(&lockLatency)
+	imageStore.RLockRepo(repo, &lockLatency)
+	defer imageStore.RUnlockRepo(repo, &lockLatency)
 
 	for _, layer := range manifestContent.Layers {
 		layerContent, err := imageStore.GetBlobContent(repo, layer.Digest)
@@ -280,8 +280,8 @@ func getNotationSignatureLayersInfo(
 
 	var lockLatency time.Time
 
-	imageStore.RLock(&lockLatency)
-	defer imageStore.RUnlock(&lockLatency)
+	imageStore.RLockRepo(repo, &lockLatency)
+	defer imageStore.RUnlockRepo(repo, &lockLatency)
 
 	layerContent, err := imageStore.GetBlobContent(repo, layer)
 	if err != nil {

pkg/storage/common/common.go

@@ -825,9 +825,6 @@ type DedupeTaskGenerator struct {
 	ImgStore storageTypes.ImageStore
 	// storage dedupe value
 	Dedupe bool
-	// store blobs paths grouped by digest
-	digest         godigest.Digest
-	duplicateBlobs []string
 	/* store processed digest, used for iterating duplicateBlobs one by one
 	and generating a task for each unprocessed one*/
 	lastDigests []godigest.Digest
@@ -866,15 +863,15 @@ func (gen *DedupeTaskGenerator) Next() (scheduler.Task, error) {
 	}
 
 	// get all blobs from storage.imageStore and group them by digest
-	gen.digest, gen.duplicateBlobs, err = gen.ImgStore.GetNextDigestWithBlobPaths(gen.repos, gen.lastDigests)
+	digest, duplicateBlobs, duplicateRepos, err := gen.ImgStore.GetNextDigestWithBlobPaths(gen.repos, gen.lastDigests)
 	if err != nil {
 		gen.Log.Error().Err(err).Str("component", "dedupe").Msg("failed to get next digest")
 
 		return nil, err
 	}
 
 	// if no digests left, then mark the task generator as done
-	if gen.digest == "" {
+	if digest == "" {
 		gen.Log.Info().Str("component", "dedupe").Msg("no digests left, finished")
 
 		gen.done = true
@@ -883,10 +880,10 @@ func (gen *DedupeTaskGenerator) Next() (scheduler.Task, error) {
 	}
 
 	// mark digest as processed before running its task
-	gen.lastDigests = append(gen.lastDigests, gen.digest)
+	gen.lastDigests = append(gen.lastDigests, digest)
 
 	// generate rebuild dedupe task for this digest
-	return newDedupeTask(gen.ImgStore, gen.digest, gen.Dedupe, gen.duplicateBlobs, gen.Log), nil
+	return newDedupeTask(gen.ImgStore, digest, gen.Dedupe, duplicateBlobs, duplicateRepos, gen.Log), nil
 }
 
 func (gen *DedupeTaskGenerator) IsDone() bool {
@@ -899,9 +896,7 @@ func (gen *DedupeTaskGenerator) IsReady() bool {
 
 func (gen *DedupeTaskGenerator) Reset() {
 	gen.lastDigests = []godigest.Digest{}
-	gen.duplicateBlobs = []string{}
 	gen.repos = []string{}
-	gen.digest = ""
 	gen.done = false
 }
 
@@ -911,19 +906,21 @@ type dedupeTask struct {
 	digest godigest.Digest
 	// blobs paths with the same digest ^
 	duplicateBlobs []string
+	duplicateRepos []string
 	dedupe         bool
 	log            zlog.Logger
 }
 
 func newDedupeTask(imgStore storageTypes.ImageStore, digest godigest.Digest, dedupe bool,
-	duplicateBlobs []string, log zlog.Logger,
+	duplicateBlobs, duplicateRepos []string, log zlog.Logger,
 ) *dedupeTask {
-	return &dedupeTask{imgStore, digest, duplicateBlobs, dedupe, log}
+	return &dedupeTask{imgStore, digest, duplicateBlobs, duplicateRepos, dedupe, log}
 }
 
 func (dt *dedupeTask) DoWork(ctx context.Context) error {
 	// run task
-	err := dt.imgStore.RunDedupeForDigest(ctx, dt.digest, dt.dedupe, dt.duplicateBlobs) //nolint: contextcheck
+	err := dt.imgStore.RunDedupeForDigest(ctx, dt.digest, dt.dedupe, dt.duplicateBlobs, //nolint: contextcheck
+		dt.duplicateRepos)
 	if err != nil {
 		// log it
 		dt.log.Error().Err(err).Str("digest", dt.digest.String()).Str("component", "dedupe").

pkg/storage/common/common_test.go

@@ -489,9 +489,9 @@ func TestDedupeGeneratorErrors(t *testing.T) {
 				return []string{"repo1", "repo2"}, nil
 			},
 			GetNextDigestWithBlobPathsFn: func(repos []string, lastDigests []godigest.Digest) (
-				godigest.Digest, []string, error,
+				godigest.Digest, []string, []string, error,
 			) {
-				return "sha256:123", []string{}, ErrTestError
+				return "sha256:123", []string{}, []string{}, ErrTestError
 			},
 		}
 

pkg/storage/gc/gc.go

@@ -107,8 +107,8 @@ func (gc GarbageCollect) cleanRepo(ctx context.Context, repo string) error {
 		return zerr.ErrRepoNotFound
 	}
 
-	gc.imgStore.Lock(&lockLatency)
-	defer gc.imgStore.Unlock(&lockLatency)
+	gc.imgStore.LockRepo(repo, &lockLatency)
+	defer gc.imgStore.UnlockRepo(repo, &lockLatency)
 
 	/* this index (which represents the index.json of this repo) is the root point from which we
 	search for dangling manifests/blobs