diff --git a/routing/dht/dht.go b/routing/dht/dht.go index 1573f3477..fb7c5a49b 100644 --- a/routing/dht/dht.go +++ b/routing/dht/dht.go @@ -102,8 +102,8 @@ func (dht *IpfsDHT) Connect(ctx context.Context, npeer peer.ID) error { return nil } -// putValueToNetwork stores the given key/value pair at the peer 'p' -func (dht *IpfsDHT) putValueToNetwork(ctx context.Context, p peer.ID, +// putValueToPeer stores the given key/value pair at the peer 'p' +func (dht *IpfsDHT) putValueToPeer(ctx context.Context, p peer.ID, key u.Key, rec *pb.Record) error { pmes := pb.NewMessage(pb.Message_PUT_VALUE, string(key), 0) @@ -237,12 +237,12 @@ func (dht *IpfsDHT) Update(ctx context.Context, p peer.ID) { } // FindLocal looks for a peer with a given ID connected to this dht and returns the peer and the table it was found in. -func (dht *IpfsDHT) FindLocal(id peer.ID) (peer.PeerInfo, *kb.RoutingTable) { +func (dht *IpfsDHT) FindLocal(id peer.ID) peer.PeerInfo { p := dht.routingTable.Find(id) if p != "" { - return dht.peerstore.PeerInfo(p), dht.routingTable + return dht.peerstore.PeerInfo(p) } - return peer.PeerInfo{}, nil + return peer.PeerInfo{} } // findPeerSingle asks peer 'p' if they know where the peer with id 'id' is @@ -256,26 +256,6 @@ func (dht *IpfsDHT) findProvidersSingle(ctx context.Context, p peer.ID, key u.Ke return dht.sendRequest(ctx, p, pmes) } -func (dht *IpfsDHT) addProviders(key u.Key, pbps []*pb.Message_Peer) []peer.ID { - peers := pb.PBPeersToPeerInfos(pbps) - - var provArr []peer.ID - for _, pi := range peers { - p := pi.ID - - // Dont add outselves to the list - if p == dht.self { - continue - } - - log.Debugf("%s adding provider: %s for %s", dht.self, p, key) - // TODO(jbenet) ensure providers is idempotent - dht.providers.AddProvider(key, p) - provArr = append(provArr, p) - } - return provArr -} - // nearestPeersToQuery returns the routing tables closest peers. func (dht *IpfsDHT) nearestPeersToQuery(pmes *pb.Message, count int) []peer.ID { key := u.Key(pmes.GetKey()) diff --git a/routing/dht/handlers.go b/routing/dht/handlers.go index e8edaa5eb..acb052248 100644 --- a/routing/dht/handlers.go +++ b/routing/dht/handlers.go @@ -39,7 +39,7 @@ func (dht *IpfsDHT) handlerForMsgType(t pb.Message_MessageType) dhtHandler { } func (dht *IpfsDHT) handleGetValue(ctx context.Context, p peer.ID, pmes *pb.Message) (*pb.Message, error) { - log.Debugf("%s handleGetValue for key: %s\n", dht.self, pmes.GetKey()) + log.Debugf("%s handleGetValue for key: %s", dht.self, pmes.GetKey()) // setup response resp := pb.NewMessage(pmes.GetType(), pmes.GetKey(), pmes.GetClusterLevel()) @@ -127,7 +127,7 @@ func (dht *IpfsDHT) handlePutValue(ctx context.Context, p peer.ID, pmes *pb.Mess } err = dht.datastore.Put(dskey, data) - log.Debugf("%s handlePutValue %v\n", dht.self, dskey) + log.Debugf("%s handlePutValue %v", dht.self, dskey) return pmes, err } @@ -137,9 +137,6 @@ func (dht *IpfsDHT) handlePing(_ context.Context, p peer.ID, pmes *pb.Message) ( } func (dht *IpfsDHT) handleFindPeer(ctx context.Context, p peer.ID, pmes *pb.Message) (*pb.Message, error) { - log.Errorf("handle find peer %s start", p) - defer log.Errorf("handle find peer %s end", p) - resp := pb.NewMessage(pmes.GetType(), "", pmes.GetClusterLevel()) var closest []peer.ID diff --git a/routing/dht/routing.go b/routing/dht/routing.go index a0334451f..8b0bb1670 100644 --- a/routing/dht/routing.go +++ b/routing/dht/routing.go @@ -50,7 +50,7 @@ func (dht *IpfsDHT) PutValue(ctx context.Context, key u.Key, value []byte) error wg.Add(1) go func(p peer.ID) { defer wg.Done() - err := dht.putValueToNetwork(ctx, p, key, rec) + err := dht.putValueToPeer(ctx, p, key, rec) if err != nil { log.Errorf("failed putting value to peer: %s", err) } @@ -125,12 +125,18 @@ func (dht *IpfsDHT) Provide(ctx context.Context, key u.Key) error { return err } + wg := sync.WaitGroup{} for p := range peers { - err := dht.putProvider(ctx, p, string(key)) - if err != nil { - log.Error(err) - } + wg.Add(1) + go func(p peer.ID) { + defer wg.Done() + err := dht.putProvider(ctx, p, string(key)) + if err != nil { + log.Error(err) + } + }(p) } + wg.Wait() return nil } @@ -144,7 +150,6 @@ func (dht *IpfsDHT) FindProviders(ctx context.Context, key u.Key) ([]peer.PeerIn } func (dht *IpfsDHT) getClosestPeers(ctx context.Context, key u.Key, count int) (<-chan peer.ID, error) { - log.Error("Get Closest Peers") tablepeers := dht.routingTable.NearestPeers(kb.ConvertKey(key), AlphaValue) if len(tablepeers) == 0 { return nil, kb.ErrLookupFailure @@ -170,15 +175,12 @@ func (dht *IpfsDHT) getClosestPeers(ctx context.Context, key u.Key, count int) ( go func() { wg.Wait() close(out) - log.Error("Closing closest peer chan") }() return out, nil } func (dht *IpfsDHT) getClosestPeersRecurse(ctx context.Context, key u.Key, p peer.ID, peers *pset.PeerSet, peerOut chan<- peer.ID) { - log.Error("closest peers recurse") - defer log.Error("closest peers recurse end") closer, err := dht.closerPeersSingle(ctx, key, p) if err != nil { log.Errorf("error getting closer peers: %s", err) @@ -204,8 +206,6 @@ func (dht *IpfsDHT) getClosestPeersRecurse(ctx context.Context, key u.Key, p pee } func (dht *IpfsDHT) closerPeersSingle(ctx context.Context, key u.Key, p peer.ID) ([]peer.ID, error) { - log.Errorf("closest peers single %s %s", p, key) - defer log.Errorf("closest peers single end %s %s", p, key) pmes, err := dht.findPeerSingle(ctx, p, peer.ID(key)) if err != nil { return nil, err @@ -236,6 +236,7 @@ func (dht *IpfsDHT) FindProvidersAsync(ctx context.Context, key u.Key, count int func (dht *IpfsDHT) findProvidersAsyncRoutine(ctx context.Context, key u.Key, count int, peerOut chan peer.PeerInfo) { defer close(peerOut) + defer log.Event(ctx, "findProviders end", &key) log.Debugf("%s FindProviders %s", dht.self, key) ps := pset.NewLimited(count) @@ -294,40 +295,11 @@ func (dht *IpfsDHT) findProvidersAsyncRoutine(ctx context.Context, key u.Key, co } } -func (dht *IpfsDHT) addPeerListAsync(ctx context.Context, k u.Key, peers []*pb.Message_Peer, ps *pset.PeerSet, count int, out chan peer.PeerInfo) { - var wg sync.WaitGroup - peerInfos := pb.PBPeersToPeerInfos(peers) - for _, pi := range peerInfos { - wg.Add(1) - go func(pi peer.PeerInfo) { - defer wg.Done() - - p := pi.ID - if err := dht.ensureConnectedToPeer(ctx, p); err != nil { - log.Errorf("%s", err) - return - } - - dht.providers.AddProvider(k, p) - if ps.TryAdd(p) { - select { - case out <- pi: - case <-ctx.Done(): - return - } - } else if ps.Size() >= count { - return - } - }(pi) - } - wg.Wait() -} - // FindPeer searches for a peer with given ID. func (dht *IpfsDHT) FindPeer(ctx context.Context, id peer.ID) (peer.PeerInfo, error) { // Check if were already connected to them - if pi, _ := dht.FindLocal(id); pi.ID != "" { + if pi := dht.FindLocal(id); pi.ID != "" { return pi, nil }