From 1a244f51543e9e141aa0d84de3e7442e908746a9 Mon Sep 17 00:00:00 2001 From: Cassandra Heart <7929478+CassOnMars@users.noreply.github.com> Date: Sat, 29 Jun 2024 19:12:53 -0500 Subject: [PATCH] v1.4.20-p1 (#251) * v1.4.20-p1 * further refinements/tuning --- go-libp2p-blossomsub/blossomsub.go | 4 +- node/config/version.go | 2 +- node/consensus/master/broadcast_messaging.go | 28 ++++++------ .../master/master_clock_consensus_engine.go | 44 +++++++------------ node/main.go | 5 ++- node/p2p/blossomsub.go | 21 ++++++++- 6 files changed, 55 insertions(+), 49 deletions(-) diff --git a/go-libp2p-blossomsub/blossomsub.go b/go-libp2p-blossomsub/blossomsub.go index 0274b5f..43c71df 100644 --- a/go-libp2p-blossomsub/blossomsub.go +++ b/go-libp2p-blossomsub/blossomsub.go @@ -214,7 +214,7 @@ func NewBlossomSubWithRouter(ctx context.Context, h host.Host, rt PubSubRouter, } // NewBlossomSubRouter returns a new BlossomSubRouter with custom parameters. -func NewBlossomSubRouter(h host.Host, params BlossomSubParams) *BlossomSubRouter { +func NewBlossomSubRouter(h host.Host, params BlossomSubParams, addrBook peerstore.AddrBook) *BlossomSubRouter { return &BlossomSubRouter{ peers: make(map[peer.ID]protocol.ID), mesh: make(map[string]map[peer.ID]struct{}), @@ -222,7 +222,7 @@ func NewBlossomSubRouter(h host.Host, params BlossomSubParams) *BlossomSubRouter lastpub: make(map[string]int64), gossip: make(map[peer.ID][]*pb.ControlIHave), control: make(map[peer.ID]*pb.ControlMessage), - cab: pstoremem.NewAddrBook(), + cab: addrBook, backoff: make(map[string]map[peer.ID]time.Time), peerhave: make(map[peer.ID]int), iasked: make(map[peer.ID]int), diff --git a/node/config/version.go b/node/config/version.go index 2a0ea18..92ef409 100644 --- a/node/config/version.go +++ b/node/config/version.go @@ -36,5 +36,5 @@ func FormatVersion(version []byte) string { } func GetPatchNumber() byte { - return 0x00 + return 0x01 } diff --git a/node/consensus/master/broadcast_messaging.go b/node/consensus/master/broadcast_messaging.go index 17f2552..9c2d334 100644 --- a/node/consensus/master/broadcast_messaging.go +++ b/node/consensus/master/broadcast_messaging.go @@ -92,16 +92,17 @@ func (e *MasterClockConsensusEngine) handleSelfTestReport( info.DifficultyMetric = report.DifficultyMetric info.MasterHeadFrame = report.MasterHeadFrame - if info.Bandwidth <= 1048576 { + if info.Bandwidth == 0 { go func() { ctx, cancel := context.WithTimeout(context.TODO(), 5*time.Minute) defer cancel() ch := e.pubSub.GetMultiaddrOfPeerStream(ctx, peerID) select { case <-ch: - go func() { - e.bandwidthTestCh <- peerID - }() + select { + case e.bandwidthTestCh <- peerID: + default: + } case <-ctx.Done(): } }() @@ -116,15 +117,16 @@ func (e *MasterClockConsensusEngine) handleSelfTestReport( for i := 0; i < len(proofs); i++ { proofs[i] = proof[i*516 : (i+1)*516] } - go func() { - e.verifyTestCh <- verifyChallenge{ - peerID: peerID, - challenge: challenge, - cores: report.Cores, - increment: report.Increment, - proof: proof, - } - }() + select { + case e.verifyTestCh <- verifyChallenge{ + peerID: peerID, + challenge: challenge, + cores: report.Cores, + increment: report.Increment, + proof: proof, + }: + default: + } return nil } diff --git a/node/consensus/master/master_clock_consensus_engine.go b/node/consensus/master/master_clock_consensus_engine.go index 6b372fc..85f6c59 100644 --- a/node/consensus/master/master_clock_consensus_engine.go +++ b/node/consensus/master/master_clock_consensus_engine.go @@ -146,7 +146,7 @@ func NewMasterClockConsensusEngine( report: report, frameValidationCh: make(chan *protobufs.ClockFrame), bandwidthTestCh: make(chan []byte), - verifyTestCh: make(chan verifyChallenge), + verifyTestCh: make(chan verifyChallenge, 4), engineConfig: engineConfig, } @@ -192,29 +192,14 @@ func (e *MasterClockConsensusEngine) Start() <-chan error { go func() { for { select { - case newFrame := <-e.frameValidationCh: - head, err := e.masterTimeReel.Head() - if err != nil { - panic(err) - } - - if head.FrameNumber > newFrame.FrameNumber || - newFrame.FrameNumber-head.FrameNumber > 128 { - e.logger.Debug( - "frame out of range, ignoring", - zap.Uint64("number", newFrame.FrameNumber), - ) - continue - } - - if err := e.frameProver.VerifyMasterClockFrame(newFrame); err != nil { - e.logger.Error("could not verify clock frame", zap.Error(err)) - continue - } - - e.masterTimeReel.Insert(newFrame, false) case peerId := <-e.bandwidthTestCh: e.performBandwidthTest(peerId) + } + } + }() + go func() { + for { + select { case verifyTest := <-e.verifyTestCh: e.performVerifyTest(verifyTest) } @@ -398,13 +383,14 @@ func (e *MasterClockConsensusEngine) Start() <-chan error { panic(err) } - e.logger.Info( - "broadcasting self-test info", - zap.Uint64("current_frame", e.report.MasterHeadFrame), - ) - - if err := e.publishMessage(e.filter, e.report); err != nil { - e.logger.Debug("error publishing message", zap.Error(err)) + if increment%30 == 0 { + e.logger.Info( + "broadcasting self-test info", + zap.Uint64("current_frame", e.report.MasterHeadFrame), + ) + if err := e.publishMessage(e.filter, e.report); err != nil { + e.logger.Debug("error publishing message", zap.Error(err)) + } } } else { skipStore = false diff --git a/node/main.go b/node/main.go index 0f61e3c..70747ca 100644 --- a/node/main.go +++ b/node/main.go @@ -227,7 +227,7 @@ func main() { fmt.Println("Signature check disabled, skipping...") } - if *memprofile != "" { + if *memprofile != "" && *core == 0 { go func() { for { time.Sleep(5 * time.Minute) @@ -241,7 +241,7 @@ func main() { }() } - if *cpuprofile != "" { + if *cpuprofile != "" && *core == 0 { f, err := os.Create(*cpuprofile) if err != nil { log.Fatal(err) @@ -438,6 +438,7 @@ func main() { } repair(*configDirectory, node) + runtime.GOMAXPROCS(1) if nodeConfig.ListenGRPCMultiaddr != "" { srv, err := rpc.NewRPCServer( diff --git a/node/p2p/blossomsub.go b/node/p2p/blossomsub.go index 97a0322..b129b2a 100644 --- a/node/p2p/blossomsub.go +++ b/node/p2p/blossomsub.go @@ -85,6 +85,16 @@ func getPeerID(p2pConfig *config.P2PConfig) peer.ID { return id } +type realclock struct{} + +func (rc realclock) Now() time.Time { + return time.Now() +} + +func (rc realclock) After(d time.Duration) <-chan time.Time { + return time.After(d) +} + func NewBlossomSub( p2pConfig *config.P2PConfig, peerstore store.Peerstore, @@ -126,7 +136,14 @@ func NewBlossomSub( opts = append(opts, libp2p.Identity(privKey)) } - ps, err := pstoreds.NewPeerstore(ctx, peerstore, pstoreds.DefaultOpts()) + ps, err := pstoreds.NewPeerstore(ctx, peerstore, pstoreds.Options{ + CacheSize: 120000, + MaxProtocols: 1024, + GCPurgeInterval: 2 * time.Hour, + GCLookaheadInterval: 0, + GCInitialDelay: 60 * time.Second, + Clock: realclock{}, + }) if err != nil { panic(err) } @@ -208,7 +225,7 @@ func NewBlossomSub( })) params := mergeDefaults(p2pConfig) - rt := blossomsub.NewBlossomSubRouter(h, params) + rt := blossomsub.NewBlossomSubRouter(h, params, ps) pubsub, err := blossomsub.NewBlossomSubWithRouter(ctx, h, rt, blossomOpts...) if err != nil { panic(err)