diff --git a/p2p/discover/portal_protocol.go b/p2p/discover/portal_protocol.go index 66e170e70c..d9a39fa452 100644 --- a/p2p/discover/portal_protocol.go +++ b/p2p/discover/portal_protocol.go @@ -169,7 +169,8 @@ func DefaultPortalProtocolConfig() *PortalProtocolConfig { type PortalProtocol struct { table *Table cachedIdsLock sync.Mutex - cachedIds map[string]*enode.Node + cachedNodes map[string]*enode.Node + cachedIds map[string]enode.ID protocolId string protocolName string @@ -213,7 +214,8 @@ func NewPortalProtocol(config *PortalProtocolConfig, protocolId portalwire.Proto closeCtx, cancelCloseCtx := context.WithCancel(context.Background()) protocol := &PortalProtocol{ - cachedIds: make(map[string]*enode.Node), + cachedNodes: make(map[string]*enode.Node), + cachedIds: make(map[string]enode.ID), protocolId: string(protocolId), protocolName: protocolId.Name(), ListenAddr: config.ListenAddr, @@ -330,11 +332,17 @@ func (p *PortalProtocol) setupUDPListening() error { p.cachedIdsLock.Lock() defer p.cachedIdsLock.Unlock() - if n, ok := p.cachedIds[addr.String()]; ok { + if n, ok := p.cachedNodes[addr.String()]; ok { //_, err := p.DiscV5.TalkRequestToID(id, addr, string(portalwire.UTPNetwork), buf) req := &v5wire.TalkRequest{Protocol: string(portalwire.Utp), Message: buf} p.DiscV5.sendFromAnotherThreadWithNode(n, netip.AddrPortFrom(netutil.IPToAddr(addr.IP), uint16(addr.Port)), req) + return len(buf), err + } else if id, ok := p.cachedIds[addr.String()]; ok { + //_, err := p.DiscV5.TalkRequestToID(id, addr, string(portalwire.UTPNetwork), buf) + req := &v5wire.TalkRequest{Protocol: string(portalwire.Utp), Message: buf} + p.DiscV5.sendFromAnotherThread(id, netip.AddrPortFrom(netutil.IPToAddr(addr.IP), uint16(addr.Port)), req) + return len(buf), err } else { p.Log.Warn("not found target node info", "ip", addr.IP.To4().String(), "port", addr.Port, "bufLength", len(buf)) @@ -392,14 +400,23 @@ func (p *PortalProtocol) cacheNode(node *enode.Node) { p.cachedIdsLock.Lock() defer p.cachedIdsLock.Unlock() addr := &net.UDPAddr{IP: node.IP(), Port: node.UDP()} - if _, ok := p.cachedIds[addr.String()]; !ok { - p.cachedIds[addr.String()] = node + if _, ok := p.cachedNodes[addr.String()]; !ok && node != nil { + p.cachedNodes[addr.String()] = node + } +} + +func (p *PortalProtocol) cacheNodeId(id enode.ID, addr *net.UDPAddr) { + p.cachedIdsLock.Lock() + defer p.cachedIdsLock.Unlock() + if (id != enode.ID{}) { + p.cachedIds[addr.String()] = id } } func (p *PortalProtocol) cacheNodeById(id enode.ID, addr *net.UDPAddr) { go func() { - if _, ok := p.cachedIds[addr.String()]; !ok { + p.cacheNodeId(id, addr) + if _, ok := p.cachedNodes[addr.String()]; !ok { n := p.ResolveNodeId(id) if n != nil { p.cacheNode(n) diff --git a/p2p/discover/v5_udp.go b/p2p/discover/v5_udp.go index 87d5cb4882..0422e9009d 100644 --- a/p2p/discover/v5_udp.go +++ b/p2p/discover/v5_udp.go @@ -604,12 +604,17 @@ func (t *UDPv5) dispatch() { c.ch = make(chan v5wire.Packet, 1) c.err = make(chan error, 1) // Assign request ID. - crand.Read(c.reqid) - c.packet.SetRequestID(c.reqid) - newNonce, _ := t.send(c.id, c.addr, c.packet, nil) - c.nonce = newNonce - t.activeCallByAuth[newNonce] = c + if tq, ok := r.msg.(*v5wire.TalkRequest); ok { + if len(tq.ReqID) == 0 { + crand.Read(c.reqid) + c.packet.SetRequestID(c.reqid) + } + } + nonce, _ := t.send(c.id, c.addr, c.packet, nil) + c.nonce = nonce + t.activeCallByAuth[nonce] = c t.startResponseTimeout(c) + //t.send(r.destID, r.destAddr, r.msg, nil) case p := <-t.packetInCh: t.handlePacket(p.Data, p.Addr) diff --git a/portalnetwork/history/history_network_test.go b/portalnetwork/history/history_network_test.go index e1bba68631..b701bcacff 100644 --- a/portalnetwork/history/history_network_test.go +++ b/portalnetwork/history/history_network_test.go @@ -140,6 +140,11 @@ func TestGetContentByKey(t *testing.T) { contentId := historyNetwork1.portalProtocol.ToContentId(headerEntry.key) err = historyNetwork1.portalProtocol.Put(headerEntry.key, contentId, headerEntry.value) require.NoError(t, err) + + header, err = historyNetwork1.GetBlockHeader(headerEntry.key[1:]) + require.NoError(t, err) + require.NotNil(t, header) + // get content from historyNetwork1 header, err = historyNetwork2.GetBlockHeader(headerEntry.key[1:]) require.NoError(t, err) diff --git a/portalnetwork/history/storage.go b/portalnetwork/history/storage.go index 7878093f6b..488d2ff362 100644 --- a/portalnetwork/history/storage.go +++ b/portalnetwork/history/storage.go @@ -452,7 +452,7 @@ func (p *ContentStorage) deleteContentOutOfRadius(radius *uint256.Int) error { return err } count, _ := res.RowsAffected() - p.log.Trace("delete %d items", count) + p.log.Trace("delete items", "count", count) return err }