From 8eab6cdf1aec9a0ef6b460ce0211c6cd42a04124 Mon Sep 17 00:00:00 2001 From: thinkAfCod Date: Sat, 11 Nov 2023 23:35:09 +0800 Subject: [PATCH] fix: update lib of utp --- go.mod | 4 +- go.sum | 4 + p2p/discover/portal_protocol.go | 55 ++++++++----- p2p/discover/portal_protocol_test.go | 116 ++++++++++++++++++++++++++- 4 files changed, 155 insertions(+), 24 deletions(-) diff --git a/go.mod b/go.mod index b4f5b129de..ece3ede4ca 100644 --- a/go.mod +++ b/go.mod @@ -54,7 +54,6 @@ require ( github.com/mattn/go-isatty v0.0.17 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-20231030043430-a1331c25fa98 github.com/peterh/liner v1.1.1-0.20190123174540-a2c9a5303de7 github.com/protolambda/bls12-381-util v0.0.0-20220416220906-d8552aa452c7 github.com/rs/cors v1.7.0 @@ -69,7 +68,7 @@ require ( golang.org/x/crypto v0.14.0 golang.org/x/exp v0.0.0-20230905200255-921286631fa9 golang.org/x/sync v0.4.0 - golang.org/x/sys v0.13.0 + golang.org/x/sys v0.14.0 golang.org/x/text v0.13.0 golang.org/x/time v0.3.0 golang.org/x/tools v0.14.0 @@ -128,6 +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/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 1d99102ab7..8f6f243848 100644 --- a/go.sum +++ b/go.sum @@ -499,6 +499,8 @@ github.com/opentracing/opentracing-go v1.1.0 h1:pWlfV3Bxv7k65HYwkikxat0+s3pV4bsq github.com/opentracing/opentracing-go v1.1.0/go.mod h1:UkNAQd3GIcIGf0SeVgPpRdFStlNbqXla1AfSYxPUl2o= github.com/optimism-java/utp-go v0.0.0-20231030043430-a1331c25fa98 h1:uxUbd8LFc24XetNFjTu9Kp9MqF2zKF92UMbcDuPxYZ8= 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/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= @@ -800,6 +802,8 @@ golang.org/x/sys v0.8.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.11.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.13.0 h1:Af8nKPmuFypiUBjVoU9V20FiaFXOcuZI21p0ycVYYGE= golang.org/x/sys v0.13.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.14.0 h1:Vz7Qs629MkJkGyHxUlRHizWJRG2j8fbQKjELVSNhy7Q= +golang.org/x/sys v0.14.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA= golang.org/x/term v0.0.0-20201117132131-f5c789dd3221/go.mod h1:Nr5EML6q2oocZ2LXRh80K7BxOlk5/8JxuGnuhpl+muw= golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo= golang.org/x/term v0.0.0-20210927222741-03fcf44c2211/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8= diff --git a/p2p/discover/portal_protocol.go b/p2p/discover/portal_protocol.go index df6bd861f8..f9634e4ac5 100644 --- a/p2p/discover/portal_protocol.go +++ b/p2p/discover/portal_protocol.go @@ -13,7 +13,6 @@ import ( "time" "github.com/VictoriaMetrics/fastcache" - "github.com/ethereum/go-ethereum/crypto" "github.com/ethereum/go-ethereum/log" "github.com/ethereum/go-ethereum/p2p/discover/portalwire" "github.com/ethereum/go-ethereum/p2p/enode" @@ -82,7 +81,8 @@ type PortalProtocol struct { nodeRadius *uint256.Int DiscV5 *UDPv5 utp *utp.Listener - utpPackets chan *utp.UdpMessage + utpSm *utp.SocketManager + packetRouter *utp.PacketRouter ListenAddr string localNode *enode.LocalNode log log.Logger @@ -163,24 +163,39 @@ func (p *PortalProtocol) setupUDPListening() (*net.UDPConn, error) { // } //} - p.utpPackets = make(chan *utp.UdpMessage, 10) - p.utp, err = utp.ListenUTPOptions("utp", (*utp.Addr)(laddr), utp.WithCustomHandler( + p.packetRouter = utp.NewSocketRouter( func(buf []byte, addr *net.UDPAddr) (int, error) { - id := crypto.Keccak256([]byte(addr.String())) - _, err := p.DiscV5.TalkRequestToID(enode.ID(id), addr, portalwire.UTPNetwork, buf) - return 0, err - }, - func() ([]byte, *net.UDPAddr, error) { - select { - case msg := <-p.utpPackets: - if msg != nil { - return msg.Buf, msg.Addr, nil - } else { - return nil, nil, errClosed + + nodes := p.table.Nodes() + var target *enode.Node + for _, node := range nodes { + if addr.Port != node.UDP() { + continue + } + 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 + } } } - }, - )) + + _, 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)) + if err != nil { + return nil, err + } + p.utp, err = utp.ListenUTPOptions("utp", (*utp.Addr)(laddr), utp.WithSocketManager(p.utpSm)) if err != nil { return nil, err @@ -458,11 +473,15 @@ func (p *PortalProtocol) handleUtpTalkRequest(id enode.ID, addr *net.UDPAddr, ms if n := p.DiscV5.getNode(id); n != nil { p.table.addSeenNode(wrapNode(n)) } - p.utpPackets <- &utp.UdpMessage{Buf: msg, Addr: addr} + if len(msg) == 0 { + fmt.Println("receive a emtpy msg") + } + p.packetRouter.ReceiveMessage(msg, addr) return []byte("") } func (p *PortalProtocol) handleTalkRequest(id enode.ID, addr *net.UDPAddr, msg []byte) []byte { + p.log.Error("handleTalkRequest", "id", id, "addr", addr) if n := p.DiscV5.getNode(id); n != nil { p.table.addSeenNode(wrapNode(n)) } diff --git a/p2p/discover/portal_protocol_test.go b/p2p/discover/portal_protocol_test.go index 240ec33935..5a63eef590 100644 --- a/p2p/discover/portal_protocol_test.go +++ b/p2p/discover/portal_protocol_test.go @@ -3,6 +3,8 @@ package discover import ( "crypto/rand" "fmt" + "github.com/optimism-java/utp-go" + "sync" "testing" "time" @@ -51,6 +53,112 @@ func setupLocalPortalNode(addr string, bootNodes []*enode.Node) (*PortalProtocol return portalProtocol, nil } +func TestPortalWireProtocolUdp(t *testing.T) { + node1, err := setupLocalPortalNode(":7777", nil) + assert.NoError(t, err) + node1.log = testlog.Logger(t, log.LvlTrace) + err = node1.Start() + assert.NoError(t, err) + + node2, err := setupLocalPortalNode(":7778", []*enode.Node{node1.localNode.Node()}) + assert.NoError(t, err) + node2.log = testlog.Logger(t, log.LvlTrace) + err = node2.Start() + assert.NoError(t, err) + + node3, err := setupLocalPortalNode(":7779", []*enode.Node{node1.localNode.Node()}) + assert.NoError(t, err) + node3.log = testlog.Logger(t, log.LvlTrace) + err = node3.Start() + assert.NoError(t, err) + time.Sleep(10 * time.Second) + + assert.Equal(t, 2, len(node1.table.Nodes())) + assert.Equal(t, 2, len(node2.table.Nodes())) + assert.Equal(t, 2, len(node3.table.Nodes())) + + rAddr, _ := utp.ResolveUTPAddr("utp", "127.0.0.1:7777") + lAddr, _ := utp.ResolveUTPAddr("utp", "127.0.0.1:7778") + + var wg sync.WaitGroup + wg.Add(4) + + cid := uint32(12) + cliSendMsgWithCid := "there are connection id : 12!" + cliSendMsgWithRandomCid := "there are connection id: random!" + + serverEchoWithCid := "accept connection sends back msg: echo" + serverEchoWithRandomCid := "ccept connection with random cid sends msg: echo" + go func() { + var acceptConn *utp.Conn + defer func() { + wg.Done() + _ = acceptConn.Close() + }() + acceptConn, err := node1.utp.AcceptUTPWithConnId(cid) + if err != nil { + panic(err) + } + buf := make([]byte, 100) + n, err := acceptConn.Read(buf) + if err != nil { + panic(err) + } + assert.Equal(t, cliSendMsgWithCid, string(buf[:n])) + acceptConn.Write([]byte(serverEchoWithCid)) + }() + go func() { + defer wg.Done() + randomConnIdConn, err := node1.utp.Accept() + if err != nil { + panic(err) + } + buf := make([]byte, 100) + n, err := randomConnIdConn.Read(buf) + if err != nil { + panic(err) + } + assert.Equal(t, cliSendMsgWithRandomCid, string(buf[:n])) + randomConnIdConn.Write([]byte(serverEchoWithRandomCid)) + }() + + go func() { + defer wg.Done() + connWithConnId, err := utp.DialUTPOptions("utp", lAddr, rAddr, utp.WithConnId(cid), utp.WithSocketManager(node2.utpSm)) + if err != nil { + panic(err) + } + _, err = connWithConnId.Write([]byte("there are connection id : 12!")) + if err != nil { + panic(err) + } + buf := make([]byte, 100) + n, err := connWithConnId.Read(buf) + if err != nil { + panic(err) + } + assert.Equal(t, serverEchoWithCid, string(buf[:n])) + }() + go func() { + defer wg.Done() + randomConnIdConn, err := utp.DialUTPOptions("utp", lAddr, rAddr, utp.WithSocketManager(node2.utpSm)) + if err != nil { + panic(err) + } + _, err = randomConnIdConn.Write([]byte(cliSendMsgWithRandomCid)) + if err != nil { + panic(err) + } + buf := make([]byte, 100) + n, err := randomConnIdConn.Read(buf) + if err != nil { + panic(err) + } + assert.Equal(t, serverEchoWithRandomCid, string(buf[:n])) + }() + wg.Wait() +} + func TestPortalWireProtocol(t *testing.T) { node1, err := setupLocalPortalNode(":7777", nil) assert.NoError(t, err) @@ -122,8 +230,8 @@ func TestPortalWireProtocol(t *testing.T) { err = node1.storage.Put([]byte("large_test_key"), largeTestContent) assert.NoError(t, err) - //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) + 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) }