Method IterateUnreferencedPacks

repo/content/content_manager_iterate.go:229–292

IterateUnreferencedPacks returns the list of unreferenced storage blobs.

(ctx context.Context, blobPrefixes []blob.ID, parallelism int, callback func(blob.Metadata) error)

// IterateUnreferencedPacks returns the list of unreferenced storage blobs.
func (bm *WriteManager) IterateUnreferencedPacks(ctx context.Context, blobPrefixes []blob.ID, parallelism int, callback func(blob.Metadata) error) error {
	usedPacks, err := bigmap.NewSet(ctx)
	if err != nil {
		return errors.Wrap(err, "new set")
	}

	defer usedPacks.Close(ctx)

	contentlog.Log(ctx, bm.log, "determining blobs in use")
	// find packs in use
	if err := bm.IteratePacks(
		ctx,
		IteratePackOptions{
			Prefixes:                           blobPrefixes,
			IncludePacksWithOnlyDeletedContent: true,
		},
		func(pi PackInfo) error {
			if pi.ContentCount > 0 {
				usedPacks.Put(ctx, []byte(pi.PackID))
			}
			return nil
		}); err != nil {
		return errors.Wrap(err, "error iterating packs")
	}

	if len(blobPrefixes) == 0 {
		blobPrefixes = PackBlobIDPrefixes
	}

	var prefixes []blob.ID

	if parallelism <= len(blobPrefixes) {
		prefixes = append(prefixes, blobPrefixes...)
	} else {
		// iterate {p,q}[0-9,a-f]
		for _, prefix := range blobPrefixes {
			for hexDigit := range 16 {
				prefixes = append(prefixes, blob.ID(fmt.Sprintf("%v%x", prefix, hexDigit)))
			}
		}
	}

	contentlog.Log1(ctx, bm.log, "scanning prefixes",
		blobparam.BlobIDList("prefixes", prefixes))

	var unusedCount atomic.Int32

	if err := blob.IterateAllPrefixesInParallel(ctx, parallelism, bm.st, prefixes,
		func(bm blob.Metadata) error {
			if usedPacks.Contains([]byte(bm.BlobID)) {
				return nil
			}

			unusedCount.Add(1)

			return callback(bm)
		}); err != nil {
		return errors.Wrap(err, "error iterating blobs")