mirror of
https://gitee.com/milvus-io/milvus.git
synced 2025-12-07 17:48:29 +08:00
2a4c00de9d
issue: #37115 the old implementation update shard cache and shard client manager at same time, which causes lots of conor case due to concurrent issue without lock. This PR decouple shard client manager from shard cache, so only shard cache will be updated if delegator changes. and make sure shard client manager will always return the right client, and create a new client if not exist. in case of client leak, shard client manager will purge client in async for every 10 minutes. --------- Signed-off-by: Wei Liu <wei.liu@zilliz.com>
67 lines
1.7 KiB
Go
67 lines
1.7 KiB
Go
package proxy
|
|
|
|
import (
|
|
"context"
|
|
"testing"
|
|
|
|
"github.com/stretchr/testify/assert"
|
|
|
|
"github.com/milvus-io/milvus/internal/mocks"
|
|
"github.com/milvus-io/milvus/internal/types"
|
|
"github.com/milvus-io/milvus/pkg/util/paramtable"
|
|
)
|
|
|
|
func TestShardClientMgr(t *testing.T) {
|
|
ctx := context.Background()
|
|
nodeInfo := nodeInfo{
|
|
nodeID: 1,
|
|
}
|
|
|
|
qn := mocks.NewMockQueryNodeClient(t)
|
|
qn.EXPECT().Close().Return(nil)
|
|
creator := func(ctx context.Context, addr string, nodeID int64) (types.QueryNodeClient, error) {
|
|
return qn, nil
|
|
}
|
|
|
|
mgr := newShardClientMgr()
|
|
mgr.SetClientCreatorFunc(creator)
|
|
_, err := mgr.GetClient(ctx, nodeInfo)
|
|
assert.Nil(t, err)
|
|
|
|
mgr.ReleaseClientRef(1)
|
|
assert.Equal(t, len(mgr.clients.data), 1)
|
|
mgr.Close()
|
|
assert.Equal(t, len(mgr.clients.data), 0)
|
|
}
|
|
|
|
func TestShardClient(t *testing.T) {
|
|
nodeInfo := nodeInfo{
|
|
nodeID: 1,
|
|
}
|
|
|
|
qn := mocks.NewMockQueryNodeClient(t)
|
|
qn.EXPECT().Close().Return(nil)
|
|
creator := func(ctx context.Context, addr string, nodeID int64) (types.QueryNodeClient, error) {
|
|
return qn, nil
|
|
}
|
|
shardClient, err := newShardClient(nodeInfo, creator)
|
|
assert.Nil(t, err)
|
|
assert.Equal(t, len(shardClient.clients), 0)
|
|
assert.Equal(t, int64(1), shardClient.refCnt.Load())
|
|
assert.Equal(t, false, shardClient.initialized.Load())
|
|
|
|
ctx := context.Background()
|
|
_, err = shardClient.getClient(ctx)
|
|
assert.Nil(t, err)
|
|
assert.Equal(t, len(shardClient.clients), paramtable.Get().ProxyCfg.QueryNodePoolingSize.GetAsInt())
|
|
assert.Equal(t, int64(2), shardClient.refCnt.Load())
|
|
assert.Equal(t, true, shardClient.initialized.Load())
|
|
|
|
shardClient.DecRef()
|
|
assert.Equal(t, int64(1), shardClient.refCnt.Load())
|
|
|
|
shardClient.DecRef()
|
|
assert.Equal(t, int64(0), shardClient.refCnt.Load())
|
|
assert.Equal(t, true, shardClient.isClosed)
|
|
}
|