diff --git a/go.mod b/go.mod index 68461bac77..253c55c7f9 100644 --- a/go.mod +++ b/go.mod @@ -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 diff --git a/go.sum b/go.sum index 82bee92d3d..f3b591ce60 100644 --- a/go.sum +++ b/go.sum @@ -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= diff --git a/p2p/discover/portal_protocol.go b/p2p/discover/portal_protocol.go index 64e13cccbc..ec96b38f4c 100644 --- a/p2p/discover/portal_protocol.go +++ b/p2p/discover/portal_protocol.go @@ -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 { diff --git a/p2p/discover/portal_protocol_test.go b/p2p/discover/portal_protocol_test.go index 06647e5ea8..b7db661257 100644 --- a/p2p/discover/portal_protocol_test.go +++ b/p2p/discover/portal_protocol_test.go @@ -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 diff --git a/p2p/discover/portal_utp.go b/p2p/discover/portal_utp.go index c635c16989..8278972f82 100644 --- a/p2p/discover/portal_utp.go +++ b/p2p/discover/portal_utp.go @@ -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("") }