forks/go-ethereum
1
0
Fork 1
mirror of https://github.com/ethereum/go-ethereum.git synced 2026-08-15 00:13:46 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Merge pull request #107 from thinkAfCod/portal_fix

Browse source 
fix: connId repeats
This commit is contained in:
thinkAfCod 2024-05-16 21:20:50 +08:00 committed by GitHub
commit 48ab327462
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
1 changed files with 10 additions and 5 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

15
p2p/discover/portal_protocol.go
View file

@ -31,6 +31,7 @@ import (
ssz "github.com/ferranbt/fastssz" ssz "github.com/ferranbt/fastssz"
"github.com/holiman/uint256" "github.com/holiman/uint256"
"github.com/optimism-java/utp-go" "github.com/optimism-java/utp-go"
"github.com/optimism-java/utp-go/libutp"
"github.com/prysmaticlabs/go-bitfield" "github.com/prysmaticlabs/go-bitfield"
"github.com/tetratelabs/wabin/leb128" "github.com/tetratelabs/wabin/leb128"
"go.uber.org/zap" "go.uber.org/zap"
@ -168,6 +169,7 @@ type PortalProtocol struct {
utp *utp.Listener utp *utp.Listener
utpSm *utp.SocketManager utpSm *utp.SocketManager
packetRouter *utp.PacketRouter packetRouter *utp.PacketRouter
connIdGen libutp.ConnIdGenerator
ListenAddr string ListenAddr string
localNode *enode.LocalNode localNode *enode.LocalNode
Log log.Logger Log log.Logger
@ -337,6 +339,7 @@ func (p *PortalProtocol) setupUDPListening() error {
} }
p.utp, err = utp.ListenUTPOptions("utp", (*utp.Addr)(laddr), utp.WithSocketManager(p.utpSm)) p.utp, err = utp.ListenUTPOptions("utp", (*utp.Addr)(laddr), utp.WithSocketManager(p.utpSm))
p.connIdGen = utp.NewConnIdGenerator()
if err != nil { if err != nil {
return err return err
} }
@ -576,7 +579,7 @@ func (p *PortalProtocol) processOffer(target *enode.Node, resp []byte, request *
laddr := p.utp.Addr().(*utp.Addr) laddr := p.utp.Addr().(*utp.Addr)
raddr := &utp.Addr{IP: target.IP(), Port: target.UDP()} raddr := &utp.Addr{IP: target.IP(), Port: target.UDP()}
conn, err = utp.DialUTPOptions("utp", laddr, raddr, utp.WithContext(connctx), utp.WithSocketManager(p.utpSm), utp.WithConnId(uint32(connId))) conn, err = utp.DialUTPOptions("utp", laddr, raddr, utp.WithContext(connctx), utp.WithSocketManager(p.utpSm), utp.WithConnId(uint32(connId)))
p.Log.Info("will connect to: ", "addr", raddr.String(), "connId", connId)
if err != nil { if err != nil {
conncancel() conncancel()
p.Log.Error("failed to dial utp connection", "err", err) p.Log.Error("failed to dial utp connection", "err", err)
@ -655,6 +658,7 @@ func (p *PortalProtocol) processContent(target *enode.Node, resp []byte) (byte,
raddr := &utp.Addr{IP: target.IP(), Port: target.UDP()} raddr := &utp.Addr{IP: target.IP(), Port: target.UDP()}
connId := binary.BigEndian.Uint16(connIdMsg.Id[:]) connId := binary.BigEndian.Uint16(connIdMsg.Id[:])
conn, err := utp.DialUTPOptions("utp", laddr, raddr, utp.WithContext(connctx), utp.WithSocketManager(p.utpSm), utp.WithConnId(uint32(connId))) conn, err := utp.DialUTPOptions("utp", laddr, raddr, utp.WithContext(connctx), utp.WithSocketManager(p.utpSm), utp.WithConnId(uint32(connId)))
p.Log.Info("will connect to: ", "addr", raddr.String(), "connId", connId)
if err != nil { if err != nil {
conncancel() conncancel()
return 0xff, nil, err return 0xff, nil, err
@ -1012,8 +1016,7 @@ func (p *PortalProtocol) handleFindContent(id enode.ID, addr *net.UDPAddr, reque
return talkRespBytes, nil return talkRespBytes, nil
} else { } else {
connIdGen := utp.NewConnIdGenerator() connId := p.connIdGen.GenCid(id, false)
connId := connIdGen.GenCid(id, false)
connIdSend := connId.SendId() connIdSend := connId.SendId()
go func(bctx context.Context) { go func(bctx context.Context) {
@ -1025,11 +1028,13 @@ func (p *PortalProtocol) handleFindContent(id enode.ID, addr *net.UDPAddr, reque
ctx, cancel := context.WithTimeout(bctx, defaultUTPConnectTimeout) ctx, cancel := context.WithTimeout(bctx, defaultUTPConnectTimeout)
var conn *utp.Conn var conn *utp.Conn
conn, err = p.utp.AcceptUTPContext(ctx, connIdSend) conn, err = p.utp.AcceptUTPContext(ctx, connIdSend)
p.Log.Info("will accept from: ", "source", addr, "connId", connId)
if err != nil { if err != nil {
p.Log.Error("failed to accept utp connection", "connId", connIdSend, "err", err) p.Log.Error("failed to accept utp connection", "connId", connIdSend, "err", err)
cancel() cancel()
return return
} }
p.Log.Info("")
cancel() cancel()
err = conn.SetWriteDeadline(time.Now().Add(defaultUTPWriteTimeout)) err = conn.SetWriteDeadline(time.Now().Add(defaultUTPWriteTimeout))
@ -1132,8 +1137,7 @@ func (p *PortalProtocol) handleOffer(id enode.ID, addr *net.UDPAddr, request *po
idBuffer := make([]byte, 2) idBuffer := make([]byte, 2)
if contentKeyBitlist.Count() != 0 { if contentKeyBitlist.Count() != 0 {
connIdGen := utp.NewConnIdGenerator() connId := p.connIdGen.GenCid(id, false)
connId := connIdGen.GenCid(id, false)
connIdSend := connId.SendId() connIdSend := connId.SendId()
go func(bctx context.Context) { go func(bctx context.Context) {
@ -1145,6 +1149,7 @@ func (p *PortalProtocol) handleOffer(id enode.ID, addr *net.UDPAddr, request *po
ctx, cancel := context.WithTimeout(bctx, defaultUTPConnectTimeout) ctx, cancel := context.WithTimeout(bctx, defaultUTPConnectTimeout)
var conn *utp.Conn var conn *utp.Conn
conn, err = p.utp.AcceptUTPContext(ctx, connIdSend) conn, err = p.utp.AcceptUTPContext(ctx, connIdSend)
p.Log.Info("will accept from: ", "source", addr, "connId", connId)
if err != nil { if err != nil {
p.Log.Error("failed to accept utp connection", "connId", connIdSend, "err", err) p.Log.Error("failed to accept utp connection", "connId", connIdSend, "err", err)
cancel() cancel()