cerc-io/plugeth
18
0
Fork 1
Code Issues 4 Pull Requests Packages Projects Releases Wiki Activity

les: drop the message if the entire p2p connection is stuck (#21033)

Browse Source 
* les: drop the message if the entire p2p connection is stuck

* les: fix lint
This commit is contained in:
gary rong 2020-05-12 16:02:15 +08:00 committed by GitHub
parent 7ace5a3a8b
commit 53cac027d0
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 2 additions and 23 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

23
les/peer.go
View File

@ -65,9 +65,6 @@ const (
// handshakeTimeout is the timeout LES handshake will be treated as failed. // handshakeTimeout is the timeout LES handshake will be treated as failed.
handshakeTimeout = 5 * time.Second handshakeTimeout = 5 * time.Second
// retrySendCachePeriod is the time interval a caching retry is performed.
retrySendCachePeriod = time.Millisecond * 100
) )
const ( const (
@ -164,24 +161,6 @@ func (p *peerCommons) queueSend(f func()) bool {
return p.sendQueue.Queue(f) return p.sendQueue.Queue(f)
} }
// mustQueueSend starts a for loop and retry the caching if failed.
// If the stopCh is closed, then it returns.
func (p *peerCommons) mustQueueSend(f func()) {
for {
// Check whether the stopCh is closed.
select {
case <-p.closeCh:
return
default:
}
// If the function is successfully cached, return.
if p.canQueue() && p.queueSend(f) {
return
}
time.Sleep(retrySendCachePeriod)
}
}
// String implements fmt.Stringer. // String implements fmt.Stringer.
func (p *peerCommons) String() string { func (p *peerCommons) String() string {
return fmt.Sprintf("Peer %s [%s]", p.id, fmt.Sprintf("les/%d", p.version)) return fmt.Sprintf("Peer %s [%s]", p.id, fmt.Sprintf("les/%d", p.version))
@ -899,7 +878,7 @@ func (p *clientPeer) updateCapacity(cap uint64) {
var kvList keyValueList var kvList keyValueList
kvList = kvList.add("flowControl/MRR", cap) kvList = kvList.add("flowControl/MRR", cap)
kvList = kvList.add("flowControl/BL", cap*bufLimitRatio) kvList = kvList.add("flowControl/BL", cap*bufLimitRatio)
p.mustQueueSend(func() { p.sendAnnounce(announceData{Update: kvList}) }) p.queueSend(func() { p.sendAnnounce(announceData{Update: kvList}) })
} }
// freezeClient temporarily puts the client in a frozen state which means all // freezeClient temporarily puts the client in a frozen state which means all

2
les/server_handler.go
View File

@ -262,7 +262,7 @@ func (h *serverHandler) handleMsg(p *clientPeer, wg *sync.WaitGroup) error {
h.server.clientPool.requestCost(p, realCost) h.server.clientPool.requestCost(p, realCost)
} }
if reply != nil { if reply != nil {
p.mustQueueSend(func() { p.queueSend(func() {
if err := reply.send(bv); err != nil { if err := reply.send(bv); err != nil {
select { select {
case p.errCh <- err: case p.errCh <- err: