mirror/erigon-pulse
1
0
Fork 0
mirror of https://gitlab.com/pulsechaincom/erigon-pulse.git synced 2024-12-25 21:17:16 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

Sentry fixes (for very frequent peer not found error) (#1918)

Browse Source 
* Extra prints

* Remove printouts

* Printing back

* Fix the stuck stage 1

* Not allow peers to connect more than once

Co-authored-by: Alex Sharp <alexsharp@Alexs-MacBook-Pro.local>
This commit is contained in:
ledgerwatch 2021-05-12 09:36:43 +01:00 committed by GitHub
parent 7f51f13804
commit 0aabf85c1d
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 11 additions and 16 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
cmd/headers/download/downloader.go
View File

@ -444,6 +444,7 @@ func NewControlServer(db ethdb.Database, nodeName string, chainConfig *params.Ch
}
func (cs *ControlServerImpl) newBlockHashes(ctx context.Context, req *proto_sentry.InboundMessage, sentry proto_sentry.SentryClient) error {
//log.Info(fmt.Sprintf("NewBlockHashes from [%s]", gointerfaces.ConvertH512ToBytes(req.PeerId)))
var request eth.NewBlockHashesPacket
if err := rlp.DecodeBytes(req.Data, &request); err != nil {
return fmt.Errorf("decode NewBlockHashes: %v", err)

18
cmd/headers/download/sentry.go
View File

@ -191,6 +191,10 @@ func MakeProtocols(ctx context.Context,
DialCandidates: dialCandidates,
Run: func(peer *p2p.Peer, rw p2p.MsgReadWriter) error {
peerID := peer.ID().String()
if _, ok := peers.Load(peerID); ok {
log.Debug(fmt.Sprintf("[%s] Peer already has connection", peerID))
return nil
}
log.Debug(fmt.Sprintf("[%s] Start with peer", peerID))
if err := handShake(ctx, statusFn(), peerID, rw, eth.ProtocolVersions[0], eth.ProtocolVersions[0]); err != nil {
return fmt.Errorf("handshake to peer %s: %v", peerID, err)
@ -778,12 +782,7 @@ func (ss *SentryServerImpl) SendMessageToRandomPeers(ctx context.Context, req *p
return true
}
if err := peerInfo.rw.WriteMsg(p2p.Msg{Code: msgcode, Size: uint32(len(req.Data.Data)), Payload: bytes.NewReader(req.Data.Data)}); err != nil {
if x, ok := ss.Peers.Load(peerID); ok {
peerInfo := x.(*PeerInfo)
if peerInfo != nil {
peerInfo.Remove()
}
}
peerInfo.Remove()
ss.Peers.Delete(peerID)
innerErr = err
return false
@ -818,12 +817,7 @@ func (ss *SentryServerImpl) SendMessageToAll(ctx context.Context, req *proto_sen
return true
}
if err := peerInfo.rw.WriteMsg(p2p.Msg{Code: msgcode, Size: uint32(len(req.Data)), Payload: bytes.NewReader(req.Data)}); err != nil {
if x, ok := ss.Peers.Load(peerID); ok {
peerInfo := x.(*PeerInfo)
if peerInfo != nil {
peerInfo.Remove()
}
}
peerInfo.Remove()
ss.Peers.Delete(peerID)
innerErr = err
return false

8
eth/stagedsync/stage_headers_new.go
View File

@ -196,6 +196,10 @@ func HeadersForward(
// if this is not an initial cycle, we need to react quickly when new headers are coming in
break
}
if initialCycle && cfg.hd.InSync() {
log.Debug("Top seen", "height", cfg.hd.TopSeenHeight())
break
}
timer = time.NewTimer(1 * time.Second)
select {
case <-ctx.Done():
@ -209,10 +213,6 @@ func HeadersForward(
case <-cfg.hd.DeliveryNotify:
log.Debug("headerLoop woken up by the incoming request")
}
if initialCycle && cfg.hd.InSync() {
log.Debug("Top seen", "height", cfg.hd.TopSeenHeight())
break
}
}
if headerInserter.AnythingDone() {
if err := s.Update(batch, headerInserter.GetHighest()); err != nil {