<commit_before>package reprovide

import (
	"context"
	"errors"

	blocks "github.com/ipfs/go-ipfs/blocks/blockstore"
	merkledag "github.com/ipfs/go-ipfs/merkledag"
	pin "github.com/ipfs/go-ipfs/pin"

	cid "<KEY>"
)

// NewBlockstoreProvider returns key provider using bstore.AllKeysChan
func NewBlockstoreProvider(bstore blocks.Blockstore) keyChanFunc {
	return func(ctx context.Context) (<-chan *cid.Cid, error) {
		return bstore.AllKeysChan(ctx)
	}
}

// NewPinnedProvider returns provider supplying pinned keys
func NewPinnedProvider(pinning pin.Pinner, dag merkledag.DAGService, onlyRoots bool) keyChanFunc {
	return func(ctx context.Context) (<-chan *cid.Cid, error) {
		set, err := pinSet(ctx, pinning, dag, onlyRoots)
		if err != nil {
			return nil, err
		}

		outCh := make(chan *cid.Cid)
		go func() {
			defer close(outCh)
			set.ForEach(func(c *cid.Cid) error {
				select {
				case <-ctx.Done():
					return errors.New("context cancelled")
				case outCh <- c:
				}
				return nil
			})
		}()

		return outCh, nil
	}
}

func pinSet(ctx context.Context, pinning pin.Pinner, dag merkledag.DAGService, onlyRoots bool) (*cid.Set, error) {
	set := cid.NewSet()
	for _, key := range pinning.DirectKeys() {
		set.Add(key)
	}

	for _, key := range pinning.RecursiveKeys() {
		set.Add(key)

		if !onlyRoots {
			err := merkledag.EnumerateChildren(ctx, dag.GetLinks, key, set.Visit)
			if err != nil {
				return nil, err
			}
		}
	}

	return set, nil
}
<commit_msg>reprovider: reduce pinned strategy i/o overhead<commit_after>package reprovide

import (
	"context"

	blocks "github.com/ipfs/go-ipfs/blocks/blockstore"
	merkledag "github.com/ipfs/go-ipfs/merkledag"
	pin "github.com/ipfs/go-ipfs/pin"

	cid "<KEY>"
)

// NewBlockstoreProvider returns key provider using bstore.AllKeysChan
func NewBlockstoreProvider(bstore blocks.Blockstore) keyChanFunc {
	return func(ctx context.Context) (<-chan *cid.Cid, error) {
		return bstore.AllKeysChan(ctx)
	}
}

// NewPinnedProvider returns provider supplying pinned keys
func NewPinnedProvider(pinning pin.Pinner, dag merkledag.DAGService, onlyRoots bool) keyChanFunc {
	return func(ctx context.Context) (<-chan *cid.Cid, error) {
		set, err := pinSet(ctx, pinning, dag, onlyRoots)
		if err != nil {
			return nil, err
		}

		outCh := make(chan *cid.Cid)
		go func() {
			defer close(outCh)
			for c := range set.new {
				select {
				case <-ctx.Done():
					return
				case outCh <- c:
				}
			}

		}()

		return outCh, nil
	}
}

func pinSet(ctx context.Context, pinning pin.Pinner, dag merkledag.DAGService, onlyRoots bool) (*streamingSet, error) {
	set := newStreamingSet()

	go func() {
		for _, key := range pinning.DirectKeys() {
			set.add(key)
		}

		for _, key := range pinning.RecursiveKeys() {
			set.add(key)

			if !onlyRoots {
				err := merkledag.EnumerateChildren(ctx, dag.GetLinks, key, set.add)
				if err != nil {
					return //TODO: propagate to chan / log?
				}
			}
		}

		close(set.new)
	}()

	return set, nil
}

type streamingSet struct {
	set map[string]struct{}
	new chan *cid.Cid
}

// NewSet initializes and returns a new Set.
func newStreamingSet() *streamingSet {
	return &streamingSet{
		set: make(map[string]struct{}),
		new: make(chan *cid.Cid),
	}
}

// has returns if the Set contains a given Cid.
func (s *streamingSet) has(c *cid.Cid) bool {
	_, ok := s.set[string(c.Bytes())]
	return ok
}

// add adds a Cid to the set only if it is
// not in it already.
func (s *streamingSet) add(c *cid.Cid) bool {
	if !s.has(c) {
		s.set[string(c.Bytes())] = struct{}{}
		s.new <- c
		return true
	}

	return false
}

{'fraction_non_alphanumeric': 0.13407510431154382, 'fraction_numerical': 0.0, 'mean_word_length': 3.4782067247820674, 'pattern_counts': {'":': 0, '<': 14, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6321177', 'n_tokens_mistral': 1484, 'n_tokens_neox': 1340, 'n_words': 334}