diff --git a/cmd/shisui/main.go b/cmd/shisui/main.go index 9a694b704f..7f1aff73e4 100644 --- a/cmd/shisui/main.go +++ b/cmd/shisui/main.go @@ -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 diff --git a/p2p/discover/portal_protocol.go b/p2p/discover/portal_protocol.go index f6cb8c7ae9..64e13cccbc 100644 --- a/p2p/discover/portal_protocol.go +++ b/p2p/discover/portal_protocol.go @@ -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 { diff --git a/p2p/discover/portal_protocol_test.go b/p2p/discover/portal_protocol_test.go index adfab1e50f..06647e5ea8 100644 --- a/p2p/discover/portal_protocol_test.go +++ b/p2p/discover/portal_protocol_test.go @@ -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) } diff --git a/p2p/discover/portal_utp.go b/p2p/discover/portal_utp.go new file mode 100644 index 0000000000..c635c16989 --- /dev/null +++ b/p2p/discover/portal_utp.go @@ -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("") +} diff --git a/p2p/discover/v5_udp.go b/p2p/discover/v5_udp.go index 72c9be5179..6383f5e4a7 100644 --- a/p2p/discover/v5_udp.go +++ b/p2p/discover/v5_udp.go @@ -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]