2022-10-11 11:39:22 +08:00
|
|
|
// Licensed to the LF AI & Data foundation under one
|
|
|
|
// or more contributor license agreements. See the NOTICE file
|
|
|
|
// distributed with this work for additional information
|
|
|
|
// regarding copyright ownership. The ASF licenses this file
|
|
|
|
// to you under the Apache License, Version 2.0 (the
|
|
|
|
// "License"); you may not use this file except in compliance
|
|
|
|
// with the License. You may obtain a copy of the License at
|
|
|
|
//
|
|
|
|
// http://www.apache.org/licenses/LICENSE-2.0
|
|
|
|
//
|
|
|
|
// Unless required by applicable law or agreed to in writing, software
|
|
|
|
// distributed under the License is distributed on an "AS IS" BASIS,
|
|
|
|
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
|
|
// See the License for the specific language governing permissions and
|
|
|
|
// limitations under the License.
|
|
|
|
|
2022-09-15 18:48:32 +08:00
|
|
|
package meta
|
|
|
|
|
|
|
|
import (
|
|
|
|
"sync"
|
|
|
|
|
2023-01-18 16:41:44 +08:00
|
|
|
"github.com/samber/lo"
|
2023-04-06 19:14:32 +08:00
|
|
|
|
|
|
|
"github.com/milvus-io/milvus/internal/proto/querypb"
|
2022-09-15 18:48:32 +08:00
|
|
|
)
|
|
|
|
|
|
|
|
type LeaderView struct {
|
|
|
|
ID int64
|
|
|
|
CollectionID int64
|
|
|
|
Channel string
|
2023-03-21 16:57:57 +08:00
|
|
|
Version int64
|
2022-09-28 12:10:54 +08:00
|
|
|
Segments map[int64]*querypb.SegmentDist
|
2022-11-07 19:37:04 +08:00
|
|
|
GrowingSegments map[int64]*Segment
|
2023-06-27 11:48:45 +08:00
|
|
|
TargetVersion int64
|
2022-09-15 18:48:32 +08:00
|
|
|
}
|
|
|
|
|
|
|
|
func (view *LeaderView) Clone() *LeaderView {
|
2022-09-28 12:10:54 +08:00
|
|
|
segments := make(map[int64]*querypb.SegmentDist)
|
2022-09-15 18:48:32 +08:00
|
|
|
for k, v := range view.Segments {
|
|
|
|
segments[k] = v
|
|
|
|
}
|
2022-11-07 19:37:04 +08:00
|
|
|
|
|
|
|
growings := make(map[int64]*Segment)
|
|
|
|
for k, v := range view.GrowingSegments {
|
|
|
|
growings[k] = v
|
|
|
|
}
|
2022-09-15 18:48:32 +08:00
|
|
|
|
|
|
|
return &LeaderView{
|
|
|
|
ID: view.ID,
|
|
|
|
CollectionID: view.CollectionID,
|
|
|
|
Channel: view.Channel,
|
2023-03-21 16:57:57 +08:00
|
|
|
Version: view.Version,
|
2022-09-15 18:48:32 +08:00
|
|
|
Segments: segments,
|
|
|
|
GrowingSegments: growings,
|
2023-06-27 11:48:45 +08:00
|
|
|
TargetVersion: view.TargetVersion,
|
2022-09-15 18:48:32 +08:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
type channelViews map[string]*LeaderView
|
|
|
|
|
|
|
|
type LeaderViewManager struct {
|
|
|
|
rwmutex sync.RWMutex
|
|
|
|
views map[int64]channelViews // LeaderID -> Views (one per shard)
|
|
|
|
}
|
|
|
|
|
|
|
|
func NewLeaderViewManager() *LeaderViewManager {
|
|
|
|
return &LeaderViewManager{
|
|
|
|
views: make(map[int64]channelViews),
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
// GetSegmentByNode returns all segments that the given node contains,
|
|
|
|
// include growing segments
|
|
|
|
func (mgr *LeaderViewManager) GetSegmentByNode(nodeID int64) []int64 {
|
|
|
|
mgr.rwmutex.RLock()
|
|
|
|
defer mgr.rwmutex.RUnlock()
|
|
|
|
|
|
|
|
segments := make([]int64, 0)
|
|
|
|
for leaderID, views := range mgr.views {
|
|
|
|
for _, view := range views {
|
2022-09-28 12:10:54 +08:00
|
|
|
for segment, version := range view.Segments {
|
|
|
|
if version.NodeID == nodeID {
|
2022-09-15 18:48:32 +08:00
|
|
|
segments = append(segments, segment)
|
|
|
|
}
|
|
|
|
}
|
|
|
|
if leaderID == nodeID {
|
2022-11-07 19:37:04 +08:00
|
|
|
segments = append(segments, lo.Keys(view.GrowingSegments)...)
|
2022-09-15 18:48:32 +08:00
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
return segments
|
|
|
|
}
|
|
|
|
|
|
|
|
// Update updates the leader's views, all views have to be with the same leader ID
|
|
|
|
func (mgr *LeaderViewManager) Update(leaderID int64, views ...*LeaderView) {
|
|
|
|
mgr.rwmutex.Lock()
|
|
|
|
defer mgr.rwmutex.Unlock()
|
|
|
|
mgr.views[leaderID] = make(channelViews, len(views))
|
|
|
|
for _, view := range views {
|
|
|
|
mgr.views[leaderID][view.Channel] = view
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
// GetSegmentDist returns the list of nodes the given segment on
|
|
|
|
func (mgr *LeaderViewManager) GetSegmentDist(segmentID int64) []int64 {
|
|
|
|
mgr.rwmutex.RLock()
|
|
|
|
defer mgr.rwmutex.RUnlock()
|
|
|
|
|
|
|
|
nodes := make([]int64, 0)
|
|
|
|
for leaderID, views := range mgr.views {
|
|
|
|
for _, view := range views {
|
2022-09-28 12:10:54 +08:00
|
|
|
version, ok := view.Segments[segmentID]
|
2022-09-15 18:48:32 +08:00
|
|
|
if ok {
|
2022-09-28 12:10:54 +08:00
|
|
|
nodes = append(nodes, version.NodeID)
|
2022-09-15 18:48:32 +08:00
|
|
|
}
|
2022-11-07 19:37:04 +08:00
|
|
|
if _, ok := view.GrowingSegments[segmentID]; ok {
|
2022-09-15 18:48:32 +08:00
|
|
|
nodes = append(nodes, leaderID)
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
return nodes
|
|
|
|
}
|
|
|
|
|
|
|
|
func (mgr *LeaderViewManager) GetSealedSegmentDist(segmentID int64) []int64 {
|
|
|
|
mgr.rwmutex.RLock()
|
|
|
|
defer mgr.rwmutex.RUnlock()
|
|
|
|
|
|
|
|
nodes := make([]int64, 0)
|
|
|
|
for _, views := range mgr.views {
|
|
|
|
for _, view := range views {
|
2022-09-28 12:10:54 +08:00
|
|
|
version, ok := view.Segments[segmentID]
|
2022-09-15 18:48:32 +08:00
|
|
|
if ok {
|
2022-09-28 12:10:54 +08:00
|
|
|
nodes = append(nodes, version.NodeID)
|
2022-09-15 18:48:32 +08:00
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
return nodes
|
|
|
|
}
|
|
|
|
|
|
|
|
func (mgr *LeaderViewManager) GetGrowingSegmentDist(segmentID int64) []int64 {
|
|
|
|
mgr.rwmutex.RLock()
|
|
|
|
defer mgr.rwmutex.RUnlock()
|
|
|
|
|
|
|
|
nodes := make([]int64, 0)
|
|
|
|
for leaderID, views := range mgr.views {
|
|
|
|
for _, view := range views {
|
2022-11-07 19:37:04 +08:00
|
|
|
if _, ok := view.GrowingSegments[segmentID]; ok {
|
2022-09-15 18:48:32 +08:00
|
|
|
nodes = append(nodes, leaderID)
|
|
|
|
break
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
return nodes
|
|
|
|
}
|
|
|
|
|
2022-09-28 12:10:54 +08:00
|
|
|
// GetLeadersByGrowingSegment returns the first leader which contains the given growing segment
|
|
|
|
func (mgr *LeaderViewManager) GetLeadersByGrowingSegment(segmentID int64) *LeaderView {
|
|
|
|
mgr.rwmutex.RLock()
|
|
|
|
defer mgr.rwmutex.RUnlock()
|
|
|
|
|
|
|
|
for _, views := range mgr.views {
|
|
|
|
for _, view := range views {
|
2022-11-07 19:37:04 +08:00
|
|
|
if _, ok := view.GrowingSegments[segmentID]; ok {
|
2022-09-28 12:10:54 +08:00
|
|
|
return view
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
return nil
|
|
|
|
}
|
|
|
|
|
2022-11-07 19:37:04 +08:00
|
|
|
// GetGrowingSegmentDistByCollectionAndNode returns all segments of the given collection and node.
|
|
|
|
func (mgr *LeaderViewManager) GetGrowingSegmentDistByCollectionAndNode(collectionID, nodeID int64) map[int64]*Segment {
|
|
|
|
mgr.rwmutex.RLock()
|
|
|
|
defer mgr.rwmutex.RUnlock()
|
|
|
|
|
|
|
|
segments := make(map[int64]*Segment, 0)
|
|
|
|
if viewsOnNode, ok := mgr.views[nodeID]; ok {
|
|
|
|
for _, view := range viewsOnNode {
|
|
|
|
if view.CollectionID == collectionID {
|
|
|
|
for ID, segment := range view.GrowingSegments {
|
|
|
|
segments[ID] = segment
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
return segments
|
|
|
|
}
|
|
|
|
|
2022-12-06 18:29:19 +08:00
|
|
|
// GetSegmentDist returns the list of nodes the given channel on
|
2022-09-15 18:48:32 +08:00
|
|
|
func (mgr *LeaderViewManager) GetChannelDist(channel string) []int64 {
|
|
|
|
mgr.rwmutex.RLock()
|
|
|
|
defer mgr.rwmutex.RUnlock()
|
|
|
|
|
|
|
|
nodes := make([]int64, 0)
|
|
|
|
for leaderID, views := range mgr.views {
|
|
|
|
_, ok := views[channel]
|
|
|
|
if ok {
|
|
|
|
nodes = append(nodes, leaderID)
|
|
|
|
}
|
|
|
|
}
|
|
|
|
return nodes
|
|
|
|
}
|
|
|
|
|
|
|
|
func (mgr *LeaderViewManager) GetLeaderView(id int64) map[string]*LeaderView {
|
|
|
|
mgr.rwmutex.RLock()
|
|
|
|
defer mgr.rwmutex.RUnlock()
|
|
|
|
|
|
|
|
return mgr.views[id]
|
|
|
|
}
|
|
|
|
|
|
|
|
func (mgr *LeaderViewManager) GetLeaderShardView(id int64, shard string) *LeaderView {
|
|
|
|
mgr.rwmutex.RLock()
|
|
|
|
defer mgr.rwmutex.RUnlock()
|
|
|
|
|
|
|
|
return mgr.views[id][shard]
|
|
|
|
}
|
|
|
|
|
|
|
|
func (mgr *LeaderViewManager) GetLeadersByShard(shard string) map[int64]*LeaderView {
|
|
|
|
mgr.rwmutex.RLock()
|
|
|
|
defer mgr.rwmutex.RUnlock()
|
|
|
|
|
|
|
|
ret := make(map[int64]*LeaderView, 0)
|
|
|
|
for _, views := range mgr.views {
|
|
|
|
view, ok := views[shard]
|
|
|
|
if ok {
|
|
|
|
ret[view.ID] = view
|
|
|
|
}
|
|
|
|
}
|
|
|
|
return ret
|
|
|
|
}
|
2023-10-19 11:14:06 +08:00
|
|
|
|
|
|
|
func (mgr *LeaderViewManager) GetLatestLeadersByReplicaShard(replica *Replica, shard string) *LeaderView {
|
|
|
|
mgr.rwmutex.RLock()
|
|
|
|
defer mgr.rwmutex.RUnlock()
|
|
|
|
|
|
|
|
var ret *LeaderView
|
|
|
|
for _, views := range mgr.views {
|
|
|
|
view, ok := views[shard]
|
|
|
|
if ok &&
|
|
|
|
replica.Contains(view.ID) &&
|
|
|
|
(ret == nil || ret.Version < view.Version) {
|
|
|
|
ret = view
|
|
|
|
}
|
|
|
|
}
|
|
|
|
return ret
|
|
|
|
}
|