diff --git a/go.mod b/go.mod index ece3ede4ca..3e73ee5c49 100644 --- a/go.mod +++ b/go.mod @@ -127,7 +127,7 @@ require ( github.com/mmcloughlin/addchain v0.4.0 // indirect github.com/naoina/go-stringutil v0.1.0 // indirect github.com/opentracing/opentracing-go v1.1.0 // indirect - github.com/optimism-java/utp-go v0.0.0-20231111152515-b2c1e9aba225 // indirect + github.com/optimism-java/utp-go v0.0.0-20231114114639-92925ba7e35e // indirect github.com/pkg/errors v0.9.1 // indirect github.com/pmezard/go-difflib v1.0.0 // indirect github.com/prometheus/client_golang v1.12.0 // indirect diff --git a/go.sum b/go.sum index 8f6f243848..c255775ea4 100644 --- a/go.sum +++ b/go.sum @@ -501,6 +501,8 @@ github.com/optimism-java/utp-go v0.0.0-20231030043430-a1331c25fa98 h1:uxUbd8LFc2 github.com/optimism-java/utp-go v0.0.0-20231030043430-a1331c25fa98/go.mod h1:DZ0jYzLzt4ZsCmhI/iqYgGFoNx45OfpEoKzXB8HVALQ= github.com/optimism-java/utp-go v0.0.0-20231111152515-b2c1e9aba225 h1:UUVmsVAv/4v0TMW3AFPwxAkuyNjKsAWVyqxfR10gMGE= github.com/optimism-java/utp-go v0.0.0-20231111152515-b2c1e9aba225/go.mod h1:DZ0jYzLzt4ZsCmhI/iqYgGFoNx45OfpEoKzXB8HVALQ= +github.com/optimism-java/utp-go v0.0.0-20231114114639-92925ba7e35e h1:61Mw2nE4trMg/Ze/0oFD5o5yKog2UuLUlbUnBS0lCh8= +github.com/optimism-java/utp-go v0.0.0-20231114114639-92925ba7e35e/go.mod h1:DZ0jYzLzt4ZsCmhI/iqYgGFoNx45OfpEoKzXB8HVALQ= github.com/pelletier/go-toml v1.2.0/go.mod h1:5z9KED0ma1S8pY6P1sdut58dfprrGBbd/94hg7ilaic= github.com/peterh/liner v1.1.1-0.20190123174540-a2c9a5303de7 h1:oYW+YCJ1pachXTQmzR3rNLYGGz4g/UgFcjb28p/viDM= github.com/peterh/liner v1.1.1-0.20190123174540-a2c9a5303de7/go.mod h1:CRroGNssyjTd/qIG2FyxByd2S8JEAZXBl4qUrZf8GS0= diff --git a/p2p/discover/portal_protocol.go b/p2p/discover/portal_protocol.go index f9634e4ac5..1159225ec2 100644 --- a/p2p/discover/portal_protocol.go +++ b/p2p/discover/portal_protocol.go @@ -7,6 +7,7 @@ import ( "encoding/binary" "errors" "fmt" + "go.uber.org/zap" "io" "net" "sort" @@ -165,7 +166,6 @@ func (p *PortalProtocol) setupUDPListening() (*net.UDPConn, error) { p.packetRouter = utp.NewSocketRouter( func(buf []byte, addr *net.UDPAddr) (int, error) { - nodes := p.table.Nodes() var target *enode.Node for _, node := range nodes { @@ -174,24 +174,28 @@ func (p *PortalProtocol) setupUDPListening() (*net.UDPConn, error) { } if addr.IP != nil && addr.IP.To4().String() == node.IP().To4().String() { target = node - p.log.Trace("target info", "ip", node.IP().To4().String(), "port", node.UDP(), "bufLength", len(buf)) + break } if addr.IP == nil { nodeIp := node.IP().To4().String() if nodeIp == "127.0.0.1" || nodeIp == "0.0.0.0" { target = node - p.log.Trace("target info", "ip", nodeIp, "port", node.UDP(), "bufLength", len(buf)) break } } } + p.log.Trace("send to target data", "ip", target.IP().String(), "port", target.UDP(), "bufLength", len(buf)) _, err := p.DiscV5.TalkRequest(target, portalwire.UTPNetwork, buf) return len(buf), err }) - p.utpSm, err = utp.NewSocketManager("utp", laddr, utp.WithPacketRouter(p.packetRouter), utp.WithBlockPacketCount(50)) + logger, err := zap.NewDevelopmentConfig().Build() + if err != nil { + return nil, err + } + p.utpSm, err = utp.NewSocketManager("utp", laddr, utp.WithLogger(logger.Named(listenAddr)), utp.WithPacketRouter(p.packetRouter), utp.WithBlockPacketCount(50), utp.WithMaxPacketSize(1145)) if err != nil { return nil, err } @@ -350,7 +354,7 @@ func (p *PortalProtocol) processContent(target *enode.Node, resp []byte) (byte, laddr := p.utp.Addr().(*utp.Addr) raddr := &utp.Addr{IP: target.IP(), Port: target.UDP()} connId := binary.BigEndian.Uint16(connIdMsg.Id[:]) - conn, err := utp.DialUTPOptions("utp", laddr, raddr, utp.WithContext(rctx), utp.WithConnId(uint32(connId))) + conn, err := utp.DialUTPOptions("utp", laddr, raddr, utp.WithContext(rctx), utp.WithSocketManager(p.utpSm), utp.WithConnId(uint32(connId))) if err != nil { rcancel() return 0xff, nil, err @@ -363,8 +367,8 @@ func (p *PortalProtocol) processContent(target *enode.Node, resp []byte) (byte, } // Read ALL the data from the connection until EOF and return it data := make([]byte, 0) + buf := make([]byte, 1024) for { - buf := make([]byte, 1024) var n int n, err = conn.Read(buf) if err != nil { @@ -473,9 +477,7 @@ func (p *PortalProtocol) handleUtpTalkRequest(id enode.ID, addr *net.UDPAddr, ms if n := p.DiscV5.getNode(id); n != nil { p.table.addSeenNode(wrapNode(n)) } - if len(msg) == 0 { - fmt.Println("receive a emtpy msg") - } + p.log.Trace("receive utp data", "addr", addr, "msg-length", len(msg)) p.packetRouter.ReceiveMessage(msg, addr) return []byte("") } @@ -695,8 +697,14 @@ func (p *PortalProtocol) handleFindContent(id enode.ID, addr *net.UDPAddr, reque ctx, cancel := context.WithTimeout(context.Background(), defaultUTPConnectTimeout) var conn *utp.Conn conn, err = p.utp.AcceptUTPContext(ctx, connIdSend) + defer func(conn *utp.Conn) { + err := conn.Close() + if err != nil { + p.log.Error("failed to close utp connection", "err", err) + } + }(conn) if err != nil { - p.log.Error("failed to accept utp connection", "err", err) + p.log.Error("failed to accept utp connection", "connId", connIdSend, "err", err) cancel() return } diff --git a/p2p/discover/portal_protocol_test.go b/p2p/discover/portal_protocol_test.go index 5a63eef590..ebef0e6e09 100644 --- a/p2p/discover/portal_protocol_test.go +++ b/p2p/discover/portal_protocol_test.go @@ -2,8 +2,11 @@ package discover import ( "crypto/rand" + "errors" "fmt" "github.com/optimism-java/utp-go" + "io" + "net" "sync" "testing" "time" @@ -88,7 +91,10 @@ func TestPortalWireProtocolUdp(t *testing.T) { cliSendMsgWithRandomCid := "there are connection id: random!" serverEchoWithCid := "accept connection sends back msg: echo" - serverEchoWithRandomCid := "ccept connection with random cid sends msg: echo" + //serverEchoWithRandomCid := "ccept connection with random cid sends msg: echo" + + largeTestContent := make([]byte, 1199) + _, err = rand.Read(largeTestContent) go func() { var acceptConn *utp.Conn defer func() { @@ -108,7 +114,11 @@ func TestPortalWireProtocolUdp(t *testing.T) { acceptConn.Write([]byte(serverEchoWithCid)) }() go func() { - defer wg.Done() + var randomConnIdConn net.Conn + defer func() { + wg.Done() + _ = randomConnIdConn.Close() + }() randomConnIdConn, err := node1.utp.Accept() if err != nil { panic(err) @@ -119,11 +129,16 @@ func TestPortalWireProtocolUdp(t *testing.T) { panic(err) } assert.Equal(t, cliSendMsgWithRandomCid, string(buf[:n])) - randomConnIdConn.Write([]byte(serverEchoWithRandomCid)) + + randomConnIdConn.Write(largeTestContent) }() go func() { - defer wg.Done() + var connWithConnId net.Conn + defer func() { + wg.Done() + //_ = connWithConnId.Close() + }() connWithConnId, err := utp.DialUTPOptions("utp", lAddr, rAddr, utp.WithConnId(cid), utp.WithSocketManager(node2.utpSm)) if err != nil { panic(err) @@ -140,7 +155,11 @@ func TestPortalWireProtocolUdp(t *testing.T) { assert.Equal(t, serverEchoWithCid, string(buf[:n])) }() go func() { - defer wg.Done() + var randomConnIdConn net.Conn + defer func() { + wg.Done() + //_ = randomConnIdConn.Close() + }() randomConnIdConn, err := utp.DialUTPOptions("utp", lAddr, rAddr, utp.WithSocketManager(node2.utpSm)) if err != nil { panic(err) @@ -149,12 +168,20 @@ func TestPortalWireProtocolUdp(t *testing.T) { if err != nil { panic(err) } - buf := make([]byte, 100) - n, err := randomConnIdConn.Read(buf) - if err != nil { - panic(err) + + data := make([]byte, 0) + buf := make([]byte, 1024) + for { + var n int + n, err = randomConnIdConn.Read(buf) + if err != nil { + if errors.Is(err, io.EOF) { + break + } + } + data = append(data, buf[:n]...) } - assert.Equal(t, serverEchoWithRandomCid, string(buf[:n])) + assert.Equal(t, largeTestContent, data) }() wg.Wait() } @@ -232,6 +259,6 @@ func TestPortalWireProtocol(t *testing.T) { flag, content, err = node2.findContent(node1.localNode.Node(), []byte("large_test_key")) assert.NoError(t, err) - assert.Equal(t, portalwire.ContentConnIdSelector, flag) assert.Equal(t, largeTestContent, content) + assert.Equal(t, portalwire.ContentConnIdSelector, flag) }