Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
milvus
milvus
提交
c8f89907
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,发现更多精彩内容 >>
未验证
提交
c8f89907
编写于
1月 17, 2023
作者:
Y
yah01
提交者:
GitHub
1月 17, 2023
浏览文件
操作
浏览文件
下载
电子邮件补丁
差异文件
Fix current target may be updated to an invalid target (#21742)
Signed-off-by:
N
yah01
<
yang.cen@zilliz.com
>
上级
81326ca0
变更
5
隐藏空白更改
内联
并排
Showing
5 changed file
with
80 addition
and
31 deletion
+80
-31
internal/querycoordv2/observers/collection_observer.go
internal/querycoordv2/observers/collection_observer.go
+5
-4
internal/querycoordv2/observers/collection_observer_test.go
internal/querycoordv2/observers/collection_observer_test.go
+16
-3
internal/querycoordv2/observers/target_observer.go
internal/querycoordv2/observers/target_observer.go
+40
-19
internal/querycoordv2/server.go
internal/querycoordv2/server.go
+6
-5
internal/querycoordv2/server_test.go
internal/querycoordv2/server_test.go
+13
-0
未找到文件。
internal/querycoordv2/observers/collection_observer.go
浏览文件 @
c8f89907
...
...
@@ -37,6 +37,7 @@ type CollectionObserver struct {
dist
*
meta
.
DistributionManager
meta
*
meta
.
Meta
targetMgr
*
meta
.
TargetManager
targetObserver
*
TargetObserver
collectionLoadedCount
map
[
int64
]
int
partitionLoadedCount
map
[
int64
]
int
...
...
@@ -47,12 +48,14 @@ func NewCollectionObserver(
dist
*
meta
.
DistributionManager
,
meta
*
meta
.
Meta
,
targetMgr
*
meta
.
TargetManager
,
targetObserver
*
TargetObserver
,
)
*
CollectionObserver
{
return
&
CollectionObserver
{
stopCh
:
make
(
chan
struct
{}),
dist
:
dist
,
meta
:
meta
,
targetMgr
:
targetMgr
,
targetObserver
:
targetObserver
,
collectionLoadedCount
:
make
(
map
[
int64
]
int
),
partitionLoadedCount
:
make
(
map
[
int64
]
int
),
}
...
...
@@ -201,9 +204,8 @@ func (ob *CollectionObserver) observeCollectionLoadStatus(collection *meta.Colle
return
}
ob
.
collectionLoadedCount
[
collection
.
GetCollectionID
()]
=
loadedCount
if
updated
.
LoadPercentage
==
100
{
if
updated
.
LoadPercentage
==
100
&&
ob
.
targetObserver
.
Check
(
updated
.
GetCollectionID
())
{
delete
(
ob
.
collectionLoadedCount
,
collection
.
GetCollectionID
())
ob
.
targetMgr
.
UpdateCollectionCurrentTarget
(
updated
.
CollectionID
)
updated
.
Status
=
querypb
.
LoadStatus_Loaded
ob
.
meta
.
CollectionManager
.
UpdateCollection
(
updated
)
...
...
@@ -265,9 +267,8 @@ func (ob *CollectionObserver) observePartitionLoadStatus(partition *meta.Partiti
return
}
ob
.
partitionLoadedCount
[
partition
.
GetPartitionID
()]
=
loadedCount
if
updated
.
LoadPercentage
==
100
{
if
updated
.
LoadPercentage
==
100
&&
ob
.
targetObserver
.
Check
(
updated
.
GetCollectionID
())
{
delete
(
ob
.
partitionLoadedCount
,
partition
.
GetPartitionID
())
ob
.
targetMgr
.
UpdateCollectionCurrentTarget
(
partition
.
GetCollectionID
(),
partition
.
GetPartitionID
())
updated
.
Status
=
querypb
.
LoadStatus_Loaded
ob
.
meta
.
CollectionManager
.
PutPartition
(
updated
)
...
...
internal/querycoordv2/observers/collection_observer_test.go
浏览文件 @
c8f89907
...
...
@@ -56,9 +56,10 @@ type CollectionObserverSuite struct {
broker
*
meta
.
MockBroker
// Dependencies
dist
*
meta
.
DistributionManager
meta
*
meta
.
Meta
targetMgr
*
meta
.
TargetManager
dist
*
meta
.
DistributionManager
meta
*
meta
.
Meta
targetMgr
*
meta
.
TargetManager
targetObserver
*
TargetObserver
// Test object
ob
*
CollectionObserver
...
...
@@ -180,18 +181,30 @@ func (suite *CollectionObserverSuite) SetupTest() {
suite
.
meta
=
meta
.
NewMeta
(
suite
.
idAllocator
,
suite
.
store
)
suite
.
broker
=
meta
.
NewMockBroker
(
suite
.
T
())
suite
.
targetMgr
=
meta
.
NewTargetManager
(
suite
.
broker
,
suite
.
meta
)
suite
.
targetObserver
=
NewTargetObserver
(
suite
.
meta
,
suite
.
targetMgr
,
suite
.
dist
,
suite
.
broker
,
)
// Test object
suite
.
ob
=
NewCollectionObserver
(
suite
.
dist
,
suite
.
meta
,
suite
.
targetMgr
,
suite
.
targetObserver
,
)
for
_
,
collection
:=
range
suite
.
collections
{
suite
.
broker
.
EXPECT
()
.
GetPartitions
(
mock
.
Anything
,
collection
)
.
Return
(
suite
.
partitions
[
collection
],
nil
)
.
Maybe
()
}
suite
.
targetObserver
.
Start
(
context
.
Background
())
suite
.
loadAll
()
}
func
(
suite
*
CollectionObserverSuite
)
TearDownTest
()
{
suite
.
targetObserver
.
Stop
()
suite
.
ob
.
Stop
()
suite
.
kv
.
Close
()
}
...
...
internal/querycoordv2/observers/target_observer.go
浏览文件 @
c8f89907
...
...
@@ -30,6 +30,11 @@ import (
"github.com/milvus-io/milvus/internal/util/typeutil"
)
type
checkRequest
struct
{
CollectionID
int64
Notifier
chan
bool
}
type
targetUpdateRequest
struct
{
CollectionID
int64
Notifier
chan
error
...
...
@@ -44,6 +49,7 @@ type TargetObserver struct {
distMgr
*
meta
.
DistributionManager
broker
meta
.
Broker
manualCheck
chan
checkRequest
nextTargetLastUpdate
map
[
int64
]
time
.
Time
updateChan
chan
targetUpdateRequest
mut
sync
.
Mutex
// Guard readyNotifiers
...
...
@@ -59,6 +65,7 @@ func NewTargetObserver(meta *meta.Meta, targetMgr *meta.TargetManager, distMgr *
targetMgr
:
targetMgr
,
distMgr
:
distMgr
,
broker
:
broker
,
manualCheck
:
make
(
chan
checkRequest
,
10
),
nextTargetLastUpdate
:
make
(
map
[
int64
]
time
.
Time
),
updateChan
:
make
(
chan
targetUpdateRequest
),
readyNotifiers
:
make
(
map
[
int64
][]
chan
struct
{}),
...
...
@@ -95,21 +102,48 @@ func (ob *TargetObserver) schedule(ctx context.Context) {
ob
.
clean
()
ob
.
tryUpdateTarget
()
case
request
:=
<-
ob
.
updateChan
:
err
:=
ob
.
updateNextTarget
(
request
.
CollectionID
)
case
req
:=
<-
ob
.
manualCheck
:
ob
.
check
(
req
.
CollectionID
)
req
.
Notifier
<-
ob
.
targetMgr
.
IsCurrentTargetExist
(
req
.
CollectionID
)
case
req
:=
<-
ob
.
updateChan
:
err
:=
ob
.
updateNextTarget
(
req
.
CollectionID
)
if
err
!=
nil
{
close
(
req
uest
.
ReadyNotifier
)
close
(
req
.
ReadyNotifier
)
}
else
{
ob
.
mut
.
Lock
()
ob
.
readyNotifiers
[
req
uest
.
CollectionID
]
=
append
(
ob
.
readyNotifiers
[
request
.
CollectionID
],
request
.
ReadyNotifier
)
ob
.
readyNotifiers
[
req
.
CollectionID
]
=
append
(
ob
.
readyNotifiers
[
req
.
CollectionID
],
req
.
ReadyNotifier
)
ob
.
mut
.
Unlock
()
}
req
uest
.
Notifier
<-
err
req
.
Notifier
<-
err
}
}
}
// Check checks whether the next target is ready,
// and updates the current target if it is,
// returns true if current target is not nil
func
(
ob
*
TargetObserver
)
Check
(
collectionID
int64
)
bool
{
notifier
:=
make
(
chan
bool
)
ob
.
manualCheck
<-
checkRequest
{
CollectionID
:
collectionID
,
Notifier
:
notifier
,
}
return
<-
notifier
}
func
(
ob
*
TargetObserver
)
check
(
collectionID
int64
)
{
if
ob
.
shouldUpdateCurrentTarget
(
collectionID
)
{
ob
.
updateCurrentTarget
(
collectionID
)
}
if
ob
.
shouldUpdateNextTarget
(
collectionID
)
{
// update next target in collection level
ob
.
updateNextTarget
(
collectionID
)
}
}
// UpdateNextTarget updates the next target,
// returns a channel which will be closed when the next target is ready,
// or returns error if failed to pull target
...
...
@@ -138,14 +172,7 @@ func (ob *TargetObserver) ReleaseCollection(collectionID int64) {
func
(
ob
*
TargetObserver
)
tryUpdateTarget
()
{
collections
:=
ob
.
meta
.
GetAll
()
for
_
,
collectionID
:=
range
collections
{
if
ob
.
shouldUpdateCurrentTarget
(
collectionID
)
{
ob
.
updateCurrentTarget
(
collectionID
)
}
if
ob
.
shouldUpdateNextTarget
(
collectionID
)
{
// update next target in collection level
ob
.
updateNextTarget
(
collectionID
)
}
ob
.
check
(
collectionID
)
}
collectionSet
:=
typeutil
.
NewUniqueSet
(
collections
...
)
...
...
@@ -199,12 +226,6 @@ func (ob *TargetObserver) updateNextTargetTimestamp(collectionID int64) {
}
func
(
ob
*
TargetObserver
)
shouldUpdateCurrentTarget
(
collectionID
int64
)
bool
{
// Collection observer will update the current target as loading done,
// avoid double updating, which will cause update current target to a unfinished next target
if
!
ob
.
targetMgr
.
IsCurrentTargetExist
(
collectionID
)
{
return
false
}
replicaNum
:=
ob
.
meta
.
CollectionManager
.
GetReplicaNumber
(
collectionID
)
// check channel first
...
...
internal/querycoordv2/server.go
浏览文件 @
c8f89907
...
...
@@ -279,11 +279,6 @@ func (s *Server) initMeta() error {
func
(
s
*
Server
)
initObserver
()
{
log
.
Info
(
"init observers"
)
s
.
collectionObserver
=
observers
.
NewCollectionObserver
(
s
.
dist
,
s
.
meta
,
s
.
targetMgr
,
)
s
.
leaderObserver
=
observers
.
NewLeaderObserver
(
s
.
dist
,
s
.
meta
,
...
...
@@ -296,6 +291,12 @@ func (s *Server) initObserver() {
s
.
dist
,
s
.
broker
,
)
s
.
collectionObserver
=
observers
.
NewCollectionObserver
(
s
.
dist
,
s
.
meta
,
s
.
targetMgr
,
s
.
targetObserver
,
)
}
func
(
s
*
Server
)
afterStart
()
{
...
...
internal/querycoordv2/server_test.go
浏览文件 @
c8f89907
...
...
@@ -35,6 +35,7 @@ import (
"github.com/milvus-io/milvus/internal/querycoordv2/dist"
"github.com/milvus-io/milvus/internal/querycoordv2/meta"
"github.com/milvus-io/milvus/internal/querycoordv2/mocks"
"github.com/milvus-io/milvus/internal/querycoordv2/observers"
"github.com/milvus-io/milvus/internal/querycoordv2/params"
"github.com/milvus-io/milvus/internal/querycoordv2/session"
"github.com/milvus-io/milvus/internal/querycoordv2/task"
...
...
@@ -401,6 +402,18 @@ func (suite *ServerSuite) hackServer() {
suite
.
server
.
balancer
,
suite
.
server
.
taskScheduler
,
)
suite
.
server
.
targetObserver
=
observers
.
NewTargetObserver
(
suite
.
server
.
meta
,
suite
.
server
.
targetMgr
,
suite
.
server
.
dist
,
suite
.
broker
,
)
suite
.
server
.
collectionObserver
=
observers
.
NewCollectionObserver
(
suite
.
server
.
dist
,
suite
.
server
.
meta
,
suite
.
server
.
targetMgr
,
suite
.
server
.
targetObserver
,
)
suite
.
broker
.
EXPECT
()
.
GetCollectionSchema
(
mock
.
Anything
,
mock
.
Anything
)
.
Return
(
&
schemapb
.
CollectionSchema
{},
nil
)
.
Maybe
()
for
_
,
collection
:=
range
suite
.
collections
{
...
...
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录