milvus-io
/
milvus
mirror of https://github.com/milvus-io/milvus.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Generate unique id with ts for sync task (#20928) 

Signed-off-by: Congqi Xia <congqi.xia@zilliz.com>

Signed-off-by: Congqi Xia <congqi.xia@zilliz.com>
pull/20914/head
congqixia 2022-12-01 20:35:18 +08:00 committed by GitHub
parent f745d7f489
commit 40abb13413
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 12 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
internal/datanode/flush_manager.go
View File

@ -124,7 +124,7 @@ func (q *orderFlushQueue) init() {
}
func (q *orderFlushQueue) getFlushTaskRunner(pos *internalpb.MsgPosition) *flushTaskRunner {
actual, loaded := q.working.LoadOrStore(string(pos.GetMsgID()), newFlushTaskRunner(q.segmentID, q.injectCh))
actual, loaded := q.working.LoadOrStore(getSyncTaskID(pos), newFlushTaskRunner(q.segmentID, q.injectCh))
t := actual.(*flushTaskRunner)
// not loaded means the task runner is new, do initializtion
if !loaded {
@ -316,7 +316,7 @@ func (m *rendezvousFlushManager) handleDeleteTask(segmentID UniqueID, task flush
if m.dropping.Load() {
// preventing separate delete, check position exists in queue first
q := m.getFlushQueue(segmentID)
_, ok := q.working.Load(string(pos.MsgID))
_, ok := q.working.Load(getSyncTaskID(pos))
// if ok, means position insert data already in queue, just handle task in normal mode
// if not ok, means the insert buf should be handle in drop mode
if !ok {
@ -550,6 +550,11 @@ func (m *rendezvousFlushManager) notifyAllFlushed() {
close(m.dropHandler.allFlushed)
}
func getSyncTaskID(pos *internalpb.MsgPosition) string {
// use msgID & timestamp to generate unique taskID, see also #20926
return fmt.Sprintf("%s%d", string(pos.GetMsgID()), pos.GetTimestamp())
}
// close cleans up all the left members
func (m *rendezvousFlushManager) close() {
m.dispatcher.Range(func(k, v interface{}) bool {

9
internal/datanode/flush_manager_test.go
View File

@ -424,11 +424,12 @@ func TestRendezvousFlushManager_dropMode(t *testing.T) {
for i := 1; i < 11; i++ {
target[int64(i)] = struct{}{}
m.flushBufferData(nil, int64(i), true, false, &internalpb.MsgPosition{
MsgID: []byte{1},
MsgID: []byte{byte(i)},
})
m.flushDelData(nil, int64(i), &internalpb.MsgPosition{
MsgID: []byte{1},
MsgID: []byte{byte(i)},
})
t.Log(i)
}
m.notifyAllFlushed()
@ -485,10 +486,10 @@ func TestRendezvousFlushManager_dropMode(t *testing.T) {
for i := 1; i < 11; i++ {
m.flushBufferData(nil, int64(i), true, false, &internalpb.MsgPosition{
MsgID: []byte{1},
MsgID: []byte{byte(i)},
})
m.flushDelData(nil, int64(i), &internalpb.MsgPosition{
MsgID: []byte{1},
MsgID: []byte{byte(i)},
})
}