Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

[Access] Refactor access connection cache #5334

Merged
merged 7 commits into from
Feb 20, 2024
Merged

[Access] Refactor access connection cache #5334

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
6794c1d
[Access] Refactor access connection cache
peterargue Feb 1, 2024
0146350
switch to rw lock
peterargue Feb 1, 2024
69b0d0b
Add rapid connection disconnection test
peterargue Feb 1, 2024
eb51f0f
Update existing rapid tests to new version
peterargue Feb 1, 2024
5b1060c
fix rapid test concurrent Draw
peterargue Feb 15, 2024
63215f0
Merge branch 'master' into petera/refactor-access-connection-cache
franklywatson Feb 17, 2024
cf4b8fd
Merge branch 'master' into petera/refactor-access-connection-cache
peterargue Feb 20, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 4 additions & 4 deletions cmd/access/node_builder/access_node_builder.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1505,12 +1505,12 @@ func (builder *FlowAccessNodeBuilder) Build() (cmd.Node, error) {
cacheSize := int(backendConfig.ConnectionPoolSize)

var connBackendCache *rpcConnection.Cache
var err error
if cacheSize > 0 {
backendCache, err := backend.NewCache(node.Logger, accessMetrics, cacheSize)
connBackendCache, err = rpcConnection.NewCache(node.Logger, accessMetrics, cacheSize)
if err != nil {
return nil, fmt.Errorf("could not initialize backend cache: %w", err)
return nil, fmt.Errorf("could not initialize connection cache: %w", err)
}
connBackendCache = rpcConnection.NewCache(backendCache, cacheSize)
}

connFactory := &rpcConnection.ConnectionFactoryImpl{
Expand All @@ -1521,9 +1521,9 @@ func (builder *FlowAccessNodeBuilder) Build() (cmd.Node, error) {
AccessMetrics: accessMetrics,
Log: node.Logger,
Manager: rpcConnection.NewManager(
connBackendCache,
node.Logger,
accessMetrics,
connBackendCache,
config.MaxMsgSize,
backendConfig.CircuitBreakerConfig,
config.CompressorName,
Expand Down
8 changes: 4 additions & 4 deletions cmd/observer/node_builder/observer_builder.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1210,12 +1210,12 @@ func (builder *ObserverServiceBuilder) enqueueRPCServer() {
cacheSize := int(backendConfig.ConnectionPoolSize)

var connBackendCache *rpcConnection.Cache
var err error
if cacheSize > 0 {
backendCache, err := backend.NewCache(node.Logger, accessMetrics, cacheSize)
connBackendCache, err = rpcConnection.NewCache(node.Logger, accessMetrics, cacheSize)
if err != nil {
return nil, fmt.Errorf("could not initialize backend cache: %w", err)
return nil, fmt.Errorf("could not initialize connection cache: %w", err)
}
connBackendCache = rpcConnection.NewCache(backendCache, cacheSize)
}

connFactory := &rpcConnection.ConnectionFactoryImpl{
Expand All @@ -1226,9 +1226,9 @@ func (builder *ObserverServiceBuilder) enqueueRPCServer() {
AccessMetrics: accessMetrics,
Log: node.Logger,
Manager: rpcConnection.NewManager(
connBackendCache,
node.Logger,
accessMetrics,
connBackendCache,
config.MaxMsgSize,
backendConfig.CircuitBreakerConfig,
config.CompressorName,
Expand Down
14 changes: 7 additions & 7 deletions consensus/hotstuff/votecollector/combined_vote_processor_v2_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -118,7 +118,7 @@ func (s *CombinedVoteProcessorV2TestSuite) TestProcess_InvalidSignatureFormat()
rapid.Check(s.T(), func(t *rapid.T) {
// create a signature with invalid length
vote := unittest.VoteForBlockFixture(s.proposal.Block, func(vote *model.Vote) {
vote.SigData = unittest.RandomBytes(generator.Draw(t, "sig-size").(int))
vote.SigData = unittest.RandomBytes(generator.Draw(t, "sig-size"))
})
err := s.processor.Process(vote)
require.Error(s.T(), err)
Expand Down Expand Up @@ -434,8 +434,8 @@ func TestCombinedVoteProcessorV2_PropertyCreatingQCCorrectness(testifyT *testing

rapid.Check(testifyT, func(t *rapid.T) {
// draw participants in range 1 <= participants <= maxParticipants
participants := rapid.Uint64Range(1, maxParticipants).Draw(t, "participants").(uint64)
beaconSignersCount := rapid.Uint64Range(participants/2+1, participants).Draw(t, "beaconSigners").(uint64)
participants := rapid.Uint64Range(1, maxParticipants).Draw(t, "participants")
beaconSignersCount := rapid.Uint64Range(participants/2+1, participants).Draw(t, "beaconSigners")
stakingSignersCount := participants - beaconSignersCount
require.Equal(t, participants, stakingSignersCount+beaconSignersCount)

Expand Down Expand Up @@ -638,20 +638,20 @@ func TestCombinedVoteProcessorV2_PropertyCreatingQCCorrectness(testifyT *testing
func TestCombinedVoteProcessorV2_PropertyCreatingQCLiveness(testifyT *testing.T) {
rapid.Check(testifyT, func(t *rapid.T) {
// draw beacon signers in range 1 <= beaconSignersCount <= 53
beaconSignersCount := rapid.Uint64Range(1, 53).Draw(t, "beaconSigners").(uint64)
beaconSignersCount := rapid.Uint64Range(1, 53).Draw(t, "beaconSigners")
// draw staking signers in range 0 <= stakingSignersCount <= 10
stakingSignersCount := rapid.Uint64Range(0, 10).Draw(t, "stakingSigners").(uint64)
stakingSignersCount := rapid.Uint64Range(0, 10).Draw(t, "stakingSigners")

stakingWeightRange, beaconWeightRange := rapid.Uint64Range(1, 10), rapid.Uint64Range(1, 10)

minRequiredWeight := uint64(0)
// draw weight for each signer randomly
stakingSigners := unittest.IdentityListFixture(int(stakingSignersCount), func(identity *flow.Identity) {
identity.InitialWeight = stakingWeightRange.Draw(t, identity.String()).(uint64)
identity.InitialWeight = stakingWeightRange.Draw(t, identity.String())
minRequiredWeight += identity.InitialWeight
})
beaconSigners := unittest.IdentityListFixture(int(beaconSignersCount), func(identity *flow.Identity) {
identity.InitialWeight = beaconWeightRange.Draw(t, identity.String()).(uint64)
identity.InitialWeight = beaconWeightRange.Draw(t, identity.String())
minRequiredWeight += identity.InitialWeight
})

Expand Down
12 changes: 6 additions & 6 deletions consensus/hotstuff/votecollector/combined_vote_processor_v3_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -434,8 +434,8 @@ func TestCombinedVoteProcessorV3_PropertyCreatingQCCorrectness(testifyT *testing

rapid.Check(testifyT, func(t *rapid.T) {
// draw participants in range 1 <= participants <= maxParticipants
participants := rapid.Uint64Range(1, maxParticipants).Draw(t, "participants").(uint64)
beaconSignersCount := rapid.Uint64Range(participants/2+1, participants).Draw(t, "beaconSigners").(uint64)
participants := rapid.Uint64Range(1, maxParticipants).Draw(t, "participants")
beaconSignersCount := rapid.Uint64Range(participants/2+1, participants).Draw(t, "beaconSigners")
stakingSignersCount := participants - beaconSignersCount
require.Equal(t, participants, stakingSignersCount+beaconSignersCount)

Expand Down Expand Up @@ -749,20 +749,20 @@ func TestCombinedVoteProcessorV3_OnlyRandomBeaconSigners(testifyT *testing.T) {
func TestCombinedVoteProcessorV3_PropertyCreatingQCLiveness(testifyT *testing.T) {
rapid.Check(testifyT, func(t *rapid.T) {
// draw beacon signers in range 1 <= beaconSignersCount <= 53
beaconSignersCount := rapid.Uint64Range(1, 53).Draw(t, "beaconSigners").(uint64)
beaconSignersCount := rapid.Uint64Range(1, 53).Draw(t, "beaconSigners")
// draw staking signers in range 0 <= stakingSignersCount <= 10
stakingSignersCount := rapid.Uint64Range(0, 10).Draw(t, "stakingSigners").(uint64)
stakingSignersCount := rapid.Uint64Range(0, 10).Draw(t, "stakingSigners")

stakingWeightRange, beaconWeightRange := rapid.Uint64Range(1, 10), rapid.Uint64Range(1, 10)

minRequiredWeight := uint64(0)
// draw weight for each signer randomly
stakingSigners := unittest.IdentityListFixture(int(stakingSignersCount), func(identity *flow.Identity) {
identity.InitialWeight = stakingWeightRange.Draw(t, identity.String()).(uint64)
identity.InitialWeight = stakingWeightRange.Draw(t, identity.String())
minRequiredWeight += identity.InitialWeight
})
beaconSigners := unittest.IdentityListFixture(int(beaconSignersCount), func(identity *flow.Identity) {
identity.InitialWeight = beaconWeightRange.Draw(t, identity.String()).(uint64)
identity.InitialWeight = beaconWeightRange.Draw(t, identity.String())
minRequiredWeight += identity.InitialWeight
})

Expand Down
2 changes: 1 addition & 1 deletion engine/access/apiproxy/access_api_proxy_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -152,9 +152,9 @@ func TestNewFlowCachedAccessAPIProxy(t *testing.T) {
AccessMetrics: metrics,
CollectionNodeGRPCTimeout: time.Second,
Manager: connection.NewManager(
nil,
unittest.Logger(),
metrics,
nil,
grpcutils.DefaultMaxMsgSize,
connection.CircuitBreakerConfig{},
grpcutils.NoCompressor,
Expand Down
21 changes: 0 additions & 21 deletions engine/access/rpc/backend/backend.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -236,27 +236,6 @@ func New(params Params) (*Backend, error) {
return b, nil
}

// NewCache constructs cache for storing connections to other nodes.
// No errors are expected during normal operations.
func NewCache(
log zerolog.Logger,
metrics module.AccessMetrics,
connectionPoolSize int,
) (*lru.Cache[string, *connection.CachedClient], error) {
cache, err := lru.NewWithEvict(connectionPoolSize, func(_ string, client *connection.CachedClient) {
go client.Close() // close is blocking, so run in a goroutine

log.Debug().Str("grpc_conn_evicted", client.Address).Msg("closing grpc connection evicted from pool")
metrics.ConnectionFromPoolEvicted()
})

if err != nil {
return nil, fmt.Errorf("could not initialize connection pool cache: %w", err)
}

return cache, nil
}

func identifierList(ids []string) (flow.IdentifierList, error) {
idList := make(flow.IdentifierList, len(ids))
for i, idStr := range ids {
Expand Down
Loading
Loading