forks/go-ethereum
1
0
Fork 1
mirror of https://github.com/ethereum/go-ethereum.git synced 2026-07-28 07:36:44 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 3

fix: abstract utp socket and remove timeout nonce

Browse source
This commit is contained in:
thinkAfCod 2024-11-08 17:03:14 +08:00 committed by Chen Kai
parent 070b861c3d
commit caa38955fd
5 changed files with 226 additions and 91 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

53
cmd/shisui/main.go
View file

@ -1,6 +1,7 @@
package main
import (
"context"
"crypto/ecdsa"
"database/sql"
"encoding/hex"
@ -243,10 +244,11 @@ func startPortalRpcServer(config Config, conn discover.UDPConn, addr string, cli
if err != nil {
return err
}
utp := discover.NewPortalUtp(context.Background(), config.Protocol, discV5, conn)
var historyNetwork *history.HistoryNetwork
if slices.Contains(config.Networks, portalwire.History.Name()) {
historyNetwork, err = initHistory(config, server, conn, localNode, discV5)
historyNetwork, err = initHistory(config, server, conn, localNode, discV5, utp)
if err != nil {
return err
}
@ -255,7 +257,7 @@ func startPortalRpcServer(config Config, conn discover.UDPConn, addr string, cli
var beaconNetwork *beacon.BeaconNetwork
if slices.Contains(config.Networks, portalwire.Beacon.Name()) {
beaconNetwork, err = initBeacon(config, server, conn, localNode, discV5)
beaconNetwork, err = initBeacon(config, server, conn, localNode, discV5, utp)
if err != nil {
return err
}
@ -264,7 +266,7 @@ func startPortalRpcServer(config Config, conn discover.UDPConn, addr string, cli
var stateNetwork *state.StateNetwork
if slices.Contains(config.Networks, portalwire.State.Name()) {
stateNetwork, err = initState(config, server, conn, localNode, discV5)
stateNetwork, err = initState(config, server, conn, localNode, discV5, utp)
if err != nil {
return err
}
@ -371,7 +373,7 @@ func doPortMapping(natm nat.Interface, ln *enode.LocalNode, addr *net.UDPAddr) {
}()
}
func initHistory(config Config, server *rpc.Server, conn discover.UDPConn, localNode *enode.LocalNode, discV5 *discover.UDPv5) (*history.HistoryNetwork, error) {
func initHistory(config Config, server *rpc.Server, conn discover.UDPConn, localNode *enode.LocalNode, discV5 *discover.UDPv5, utp *discover.PortalUtp) (*history.HistoryNetwork, error) {
networkName := portalwire.History.Name()
db, err := history.NewDB(config.DataDir, networkName)
if err != nil {
@ -388,7 +390,18 @@ func initHistory(config Config, server *rpc.Server, conn discover.UDPConn, local
}
contentQueue := make(chan *discover.ContentElement, 50)
protocol, err := discover.NewPortalProtocol(config.Protocol, portalwire.History, config.PrivateKey, conn, localNode, discV5, contentStorage, contentQueue)
protocol, err := discover.NewPortalProtocol(
config.Protocol,
portalwire.History,
config.PrivateKey,
conn,
localNode,
discV5,
contentStorage,
contentQueue,
func(p *discover.PortalProtocol) {
p.Utp = utp
})
if err != nil {
return nil, err
@ -407,7 +420,7 @@ func initHistory(config Config, server *rpc.Server, conn discover.UDPConn, local
return historyNetwork, historyNetwork.Start()
}
func initBeacon(config Config, server *rpc.Server, conn discover.UDPConn, localNode *enode.LocalNode, discV5 *discover.UDPv5) (*beacon.BeaconNetwork, error) {
func initBeacon(config Config, server *rpc.Server, conn discover.UDPConn, localNode *enode.LocalNode, discV5 *discover.UDPv5, utp *discover.PortalUtp) (*beacon.BeaconNetwork, error) {
dbPath := path.Join(config.DataDir, "beacon")
err := os.MkdirAll(dbPath, 0755)
if err != nil {
@ -430,7 +443,18 @@ func initBeacon(config Config, server *rpc.Server, conn discover.UDPConn, localN
}
contentQueue := make(chan *discover.ContentElement, 50)
protocol, err := discover.NewPortalProtocol(config.Protocol, portalwire.Beacon, config.PrivateKey, conn, localNode, discV5, contentStorage, contentQueue)
protocol, err := discover.NewPortalProtocol(
config.Protocol,
portalwire.Beacon,
config.PrivateKey,
conn,
localNode,
discV5,
contentStorage,
contentQueue,
func(p *discover.PortalProtocol) {
p.Utp = utp
})
if err != nil {
return nil, err
@ -447,7 +471,7 @@ func initBeacon(config Config, server *rpc.Server, conn discover.UDPConn, localN
return beaconNetwork, beaconNetwork.Start()
}
func initState(config Config, server *rpc.Server, conn discover.UDPConn, localNode *enode.LocalNode, discV5 *discover.UDPv5) (*state.StateNetwork, error) {
func initState(config Config, server *rpc.Server, conn discover.UDPConn, localNode *enode.LocalNode, discV5 *discover.UDPv5, utp *discover.PortalUtp) (*state.StateNetwork, error) {
networkName := portalwire.State.Name()
db, err := history.NewDB(config.DataDir, networkName)
if err != nil {
@ -465,7 +489,18 @@ func initState(config Config, server *rpc.Server, conn discover.UDPConn, localNo
stateStore := state.NewStateStorage(contentStorage, db)
contentQueue := make(chan *discover.ContentElement, 50)
protocol, err := discover.NewPortalProtocol(config.Protocol, portalwire.State, config.PrivateKey, conn, localNode, discV5, stateStore, contentQueue)
protocol, err := discover.NewPortalProtocol(
config.Protocol,
portalwire.State,
config.PrivateKey,
conn,
localNode,
discV5,
stateStore,
contentQueue,
func(p *discover.PortalProtocol) {
p.Utp = utp
})
if err != nil {
return nil, err

91
p2p/discover/portal_protocol.go
View file

@ -13,21 +13,18 @@ import (
"math/big"
"math/rand"
"net"
"net/netip"
"slices"
"sort"
"sync"
"sync/atomic"
"time"
"github.com/VictoriaMetrics/fastcache"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/common/hexutil"
"github.com/ethereum/go-ethereum/common/mclock"
"github.com/ethereum/go-ethereum/metrics"
"github.com/ethereum/go-ethereum/p2p/discover/v5wire"
"github.com/VictoriaMetrics/fastcache"
"github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/metrics"
"github.com/ethereum/go-ethereum/p2p/discover/portalwire"
"github.com/ethereum/go-ethereum/p2p/enode"
"github.com/ethereum/go-ethereum/p2p/enr"
@ -41,7 +38,6 @@ import (
"github.com/optimism-java/utp-go/libutp"
"github.com/prysmaticlabs/go-bitfield"
"github.com/tetratelabs/wabin/leb128"
"go.uber.org/zap"
)
const (
@ -173,11 +169,6 @@ type PortalProtocol struct {
protocolName string
DiscV5 *UDPv5
utp *utp.Listener
utpSm *utp.SocketManager
packetRouter *utp.PacketRouter
connIdGen libutp.ConnIdGenerator
ListenAddr string
localNode *enode.LocalNode
Log log.Logger
PrivateKey *ecdsa.PrivateKey
@ -185,6 +176,9 @@ type PortalProtocol struct {
BootstrapNodes []*enode.Node
conn UDPConn
Utp *PortalUtp
connIdGen libutp.ConnIdGenerator
validSchemes enr.IdentityScheme
radiusCache *fastcache.Cache
closeCtx context.Context
@ -213,7 +207,6 @@ func NewPortalProtocol(config *PortalProtocolConfig, protocolId portalwire.Proto
protocol := &PortalProtocol{
protocolId: string(protocolId),
protocolName: protocolId.Name(),
ListenAddr: config.ListenAddr,
Log: log.New("protocol", protocolId.Name()),
PrivateKey: privateKey,
NetRestrict: config.NetRestrict,
@ -231,6 +224,7 @@ func NewPortalProtocol(config *PortalProtocolConfig, protocolId portalwire.Proto
DiscV5: discV5,
NAT: config.NAT,
clock: config.clock,
connIdGen: libutp.NewConnIdGenerator(),
}
for _, opt := range opts {
@ -253,7 +247,10 @@ func (p *PortalProtocol) Start() error {
}
p.DiscV5.RegisterTalkHandler(p.protocolId, p.handleTalkRequest)
p.DiscV5.RegisterTalkHandler(string(portalwire.Utp), p.handleUtpTalkRequest)
err = p.Utp.Start()
if err != nil {
return err
}
go p.table.loop()
@ -271,10 +268,7 @@ func (p *PortalProtocol) Stop() {
p.cancelCloseCtx()
p.table.close()
p.DiscV5.Close()
err := p.utp.Close()
if err != nil {
p.Log.Error("failed to close utp listener", "err", err)
}
p.Utp.Stop()
}
func (p *PortalProtocol) RoutingTableInfo() [][]string {
p.table.mutex.Lock()
@ -320,46 +314,6 @@ func (p *PortalProtocol) setupUDPListening() error {
port: laddr.Port,
}
}
var err error
p.packetRouter = utp.NewPacketRouter(
func(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))
if n, ok := p.DiscV5.GetCachedNode(addr.String()); ok {
//_, err := p.DiscV5.TalkRequestToID(id, addr, string(portalwire.UTPNetwork), buf)
req := &v5wire.TalkRequest{Protocol: string(portalwire.Utp), Message: buf}
p.DiscV5.sendFromAnotherThreadWithNode(n, netip.AddrPortFrom(netutil.IPToAddr(addr.IP), uint16(addr.Port)), req)
return len(buf), err
} else {
p.Log.Warn("not found target node info", "ip", addr.IP.To4().String(), "port", addr.Port, "bufLength", len(buf))
return 0, fmt.Errorf("not found target node id")
}
})
ctx := context.Background()
var logger *zap.Logger
if p.Log.Enabled(ctx, log.LevelDebug) || p.Log.Enabled(ctx, log.LevelTrace) {
logger, err = zap.NewDevelopmentConfig().Build()
} else {
logger, err = zap.NewProductionConfig().Build()
}
if err != nil {
return err
}
p.utpSm, err = utp.NewSocketManagerWithOptions("utp", laddr, utp.WithLogger(logger.Named(p.ListenAddr)), utp.WithPacketRouter(p.packetRouter), utp.WithMaxPacketSize(1145))
if err != nil {
return err
}
p.utp, err = utp.ListenUTPOptions("utp", (*utp.Addr)(laddr), utp.WithSocketManager(p.utpSm))
p.connIdGen = utp.NewConnIdGenerator()
if err != nil {
return err
}
return nil
}
@ -628,10 +582,7 @@ func (p *PortalProtocol) processOffer(target *enode.Node, resp []byte, request *
}
connctx, conncancel := context.WithTimeout(ctx, defaultUTPConnectTimeout)
laddr := p.utp.Addr().(*utp.Addr)
raddr := &utp.Addr{IP: target.IP(), Port: target.UDP()}
p.Log.Info("will connect to: ", "addr", raddr.String(), "connId", connId)
conn, err = utp.DialUTPOptions("utp", laddr, raddr, utp.WithContext(connctx), utp.WithSocketManager(p.utpSm), utp.WithConnId(uint32(connId)))
conn, err = p.Utp.DialWithCid(connctx, target, connId)
conncancel()
if err != nil {
if metrics.Enabled {
@ -720,11 +671,8 @@ func (p *PortalProtocol) processContent(target *enode.Node, resp []byte) (byte,
log.Debug("Node added to replacements list", "protocol", p.protocolName, "node", target.IP(), "port", target.UDP())
}
connctx, conncancel := context.WithTimeout(p.closeCtx, defaultUTPConnectTimeout)
laddr := p.utp.Addr().(*utp.Addr)
raddr := &utp.Addr{IP: target.IP(), Port: target.UDP()}
connId := binary.BigEndian.Uint16(connIdMsg.Id[:])
p.Log.Info("will connect to: ", "addr", raddr.String(), "connId", connId)
conn, err := utp.DialUTPOptions("utp", laddr, raddr, utp.WithContext(connctx), utp.WithSocketManager(p.utpSm), utp.WithConnId(uint32(connId)))
conn, err := p.Utp.DialWithCid(connctx, target, connId)
defer func() {
if conn == nil {
if metrics.Enabled {
@ -886,15 +834,6 @@ func (p *PortalProtocol) processPong(target *enode.Node, resp []byte) (*portalwi
return pong, nil
}
func (p *PortalProtocol) handleUtpTalkRequest(id enode.ID, addr *net.UDPAddr, msg []byte) []byte {
if n := p.DiscV5.getNode(id); n != nil {
p.table.addInboundNode(n)
}
p.Log.Trace("receive utp data", "addr", addr, "msg-length", len(msg))
p.packetRouter.ReceiveMessage(msg, addr)
return []byte("")
}
func (p *PortalProtocol) handleTalkRequest(id enode.ID, addr *net.UDPAddr, msg []byte) []byte {
if n := p.DiscV5.getNode(id); n != nil {
p.table.addInboundNode(n)
@ -1168,7 +1107,7 @@ func (p *PortalProtocol) handleFindContent(id enode.ID, addr *net.UDPAddr, reque
default:
p.Log.Debug("will accept find content conn from: ", "source", addr, "connId", connId)
connectCtx, cancel = context.WithTimeout(bctx, defaultUTPConnectTimeout)
conn, err = p.utp.AcceptUTPContext(connectCtx, connectionId.SendId())
conn, err = p.Utp.AcceptWithCid(connectCtx, id, uint16(connectionId.SendId()))
cancel()
if err != nil {
if metrics.Enabled {
@ -1300,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.AcceptUTPContext(connectCtx, connectionId.SendId())
conn, err = p.Utp.AcceptWithCid(connectCtx, id, uint16(connectionId.SendId()))
cancel()
if err != nil {
if metrics.Enabled {

28
p2p/discover/portal_protocol_test.go
View file

@ -83,9 +83,21 @@ func setupLocalPortalNode(addr string, bootNodes []*enode.Node) (*PortalProtocol
if err != nil {
return nil, err
}
utpSocket := NewPortalUtp(context.Background(), conf, discV5, conn)
contentQueue := make(chan *ContentElement, 50)
portalProtocol, err := NewPortalProtocol(conf, portalwire.History, privKey, conn, localNode, discV5, &storage.MockStorage{Db: make(map[string][]byte)}, contentQueue)
portalProtocol, err := NewPortalProtocol(
conf,
portalwire.History,
privKey,
conn,
localNode,
discV5,
&storage.MockStorage{Db: make(map[string][]byte)},
contentQueue,
func(p *PortalProtocol) {
p.Utp = utpSocket
})
if err != nil {
return nil, err
}
@ -119,8 +131,12 @@ func TestPortalWireProtocolUdp(t *testing.T) {
node1Addr, _ := utp.ResolveUTPAddr("utp", udpAddrStr1)
node2Addr, _ := utp.ResolveUTPAddr("utp", udpAddrStr2)
fmt.Println(udpAddrStr1)
fmt.Println(udpAddrStr2)
fmt.Println(node1Addr)
fmt.Println(node2Addr)
cid := uint32(12)
cid := uint16(12)
cliSendMsgWithCid := "there are connection id : 12!"
cliSendMsgWithRandomCid := "there are connection id: random!"
@ -140,7 +156,7 @@ func TestPortalWireProtocolUdp(t *testing.T) {
workGroup.Done()
_ = acceptConn.Close()
}()
acceptConn, err := node1.utp.AcceptUTPWithConnId(cid)
acceptConn, err := node1.Utp.AcceptWithCid(context.Background(), node2.localNode.ID(), cid)
if err != nil {
panic(err)
}
@ -162,7 +178,7 @@ func TestPortalWireProtocolUdp(t *testing.T) {
workGroup.Done()
_ = randomConnIdConn.Close()
}()
randomConnIdConn, err := node1.utp.Accept()
randomConnIdConn, err := node1.Utp.Accept(context.Background())
if err != nil {
panic(err)
}
@ -187,7 +203,7 @@ func TestPortalWireProtocolUdp(t *testing.T) {
_ = connWithConnId.Close()
}
}()
connWithConnId, err := utp.DialUTPOptions("utp", node2Addr, node1Addr, utp.WithConnId(cid), utp.WithSocketManager(node2.utpSm))
connWithConnId, err = node2.Utp.DialWithCid(context.Background(), node1.localNode.Node(), cid)
if err != nil {
panic(err)
}
@ -210,7 +226,7 @@ func TestPortalWireProtocolUdp(t *testing.T) {
_ = randomConnIdConn.Close()
}
}()
randomConnIdConn, err := utp.DialUTPOptions("utp", node2Addr, node1Addr, utp.WithSocketManager(node2.utpSm))
randomConnIdConn, err = node2.Utp.Dial(context.Background(), node1.localNode.Node())
if err != nil && err != io.EOF {
panic(err)
}

143
p2p/discover/portal_utp.go Normal file
View file

@ -0,0 +1,143 @@
package discover
import (
"context"
"fmt"
"net"
"net/netip"
"sync"
"github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/p2p/discover/portalwire"
"github.com/ethereum/go-ethereum/p2p/discover/v5wire"
"github.com/ethereum/go-ethereum/p2p/enode"
"github.com/ethereum/go-ethereum/p2p/netutil"
"github.com/optimism-java/utp-go"
"go.uber.org/zap"
)
type PortalUtp struct {
ctx context.Context
log log.Logger
discV5 *UDPv5
conn UDPConn
ListenAddr string
listener *utp.Listener
utpSm *utp.SocketManager
packetRouter *utp.PacketRouter
lAddr *utp.Addr
startOnce sync.Once
}
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,
}
}
func (p *PortalUtp) Start() error {
errCh := make(chan error, 1)
p.startOnce.Do(func() {
defer func() {
close(errCh)
}()
laddr := p.getLocalAddr()
p.packetRouter = utp.NewPacketRouter(p.packetRouterFunc)
var logger *zap.Logger
var err error
if p.log.Enabled(p.ctx, log.LevelDebug) || p.log.Enabled(p.ctx, log.LevelTrace) {
logger, err = zap.NewDevelopmentConfig().Build()
} else {
logger, err = zap.NewProductionConfig().Build()
}
if err != nil {
errCh <- err
return
}
p.utpSm, err = utp.NewSocketManagerWithOptions(
"utp",
laddr,
utp.WithContext(p.ctx),
utp.WithLogger(logger.Named(p.ListenAddr)),
utp.WithPacketRouter(p.packetRouter),
utp.WithMaxPacketSize(1145))
if err != nil {
errCh <- err
return
}
p.listener, err = utp.ListenUTPOptions("utp", (*utp.Addr)(laddr), utp.WithSocketManager(p.utpSm))
if err != nil {
errCh <- err
return
}
p.lAddr = p.listener.Addr().(*utp.Addr)
// register discv5 listener
p.discV5.RegisterTalkHandler(string(portalwire.Utp), p.handleUtpTalkRequest)
})
return <-errCh
}
func (p *PortalUtp) Stop() {
err := p.listener.Close()
if err != nil {
p.log.Error("close utp listener has error", "error", err)
}
p.discV5.Close()
}
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)))
return conn, err
}
func (p *PortalUtp) Dial(ctx context.Context, dest *enode.Node) (net.Conn, error) {
raddr := &utp.Addr{IP: dest.IP(), Port: dest.UDP()}
p.log.Info("will connect to: ", "addr", raddr.String())
conn, err := utp.DialUTPOptions("utp", p.lAddr, raddr, utp.WithContext(ctx), utp.WithSocketManager(p.utpSm))
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) Accept(ctx context.Context) (*utp.Conn, error) {
return p.listener.AcceptUTPContext(ctx, 0)
}
func (p *PortalUtp) getLocalAddr() *net.UDPAddr {
laddr := p.conn.LocalAddr().(*net.UDPAddr)
p.log.Debug("UDP listener up", "addr", laddr)
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))
if n, ok := p.discV5.GetCachedNode(addr.String()); ok {
//_, err := p.DiscV5.TalkRequestToID(id, addr, string(portalwire.UTPNetwork), buf)
req := &v5wire.TalkRequest{Protocol: string(portalwire.Utp), Message: buf}
p.discV5.sendFromAnotherThreadWithNode(n, netip.AddrPortFrom(netutil.IPToAddr(addr.IP), uint16(addr.Port)), req)
return len(buf), nil
} else {
p.log.Warn("not found target node info", "ip", addr.IP.To4().String(), "port", addr.Port, "bufLength", len(buf))
return 0, fmt.Errorf("not found target node id")
}
}
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)
return []byte("")
}

2
p2p/discover/v5_udp.go
View file

@ -588,6 +588,8 @@ func (t *UDPv5) dispatch() {
if ct.c == active && ct.timer == active.timeout {
ct.c.err <- errTimeout
}
delete(t.activeCallByAuth, ct.c.nonce)
ct.c.timeout.Stop()
case c := <-t.callDoneCh:
active := t.activeCallByNode[c.id]