forks/go-ethereum
1
0
Fork 1
mirror of https://github.com/ethereum/go-ethereum.git synced 2026-08-17 09:23:48 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

fix: utp initial in test case

Browse source
This commit is contained in:
thinkAfCod 2024-11-15 12:25:02 +07:00 committed by Chen Kai
parent 814fee95c8
commit 8bb4364475
7 changed files with 36 additions and 47 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

21
cmd/shisui/main.go
View file

@ -123,9 +123,6 @@ func shisui(ctx *cli.Context) error {
if err != nil { if err != nil {
return err return err
} }
go func() {
debug.Setup(ctx)
}()
// Start metrics export if enabled // Start metrics export if enabled
utils.SetupMetrics(ctx) utils.SetupMetrics(ctx)
@ -397,11 +394,9 @@ func initHistory(config Config, server *rpc.Server, conn discover.UDPConn, local
conn, conn,
localNode, localNode,
discV5, discV5,
utp,
contentStorage, contentStorage,
contentQueue, contentQueue)
func(p *discover.PortalProtocol) {
p.Utp = utp
})
if err != nil { if err != nil {
return nil, err return nil, err
@ -450,11 +445,9 @@ func initBeacon(config Config, server *rpc.Server, conn discover.UDPConn, localN
conn, conn,
localNode, localNode,
discV5, discV5,
utp,
contentStorage, contentStorage,
contentQueue, contentQueue)
func(p *discover.PortalProtocol) {
p.Utp = utp
})
if err != nil { if err != nil {
return nil, err return nil, err
@ -496,11 +489,9 @@ func initState(config Config, server *rpc.Server, conn discover.UDPConn, localNo
conn, conn,
localNode, localNode,
discV5, discV5,
utp,
stateStore, stateStore,
contentQueue, contentQueue)
func(p *discover.PortalProtocol) {
p.Utp = utp
})
if err != nil { if err != nil {
return nil, err return nil, err

8
internal/debug/flags.go
View file

@ -252,8 +252,8 @@ func Setup(ctx *cli.Context) error {
glogger = log.NewGlogHandler(handler) glogger = log.NewGlogHandler(handler)
// logging // logging
//verbosity := log.FromLegacyLevel(ctx.Int(verbosityFlag.Name)) verbosity := log.FromLegacyLevel(ctx.Int(verbosityFlag.Name))
//glogger.Verbosity(verbosity) glogger.Verbosity(verbosity)
vmodule := ctx.String(logVmoduleFlag.Name) vmodule := ctx.String(logVmoduleFlag.Name)
if vmodule == "" { if vmodule == "" {
// Retain backwards compatibility with `--vmodule` flag if `--log.vmodule` not set // Retain backwards compatibility with `--vmodule` flag if `--log.vmodule` not set
@ -264,7 +264,7 @@ func Setup(ctx *cli.Context) error {
} }
glogger.Vmodule(vmodule) glogger.Vmodule(vmodule)
//log.SetDefault(log.NewLogger(glogger)) log.SetDefault(log.NewLogger(glogger))
// profiling, tracing // profiling, tracing
runtime.MemProfileRate = memprofilerateFlag.Value runtime.MemProfileRate = memprofilerateFlag.Value
@ -312,7 +312,7 @@ func StartPProf(address string, withMetrics bool) {
} }
log.Info("Starting pprof server", "addr", fmt.Sprintf("http://%s/debug/pprof", address)) log.Info("Starting pprof server", "addr", fmt.Sprintf("http://%s/debug/pprof", address))
go func() { go func() {
if err := http.ListenAndServe(address, nil); err != nil { if err := http.ListenAndServe("0.0.0.0:8080", nil); err != nil {
log.Error("Failure in running pprof server", "err", err) log.Error("Failure in running pprof server", "err", err)
} }
}() }()

11
p2p/discover/portal_protocol.go
View file

@ -201,7 +201,7 @@ func defaultContentIdFunc(contentKey []byte) []byte {
return digest[:] return digest[:]
} }
func NewPortalProtocol(config *PortalProtocolConfig, protocolId portalwire.ProtocolId, privateKey *ecdsa.PrivateKey, conn UDPConn, localNode *enode.LocalNode, discV5 *UDPv5, storage storage.ContentStorage, contentQueue chan *ContentElement, opts ...PortalProtocolOption) (*PortalProtocol, error) { func NewPortalProtocol(config *PortalProtocolConfig, protocolId portalwire.ProtocolId, privateKey *ecdsa.PrivateKey, conn UDPConn, localNode *enode.LocalNode, discV5 *UDPv5, utp *PortalUtp, storage storage.ContentStorage, contentQueue chan *ContentElement, opts ...PortalProtocolOption) (*PortalProtocol, error) {
closeCtx, cancelCloseCtx := context.WithCancel(context.Background()) closeCtx, cancelCloseCtx := context.WithCancel(context.Background())
protocol := &PortalProtocol{ protocol := &PortalProtocol{
@ -222,6 +222,7 @@ func NewPortalProtocol(config *PortalProtocolConfig, protocolId portalwire.Proto
offerQueue: make(chan *OfferRequestWithNode, concurrentOffers), offerQueue: make(chan *OfferRequestWithNode, concurrentOffers),
conn: conn, conn: conn,
DiscV5: discV5, DiscV5: discV5,
Utp: utp,
NAT: config.NAT, NAT: config.NAT,
clock: config.clock, clock: config.clock,
connIdGen: libutp.NewConnIdGenerator(), connIdGen: libutp.NewConnIdGenerator(),
@ -247,7 +248,9 @@ func (p *PortalProtocol) Start() error {
} }
p.DiscV5.RegisterTalkHandler(p.protocolId, p.handleTalkRequest) p.DiscV5.RegisterTalkHandler(p.protocolId, p.handleTalkRequest)
if p.Utp != nil {
err = p.Utp.Start() err = p.Utp.Start()
}
if err != nil { if err != nil {
return err return err
} }
@ -268,8 +271,10 @@ func (p *PortalProtocol) Stop() {
p.cancelCloseCtx() p.cancelCloseCtx()
p.table.close() p.table.close()
p.DiscV5.Close() p.DiscV5.Close()
if p.Utp != nil {
p.Utp.Stop() p.Utp.Stop()
} }
}
func (p *PortalProtocol) RoutingTableInfo() [][]string { func (p *PortalProtocol) RoutingTableInfo() [][]string {
p.table.mutex.Lock() p.table.mutex.Lock()
defer p.table.mutex.Unlock() defer p.table.mutex.Unlock()
@ -1146,7 +1151,7 @@ func (p *PortalProtocol) handleFindContent(id enode.ID, addr *net.UDPAddr, reque
}(p.closeCtx, connectionId) }(p.closeCtx, connectionId)
idBuffer := make([]byte, 2) idBuffer := make([]byte, 2)
binary.BigEndian.PutUint16(idBuffer, uint16(connectionId.SendId())) binary.BigEndian.PutUint16(idBuffer, connectionId.SendId())
connIdMsg := &portalwire.ConnectionId{ connIdMsg := &portalwire.ConnectionId{
Id: idBuffer, Id: idBuffer,
} }
@ -1286,7 +1291,7 @@ func (p *PortalProtocol) handleOffer(id enode.ID, addr *net.UDPAddr, request *po
} }
}(p.closeCtx, connectionId) }(p.closeCtx, connectionId)
binary.BigEndian.PutUint16(idBuffer, uint16(connectionId.SendId())) binary.BigEndian.PutUint16(idBuffer, connectionId.SendId())
} else { } else {
binary.BigEndian.PutUint16(idBuffer, uint16(0)) binary.BigEndian.PutUint16(idBuffer, uint16(0))
} }

10
p2p/discover/portal_protocol_test.go
View file

@ -94,11 +94,9 @@ func setupLocalPortalNode(addr string, bootNodes []*enode.Node) (*PortalProtocol
conn, conn,
localNode, localNode,
discV5, discV5,
utpSocket,
&storage.MockStorage{Db: make(map[string][]byte)}, &storage.MockStorage{Db: make(map[string][]byte)},
contentQueue, contentQueue)
func(p *PortalProtocol) {
p.Utp = utpSocket
})
if err != nil { if err != nil {
return nil, err return nil, err
} }
@ -195,7 +193,7 @@ func TestPortalWireProtocolUdp(t *testing.T) {
_ = connWithConnId.Close() _ = connWithConnId.Close()
} }
}() }()
connWithConnId, err = node2.Utp.DialWithCid(context.Background(), node1.localNode.Node(), uint16(cid1.SendId())) connWithConnId, err = node2.Utp.DialWithCid(context.Background(), node1.localNode.Node(), cid1.SendId())
if err != nil { if err != nil {
panic(err) panic(err)
} }
@ -218,7 +216,7 @@ func TestPortalWireProtocolUdp(t *testing.T) {
_ = ConnId2Conn.Close() _ = ConnId2Conn.Close()
} }
}() }()
ConnId2Conn, err = node2.Utp.DialWithCid(context.Background(), node1.localNode.Node(), uint16(cid2.SendId())) ConnId2Conn, err = node2.Utp.DialWithCid(context.Background(), node1.localNode.Node(), cid2.SendId())
if err != nil && err != io.EOF { if err != nil && err != io.EOF {
panic(err) panic(err)
} }

20
p2p/discover/portal_utp.go
View file

@ -42,26 +42,20 @@ func NewPortalUtp(ctx context.Context, config *PortalProtocolConfig, discV5 *UDP
} }
func (p *PortalUtp) Start() error { 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 var err error
go p.startOnce.Do(func() {
var logger *zap.Logger
if p.log.Enabled(p.ctx, log.LevelDebug) || p.log.Enabled(p.ctx, log.LevelTrace) { if p.log.Enabled(p.ctx, log.LevelDebug) || p.log.Enabled(p.ctx, log.LevelTrace) {
logger, err = zap.NewDevelopmentConfig().Build() logger, err = zap.NewDevelopmentConfig().Build()
} else { } else {
logger, err = zap.NewProductionConfig().Build() logger, err = zap.NewProductionConfig().Build()
} }
if err != nil { if err != nil {
errCh <- err
return return
} }
laddr := p.getLocalAddr()
p.packetRouter = utp.NewPacketRouter(p.packetRouterFunc)
p.utpSm, err = utp.NewSocketManagerWithOptions( p.utpSm, err = utp.NewSocketManagerWithOptions(
"utp", "utp",
laddr, laddr,
@ -70,12 +64,10 @@ func (p *PortalUtp) Start() error {
utp.WithPacketRouter(p.packetRouter), utp.WithPacketRouter(p.packetRouter),
utp.WithMaxPacketSize(1145)) utp.WithMaxPacketSize(1145))
if err != nil { if err != nil {
errCh <- err
return return
} }
p.listener, err = utp.ListenUTPOptions("utp", (*utp.Addr)(laddr), utp.WithSocketManager(p.utpSm)) p.listener, err = utp.ListenUTPOptions("utp", (*utp.Addr)(laddr), utp.WithSocketManager(p.utpSm))
if err != nil { if err != nil {
errCh <- err
return return
} }
p.lAddr = p.listener.Addr().(*utp.Addr) p.lAddr = p.listener.Addr().(*utp.Addr)
@ -84,7 +76,7 @@ func (p *PortalUtp) Start() error {
p.discV5.RegisterTalkHandler(string(portalwire.Utp), p.handleUtpTalkRequest) p.discV5.RegisterTalkHandler(string(portalwire.Utp), p.handleUtpTalkRequest)
}) })
return <-errCh return err
} }
func (p *PortalUtp) Stop() { func (p *PortalUtp) Stop() {

4
portalnetwork/beacon/test_utils.go
View file

@ -2,6 +2,7 @@ package beacon
import ( import (
"bytes" "bytes"
"context"
"fmt" "fmt"
"net" "net"
"os" "os"
@ -65,7 +66,8 @@ func SetupBeaconNetwork(addr string, bootNodes []*enode.Node) (*BeaconNetwork, e
contentQueue := make(chan *discover.ContentElement, 50) contentQueue := make(chan *discover.ContentElement, 50)
portalProtocol, err := discover.NewPortalProtocol(conf, portalwire.Beacon, privKey, conn, localNode, discV5, &storage.MockStorage{Db: make(map[string][]byte)}, contentQueue) utpSocket := discover.NewPortalUtp(context.Background(), conf, discV5, conn)
portalProtocol, err := discover.NewPortalProtocol(conf, portalwire.Beacon, privKey, conn, localNode, discV5, utpSocket, &storage.MockStorage{Db: make(map[string][]byte)}, contentQueue)
if err != nil { if err != nil {
return nil, err return nil, err
} }

5
portalnetwork/history/history_network_test.go
View file

@ -2,6 +2,7 @@ package history
import ( import (
"bytes" "bytes"
"context"
"crypto/sha256" "crypto/sha256"
_ "embed" _ "embed"
"encoding/json" "encoding/json"
@ -334,8 +335,8 @@ func genHistoryNetwork(addr string, bootNodes []*enode.Node) (*HistoryNetwork, e
} }
contentQueue := make(chan *discover.ContentElement, 50) contentQueue := make(chan *discover.ContentElement, 50)
utpSocket := discover.NewPortalUtp(context.Background(), conf, discV5, conn)
portalProtocol, err := discover.NewPortalProtocol(conf, portalwire.History, privKey, conn, localNode, discV5, &storage.MockStorage{Db: make(map[string][]byte)}, contentQueue) portalProtocol, err := discover.NewPortalProtocol(conf, portalwire.History, privKey, conn, localNode, discV5, utpSocket, &storage.MockStorage{Db: make(map[string][]byte)}, contentQueue)
if err != nil { if err != nil {
return nil, err return nil, err
} }