forks/go-ethereum
1
0
Fork 1
mirror of https://github.com/ethereum/go-ethereum.git synced 2026-07-26 14:46:42 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

remove write channel

Browse source
This commit is contained in:
thinkAfCod 2025-05-19 14:44:55 +08:00
parent e5d979e582
commit 33a000cb3e
1 changed files with 17 additions and 41 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

58
p2p/discover/v5_udp.go
View file

@ -97,7 +97,6 @@ type UDPv5 struct {
sendCh chan sendRequest sendCh chan sendRequest
sendNoRespCh chan *sendNoRespRequest sendNoRespCh chan *sendNoRespRequest
unhandled chan<- ReadPacket unhandled chan<- ReadPacket
writeCh chan pendingWrite // New channel for outgoing packets
// state of dispatch // state of dispatch
codec codecV5 codec codecV5
@ -113,12 +112,6 @@ type UDPv5 struct {
wg sync.WaitGroup 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 { type sendRequest struct {
destID enode.ID destID enode.ID
destAddr netip.AddrPort destAddr netip.AddrPort
@ -166,7 +159,6 @@ func ListenV5(conn UDPConn, ln *enode.LocalNode, cfg Config) (*UDPv5, error) {
t.wg.Add(3) t.wg.Add(3)
go t.readLoop() go t.readLoop()
go t.dispatch() go t.dispatch()
go t.writeLoop()
return t, nil return t, nil
} }
@ -186,14 +178,13 @@ func newUDPv5(conn UDPConn, ln *enode.LocalNode, cfg Config) (*UDPv5, error) {
clock: cfg.Clock, clock: cfg.Clock,
respTimeout: cfg.V5RespTimeout, respTimeout: cfg.V5RespTimeout,
// channels into dispatch // channels into dispatch
packetInCh: make(chan ReadPacket, 64), packetInCh: make(chan ReadPacket, 1),
callCh: make(chan *callV5), callCh: make(chan *callV5),
callDoneCh: make(chan *callV5), callDoneCh: make(chan *callV5),
sendCh: make(chan sendRequest), sendCh: make(chan sendRequest),
sendNoRespCh: make(chan *sendNoRespRequest), sendNoRespCh: make(chan *sendNoRespRequest),
respTimeoutCh: make(chan *callTimeout), respTimeoutCh: make(chan *callTimeout),
unhandled: cfg.Unhandled, unhandled: cfg.Unhandled,
writeCh: make(chan pendingWrite, 32), // Buffered channel for outgoing packets
// state of dispatch // state of dispatch
codec: v5wire.NewCodec(ln, cfg.PrivateKey, cfg.Clock, cfg.V5ProtocolID), codec: v5wire.NewCodec(ln, cfg.PrivateKey, cfg.Clock, cfg.V5ProtocolID),
activeCallByNode: make(map[enode.ID]*callV5), activeCallByNode: make(map[enode.ID]*callV5),
@ -625,7 +616,6 @@ func (t *UDPv5) dispatch() {
t.handlePacket(p.Data, p.Addr) t.handlePacket(p.Data, p.Addr)
case <-t.closeCtx.Done(): case <-t.closeCtx.Done():
close(t.writeCh)
for id, queue := range t.callQueue { for id, queue := range t.callQueue {
for _, c := range queue { for _, c := range queue {
c.err <- errClosed c.err <- errClosed
@ -766,40 +756,26 @@ func (t *UDPv5) send(toID enode.ID, toAddr netip.AddrPort, packet v5wire.Packet,
return nonce, err return nonce, err
} }
_, err = t.conn.WriteToUDPAddrPort(enc, toAddr)
t.log.Trace(">> "+packet.Name(), t.logcontext...) t.log.Trace(">> "+packet.Name(), t.logcontext...)
return nonce, err
dataForSend := make([]byte, 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():
return nonce, errClosed
}
} }
// writeLoop runs in its own goroutine and sends packets from the writeCh to the network. // writeLoop runs in its own goroutine and sends packets from the writeCh to the network.
func (t *UDPv5) writeLoop() { //func (t *UDPv5) writeLoop() {
defer t.wg.Done() // defer t.wg.Done()
for pw := range t.writeCh { // Loop continues until writeCh is closed and empty. // for pw := range t.writeCh { // Loop continues until writeCh is closed and empty.
_, err := t.conn.WriteToUDPAddrPort(pw.data, pw.toAddr) // _, err := t.conn.WriteToUDPAddrPort(pw.data, pw.toAddr)
if netutil.IsTemporaryError(err) { // if netutil.IsTemporaryError(err) {
t.log.Debug("Temporary UDP write error", "addr", pw.toAddr, "err", err) // t.log.Debug("Temporary UDP write error", "addr", pw.toAddr, "err", err)
} else if err != nil { // } else if err != nil {
if !errors.Is(err, net.ErrClosed) || !errors.Is(err, io.EOF) { // if !errors.Is(err, net.ErrClosed) || !errors.Is(err, io.EOF) {
t.log.Warn("UDP write error", "addr", pw.toAddr, "err", err) // t.log.Warn("UDP write error", "addr", pw.toAddr, "err", err)
} // }
return // return
} // }
} // }
} //}
// readLoop runs in its own goroutine and reads packets from the network. // readLoop runs in its own goroutine and reads packets from the network.
func (t *UDPv5) readLoop() { func (t *UDPv5) readLoop() {