fix: update lib of utp

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

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=

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.packetRouter = utp.NewSocketRouter(
-	p.utp, err = utp.ListenUTPOptions("utp", (*utp.Addr)(laddr), utp.WithCustomHandler(
 		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)
+			nodes := p.table.Nodes()
-			return 0, err
+			var target *enode.Node
-		},
+			for _, node := range nodes {
-		func() ([]byte, *net.UDPAddr, error) {
+				if addr.Port != node.UDP() {
-			select {
+					continue
-			case msg := <-p.utpPackets:
+				}
-				if msg != nil {
+				if addr.IP != nil && addr.IP.To4().String() == node.IP().To4().String() {
-					return msg.Buf, msg.Addr, nil
+					target = node
-				} else {
+					p.log.Trace("target info", "ip", node.IP().To4().String(), "port", node.UDP(), "bufLength", len(buf))
-					return nil, nil, errClosed
+					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))
 	}

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"))
+	flag, content, err = node2.findContent(node1.localNode.Node(), []byte("large_test_key"))
-	//assert.NoError(t, err)
+	assert.NoError(t, err)
-	//assert.Equal(t, portalwire.ContentConnIdSelector, flag)
+	assert.Equal(t, portalwire.ContentConnIdSelector, flag)
-	//assert.Equal(t, largeTestContent, content)
+	assert.Equal(t, largeTestContent, content)
 }