responsecache.go 2.94 KB
Newer Older
1 2 3 4 5 6
package responsecache

import (
	"fmt"
	"sync"

7
	blocks "github.com/ipfs/go-block-format"
Hannah Howard's avatar
Hannah Howard committed
8
	logging "github.com/ipfs/go-log"
9
	"github.com/ipld/go-ipld-prime"
10
	cidlink "github.com/ipld/go-ipld-prime/linking/cid"
Hannah Howard's avatar
Hannah Howard committed
11 12 13 14

	"github.com/ipfs/go-graphsync"
	"github.com/ipfs/go-graphsync/linktracker"
	"github.com/ipfs/go-graphsync/metadata"
15 16
)

17 18
var log = logging.Logger("graphsync")

19 20
// UnverifiedBlockStore is an interface for storing blocks
// as they come in and removing them as they are verified
21 22
type UnverifiedBlockStore interface {
	PruneBlocks(func(ipld.Link) bool)
23
	PruneBlock(ipld.Link)
24 25 26 27
	VerifyBlock(ipld.Link) ([]byte, error)
	AddUnverifiedBlock(ipld.Link, []byte)
}

28 29
// ResponseCache maintains a store of unverified blocks and response
// data about links for loading, and prunes blocks as needed.
30 31 32 33 34 35 36
type ResponseCache struct {
	responseCacheLk sync.RWMutex

	linkTracker          *linktracker.LinkTracker
	unverifiedBlockStore UnverifiedBlockStore
}

37
// New initializes a new ResponseCache using the given unverified block store.
38 39 40 41 42 43 44
func New(unverifiedBlockStore UnverifiedBlockStore) *ResponseCache {
	return &ResponseCache{
		linkTracker:          linktracker.New(),
		unverifiedBlockStore: unverifiedBlockStore,
	}
}

45 46
// FinishRequest indicate there is no more need to track blocks tied to this
// response
47
func (rc *ResponseCache) FinishRequest(requestID graphsync.RequestID) {
48 49 50 51 52 53 54 55 56
	rc.responseCacheLk.Lock()
	rc.linkTracker.FinishRequest(requestID)

	rc.unverifiedBlockStore.PruneBlocks(func(link ipld.Link) bool {
		return rc.linkTracker.BlockRefCount(link) == 0
	})
	rc.responseCacheLk.Unlock()
}

57
// AttemptLoad attempts to laod the given block from the cache
58
func (rc *ResponseCache) AttemptLoad(requestID graphsync.RequestID, link ipld.Link) ([]byte, error) {
59 60 61
	rc.responseCacheLk.Lock()
	defer rc.responseCacheLk.Unlock()
	if rc.linkTracker.IsKnownMissingLink(requestID, link) {
62
		return nil, fmt.Errorf("Remote Peer Is Missing Block: %s", link.String())
63 64 65 66 67
	}
	data, _ := rc.unverifiedBlockStore.VerifyBlock(link)
	return data, nil
}

68 69
// ProcessResponse processes incoming response data, adding unverified blocks,
// and tracking link metadata from a remote peer
70
func (rc *ResponseCache) ProcessResponse(responses map[graphsync.RequestID]metadata.Metadata,
71 72 73 74
	blks []blocks.Block) {
	rc.responseCacheLk.Lock()

	for _, block := range blks {
75
		log.Debugf("Received block from network: %s", block.Cid().String())
76 77 78 79 80
		rc.unverifiedBlockStore.AddUnverifiedBlock(cidlink.Link{Cid: block.Cid()}, block.RawData())
	}

	for requestID, md := range responses {
		for _, item := range md {
81
			log.Debugf("Traverse link %s on request ID %d", item.Link.String(), requestID)
82
			rc.linkTracker.RecordLinkTraversal(requestID, cidlink.Link{Cid: item.Link}, item.BlockPresent)
83 84 85 86
		}
	}

	// prune unused blocks right away
87 88 89 90 91
	for _, block := range blks {
		if rc.linkTracker.BlockRefCount(cidlink.Link{Cid: block.Cid()}) == 0 {
			rc.unverifiedBlockStore.PruneBlock(cidlink.Link{Cid: block.Cid()})
		}
	}
92 93 94

	rc.responseCacheLk.Unlock()
}