PEX: integrate with send throttling

diff --git a/peerconn.go b/peerconn.go
index 1be1289192165dd095f94850605f5f28d8212acc..052dcdc0dbd91cee962af6a9e0f122cd77948f8e 100644 (file)
--- a/peerconn.go
+++ b/peerconn.go
@@ -564,7 +564,9 @@ func (cn *PeerConn) fillWriteBuffer(msg func(pp.Message) bool) {
                cn.requestsLowWater = len(cn.requests) / 2
        }
        if cn.pex.IsEnabled() {
-               cn.pex.Share(msg) // gated internally
+               if flow := cn.pex.Share(msg); !flow {
+                       return
+               }
        }
        cn.upload(msg)
 }

diff --git a/pexconn.go b/pexconn.go
index 51aac7e9f6842b8cf84689322cc501d0af3cac5c..e70ed105de502f3aec81a4dd8776dec280d3e588 100644 (file)
--- a/pexconn.go
+++ b/pexconn.go
@@ -68,20 +68,22 @@ func (s *pexConnState) genmsg() *pp.PexMsg {
 }
 
 // Share is called from the writer goroutine if when it is woken up with the write buffers empty
-func (s *pexConnState) Share(postfn messageWriter) {
+// Returns whether there's more room on the send buffer to write to.
+func (s *pexConnState) Share(postfn messageWriter) bool {
        select {
        case <-s.gate:
                if tx := s.genmsg(); tx != nil {
                        s.dbg.Print("sending PEX message: ", tx)
-                       postfn(tx.Message(s.xid))
+                       flow := postfn(tx.Message(s.xid))
                        s.sched(pexInterval)
+                       return flow
                } else {
                        // no PEX to send this time - try again shortly
                        s.sched(pexRetryDelay)
                }
        default:
-               return
        }
+       return true
 }
 
 // Recv is called from the reader goroutine