Skip to content

Commit

Permalink
feat(dht): provider records use multihashes instead of CIDs
Browse files Browse the repository at this point in the history
  • Loading branch information
@aschmahmann
aschmahmann committed Dec 12, 2019
1 parent 5308127 commit 750a532
Show file tree
Hide file tree
Showing 7 changed files with 146 additions and 111 deletions.
7 changes: 4 additions & 3 deletions dht.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -374,11 +374,12 @@ func (dht *IpfsDHT) findPeerSingle(ctx context.Context, p peer.ID, id peer.ID) (
}
}

func (dht *IpfsDHT) findProvidersSingle(ctx context.Context, p peer.ID, key cid.Cid) (*pb.Message, error) {
eip := logger.EventBegin(ctx, "findProvidersSingle", p, key)
func (dht *IpfsDHT) findProvidersSingle(ctx context.Context, p peer.ID, keyCid cid.Cid) (*pb.Message, error) {
eip := logger.EventBegin(ctx, "findProvidersSingle", p, keyCid)
defer eip.Done()

pmes := pb.NewMessage(pb.Message_GET_PROVIDERS, key.Bytes(), 0)
key := keyCid.Hash()
pmes := pb.NewMessage(pb.Message_GET_PROVIDERS, key, 0)
resp, err := dht.sendRequest(ctx, p, pmes)
switch err {
case nil:
Expand Down
33 changes: 27 additions & 6 deletions dht_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@ package dht
import (
"bytes"
"context"
"encoding/binary"
"errors"
"fmt"
"math/rand"
Expand All @@ -15,6 +16,7 @@ import (
"github.com/libp2p/go-libp2p-core/peer"
"github.com/libp2p/go-libp2p-core/peerstore"
"github.com/libp2p/go-libp2p-core/routing"
"github.com/multiformats/go-multihash"
"github.com/multiformats/go-multistream"

"golang.org/x/xerrors"
Expand Down Expand Up @@ -42,8 +44,26 @@ func init() {
for i := 0; i < 100; i++ {
v := fmt.Sprintf("%d -- value", i)

mhv := u.Hash([]byte(v))
testCaseCids = append(testCaseCids, cid.NewCidV0(mhv))
var newCid cid.Cid
switch i % 3 {
case 0:
mhv := u.Hash([]byte(v))
newCid = cid.NewCidV0(mhv)
case 1:
mhv := u.Hash([]byte(v))
newCid = cid.NewCidV1(cid.DagCBOR, mhv)
case 2:
rawMh := make([]byte, 12)
binary.PutUvarint(rawMh, cid.Raw)
binary.PutUvarint(rawMh[1:], 10)
copy(rawMh[2:], []byte(v)[:10])
_, mhv, err := multihash.MHFromBytes(rawMh)
if err != nil {
panic(err)
}
newCid = cid.NewCidV1(cid.Raw, mhv)
}
testCaseCids = append(testCaseCids, newCid)
}
}

Expand Down Expand Up @@ -593,7 +613,7 @@ func TestLocalProvides(t *testing.T) {

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

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

provs, err := b.FindProviders(ctx, c)
Expand Down Expand Up @@ -1504,6 +1524,7 @@ func TestFindClosestPeers(t *testing.T) {

func TestProvideDisabled(t *testing.T) {
k := testCaseCids[0]
kHash := k.Hash()
for i := 0; i < 3; i++ {
enabledA := (i & 0x1) > 0
enabledB := (i & 0x2) > 0
Expand Down Expand Up @@ -1544,7 +1565,7 @@ func TestProvideDisabled(t *testing.T) {
if err != routing.ErrNotSupported {
t.Fatal("get should have failed on node B")
}
provs := dhtB.providers.GetProviders(ctx, k)
provs := dhtB.providers.GetProviders(ctx, kHash)
if len(provs) != 0 {
t.Fatal("node B should not have found local providers")
}
Expand All @@ -1560,7 +1581,7 @@ func TestProvideDisabled(t *testing.T) {
t.Fatal("node A should not have found providers")
}
}
provAddrs := dhtA.providers.GetProviders(ctx, k)
provAddrs := dhtA.providers.GetProviders(ctx, kHash)
if len(provAddrs) != 0 {
t.Fatal("node A should not have found local providers")
}
Expand Down
1 change: 1 addition & 0 deletions go.mod
View file
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,7 @@ require (
github.com/multiformats/go-base32 v0.0.3
github.com/multiformats/go-multiaddr v0.2.0
github.com/multiformats/go-multiaddr-dns v0.2.0
github.com/multiformats/go-multihash v0.0.10
github.com/multiformats/go-multistream v0.1.0
github.com/stretchr/testify v1.4.0
go.opencensus.io v0.22.2
Expand Down
23 changes: 12 additions & 11 deletions handlers.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -12,8 +12,9 @@ import (
"github.com/libp2p/go-libp2p-core/peerstore"
pstore "github.com/libp2p/go-libp2p-peerstore"

mh "github.com/multiformats/go-multihash"

"github.com/gogo/protobuf/proto"
"github.com/ipfs/go-cid"
ds "github.com/ipfs/go-datastore"
u "github.com/ipfs/go-ipfs-util"
pb "github.com/libp2p/go-libp2p-kad-dht/pb"
Expand Down Expand Up @@ -317,26 +318,26 @@ func (dht *IpfsDHT) handleGetProviders(ctx context.Context, p peer.ID, pmes *pb.
logger.SetTag(ctx, "peer", p)

resp := pb.NewMessage(pmes.GetType(), pmes.GetKey(), pmes.GetClusterLevel())
c, err := cid.Cast([]byte(pmes.GetKey()))
_, h, err := mh.MHFromBytes(pmes.GetKey())
if err != nil {
return nil, err
}
logger.SetTag(ctx, "key", c)
logger.SetTag(ctx, "key", h)

// debug logging niceness.
reqDesc := fmt.Sprintf("%s handleGetProviders(%s, %s): ", dht.self, p, c)
reqDesc := fmt.Sprintf("%s handleGetProviders(%s, %s): ", dht.self, p, h)
logger.Debugf("%s begin", reqDesc)
defer logger.Debugf("%s end", reqDesc)

// check if we have this value, to add ourselves as provider.
has, err := dht.datastore.Has(convertToDsKey(c.Bytes()))
has, err := dht.datastore.Has(convertToDsKey(h))
if err != nil && err != ds.ErrNotFound {
logger.Debugf("unexpected datastore error: %v\n", err)
has = false
}

// setup providers
providers := dht.providers.GetProviders(ctx, c)
providers := dht.providers.GetProviders(ctx, h)
if has {
providers = append(providers, dht.self)
logger.Debugf("%s have the value. added self as provider", reqDesc)
Expand Down Expand Up @@ -366,13 +367,13 @@ func (dht *IpfsDHT) handleAddProvider(ctx context.Context, p peer.ID, pmes *pb.M
defer func() { logger.FinishWithErr(ctx, _err) }()
logger.SetTag(ctx, "peer", p)

c, err := cid.Cast([]byte(pmes.GetKey()))
_, h, err := mh.MHFromBytes(pmes.GetKey())
if err != nil {
return nil, err
}
logger.SetTag(ctx, "key", c)
logger.SetTag(ctx, "key", h)

logger.Debugf("%s adding %s as a provider for '%s'\n", dht.self, p, c)
logger.Debugf("%s adding %s as a provider for '%s'\n", dht.self, p, h)

// add provider should use the address given in the message
pinfos := pb.PBPeersToPeerInfos(pmes.GetProviderPeers())
Expand All @@ -389,12 +390,12 @@ func (dht *IpfsDHT) handleAddProvider(ctx context.Context, p peer.ID, pmes *pb.M
continue
}

logger.Debugf("received provider %s for %s (addrs: %s)", p, c, pi.Addrs)
logger.Debugf("received provider %s for %s (addrs: %s)", p, h, pi.Addrs)
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.providers.AddProvider(ctx, c, p)
dht.providers.AddProvider(ctx, h, p)
}

return nil, nil
Expand Down
33 changes: 17 additions & 16 deletions providers/providers.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -9,8 +9,9 @@ import (

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

mh "github.com/multiformats/go-multihash"

lru "github.com/hashicorp/golang-lru/simplelru"
cid "github.com/ipfs/go-cid"
ds "github.com/ipfs/go-datastore"
autobatch "github.com/ipfs/go-datastore/autobatch"
dsq "github.com/ipfs/go-datastore/query"
Expand Down Expand Up @@ -47,12 +48,12 @@ type providerSet struct {
}

type addProv struct {
k cid.Cid
k mh.Multihash
val peer.ID
}

type getProv struct {
k cid.Cid
k mh.Multihash
resp chan []peer.ID
}

Expand All @@ -76,24 +77,24 @@ func NewProviderManager(ctx context.Context, local peer.ID, dstore ds.Batching)

const providersKeyPrefix = "/providers/"

func mkProvKey(k cid.Cid) string {
return providersKeyPrefix + base32.RawStdEncoding.EncodeToString(k.Bytes())
func mkProvKey(k mh.Multihash) string {
return providersKeyPrefix + base32.RawStdEncoding.EncodeToString(k)
}

func (pm *ProviderManager) Process() goprocess.Process {
return pm.proc
}

func (pm *ProviderManager) providersForKey(k cid.Cid) ([]peer.ID, error) {
func (pm *ProviderManager) providersForKey(k mh.Multihash) ([]peer.ID, error) {
pset, err := pm.getProvSet(k)
if err != nil {
return nil, err
}
return pset.providers, nil
}

func (pm *ProviderManager) getProvSet(k cid.Cid) (*providerSet, error) {
cached, ok := pm.providers.Get(k)
func (pm *ProviderManager) getProvSet(k mh.Multihash) (*providerSet, error) {
cached, ok := pm.providers.Get(string(k))
if ok {
return cached.(*providerSet), nil
}
Expand All @@ -104,13 +105,13 @@ func (pm *ProviderManager) getProvSet(k cid.Cid) (*providerSet, error) {
}

if len(pset.providers) > 0 {
pm.providers.Add(k, pset)
pm.providers.Add(string(k), pset)
}

return pset, nil
}

func loadProvSet(dstore ds.Datastore, k cid.Cid) (*providerSet, error) {
func loadProvSet(dstore ds.Datastore, k mh.Multihash) (*providerSet, error) {
res, err := dstore.Query(dsq.Query{Prefix: mkProvKey(k)})
if err != nil {
return nil, err
Expand Down Expand Up @@ -174,20 +175,20 @@ func readTimeValue(data []byte) (time.Time, error) {
return time.Unix(0, nsec), nil
}

func (pm *ProviderManager) addProv(k cid.Cid, p peer.ID) error {
func (pm *ProviderManager) addProv(k mh.Multihash, p peer.ID) error {
now := time.Now()
if provs, ok := pm.providers.Get(k); ok {
if provs, ok := pm.providers.Get(string(k)); ok {
provs.(*providerSet).setVal(p, now)
} // else not cached, just write through

return writeProviderEntry(pm.dstore, k, p, now)
}

func mkProvKeyFor(k cid.Cid, p peer.ID) string {
func mkProvKeyFor(k mh.Multihash, p peer.ID) string {
return mkProvKey(k) + "/" + base32.RawStdEncoding.EncodeToString([]byte(p))
}

func writeProviderEntry(dstore ds.Datastore, k cid.Cid, p peer.ID, t time.Time) error {
func writeProviderEntry(dstore ds.Datastore, k mh.Multihash, p peer.ID, t time.Time) error {
dsk := mkProvKeyFor(k, p)

buf := make([]byte, 16)
Expand Down Expand Up @@ -300,7 +301,7 @@ func (pm *ProviderManager) run(proc goprocess.Process) {
}

// AddProvider adds a provider.
func (pm *ProviderManager) AddProvider(ctx context.Context, k cid.Cid, val peer.ID) {
func (pm *ProviderManager) AddProvider(ctx context.Context, k mh.Multihash, val peer.ID) {
prov := &addProv{
k: k,
val: val,
Expand All @@ -313,7 +314,7 @@ func (pm *ProviderManager) AddProvider(ctx context.Context, k cid.Cid, val peer.

// 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 cid.Cid) []peer.ID {
func (pm *ProviderManager) GetProviders(ctx context.Context, k mh.Multihash) []peer.ID {
gp := &getProv{
k: k,
resp: make(chan []peer.ID, 1), // buffered to prevent sender from blocking
Expand Down
Loading

0 comments on commit 750a532

Please sign in to comment.