From bf12305507608f64ff06b81dfb650281f4044bfd Mon Sep 17 00:00:00 2001 From: grapebaba <281165273@qq.com> Date: Thu, 26 Oct 2023 13:24:27 +0800 Subject: [PATCH] feat:handle find content message Signed-off-by: grapebaba <281165273@qq.com> --- go.mod | 4 + go.sum | 13 ++- p2p/discover/portal_protocol.go | 166 +++++++++++++++++++++++++++- p2p/discover/portal_storage.go | 9 ++ p2p/discover/portalwire/messages.go | 10 +- 5 files changed, 197 insertions(+), 5 deletions(-) create mode 100644 p2p/discover/portal_storage.go diff --git a/go.mod b/go.mod index 3f8b2194df..d291f28b32 100644 --- a/go.mod +++ b/go.mod @@ -128,6 +128,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-20231024092003-1dd76611b5f2 // 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 @@ -140,6 +141,9 @@ require ( github.com/tklauser/go-sysconf v0.3.12 // indirect github.com/tklauser/numcpus v0.6.1 // indirect github.com/xrash/smetrics v0.0.0-20201216005158-039620a65673 // indirect + go.uber.org/atomic v1.7.0 // indirect + go.uber.org/multierr v1.6.0 // indirect + go.uber.org/zap v1.19.0 // indirect golang.org/x/mod v0.12.0 // indirect golang.org/x/net v0.17.0 // indirect google.golang.org/protobuf v1.27.1 // indirect diff --git a/go.sum b/go.sum index ef3b839b0d..78edd4a180 100644 --- a/go.sum +++ b/go.sum @@ -93,6 +93,7 @@ github.com/aws/aws-sdk-go-v2/service/sts v1.23.2/go.mod h1:Eows6e1uQEsc4ZaHANmsP github.com/aws/smithy-go v1.15.0 h1:PS/durmlzvAFpQHDs4wi4sNNP9ExsqZh6IlfdHXgKK8= github.com/aws/smithy-go v1.15.0/go.mod h1:Tg+OJXh4MB2R/uN61Ko2f6hTZwB/ZYGOtib8J3gBHzA= github.com/aymerick/raymond v2.0.3-0.20180322193309-b565731e1464+incompatible/go.mod h1:osfaiScAUVup+UC9Nfq76eWqDhXlp+4UYaA8uhTBO6g= +github.com/benbjohnson/clock v1.1.0/go.mod h1:J11/hYXuz8f4ySSvYwY0FKfm+ezbsZBKZxNJlLklBHA= github.com/beorn7/perks v0.0.0-20180321164747-3a771d992973/go.mod h1:Dwedo/Wpr24TaqPxmxbtue+5NUziq4I4S80YR8gNf3Q= github.com/beorn7/perks v1.0.0/go.mod h1:KWe93zE9D1o94FZ5RNwFwVgaQK1VOXiVxmqh+CedLV8= github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM= @@ -495,6 +496,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-20231024092003-1dd76611b5f2 h1:1H9unjvDxqDVTF9rh99sfolMtp8zYCUNAq+aoeGCxLA= +github.com/optimism-java/utp-go v0.0.0-20231024092003-1dd76611b5f2/go.mod h1:ohCuwoc66lfiNpo2Wk22zV07IbB0gte8+TYiclv9an4= 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= @@ -586,7 +589,6 @@ github.com/tyler-smith/go-bip39 v1.1.0/go.mod h1:gUYDtqQw1JS3ZJ8UWVcGTGqqr6YIN3C github.com/ugorji/go v1.1.4/go.mod h1:uQMGLiO92mf5W77hV/PUCpI3pbzQx3CRekS0kk+RGrc= github.com/ugorji/go/codec v0.0.0-20181204163529-d75b2dcb6bc8/go.mod h1:VFNgLljTbGfSG7qAOspJ7OScBnGdDN/yBr0sguwnwf0= github.com/umbracle/gohashtree v0.0.2-alpha.0.20230207094856-5b775a815c10 h1:CQh33pStIp/E30b7TxDlXfM0145bn2e8boI30IxAhTg= -github.com/urfave/cli/v2 v2.3.0/go.mod h1:LJmUH05zAU44vOAcrfzZQKsZbVcdbOG8rtL3/XcUArI= 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/urfave/negroni v1.0.0/go.mod h1:Meg73S6kFm/4PpbYdq35yYWoCZ9mS/YSx+lKnmiohz4= @@ -615,8 +617,15 @@ go.opencensus.io v0.22.0/go.mod h1:+kGneAE2xo2IficOXnaByMWTGM9T73dGwxeWcUqIpI8= go.opencensus.io v0.22.2/go.mod h1:yxeiOL68Rb0Xd1ddK5vPZ/oVn4vY4Ynel7k9FzqtOIw= go.opencensus.io v0.22.3/go.mod h1:yxeiOL68Rb0Xd1ddK5vPZ/oVn4vY4Ynel7k9FzqtOIw= go.opencensus.io v0.22.4/go.mod h1:yxeiOL68Rb0Xd1ddK5vPZ/oVn4vY4Ynel7k9FzqtOIw= +go.uber.org/atomic v1.7.0 h1:ADUqmZGgLDDfbSL9ZmPxKTybcoEYHgpYfELNoN+7hsw= +go.uber.org/atomic v1.7.0/go.mod h1:fEN4uk6kAWBTFdckzkM89CLk9XfWZrxpCo0nPH17wJc= go.uber.org/automaxprocs v1.5.2 h1:2LxUOGiR3O6tw8ui5sZa2LAaHnsviZdVOUZw4fvbnME= go.uber.org/automaxprocs v1.5.2/go.mod h1:eRbA25aqJrxAbsLO0xy5jVwPt7FQnRgjW+efnwa1WM0= +go.uber.org/goleak v1.1.10/go.mod h1:8a7PlsEVH3e/a/GLqe5IIrQx6GzcnRmZEufDUTk4A7A= +go.uber.org/multierr v1.6.0 h1:y6IPFStTAIT5Ytl7/XYmHvzXQ7S3g/IeZW9hyZ5thw4= +go.uber.org/multierr v1.6.0/go.mod h1:cdWPpRnG4AhwMwsgIHip0KRBQjJy5kYEpYjJxpXp9iU= +go.uber.org/zap v1.19.0 h1:mZQZefskPPCMIBCSEH0v2/iUqqLrYtaeqwD6FUGUnFE= +go.uber.org/zap v1.19.0/go.mod h1:xg/QME4nWcxGxrpdeYfq7UvYrLh66cuVKdrbD1XF/NI= golang.org/x/crypto v0.0.0-20180904163835-0709b304e793/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4= golang.org/x/crypto v0.0.0-20181203042331-505ab145d0a9/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4= golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w= @@ -832,6 +841,7 @@ golang.org/x/tools v0.0.0-20190628153133-6cdbf07be9d0/go.mod h1:/rFqwRUd4F7ZHNgw golang.org/x/tools v0.0.0-20190816200558-6889da9d5479/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo= golang.org/x/tools v0.0.0-20190911174233-4f2ddba30aff/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo= golang.org/x/tools v0.0.0-20191012152004-8de300cfc20a/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo= +golang.org/x/tools v0.0.0-20191108193012-7d206e10da11/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo= golang.org/x/tools v0.0.0-20191113191852-77e3bb0ad9e7/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo= golang.org/x/tools v0.0.0-20191115202509-3a792d9c32b2/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo= golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo= @@ -970,6 +980,7 @@ gopkg.in/yaml.v2 v2.3.0/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI= gopkg.in/yaml.v2 v2.4.0 h1:D8xgwECY7CYvx+Y2n4sBz93Jn9JRvxdiyyo8CTfuKaY= gopkg.in/yaml.v2 v2.4.0/go.mod h1:RDklbk79AGWmwhnvt/jBztapEOGDOx6ZbXqjP6csGnQ= gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= +gopkg.in/yaml.v3 v3.0.0-20210107192922-496545a6307b/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= gotest.tools/v3 v3.5.1 h1:EENdUnS3pdur5nybKYIh2Vfgc8IUNBjxDPSjtiJcOzU= diff --git a/p2p/discover/portal_protocol.go b/p2p/discover/portal_protocol.go index d0682cc2f4..35dbbd66dc 100644 --- a/p2p/discover/portal_protocol.go +++ b/p2p/discover/portal_protocol.go @@ -4,9 +4,11 @@ import ( "context" "crypto/ecdsa" crand "crypto/rand" + "encoding/binary" "errors" "fmt" "net" + "sort" "time" "github.com/VictoriaMetrics/fastcache" @@ -18,6 +20,7 @@ import ( "github.com/ethereum/go-ethereum/rlp" ssz "github.com/ferranbt/fastssz" "github.com/holiman/uint256" + "github.com/optimism-java/utp-go" ) const ( @@ -37,6 +40,12 @@ const ( 9 + // request id (max = 8) + 1 byte from rlp encoding byte string 3 + // rlp encoding response byte string, max length in 2 bytes 16 // HMAC + + portalFindnodesResultLimit = 32 + + defaultUTPAcceptTimeout = 15 * time.Second + + defaultUTPWriteTimeout = 60 * time.Second ) type PortalProtocolConfig struct { @@ -68,6 +77,7 @@ type PortalProtocol struct { nodeRadius *uint256.Int DiscV5 *UDPv5 + utp *utp.Listener ListenAddr string localNode *enode.LocalNode log log.Logger @@ -80,6 +90,7 @@ type PortalProtocol struct { radiusCache *fastcache.Cache closeCtx context.Context cancelCloseCtx context.CancelFunc + storage Storage } func NewPortalProtocol(config *PortalProtocolConfig, protocolId string, privateKey *ecdsa.PrivateKey) (*PortalProtocol, error) { @@ -108,7 +119,6 @@ func NewPortalProtocol(config *PortalProtocolConfig, protocolId string, privateK } return protocol, nil - } func (p *PortalProtocol) Start() error { @@ -146,6 +156,10 @@ func (p *PortalProtocol) setupUDPListening() (*net.UDPConn, error) { // } //} + p.utp, err = utp.ListenUTP("udp", (*utp.Addr)(laddr)) + if err != nil { + return nil, err + } return conn, nil } @@ -169,6 +183,10 @@ func (p *PortalProtocol) setupDiscV5AndTable() error { } p.table, err = newMeteredTable(p, p.localNode.Database(), cfg) + if err != nil { + return err + } + return nil } @@ -308,6 +326,22 @@ func (p *PortalProtocol) handleTalkRequest(id enode.ID, addr *net.UDPAddr, msg [ return nil } + return resp + case portalwire.FINDCONTENT: + findContentRequest := &portalwire.FindContent{} + err := findContentRequest.UnmarshalSSZ(msg[1:]) + if err != nil { + p.log.Error("failed to unmarshal find content request", "err", err) + return nil + } + + p.log.Trace("received find content request", "protocol", p.protocolId, "source", id, "findContentRequest", findContentRequest) + resp, err := p.handleFindContent(id, addr, findContentRequest) + if err != nil { + p.log.Error("failed to handle find content request", "err", err) + return nil + } + return resp } @@ -362,7 +396,7 @@ func (p *PortalProtocol) handleFindNodes(fromAddr *net.UDPAddr, request *portalw distances[i] = uint(ssz.UnmarshallUint16(distance[:])) } - nodes := p.DiscV5.collectTableNodes(fromAddr.IP, distances, 64) + nodes := p.DiscV5.collectTableNodes(fromAddr.IP, distances, portalFindnodesResultLimit) nodesOverhead := 1 + 1 + 4 // msg id + total + container offset maxPayloadSize := maxPacketSize - talkRespOverhead - nodesOverhead @@ -388,6 +422,117 @@ func (p *PortalProtocol) handleFindNodes(fromAddr *net.UDPAddr, request *portalw return talkRespBytes, nil } +func (p *PortalProtocol) handleFindContent(id enode.ID, addr *net.UDPAddr, request *portalwire.FindContent) ([]byte, error) { + contentOverhead := 1 + 1 // msg id + SSZ Union selector + maxPayloadSize := maxPacketSize - talkRespOverhead - contentOverhead + enrOverhead := 4 //per added ENR, 4 bytes offset overhead + var err error + + contentId := p.storage.ContentId(request.ContentKey) + if contentId == nil { + return nil, fmt.Errorf("content not found") + } + + var content []byte + content, err = p.storage.Get(request.ContentKey, contentId) + if err != nil { + return nil, err + } + + if content == nil { + closestNodes := p.findNodesCloseToContent(contentId) + for i, n := range closestNodes { + if n.ID() == id { + closestNodes = append(closestNodes[:i], closestNodes[i+1:]...) + break + } + } + + enrs := p.truncateNodes(closestNodes, maxPayloadSize, enrOverhead) + + enrsMsg := &portalwire.Enrs{ + Enrs: enrs, + } + + p.log.Trace("Sending enrs content response", "protocol", p.protocolId, "source", addr, "enrs", enrsMsg) + var enrsMsgBytes []byte + enrsMsgBytes, err = enrsMsg.MarshalSSZ() + if err != nil { + return nil, err + } + + contentMsgBytes := make([]byte, 0, len(enrsMsgBytes)+1) + contentMsgBytes = append(contentMsgBytes, portalwire.ContentEnrsSelector) + contentMsgBytes = append(contentMsgBytes, enrsMsgBytes...) + + talkRespBytes := make([]byte, 0, len(contentMsgBytes)+1) + talkRespBytes = append(talkRespBytes, portalwire.CONTENT) + talkRespBytes = append(talkRespBytes, contentMsgBytes...) + + return talkRespBytes, nil + } else if len(content) <= maxPayloadSize { + contentMsgBytes := make([]byte, 0, len(content)+1) + contentMsgBytes = append(contentMsgBytes, portalwire.ContentRawSelector) + contentMsgBytes = append(contentMsgBytes, content...) + + talkRespBytes := make([]byte, 0, len(contentMsgBytes)+1) + talkRespBytes = append(talkRespBytes, portalwire.CONTENT) + talkRespBytes = append(talkRespBytes, contentMsgBytes...) + + return talkRespBytes, nil + } else { + connIdGen := utp.NewConnIdGenerator() + connId := connIdGen.GenCid(id, false) + connIdSend := connId.SendId() + + go func() { + ctx, cancel := context.WithTimeout(context.Background(), defaultUTPAcceptTimeout) + var conn *utp.Conn + conn, err = p.utp.AcceptUTPContext(ctx, connIdSend) + if err != nil { + p.log.Error("failed to accept utp connection", "err", err) + cancel() + return + } + cancel() + + wctx, wcancel := context.WithTimeout(context.Background(), defaultUTPWriteTimeout) + var n int + n, err = conn.WriteContext(wctx, content) + if err != nil { + p.log.Error("failed to write content to utp connection", "err", err) + wcancel() + return + } + wcancel() + p.log.Trace("wrote content size to utp connection", "n", n) + }() + + idBuffer := make([]byte, 2) + binary.BigEndian.PutUint16(idBuffer, uint16(connIdSend)) + connIdMsg := &portalwire.ConnectionId{ + Id: idBuffer, + } + + p.log.Trace("Sending connection id content response", "protocol", p.protocolId, "source", addr, "connId", connIdMsg) + var connIdMsgBytes []byte + connIdMsgBytes, err = connIdMsg.MarshalSSZ() + if err != nil { + return nil, err + } + + contentMsgBytes := make([]byte, 0, len(connIdMsgBytes)+1) + contentMsgBytes = append(contentMsgBytes, portalwire.ContentConnIdSelector) + contentMsgBytes = append(contentMsgBytes, connIdMsgBytes...) + + talkRespBytes := make([]byte, 0, len(contentMsgBytes)+1) + talkRespBytes = append(talkRespBytes, portalwire.CONTENT) + talkRespBytes = append(talkRespBytes, contentMsgBytes...) + + return talkRespBytes, nil + } +} + func (p *PortalProtocol) Self() *enode.Node { return p.DiscV5.LocalNode().Node() } @@ -514,7 +659,7 @@ func (p *PortalProtocol) lookupWorker(destNode *node, target enode.ID) ([]*node, } for _, n := range r { if n.ID() != p.Self().ID() { - nodes.push(wrapNode(n), findnodeResultLimit) + nodes.push(wrapNode(n), portalFindnodesResultLimit) } } return nodes.entries, err @@ -539,3 +684,18 @@ func (p *PortalProtocol) truncateNodes(nodes []*enode.Node, maxSize int, enrOver } return res } + +func (p *PortalProtocol) findNodesCloseToContent(contentId []byte) []*enode.Node { + allNodes := p.table.Nodes() + sort.Slice(allNodes, func(i, j int) bool { + return enode.LogDist(allNodes[i].ID(), enode.ID(contentId)) < enode.LogDist(allNodes[j].ID(), enode.ID(contentId)) + }) + + if len(allNodes) > portalFindnodesResultLimit { + allNodes = allNodes[:portalFindnodesResultLimit] + } else { + allNodes = allNodes[:] + } + + return allNodes +} diff --git a/p2p/discover/portal_storage.go b/p2p/discover/portal_storage.go new file mode 100644 index 0000000000..d603cf5f79 --- /dev/null +++ b/p2p/discover/portal_storage.go @@ -0,0 +1,9 @@ +package discover + +type Storage interface { + ContentId(contentKey []byte) []byte + + Get(contentKey []byte, contentId []byte) ([]byte, error) + + Put(contentKey []byte, content []byte) error +} diff --git a/p2p/discover/portalwire/messages.go b/p2p/discover/portalwire/messages.go index 7773998d92..f436f39b76 100644 --- a/p2p/discover/portalwire/messages.go +++ b/p2p/discover/portalwire/messages.go @@ -1,6 +1,8 @@ package portalwire -import "github.com/ethereum/go-ethereum/common/hexutil" +import ( + "github.com/ethereum/go-ethereum/common/hexutil" +) // Protocol IDs for the portal protocol. const ( @@ -26,6 +28,12 @@ const ( ACCEPT byte = 0x07 ) +const ( + ContentConnIdSelector byte = 0x00 + ContentRawSelector byte = 0x01 + ContentEnrsSelector byte = 0x02 +) + // Request messages for the portal protocol. type ( PingPongCustomData struct {