Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

custom ProviderManager that brokers AddrInfos #751

Merged
merged 8 commits into from
Oct 26, 2021
Merged

custom ProviderManager that brokers AddrInfos #751

Show file tree
Hide file tree
Changes from 5 commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
427889a
first draft of ProviderManager that brokers AddrInfos
petar Oct 18, 2021
a86b4a2
fix dht_test
petar Oct 18, 2021
b11b187
fix provider manager test
petar Oct 18, 2021
78eec1c
fix derecations
petar Oct 18, 2021
60647d7
go mod tidy
petar Oct 18, 2021
0361e76
add error to provider methods
petar Oct 20, 2021
0e473c3
remove ProvidersOptions from DHT
petar Oct 20, 2021
bdcaf4c
naming
petar Oct 22, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
24 changes: 17 additions & 7 deletions dht.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -84,8 +84,8 @@ type IpfsDHT struct {
datastore ds.Datastore // Local data

routingTable *kb.RoutingTable // Array of routing tables for differently distanced nodes
// ProviderManager stores & manages the provider records for this Dht peer.
ProviderManager *providers.ProviderManager
// providerStore stores & manages the provider records for this Dht peer.
providerStore providers.ProviderStore

// manages Routing Table refresh
rtRefreshManager *rtrefresh.RtRefreshManager
Expand Down Expand Up @@ -221,7 +221,9 @@ func New(ctx context.Context, h host.Host, options ...Option) (*IpfsDHT, error)
}
dht.proc.Go(sn.subscribe)
// handle providers
dht.proc.AddChild(dht.ProviderManager.Process())
if mgr, ok := dht.providerStore.(interface{ Process() goprocess.Process }); ok {
dht.proc.AddChild(mgr.Process())
}

// go-routine to make sure we ALWAYS have RT peer addresses in the peerstore
// since RT membership is decoupled from connectivity
Expand Down Expand Up @@ -338,11 +340,14 @@ func makeDHT(ctx context.Context, h host.Host, cfg dhtcfg.Config) (*IpfsDHT, err
// the DHT context should be done when the process is closed
dht.ctx = goprocessctx.WithProcessClosing(ctxTags, dht.proc)

pm, err := providers.NewProviderManager(dht.ctx, h.ID(), cfg.Datastore, cfg.ProvidersOptions...)
if err != nil {
return nil, err
if cfg.ProviderStore != nil {
dht.providerStore = cfg.ProviderStore
} else {
dht.providerStore, err = providers.NewProviderManager(dht.ctx, h.ID(), dht.peerstore, cfg.Datastore, cfg.ProvidersOptions...)
if err != nil {
return nil, fmt.Errorf("initializing default provider manager (%v)", err)
}
}
dht.ProviderManager = pm

dht.rtFreezeTimeout = rtFreezeTimeout

Expand Down Expand Up @@ -413,6 +418,11 @@ func makeRoutingTable(dht *IpfsDHT, cfg dhtcfg.Config, maxLastSuccessfulOutbound
return rt, err
}

// ProviderStore returns the provider storage object for storing and retrieving provider records.
func (dht *IpfsDHT) ProviderStore() providers.ProviderStore {
return dht.providerStore
}

// GetRoutingTableDiversityStats returns the diversity stats for the Routing Table.
func (dht *IpfsDHT) GetRoutingTableDiversityStats() []peerdiversity.CplDiversityStats {
return dht.routingTable.GetDiversityStats()
Expand Down
8 changes: 8 additions & 0 deletions dht_options.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -36,6 +36,14 @@ const DefaultPrefix protocol.ID = "/ipfs"

type Option = dhtcfg.Option

// ProviderStore sets the provider storage manager.
func ProviderStore(ps providers.ProviderStore) Option {
return func(c *dhtcfg.Config) error {
c.ProviderStore = ps
return nil
}
}

// RoutingTableLatencyTolerance sets the maximum acceptable latency for peers
// in the routing table's cluster.
func RoutingTableLatencyTolerance(latency time.Duration) Option {
Expand Down
8 changes: 4 additions & 4 deletions dht_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -592,7 +592,7 @@ func TestLocalProvides(t *testing.T) {

for _, c := range testCaseCids {
for i := 0; i < 3; i++ {
provs := dhts[i].ProviderManager.GetProviders(ctx, c.Hash())
provs := dhts[i].ProviderStore().GetProviders(ctx, c.Hash())
if len(provs) > 0 {
t.Fatal("shouldnt know this")
}
Expand Down Expand Up @@ -1285,7 +1285,7 @@ func TestClientModeConnect(t *testing.T) {

c := testCaseCids[0]
p := peer.ID("TestPeer")
a.ProviderManager.AddProvider(ctx, c.Hash(), p)
a.ProviderStore().AddProvider(ctx, c.Hash(), peer.AddrInfo{ID: p})
time.Sleep(time.Millisecond * 5) // just in case...

provs, err := b.FindProviders(ctx, c)
Expand Down Expand Up @@ -1548,7 +1548,7 @@ func TestProvideDisabled(t *testing.T) {
if err != routing.ErrNotSupported {
t.Fatal("get should have failed on node B")
}
provs := dhtB.ProviderManager.GetProviders(ctx, kHash)
provs := dhtB.ProviderStore().GetProviders(ctx, kHash)
if len(provs) != 0 {
t.Fatal("node B should not have found local providers")
}
Expand All @@ -1564,7 +1564,7 @@ func TestProvideDisabled(t *testing.T) {
t.Fatal("node A should not have found providers")
}
}
provAddrs := dhtA.ProviderManager.GetProviders(ctx, kHash)
provAddrs := dhtA.ProviderStore().GetProviders(ctx, kHash)
if len(provAddrs) != 0 {
t.Fatal("node A should not have found local providers")
}
Expand Down
9 changes: 4 additions & 5 deletions fullrt/dht.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -139,7 +139,7 @@ func NewFullRT(h host.Host, protocolPrefix protocol.ID, options ...Option) (*Ful

ctx, cancel := context.WithCancel(context.Background())

pm, err := providers.NewProviderManager(ctx, h.ID(), dhtcfg.Datastore)
pm, err := providers.NewProviderManager(ctx, h.ID(), h.Peerstore(), dhtcfg.Datastore)
if err != nil {
cancel()
return nil, err
Expand Down Expand Up @@ -762,7 +762,7 @@ func (dht *FullRT) Provide(ctx context.Context, key cid.Cid, brdcst bool) (err e
logger.Debugw("providing", "cid", key, "mh", internal.LoggableProviderRecordBytes(keyMH))

// add self locally
dht.ProviderManager.AddProvider(ctx, keyMH, dht.h.ID())
dht.ProviderManager.AddProvider(ctx, keyMH, peer.AddrInfo{ID: dht.h.ID()})
if !brdcst {
return nil
}
Expand Down Expand Up @@ -1212,10 +1212,9 @@ func (dht *FullRT) findProvidersAsyncRoutine(ctx context.Context, key multihash.
provs := dht.ProviderManager.GetProviders(ctx, key)
for _, p := range provs {
// NOTE: Assuming that this list of peers is unique
if ps.TryAdd(p) {
pi := dht.h.Peerstore().PeerInfo(p)
if ps.TryAdd(p.ID) {
select {
case peerOut <- pi:
case peerOut <- p:
case <-ctx.Done():
return
}
Expand Down
16 changes: 3 additions & 13 deletions handlers.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,7 +8,6 @@ import (
"time"

"github.com/libp2p/go-libp2p-core/peer"
"github.com/libp2p/go-libp2p-core/peerstore"
pstore "github.com/libp2p/go-libp2p-peerstore"

"github.com/gogo/protobuf/proto"
Expand Down Expand Up @@ -318,13 +317,8 @@ func (dht *IpfsDHT) handleGetProviders(ctx context.Context, p peer.ID, pmes *pb.
resp := pb.NewMessage(pmes.GetType(), pmes.GetKey(), pmes.GetClusterLevel())

// setup providers
providers := dht.ProviderManager.GetProviders(ctx, key)

if len(providers) > 0 {
// TODO: pstore.PeerInfos should move to core (=> peerstore.AddrInfos).
infos := pstore.PeerInfos(dht.peerstore, providers)
resp.ProviderPeers = pb.PeerInfosToPBPeers(dht.host.Network(), infos)
}
providers := dht.providerStore.GetProviders(ctx, key)
resp.ProviderPeers = pb.PeerInfosToPBPeers(dht.host.Network(), providers)

// Also send closer peers.
closer := dht.betterPeersToQuery(pmes, p, dht.bucketSize)
Expand Down Expand Up @@ -362,11 +356,7 @@ func (dht *IpfsDHT) handleAddProvider(ctx context.Context, p peer.ID, pmes *pb.M
continue
}

if pi.ID != dht.self { // don't add own addrs.
// add the received addresses to our peerstore.
dht.peerstore.AddAddrs(pi.ID, pi.Addrs, peerstore.ProviderAddrTTL)
}
dht.ProviderManager.AddProvider(ctx, key, p)
dht.providerStore.AddProvider(ctx, key, peer.AddrInfo{ID: p})
}

return nil, nil
Expand Down
3 changes: 2 additions & 1 deletion internal/config/config.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -44,7 +44,8 @@ type Config struct {
MaxRecordAge time.Duration
EnableProviders bool
EnableValues bool
ProvidersOptions []providers.Option
ProvidersOptions []providers.Option //XXX: get rid?
petar marked this conversation as resolved.
Show resolved Hide resolved
ProviderStore providers.ProviderStore
QueryPeerFilter QueryFilterFunc

RoutingTable struct {
Expand Down
25 changes: 20 additions & 5 deletions providers/providers_manager.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,8 @@ import (
"time"

"github.com/libp2p/go-libp2p-core/peer"
peerstore "github.com/libp2p/go-libp2p-core/peerstore"
peerstore_legacy "github.com/libp2p/go-libp2p-peerstore"
petar marked this conversation as resolved.
Show resolved Hide resolved

lru "github.com/hashicorp/golang-lru/simplelru"
ds "github.com/ipfs/go-datastore"
Expand All @@ -30,12 +32,20 @@ var lruCacheSize = 256
var batchBufferSize = 256
var log = logging.Logger("providers")

// ProviderStore represents a store that associates peers and their addresses to keys.
type ProviderStore interface {
AddProvider(ctx context.Context, key []byte, prov peer.AddrInfo)
GetProviders(ctx context.Context, key []byte) []peer.AddrInfo
petar marked this conversation as resolved.
Show resolved Hide resolved
}

// ProviderManager adds and pulls providers out of the datastore,
// caching them in between
type ProviderManager struct {
self peer.ID
// all non channel fields are meant to be accessed only within
// the run method
cache lru.LRUCache
pstore peerstore.Peerstore
dstore *autobatch.Datastore

newprovs chan *addProv
Expand Down Expand Up @@ -86,10 +96,12 @@ type getProv struct {
}

// NewProviderManager constructor
func NewProviderManager(ctx context.Context, local peer.ID, dstore ds.Batching, opts ...Option) (*ProviderManager, error) {
func NewProviderManager(ctx context.Context, local peer.ID, ps peerstore.Peerstore, dstore ds.Batching, opts ...Option) (*ProviderManager, error) {
aschmahmann marked this conversation as resolved.
Show resolved Hide resolved
pm := new(ProviderManager)
pm.self = local
pm.getprovs = make(chan *getProv)
pm.newprovs = make(chan *addProv)
pm.pstore = ps
pm.dstore = autobatch.NewAutoBatching(dstore, batchBufferSize)
cache, err := lru.NewLRU(lruCacheSize, nil)
if err != nil {
Expand Down Expand Up @@ -214,10 +226,13 @@ func (pm *ProviderManager) run(proc goprocess.Process) {
}

// AddProvider adds a provider
func (pm *ProviderManager) AddProvider(ctx context.Context, k []byte, val peer.ID) {
func (pm *ProviderManager) AddProvider(ctx context.Context, k []byte, provInfo peer.AddrInfo) {
if provInfo.ID != pm.self { // don't add own addrs.
pm.pstore.AddAddrs(provInfo.ID, provInfo.Addrs, peerstore.ProviderAddrTTL)
}
prov := &addProv{
key: k,
val: val,
val: provInfo.ID,
}
select {
case pm.newprovs <- prov:
Expand Down Expand Up @@ -255,7 +270,7 @@ func mkProvKey(k []byte) string {

// GetProviders returns the set of providers for the given key.
// This method _does not_ copy the set. Do not modify it.
func (pm *ProviderManager) GetProviders(ctx context.Context, k []byte) []peer.ID {
func (pm *ProviderManager) GetProviders(ctx context.Context, k []byte) []peer.AddrInfo {
gp := &getProv{
key: k,
resp: make(chan []peer.ID, 1), // buffered to prevent sender from blocking
Expand All @@ -269,7 +284,7 @@ func (pm *ProviderManager) GetProviders(ctx context.Context, k []byte) []peer.ID
case <-ctx.Done():
return nil
case peers := <-gp.resp:
return peers
return peerstore_legacy.PeerInfos(pm.pstore, peers)
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think you can use peerstore.AddrInfos here, no?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

No. The newer peerstore does not have this function.

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It's right here https://github.com/libp2p/go-libp2p-core/blob/8bf73f7e2f7edb455ea043ef6d65b9542c99c9ea/peerstore/helpers.go#L8

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It's only present in go-libp2p-core v0.9.0 and up, however bumping to this release introduces a ton of other problems. Many of the libraries that this repo depends on, as well as the tests in the repo, are incompatible with v0.9.0 and up.

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

hmmm... ok I guess. We're going to be using the later go-libp2p-core anyway in places like go-ipfs, but we can leave this for now.

In that case I wouldn't call the other one peerstore_legacy, it's where the common peerstore implementations live and definitely isn't legacy. Perhaps peerstore_impl or something like that.

}
}

Expand Down
Loading