forks/go-ethereum
1
0
Fork 1
mirror of https://github.com/ethereum/go-ethereum.git synced 2026-07-28 07:36:44 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 3

fix: upgrade utp-go

Browse source
This commit is contained in:
thinkAfCod 2024-11-08 23:40:20 +08:00 committed by Chen Kai
parent caa38955fd
commit c394b097a5
5 changed files with 48 additions and 50 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

3
go.mod
View file

@ -54,7 +54,7 @@ require (
github.com/mattn/go-sqlite3 v1.14.18
github.com/naoina/toml v0.1.2-0.20170918210437-9fafd6967416
github.com/olekukonko/tablewriter v0.0.5
github.com/optimism-java/utp-go v0.0.0-20241023035141-295a86339e8b
github.com/optimism-java/utp-go v0.0.0-20241110145701-0f0eebf881b3
github.com/peterh/liner v1.1.1-0.20190123174540-a2c9a5303de7
github.com/pion/stun v0.6.1
github.com/protolambda/bls12-381-util v0.1.0
@ -149,6 +149,7 @@ require (
github.com/russross/blackfriday/v2 v2.1.0 // indirect
github.com/tklauser/go-sysconf v0.3.12 // indirect
github.com/tklauser/numcpus v0.6.1 // indirect
github.com/valyala/fastrand v1.1.0 // indirect
github.com/xrash/smetrics v0.0.0-20201216005158-039620a65673 // indirect
go.uber.org/multierr v1.11.0 // indirect
golang.org/x/mod v0.17.0 // indirect

6
go.sum
View file

@ -421,8 +421,8 @@ github.com/onsi/gomega v1.10.1 h1:o0+MgICZLuZ7xjH7Vx6zS/zcu93/BEp1VwkIW1mEXCE=
github.com/onsi/gomega v1.10.1/go.mod h1:iN09h71vgCQne3DLsj+A5owkum+a2tYe+TOCB1ybHNo=
github.com/opentracing/opentracing-go v1.1.0 h1:pWlfV3Bxv7k65HYwkikxat0+s3pV4bsqf19k25Ur8rU=
github.com/opentracing/opentracing-go v1.1.0/go.mod h1:UkNAQd3GIcIGf0SeVgPpRdFStlNbqXla1AfSYxPUl2o=
github.com/optimism-java/utp-go v0.0.0-20241023035141-295a86339e8b h1:WAb+ccelzvqShjfzv2mQQMNK0BTmSW1fzDlE8WrMYpQ=
github.com/optimism-java/utp-go v0.0.0-20241023035141-295a86339e8b/go.mod h1:DZ0jYzLzt4ZsCmhI/iqYgGFoNx45OfpEoKzXB8HVALQ=
github.com/optimism-java/utp-go v0.0.0-20241110145701-0f0eebf881b3 h1:KfAZ//Sxrqulozmw4QoC8jY3h4I5diWXWPmZ1gptvGY=
github.com/optimism-java/utp-go v0.0.0-20241110145701-0f0eebf881b3/go.mod h1:dJZNMUlyNpjM2VkUEHhmFprLei6gCg3r7U9qj9MmJNQ=
github.com/optimism-java/zrnt v0.32.4-0.20240415084906-d9dbf06b32f7 h1:ZTQWXQ8xblCRUXhZs3h5qrBMSAHe8iNH7BG7a7IVFlI=
github.com/optimism-java/zrnt v0.32.4-0.20240415084906-d9dbf06b32f7/go.mod h1:A0fezkp9Tt3GBLATSPIbuY4ywYESyAuc/FFmPKg8Lqs=
github.com/peterh/liner v1.1.1-0.20190123174540-a2c9a5303de7 h1:oYW+YCJ1pachXTQmzR3rNLYGGz4g/UgFcjb28p/viDM=
@ -524,6 +524,8 @@ github.com/tklauser/numcpus v0.6.1/go.mod h1:1XfjsgE2zo8GVw7POkMbHENHzVg3GzmoZ9f
github.com/urfave/cli/v2 v2.25.7 h1:VAzn5oq403l5pHjc4OhD54+XGO9cdKVL/7lDjF+iKUs=
github.com/urfave/cli/v2 v2.25.7/go.mod h1:8qnjx1vcq5s2/wpsqoZFndg2CE5tNFyrTvS6SinrnYQ=
github.com/valyala/bytebufferpool v1.0.0/go.mod h1:6bBcMArwyJ5K/AmCkWv1jt77kVWyCJ6HpOuEn7z0Csc=
github.com/valyala/fastrand v1.1.0 h1:f+5HkLW4rsgzdNoleUOB69hyT9IlD2ZQh9GyDMfb5G8=
github.com/valyala/fastrand v1.1.0/go.mod h1:HWqCzkrkg6QXT8V2EXWvXCoow7vLwOFN002oeRzjapQ=
github.com/valyala/fasttemplate v1.0.1/go.mod h1:UQGH1tvbgY+Nz5t2n7tXsz52dQxojPUpymEIMZ47gx8=
github.com/valyala/fasttemplate v1.2.1/go.mod h1:KHLXt3tVN2HBp8eijSv/kGJopbvo7S+qRAEEKiv+SiQ=
github.com/xrash/smetrics v0.0.0-20201216005158-039620a65673 h1:bAn7/zixMGCfxrRTfdpNzjtPYqr8smhKouy9mxVdGPU=

10
p2p/discover/portal_protocol.go
View file

@ -582,7 +582,7 @@ func (p *PortalProtocol) processOffer(target *enode.Node, resp []byte, request *
}
connctx, conncancel := context.WithTimeout(ctx, defaultUTPConnectTimeout)
conn, err = p.Utp.DialWithCid(connctx, target, connId)
conn, err = p.Utp.DialWithCid(connctx, target, libutp.ReceConnId(connId).SendId())
conncancel()
if err != nil {
if metrics.Enabled {
@ -672,7 +672,7 @@ func (p *PortalProtocol) processContent(target *enode.Node, resp []byte) (byte,
}
connctx, conncancel := context.WithTimeout(p.closeCtx, defaultUTPConnectTimeout)
connId := binary.BigEndian.Uint16(connIdMsg.Id[:])
conn, err := p.Utp.DialWithCid(connctx, target, connId)
conn, err := p.Utp.DialWithCid(connctx, target, libutp.ReceConnId(connId).SendId())
defer func() {
if conn == nil {
if metrics.Enabled {
@ -1105,9 +1105,9 @@ func (p *PortalProtocol) handleFindContent(id enode.ID, addr *net.UDPAddr, reque
case <-bctx.Done():
return
default:
p.Log.Debug("will accept find content conn from: ", "source", addr, "connId", connId)
p.Log.Debug("will accept find content conn from: ", "nodeId", id.String(), "source", addr, "connId", connId)
connectCtx, cancel = context.WithTimeout(bctx, defaultUTPConnectTimeout)
conn, err = p.Utp.AcceptWithCid(connectCtx, id, uint16(connectionId.SendId()))
conn, err = p.Utp.AcceptWithCid(connectCtx, id, connectionId)
cancel()
if err != nil {
if metrics.Enabled {
@ -1239,7 +1239,7 @@ func (p *PortalProtocol) handleOffer(id enode.ID, addr *net.UDPAddr, request *po
default:
p.Log.Debug("will accept offer conn from: ", "source", addr, "connId", connId)
connectCtx, cancel = context.WithTimeout(bctx, defaultUTPConnectTimeout)
conn, err = p.Utp.AcceptWithCid(connectCtx, id, uint16(connectionId.SendId()))
conn, err = p.Utp.AcceptWithCid(connectCtx, id, connectionId)
cancel()
if err != nil {
if metrics.Enabled {

50
p2p/discover/portal_protocol_test.go
View file

@ -13,6 +13,7 @@ import (
"github.com/ethereum/go-ethereum/portalnetwork/storage"
"github.com/optimism-java/utp-go"
"github.com/optimism-java/utp-go/libutp"
"github.com/prysmaticlabs/go-bitfield"
"golang.org/x/exp/slices"
@ -126,19 +127,10 @@ func TestPortalWireProtocolUdp(t *testing.T) {
assert.NoError(t, err)
time.Sleep(12 * time.Second)
udpAddrStr1 := fmt.Sprintf("%s:%d", node1.localNode.Node().IP(), node1.localNode.Node().UDP())
udpAddrStr2 := fmt.Sprintf("%s:%d", node2.localNode.Node().IP(), node2.localNode.Node().UDP())
node1Addr, _ := utp.ResolveUTPAddr("utp", udpAddrStr1)
node2Addr, _ := utp.ResolveUTPAddr("utp", udpAddrStr2)
fmt.Println(udpAddrStr1)
fmt.Println(udpAddrStr2)
fmt.Println(node1Addr)
fmt.Println(node2Addr)
cid := uint16(12)
cliSendMsgWithCid := "there are connection id : 12!"
cliSendMsgWithRandomCid := "there are connection id: random!"
cid1 := libutp.ReceConnId(12)
cid2 := libutp.ReceConnId(116)
cliSendMsgWithCid1 := "there are connection id : 12!"
cliSendMsgWithCid2 := "there are connection id: 116!"
serverEchoWithCid := "accept connection sends back msg: echo"
@ -156,7 +148,7 @@ func TestPortalWireProtocolUdp(t *testing.T) {
workGroup.Done()
_ = acceptConn.Close()
}()
acceptConn, err := node1.Utp.AcceptWithCid(context.Background(), node2.localNode.ID(), cid)
acceptConn, err := node1.Utp.AcceptWithCid(context.Background(), node2.localNode.ID(), cid1)
if err != nil {
panic(err)
}
@ -166,30 +158,30 @@ func TestPortalWireProtocolUdp(t *testing.T) {
if err != nil && err != io.EOF {
panic(err)
}
assert.Equal(t, cliSendMsgWithCid, string(buf[:n]))
assert.Equal(t, cliSendMsgWithCid1, string(buf[:n]))
_, err = acceptConn.Write([]byte(serverEchoWithCid))
if err != nil {
panic(err)
}
}()
go func() {
var randomConnIdConn net.Conn
var connId2Conn net.Conn
defer func() {
workGroup.Done()
_ = randomConnIdConn.Close()
_ = connId2Conn.Close()
}()
randomConnIdConn, err := node1.Utp.Accept(context.Background())
connId2Conn, err := node1.Utp.AcceptWithCid(context.Background(), node2.localNode.ID(), cid2)
if err != nil {
panic(err)
}
buf := make([]byte, 100)
n, err := randomConnIdConn.Read(buf)
n, err := connId2Conn.Read(buf)
if err != nil && err != io.EOF {
panic(err)
}
assert.Equal(t, cliSendMsgWithRandomCid, string(buf[:n]))
assert.Equal(t, cliSendMsgWithCid2, string(buf[:n]))
_, err = randomConnIdConn.Write(largeTestContent)
_, err = connId2Conn.Write(largeTestContent)
if err != nil {
panic(err)
}
@ -203,11 +195,11 @@ func TestPortalWireProtocolUdp(t *testing.T) {
_ = connWithConnId.Close()
}
}()
connWithConnId, err = node2.Utp.DialWithCid(context.Background(), node1.localNode.Node(), cid)
connWithConnId, err = node2.Utp.DialWithCid(context.Background(), node1.localNode.Node(), uint16(cid1.SendId()))
if err != nil {
panic(err)
}
_, err = connWithConnId.Write([]byte("there are connection id : 12!"))
_, err = connWithConnId.Write([]byte(cliSendMsgWithCid1))
if err != nil && err != io.EOF {
panic(err)
}
@ -219,18 +211,18 @@ func TestPortalWireProtocolUdp(t *testing.T) {
assert.Equal(t, serverEchoWithCid, string(buf[:n]))
}()
go func() {
var randomConnIdConn net.Conn
var ConnId2Conn net.Conn
defer func() {
workGroup.Done()
if randomConnIdConn != nil {
_ = randomConnIdConn.Close()
if ConnId2Conn != nil {
_ = ConnId2Conn.Close()
}
}()
randomConnIdConn, err = node2.Utp.Dial(context.Background(), node1.localNode.Node())
ConnId2Conn, err = node2.Utp.DialWithCid(context.Background(), node1.localNode.Node(), uint16(cid2.SendId()))
if err != nil && err != io.EOF {
panic(err)
}
_, err = randomConnIdConn.Write([]byte(cliSendMsgWithRandomCid))
_, err = ConnId2Conn.Write([]byte(cliSendMsgWithCid2))
if err != nil {
panic(err)
}
@ -239,7 +231,7 @@ func TestPortalWireProtocolUdp(t *testing.T) {
buf := make([]byte, 1024)
for {
var n int
n, err = randomConnIdConn.Read(buf)
n, err = ConnId2Conn.Read(buf)
if err != nil {
if errors.Is(err, io.EOF) {
break

29
p2p/discover/portal_utp.go
View file

@ -13,6 +13,7 @@ import (
"github.com/ethereum/go-ethereum/p2p/enode"
"github.com/ethereum/go-ethereum/p2p/netutil"
"github.com/optimism-java/utp-go"
"github.com/optimism-java/utp-go/libutp"
"go.uber.org/zap"
)
@ -32,10 +33,11 @@ type PortalUtp struct {
func NewPortalUtp(ctx context.Context, config *PortalProtocolConfig, discV5 *UDPv5, conn UDPConn) *PortalUtp {
return &PortalUtp{
ctx: ctx,
log: log.New("protocol", "utp"),
discV5: discV5,
conn: conn,
ctx: ctx,
log: log.New("protocol", "utp", "local", conn.LocalAddr().String()),
discV5: discV5,
conn: conn,
ListenAddr: config.ListenAddr,
}
}
@ -95,8 +97,8 @@ func (p *PortalUtp) Stop() {
func (p *PortalUtp) DialWithCid(ctx context.Context, dest *enode.Node, connId uint16) (net.Conn, error) {
raddr := &utp.Addr{IP: dest.IP(), Port: dest.UDP()}
p.log.Info("will connect to: ", "addr", raddr.String(), "connId", connId)
conn, err := utp.DialUTPOptions("utp", p.lAddr, raddr, utp.WithContext(ctx), utp.WithSocketManager(p.utpSm), utp.WithConnId(uint32(connId)))
p.log.Debug("will connect to: ", "nodeId", dest.ID().String(), "connId", connId)
conn, err := utp.DialUTPOptions("utp", p.lAddr, raddr, utp.WithContext(ctx), utp.WithSocketManager(p.utpSm), utp.WithConnId(connId))
return conn, err
}
@ -107,12 +109,13 @@ func (p *PortalUtp) Dial(ctx context.Context, dest *enode.Node) (net.Conn, error
return conn, err
}
func (p *PortalUtp) AcceptWithCid(ctx context.Context, nodeId enode.ID, cid uint16) (*utp.Conn, error) {
return p.listener.AcceptUTPContext(ctx, uint32(cid))
func (p *PortalUtp) AcceptWithCid(ctx context.Context, nodeId enode.ID, cid *libutp.ConnId) (*utp.Conn, error) {
p.log.Debug("will accept from: ", "nodeId", nodeId.String(), "sendId", cid.SendId(), "recvId", cid.RecvId())
return p.listener.AcceptUTPContext(ctx, nodeId, cid)
}
func (p *PortalUtp) Accept(ctx context.Context) (*utp.Conn, error) {
return p.listener.AcceptUTPContext(ctx, 0)
return p.listener.AcceptUTPContext(ctx, enode.ID{}, nil)
}
func (p *PortalUtp) getLocalAddr() *net.UDPAddr {
@ -121,8 +124,8 @@ func (p *PortalUtp) getLocalAddr() *net.UDPAddr {
return laddr
}
func (p *PortalUtp) packetRouterFunc(buf []byte, addr *net.UDPAddr) (int, error) {
p.log.Info("will send to target data", "ip", addr.IP.To4().String(), "port", addr.Port, "bufLength", len(buf))
func (p *PortalUtp) packetRouterFunc(buf []byte, id enode.ID, addr *net.UDPAddr) (int, error) {
p.log.Info("will send to target data", "nodeId", id.String(), "ip", addr.IP.To4().String(), "port", addr.Port, "bufLength", len(buf))
if n, ok := p.discV5.GetCachedNode(addr.String()); ok {
//_, err := p.DiscV5.TalkRequestToID(id, addr, string(portalwire.UTPNetwork), buf)
@ -137,7 +140,7 @@ func (p *PortalUtp) packetRouterFunc(buf []byte, addr *net.UDPAddr) (int, error)
}
func (p *PortalUtp) handleUtpTalkRequest(id enode.ID, addr *net.UDPAddr, msg []byte) []byte {
p.log.Trace("receive utp data", "addr", addr, "msg-length", len(msg))
p.packetRouter.ReceiveMessage(msg, addr)
p.log.Trace("receive utp data", "nodeId", id.String(), "addr", addr, "msg-length", len(msg))
p.packetRouter.ReceiveMessage(msg, &utp.NodeInfo{Id: id, Addr: addr})
return []byte("")
}