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

fix: connection memory leak

Browse source
This commit is contained in:
thinkAfCod 2024-11-04 17:38:10 +08:00 committed by Chen Kai
parent 7815dab7af
commit 685123294b
1 changed files with 32 additions and 27 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

59
p2p/discover/portal_protocol.go
View file

@ -1145,36 +1145,38 @@ func (p *PortalProtocol) handleFindContent(id enode.ID, addr *net.UDPAddr, reque
return talkRespBytes, nil return talkRespBytes, nil
} else { } else {
connId := p.connIdGen.GenCid(id, false) connectionId := p.connIdGen.GenCid(id, false)
connIdSend := connId.SendId()
go func(bctx context.Context) { go func(bctx context.Context, connId *libutp.ConnId) {
var conn *utp.Conn var conn *utp.Conn
defer func() { var connectCtx context.Context
var cancel context.CancelFunc
defer func(clsConn *utp.Conn) {
p.connIdGen.Remove(connId) p.connIdGen.Remove(connId)
if conn == nil { if clsConn == nil {
return return
} }
err := conn.Close() err := clsConn.Close()
if err != nil { if err != nil {
p.Log.Error("failed to close connection", "err", err) p.Log.Error("failed to close connection", "err", err)
} }
}() }(conn)
connectCtx, cancel = context.WithTimeout(bctx, defaultUTPConnectTimeout)
defer func(cancelFunc context.CancelFunc) {
cancelFunc()
}(cancel)
for { for {
select { select {
case <-bctx.Done(): case <-bctx.Done():
return return
default: default:
ctx, cancel := context.WithTimeout(bctx, defaultUTPConnectTimeout)
p.Log.Debug("will accept find content conn from: ", "source", addr, "connId", connId) p.Log.Debug("will accept find content conn from: ", "source", addr, "connId", connId)
conn, err = p.utp.AcceptUTPContext(ctx, connIdSend) conn, err = p.utp.AcceptUTPContext(connectCtx, connectionId.SendId())
cancel()
if err != nil { if err != nil {
if metrics.Enabled { if metrics.Enabled {
p.portalMetrics.utpOutFailConn.Inc(1) p.portalMetrics.utpOutFailConn.Inc(1)
} }
p.Log.Error("failed to accept utp connection for handle find content", "connId", connIdSend, "err", err) p.Log.Error("failed to accept utp connection for handle find content", "connId", connectionId.SendId(), "err", err)
return return
} }
@ -1204,10 +1206,10 @@ func (p *PortalProtocol) handleFindContent(id enode.ID, addr *net.UDPAddr, reque
return return
} }
} }
}(p.closeCtx) }(p.closeCtx, connectionId)
idBuffer := make([]byte, 2) idBuffer := make([]byte, 2)
binary.BigEndian.PutUint16(idBuffer, uint16(connIdSend)) binary.BigEndian.PutUint16(idBuffer, uint16(connectionId.SendId()))
connIdMsg := &portalwire.ConnectionId{ connIdMsg := &portalwire.ConnectionId{
Id: idBuffer, Id: idBuffer,
} }
@ -1277,36 +1279,39 @@ func (p *PortalProtocol) handleOffer(id enode.ID, addr *net.UDPAddr, request *po
idBuffer := make([]byte, 2) idBuffer := make([]byte, 2)
if contentKeyBitlist.Count() != 0 { if contentKeyBitlist.Count() != 0 {
connId := p.connIdGen.GenCid(id, false) connectionId := p.connIdGen.GenCid(id, false)
connIdSend := connId.SendId()
go func(bctx context.Context) { go func(bctx context.Context, connId *libutp.ConnId) {
var conn *utp.Conn var conn *utp.Conn
defer func() { var connectCtx context.Context
var cancel context.CancelFunc
defer func(clsConn *utp.Conn) {
p.connIdGen.Remove(connId) p.connIdGen.Remove(connId)
if conn == nil { if clsConn == nil {
return return
} }
err := conn.Close() err := clsConn.Close()
if err != nil { if err != nil {
p.Log.Error("failed to close connection", "err", err) p.Log.Error("failed to close connection", "err", err)
} }
}() }(conn)
connectCtx, cancel = context.WithTimeout(bctx, defaultUTPConnectTimeout)
defer func(cancelFunc context.CancelFunc) {
cancelFunc()
}(cancel)
for { for {
select { select {
case <-bctx.Done(): case <-bctx.Done():
return return
default: default:
ctx, cancel := context.WithTimeout(bctx, defaultUTPConnectTimeout)
p.Log.Debug("will accept offer conn from: ", "source", addr, "connId", connId) p.Log.Debug("will accept offer conn from: ", "source", addr, "connId", connId)
conn, err = p.utp.AcceptUTPContext(ctx, connIdSend) conn, err = p.utp.AcceptUTPContext(connectCtx, connectionId.SendId())
cancel() cancel()
if err != nil { if err != nil {
if metrics.Enabled { if metrics.Enabled {
p.portalMetrics.utpInFailConn.Inc(1) p.portalMetrics.utpInFailConn.Inc(1)
} }
p.Log.Error("failed to accept utp connection for handle offer", "connId", connIdSend, "err", err) p.Log.Error("failed to accept utp connection for handle offer", "connId", connectionId.SendId(), "err", err)
return return
} }
@ -1345,9 +1350,9 @@ func (p *PortalProtocol) handleOffer(id enode.ID, addr *net.UDPAddr, request *po
return return
} }
} }
}(p.closeCtx) }(p.closeCtx, connectionId)
binary.BigEndian.PutUint16(idBuffer, uint16(connIdSend)) binary.BigEndian.PutUint16(idBuffer, uint16(connectionId.SendId()))
} else { } else {
binary.BigEndian.PutUint16(idBuffer, uint16(0)) binary.BigEndian.PutUint16(idBuffer, uint16(0))
} }