Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
milvus
milvus
提交
9bbff86c
M
milvus
项目概览
milvus
/
milvus
接近 1 年 前同步成功
通知
261
Star
22476
Fork
2472
代码
文件
提交
分支
Tags
贡献者
分支图
Diff
Issue
0
列表
看板
标记
里程碑
合并请求
0
DevOps
流水线
流水线任务
计划
Wiki
0
Wiki
分析
仓库
DevOps
项目成员
Pages
M
milvus
项目概览
项目概览
详情
发布
仓库
仓库
文件
提交
分支
标签
贡献者
分支图
比较
Issue
0
Issue
0
列表
看板
标记
里程碑
合并请求
0
合并请求
0
Pages
DevOps
DevOps
流水线
流水线任务
计划
分析
分析
仓库分析
DevOps
Wiki
0
Wiki
成员
成员
收起侧边栏
关闭侧边栏
动态
分支图
创建新Issue
流水线任务
提交
Issue看板
体验新版 GitCode,发现更多精彩内容 >>
未验证
提交
9bbff86c
编写于
9月 16, 2021
作者:
D
dragondriver
提交者:
GitHub
9月 16, 2021
浏览文件
操作
浏览文件
下载
电子邮件补丁
差异文件
Fix data race of rootcoord grpc client (#8078)
Signed-off-by:
N
dragondriver
<
jiquan.long@zilliz.com
>
上级
690a3604
变更
1
隐藏空白更改
内联
并排
Showing
1 changed file
with
38 addition
and
25 deletion
+38
-25
internal/distributed/rootcoord/client/client.go
internal/distributed/rootcoord/client/client.go
+38
-25
未找到文件。
internal/distributed/rootcoord/client/client.go
浏览文件 @
9bbff86c
...
...
@@ -15,6 +15,7 @@ import (
"context"
"errors"
"fmt"
"sync"
"time"
grpc_middleware
"github.com/grpc-ecosystem/go-grpc-middleware"
...
...
@@ -41,8 +42,9 @@ type GrpcClient struct {
ctx
context
.
Context
cancel
context
.
CancelFunc
grpcClient
rootcoordpb
.
RootCoordClient
conn
*
grpc
.
ClientConn
grpcClient
rootcoordpb
.
RootCoordClient
conn
*
grpc
.
ClientConn
grpcClientMtx
sync
.
RWMutex
sess
*
sessionutil
.
Session
addr
string
...
...
@@ -136,10 +138,21 @@ func (c *GrpcClient) connect(retryOptions ...retry.Option) error {
return
err
}
log
.
Debug
(
"RootCoordClient try reconnect success"
)
c
.
grpcClientMtx
.
Lock
()
defer
c
.
grpcClientMtx
.
Unlock
()
c
.
grpcClient
=
rootcoordpb
.
NewRootCoordClient
(
c
.
conn
)
return
nil
}
func
(
c
*
GrpcClient
)
getGrpcClient
()
rootcoordpb
.
RootCoordClient
{
c
.
grpcClientMtx
.
RLock
()
defer
c
.
grpcClientMtx
.
RUnlock
()
return
c
.
grpcClient
}
func
(
c
*
GrpcClient
)
Start
()
error
{
return
nil
}
...
...
@@ -174,13 +187,13 @@ func (c *GrpcClient) recall(caller func() (interface{}, error)) (interface{}, er
// GetComponentStates TODO: timeout need to be propagated through ctx
func
(
c
*
GrpcClient
)
GetComponentStates
(
ctx
context
.
Context
)
(
*
internalpb
.
ComponentStates
,
error
)
{
ret
,
err
:=
c
.
recall
(
func
()
(
interface
{},
error
)
{
return
c
.
g
rpcClient
.
GetComponentStates
(
ctx
,
&
internalpb
.
GetComponentStatesRequest
{})
return
c
.
g
etGrpcClient
()
.
GetComponentStates
(
ctx
,
&
internalpb
.
GetComponentStatesRequest
{})
})
return
ret
.
(
*
internalpb
.
ComponentStates
),
err
}
func
(
c
*
GrpcClient
)
GetTimeTickChannel
(
ctx
context
.
Context
)
(
*
milvuspb
.
StringResponse
,
error
)
{
ret
,
err
:=
c
.
recall
(
func
()
(
interface
{},
error
)
{
return
c
.
g
rpcClient
.
GetTimeTickChannel
(
ctx
,
&
internalpb
.
GetTimeTickChannelRequest
{})
return
c
.
g
etGrpcClient
()
.
GetTimeTickChannel
(
ctx
,
&
internalpb
.
GetTimeTickChannelRequest
{})
})
return
ret
.
(
*
milvuspb
.
StringResponse
),
err
}
...
...
@@ -188,7 +201,7 @@ func (c *GrpcClient) GetTimeTickChannel(ctx context.Context) (*milvuspb.StringRe
// GetStatisticsChannel just define a channel, not used currently
func
(
c
*
GrpcClient
)
GetStatisticsChannel
(
ctx
context
.
Context
)
(
*
milvuspb
.
StringResponse
,
error
)
{
ret
,
err
:=
c
.
recall
(
func
()
(
interface
{},
error
)
{
return
c
.
g
rpcClient
.
GetStatisticsChannel
(
ctx
,
&
internalpb
.
GetStatisticsChannelRequest
{})
return
c
.
g
etGrpcClient
()
.
GetStatisticsChannel
(
ctx
,
&
internalpb
.
GetStatisticsChannelRequest
{})
})
return
ret
.
(
*
milvuspb
.
StringResponse
),
err
}
...
...
@@ -196,61 +209,61 @@ func (c *GrpcClient) GetStatisticsChannel(ctx context.Context) (*milvuspb.String
//DDL request
func
(
c
*
GrpcClient
)
CreateCollection
(
ctx
context
.
Context
,
in
*
milvuspb
.
CreateCollectionRequest
)
(
*
commonpb
.
Status
,
error
)
{
ret
,
err
:=
c
.
recall
(
func
()
(
interface
{},
error
)
{
return
c
.
g
rpcClient
.
CreateCollection
(
ctx
,
in
)
return
c
.
g
etGrpcClient
()
.
CreateCollection
(
ctx
,
in
)
})
return
ret
.
(
*
commonpb
.
Status
),
err
}
func
(
c
*
GrpcClient
)
DropCollection
(
ctx
context
.
Context
,
in
*
milvuspb
.
DropCollectionRequest
)
(
*
commonpb
.
Status
,
error
)
{
ret
,
err
:=
c
.
recall
(
func
()
(
interface
{},
error
)
{
return
c
.
g
rpcClient
.
DropCollection
(
ctx
,
in
)
return
c
.
g
etGrpcClient
()
.
DropCollection
(
ctx
,
in
)
})
return
ret
.
(
*
commonpb
.
Status
),
err
}
func
(
c
*
GrpcClient
)
HasCollection
(
ctx
context
.
Context
,
in
*
milvuspb
.
HasCollectionRequest
)
(
*
milvuspb
.
BoolResponse
,
error
)
{
ret
,
err
:=
c
.
recall
(
func
()
(
interface
{},
error
)
{
return
c
.
g
rpcClient
.
HasCollection
(
ctx
,
in
)
return
c
.
g
etGrpcClient
()
.
HasCollection
(
ctx
,
in
)
})
return
ret
.
(
*
milvuspb
.
BoolResponse
),
err
}
func
(
c
*
GrpcClient
)
DescribeCollection
(
ctx
context
.
Context
,
in
*
milvuspb
.
DescribeCollectionRequest
)
(
*
milvuspb
.
DescribeCollectionResponse
,
error
)
{
ret
,
err
:=
c
.
recall
(
func
()
(
interface
{},
error
)
{
return
c
.
g
rpcClient
.
DescribeCollection
(
ctx
,
in
)
return
c
.
g
etGrpcClient
()
.
DescribeCollection
(
ctx
,
in
)
})
return
ret
.
(
*
milvuspb
.
DescribeCollectionResponse
),
err
}
func
(
c
*
GrpcClient
)
ShowCollections
(
ctx
context
.
Context
,
in
*
milvuspb
.
ShowCollectionsRequest
)
(
*
milvuspb
.
ShowCollectionsResponse
,
error
)
{
ret
,
err
:=
c
.
recall
(
func
()
(
interface
{},
error
)
{
return
c
.
g
rpcClient
.
ShowCollections
(
ctx
,
in
)
return
c
.
g
etGrpcClient
()
.
ShowCollections
(
ctx
,
in
)
})
return
ret
.
(
*
milvuspb
.
ShowCollectionsResponse
),
err
}
func
(
c
*
GrpcClient
)
CreatePartition
(
ctx
context
.
Context
,
in
*
milvuspb
.
CreatePartitionRequest
)
(
*
commonpb
.
Status
,
error
)
{
ret
,
err
:=
c
.
recall
(
func
()
(
interface
{},
error
)
{
return
c
.
g
rpcClient
.
CreatePartition
(
ctx
,
in
)
return
c
.
g
etGrpcClient
()
.
CreatePartition
(
ctx
,
in
)
})
return
ret
.
(
*
commonpb
.
Status
),
err
}
func
(
c
*
GrpcClient
)
DropPartition
(
ctx
context
.
Context
,
in
*
milvuspb
.
DropPartitionRequest
)
(
*
commonpb
.
Status
,
error
)
{
ret
,
err
:=
c
.
recall
(
func
()
(
interface
{},
error
)
{
return
c
.
g
rpcClient
.
DropPartition
(
ctx
,
in
)
return
c
.
g
etGrpcClient
()
.
DropPartition
(
ctx
,
in
)
})
return
ret
.
(
*
commonpb
.
Status
),
err
}
func
(
c
*
GrpcClient
)
HasPartition
(
ctx
context
.
Context
,
in
*
milvuspb
.
HasPartitionRequest
)
(
*
milvuspb
.
BoolResponse
,
error
)
{
ret
,
err
:=
c
.
recall
(
func
()
(
interface
{},
error
)
{
return
c
.
g
rpcClient
.
HasPartition
(
ctx
,
in
)
return
c
.
g
etGrpcClient
()
.
HasPartition
(
ctx
,
in
)
})
return
ret
.
(
*
milvuspb
.
BoolResponse
),
err
}
func
(
c
*
GrpcClient
)
ShowPartitions
(
ctx
context
.
Context
,
in
*
milvuspb
.
ShowPartitionsRequest
)
(
*
milvuspb
.
ShowPartitionsResponse
,
error
)
{
ret
,
err
:=
c
.
recall
(
func
()
(
interface
{},
error
)
{
return
c
.
g
rpcClient
.
ShowPartitions
(
ctx
,
in
)
return
c
.
g
etGrpcClient
()
.
ShowPartitions
(
ctx
,
in
)
})
return
ret
.
(
*
milvuspb
.
ShowPartitionsResponse
),
err
}
...
...
@@ -258,21 +271,21 @@ func (c *GrpcClient) ShowPartitions(ctx context.Context, in *milvuspb.ShowPartit
// CreateIndex index builder service
func
(
c
*
GrpcClient
)
CreateIndex
(
ctx
context
.
Context
,
in
*
milvuspb
.
CreateIndexRequest
)
(
*
commonpb
.
Status
,
error
)
{
ret
,
err
:=
c
.
recall
(
func
()
(
interface
{},
error
)
{
return
c
.
g
rpcClient
.
CreateIndex
(
ctx
,
in
)
return
c
.
g
etGrpcClient
()
.
CreateIndex
(
ctx
,
in
)
})
return
ret
.
(
*
commonpb
.
Status
),
err
}
func
(
c
*
GrpcClient
)
DropIndex
(
ctx
context
.
Context
,
in
*
milvuspb
.
DropIndexRequest
)
(
*
commonpb
.
Status
,
error
)
{
ret
,
err
:=
c
.
recall
(
func
()
(
interface
{},
error
)
{
return
c
.
g
rpcClient
.
DropIndex
(
ctx
,
in
)
return
c
.
g
etGrpcClient
()
.
DropIndex
(
ctx
,
in
)
})
return
ret
.
(
*
commonpb
.
Status
),
err
}
func
(
c
*
GrpcClient
)
DescribeIndex
(
ctx
context
.
Context
,
in
*
milvuspb
.
DescribeIndexRequest
)
(
*
milvuspb
.
DescribeIndexResponse
,
error
)
{
ret
,
err
:=
c
.
recall
(
func
()
(
interface
{},
error
)
{
return
c
.
g
rpcClient
.
DescribeIndex
(
ctx
,
in
)
return
c
.
g
etGrpcClient
()
.
DescribeIndex
(
ctx
,
in
)
})
return
ret
.
(
*
milvuspb
.
DescribeIndexResponse
),
err
}
...
...
@@ -280,14 +293,14 @@ func (c *GrpcClient) DescribeIndex(ctx context.Context, in *milvuspb.DescribeInd
// AllocTimestamp global timestamp allocator
func
(
c
*
GrpcClient
)
AllocTimestamp
(
ctx
context
.
Context
,
in
*
rootcoordpb
.
AllocTimestampRequest
)
(
*
rootcoordpb
.
AllocTimestampResponse
,
error
)
{
ret
,
err
:=
c
.
recall
(
func
()
(
interface
{},
error
)
{
return
c
.
g
rpcClient
.
AllocTimestamp
(
ctx
,
in
)
return
c
.
g
etGrpcClient
()
.
AllocTimestamp
(
ctx
,
in
)
})
return
ret
.
(
*
rootcoordpb
.
AllocTimestampResponse
),
err
}
func
(
c
*
GrpcClient
)
AllocID
(
ctx
context
.
Context
,
in
*
rootcoordpb
.
AllocIDRequest
)
(
*
rootcoordpb
.
AllocIDResponse
,
error
)
{
ret
,
err
:=
c
.
recall
(
func
()
(
interface
{},
error
)
{
return
c
.
g
rpcClient
.
AllocID
(
ctx
,
in
)
return
c
.
g
etGrpcClient
()
.
AllocID
(
ctx
,
in
)
})
return
ret
.
(
*
rootcoordpb
.
AllocIDResponse
),
err
}
...
...
@@ -295,7 +308,7 @@ func (c *GrpcClient) AllocID(ctx context.Context, in *rootcoordpb.AllocIDRequest
// UpdateChannelTimeTick used to handle ChannelTimeTickMsg
func
(
c
*
GrpcClient
)
UpdateChannelTimeTick
(
ctx
context
.
Context
,
in
*
internalpb
.
ChannelTimeTickMsg
)
(
*
commonpb
.
Status
,
error
)
{
ret
,
err
:=
c
.
recall
(
func
()
(
interface
{},
error
)
{
return
c
.
g
rpcClient
.
UpdateChannelTimeTick
(
ctx
,
in
)
return
c
.
g
etGrpcClient
()
.
UpdateChannelTimeTick
(
ctx
,
in
)
})
return
ret
.
(
*
commonpb
.
Status
),
err
}
...
...
@@ -303,33 +316,33 @@ func (c *GrpcClient) UpdateChannelTimeTick(ctx context.Context, in *internalpb.C
// DescribeSegment receiver time tick from proxy service, and put it into this channel
func
(
c
*
GrpcClient
)
DescribeSegment
(
ctx
context
.
Context
,
in
*
milvuspb
.
DescribeSegmentRequest
)
(
*
milvuspb
.
DescribeSegmentResponse
,
error
)
{
ret
,
err
:=
c
.
recall
(
func
()
(
interface
{},
error
)
{
return
c
.
g
rpcClient
.
DescribeSegment
(
ctx
,
in
)
return
c
.
g
etGrpcClient
()
.
DescribeSegment
(
ctx
,
in
)
})
return
ret
.
(
*
milvuspb
.
DescribeSegmentResponse
),
err
}
func
(
c
*
GrpcClient
)
ShowSegments
(
ctx
context
.
Context
,
in
*
milvuspb
.
ShowSegmentsRequest
)
(
*
milvuspb
.
ShowSegmentsResponse
,
error
)
{
ret
,
err
:=
c
.
recall
(
func
()
(
interface
{},
error
)
{
return
c
.
g
rpcClient
.
ShowSegments
(
ctx
,
in
)
return
c
.
g
etGrpcClient
()
.
ShowSegments
(
ctx
,
in
)
})
return
ret
.
(
*
milvuspb
.
ShowSegmentsResponse
),
err
}
func
(
c
*
GrpcClient
)
ReleaseDQLMessageStream
(
ctx
context
.
Context
,
in
*
proxypb
.
ReleaseDQLMessageStreamRequest
)
(
*
commonpb
.
Status
,
error
)
{
ret
,
err
:=
c
.
recall
(
func
()
(
interface
{},
error
)
{
return
c
.
g
rpcClient
.
ReleaseDQLMessageStream
(
ctx
,
in
)
return
c
.
g
etGrpcClient
()
.
ReleaseDQLMessageStream
(
ctx
,
in
)
})
return
ret
.
(
*
commonpb
.
Status
),
err
}
func
(
c
*
GrpcClient
)
SegmentFlushCompleted
(
ctx
context
.
Context
,
in
*
datapb
.
SegmentFlushCompletedMsg
)
(
*
commonpb
.
Status
,
error
)
{
ret
,
err
:=
c
.
recall
(
func
()
(
interface
{},
error
)
{
return
c
.
g
rpcClient
.
SegmentFlushCompleted
(
ctx
,
in
)
return
c
.
g
etGrpcClient
()
.
SegmentFlushCompleted
(
ctx
,
in
)
})
return
ret
.
(
*
commonpb
.
Status
),
err
}
func
(
c
*
GrpcClient
)
GetMetrics
(
ctx
context
.
Context
,
in
*
milvuspb
.
GetMetricsRequest
)
(
*
milvuspb
.
GetMetricsResponse
,
error
)
{
ret
,
err
:=
c
.
recall
(
func
()
(
interface
{},
error
)
{
return
c
.
g
rpcClient
.
GetMetrics
(
ctx
,
in
)
return
c
.
g
etGrpcClient
()
.
GetMetrics
(
ctx
,
in
)
})
return
ret
.
(
*
milvuspb
.
GetMetricsResponse
),
err
}
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录