mirror of
https://github.com/kaspanet/kaspad.git
synced 2025-06-05 13:46:42 +00:00
Fix RPC client memory/goroutine leak (#2122)
* Showcase the RPC client memory leak * Fixes an RPC client goroutine leak by properly closing the underlying connection
This commit is contained in:
parent
715cb3b1ac
commit
9ee409afaa
@ -22,6 +22,7 @@ type OnDisconnectedHandler func()
|
|||||||
// GRPCClient is a gRPC-based RPC client
|
// GRPCClient is a gRPC-based RPC client
|
||||||
type GRPCClient struct {
|
type GRPCClient struct {
|
||||||
stream protowire.RPC_MessageStreamClient
|
stream protowire.RPC_MessageStreamClient
|
||||||
|
connection *grpc.ClientConn
|
||||||
onErrorHandler OnErrorHandler
|
onErrorHandler OnErrorHandler
|
||||||
onDisconnectedHandler OnDisconnectedHandler
|
onDisconnectedHandler OnDisconnectedHandler
|
||||||
}
|
}
|
||||||
@ -43,7 +44,12 @@ func Connect(address string) (*GRPCClient, error) {
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, errors.Wrapf(err, "error getting client stream for %s", address)
|
return nil, errors.Wrapf(err, "error getting client stream for %s", address)
|
||||||
}
|
}
|
||||||
return &GRPCClient{stream: stream}, nil
|
return &GRPCClient{stream: stream, connection: gRPCConnection}, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// Close closes the underlying grpc connection
|
||||||
|
func (c *GRPCClient) Close() error {
|
||||||
|
return c.connection.Close()
|
||||||
}
|
}
|
||||||
|
|
||||||
// Disconnect disconnects from the RPC server
|
// Disconnect disconnects from the RPC server
|
||||||
|
@ -143,6 +143,9 @@ func (c *RPCClient) handleClientDisconnected() {
|
|||||||
}
|
}
|
||||||
|
|
||||||
func (c *RPCClient) handleClientError(err error) {
|
func (c *RPCClient) handleClientError(err error) {
|
||||||
|
if atomic.LoadUint32(&c.isClosed) == 1 {
|
||||||
|
return
|
||||||
|
}
|
||||||
log.Warnf("Received error from client: %s", err)
|
log.Warnf("Received error from client: %s", err)
|
||||||
c.handleClientDisconnected()
|
c.handleClientDisconnected()
|
||||||
}
|
}
|
||||||
@ -159,7 +162,7 @@ func (c *RPCClient) Close() error {
|
|||||||
return errors.Errorf("Cannot close a client that had already been closed")
|
return errors.Errorf("Cannot close a client that had already been closed")
|
||||||
}
|
}
|
||||||
c.rpcRouter.router.Close()
|
c.rpcRouter.router.Close()
|
||||||
return nil
|
return c.GRPCClient.Close()
|
||||||
}
|
}
|
||||||
|
|
||||||
// Address returns the address the RPC client connected to
|
// Address returns the address the RPC client connected to
|
||||||
|
@ -2,6 +2,7 @@ package integration
|
|||||||
|
|
||||||
import (
|
import (
|
||||||
"github.com/kaspanet/kaspad/infrastructure/config"
|
"github.com/kaspanet/kaspad/infrastructure/config"
|
||||||
|
"runtime"
|
||||||
"testing"
|
"testing"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
@ -26,6 +27,37 @@ func newTestRPCClient(rpcAddress string) (*testRPCClient, error) {
|
|||||||
}, nil
|
}, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func connectAndClose(rpcAddress string) error {
|
||||||
|
client, err := rpcclient.NewRPCClient(rpcAddress)
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
defer client.Close()
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestRPCClientGoroutineLeak(t *testing.T) {
|
||||||
|
_, teardown := setupHarness(t, &harnessParams{
|
||||||
|
p2pAddress: p2pAddress1,
|
||||||
|
rpcAddress: rpcAddress1,
|
||||||
|
miningAddress: miningAddress1,
|
||||||
|
miningAddressPrivateKey: miningAddress1PrivateKey,
|
||||||
|
})
|
||||||
|
defer teardown()
|
||||||
|
numGoroutinesBefore := runtime.NumGoroutine()
|
||||||
|
for i := 1; i < 100; i++ {
|
||||||
|
err := connectAndClose(rpcAddress1)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("Failed to set up an RPC client: %s", err)
|
||||||
|
}
|
||||||
|
time.Sleep(10 * time.Millisecond)
|
||||||
|
if runtime.NumGoroutine() > numGoroutinesBefore+10 {
|
||||||
|
t.Fatalf("Number of goroutines is increasing for each RPC client open (%d -> %d), which indicates a memory leak",
|
||||||
|
numGoroutinesBefore, runtime.NumGoroutine())
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
func TestRPCMaxInboundConnections(t *testing.T) {
|
func TestRPCMaxInboundConnections(t *testing.T) {
|
||||||
harness, teardown := setupHarness(t, &harnessParams{
|
harness, teardown := setupHarness(t, &harnessParams{
|
||||||
p2pAddress: p2pAddress1,
|
p2pAddress: p2pAddress1,
|
||||||
|
Loading…
x
Reference in New Issue
Block a user