mirror of
https://github.com/ethereum/go-ethereum.git
synced 2026-07-26 06:36:43 +00:00
make send in separate goroutine
Signed-off-by: Chen Kai <281165273grape@gmail.com>
This commit is contained in:
parent
6617c338d8
commit
62d6517c06
3 changed files with 62 additions and 3 deletions
1
go.mod
1
go.mod
|
|
@ -118,6 +118,7 @@ require (
|
|||
github.com/klauspost/cpuid/v2 v2.0.9 // indirect
|
||||
github.com/kr/pretty v0.3.1 // indirect
|
||||
github.com/kr/text v0.2.0 // indirect
|
||||
github.com/libp2p/go-buffer-pool v0.1.0 // indirect
|
||||
github.com/mattn/go-runewidth v0.0.13 // indirect
|
||||
github.com/matttproud/golang_protobuf_extensions v1.0.4 // indirect
|
||||
github.com/minio/sha256-simd v1.0.0 // indirect
|
||||
|
|
|
|||
2
go.sum
2
go.sum
|
|
@ -238,6 +238,8 @@ github.com/labstack/echo/v4 v4.2.1/go.mod h1:AA49e0DZ8kk5jTOOCKNuPR6oTnBS0dYiM4F
|
|||
github.com/labstack/gommon v0.3.0/go.mod h1:MULnywXg0yavhxWKc+lOruYdAhDwPK9wf0OL7NoOu+k=
|
||||
github.com/leanovate/gopter v0.2.11 h1:vRjThO1EKPb/1NsDXuDrzldR28RLkBflWYcU9CvzWu4=
|
||||
github.com/leanovate/gopter v0.2.11/go.mod h1:aK3tzZP/C+p1m3SPRE4SYZFGP7jjkuSI4f7Xvpt0S9c=
|
||||
github.com/libp2p/go-buffer-pool v0.1.0 h1:oK4mSFcQz7cTQIfqbe4MIj9gLW+mnanjyFtc6cdF0Y8=
|
||||
github.com/libp2p/go-buffer-pool v0.1.0/go.mod h1:N+vh8gMqimBzdKkSMVuydVDq+UV5QTWy5HSiZacSbPg=
|
||||
github.com/mailru/easyjson v0.0.0-20190614124828-94de47d64c63/go.mod h1:C1wdFJiN94OJF2b5HbByQZoLdCWB1Yqtg26g4irojpc=
|
||||
github.com/mailru/easyjson v0.0.0-20190626092158-b2ccc519800e/go.mod h1:C1wdFJiN94OJF2b5HbByQZoLdCWB1Yqtg26g4irojpc=
|
||||
github.com/matryer/moq v0.0.0-20190312154309-6cfb0558e1bd/go.mod h1:9ELz6aaclSIGnZBoaSLZ3NAl1VTufbOrXBPvtcy6WiQ=
|
||||
|
|
|
|||
|
|
@ -37,6 +37,7 @@ import (
|
|||
"github.com/ethereum/go-ethereum/p2p/enode"
|
||||
"github.com/ethereum/go-ethereum/p2p/enr"
|
||||
"github.com/ethereum/go-ethereum/p2p/netutil"
|
||||
bufferpool "github.com/libp2p/go-buffer-pool"
|
||||
)
|
||||
|
||||
const (
|
||||
|
|
@ -99,6 +100,7 @@ type UDPv5 struct {
|
|||
sendCh chan sendRequest
|
||||
sendNoRespCh chan *sendNoRespRequest
|
||||
unhandled chan<- ReadPacket
|
||||
writeCh chan pendingWrite // New channel for outgoing packets
|
||||
|
||||
// state of dispatch
|
||||
codec codecV5
|
||||
|
|
@ -114,6 +116,12 @@ type UDPv5 struct {
|
|||
wg sync.WaitGroup
|
||||
}
|
||||
|
||||
// pendingWrite holds data for a packet to be sent by the writeLoop.
|
||||
type pendingWrite struct {
|
||||
toAddr netip.AddrPort
|
||||
data []byte
|
||||
}
|
||||
|
||||
type sendRequest struct {
|
||||
destID enode.ID
|
||||
destAddr netip.AddrPort
|
||||
|
|
@ -158,9 +166,10 @@ func ListenV5(conn UDPConn, ln *enode.LocalNode, cfg Config) (*UDPv5, error) {
|
|||
return nil, err
|
||||
}
|
||||
go t.tab.loop()
|
||||
t.wg.Add(2)
|
||||
t.wg.Add(3)
|
||||
go t.readLoop()
|
||||
go t.dispatch()
|
||||
go t.writeLoop()
|
||||
return t, nil
|
||||
}
|
||||
|
||||
|
|
@ -188,6 +197,7 @@ func newUDPv5(conn UDPConn, ln *enode.LocalNode, cfg Config) (*UDPv5, error) {
|
|||
sendNoRespCh: make(chan *sendNoRespRequest),
|
||||
respTimeoutCh: make(chan *callTimeout),
|
||||
unhandled: cfg.Unhandled,
|
||||
writeCh: make(chan pendingWrite, 128), // Buffered channel for outgoing packets
|
||||
// state of dispatch
|
||||
codec: v5wire.NewCodec(ln, cfg.PrivateKey, cfg.Clock, cfg.V5ProtocolID),
|
||||
activeCallByNode: make(map[enode.ID]*callV5),
|
||||
|
|
@ -631,6 +641,7 @@ func (t *UDPv5) dispatch() {
|
|||
|
||||
case <-t.closeCtx.Done():
|
||||
close(t.readNextCh)
|
||||
close(t.writeCh)
|
||||
for id, queue := range t.callQueue {
|
||||
for _, c := range queue {
|
||||
c.err <- errClosed
|
||||
|
|
@ -770,10 +781,55 @@ func (t *UDPv5) send(toID enode.ID, toAddr netip.AddrPort, packet v5wire.Packet,
|
|||
t.log.Warn(">> "+packet.Name(), t.logcontext...)
|
||||
return nonce, err
|
||||
}
|
||||
|
||||
t.logcontext = append(t.logcontext, "rawpacket", hexutil.Encode(enc))
|
||||
_, err = t.conn.WriteToUDPAddrPort(enc, toAddr)
|
||||
t.log.Trace(">> "+packet.Name(), t.logcontext...)
|
||||
return nonce, err
|
||||
|
||||
dataForSend := bufferpool.Get(len(enc))
|
||||
copy(dataForSend, enc)
|
||||
|
||||
pw := pendingWrite{
|
||||
toAddr: toAddr,
|
||||
data: dataForSend, // codec.Encode should return a new slice, safe to pass directly.
|
||||
}
|
||||
|
||||
select {
|
||||
case t.writeCh <- pw:
|
||||
// Packet successfully queued.
|
||||
return nonce, nil
|
||||
case <-t.closeCtx.Done():
|
||||
bufferpool.Put(dataForSend)
|
||||
return nonce, errClosed
|
||||
}
|
||||
}
|
||||
|
||||
// writeLoop runs in its own goroutine and sends packets from the writeCh to the network.
|
||||
func (t *UDPv5) writeLoop() {
|
||||
defer t.wg.Done()
|
||||
for pw := range t.writeCh { // Loop continues until writeCh is closed and empty.
|
||||
_, err := t.conn.WriteToUDPAddrPort(pw.data, pw.toAddr)
|
||||
if err != nil {
|
||||
// Generic error logging, as we don't have packetName or rich context here.
|
||||
select {
|
||||
case <-t.closeCtx.Done():
|
||||
// Log trace level if error occurs during or after shutdown initiation.
|
||||
t.log.Trace("UDP write error during/after shutdown", "addr", pw.toAddr, "err", err)
|
||||
default:
|
||||
// Not closing, so it's a more unexpected error.
|
||||
if netutil.IsTemporaryError(err) {
|
||||
t.log.Debug("Temporary UDP write error", "addr", pw.toAddr, "err", err)
|
||||
} else if !errors.Is(err, net.ErrClosed) && !errors.Is(err, io.EOF) { // Avoid logging common "closed" errors if not caught by closeCtx.
|
||||
t.log.Warn("UDP write error", "addr", pw.toAddr, "err", err)
|
||||
}
|
||||
}
|
||||
} else {
|
||||
// Minimal trace log confirming the actual send.
|
||||
// The detailed packet-specific trace was done in the 'send' method.
|
||||
t.log.Trace("UDP packet data sent", "addr", pw.toAddr, "len", len(pw.data))
|
||||
}
|
||||
|
||||
bufferpool.Put(pw.data)
|
||||
}
|
||||
}
|
||||
|
||||
// readLoop runs in its own goroutine and reads packets from the network.
|
||||
|
|
|
|||
Loading…
Reference in a new issue