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

Fixbug: QueryNode search exceptions will cause memory leak (#15305) 

Signed-off-by: zhenshan.cao <zhenshan.cao@zilliz.com>
pull/15375/head
zhenshan.cao 2022-01-24 21:19:39 +08:00 committed by GitHub
parent 788ff29443
commit 5bc6ec7250
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 9 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

17
internal/querynode/query_collection.go
View File

@ -1011,6 +1011,9 @@ func (q *queryCollection) search(msg queryMsg) error {
return err
}
}
defer plan.delete()
topK := plan.getTopK()
if topK == 0 {
return fmt.Errorf("limit must be greater than 0, msgID = %d", searchMsg.ID())
@ -1023,6 +1026,8 @@ func (q *queryCollection) search(msg queryMsg) error {
if err != nil {
return err
}
defer searchReq.delete()
queryNum := searchReq.getNumOfQuery()
searchRequests := make([]*searchRequest, 0)
searchRequests = append(searchRequests, searchReq)
@ -1048,7 +1053,9 @@ func (q *queryCollection) search(msg queryMsg) error {
}
searchResults := make([]*SearchResult, 0)
defer func() {
deleteSearchResults(searchResults)
}()
// historical search
log.Debug("historical search start", zap.Int64("msgID", searchMsg.ID()))
hisSearchResults, sealedSegmentSearched, sealedPartitionSearched, err := q.historical.search(searchRequests, collection.id, searchMsg.PartitionIDs, plan, travelTimestamp)
@ -1061,7 +1068,6 @@ func (q *queryCollection) search(msg queryMsg) error {
log.Debug("streaming search start", zap.Int64("msgID", searchMsg.ID()))
for _, channel := range collection.getVChannels() {
var strSearchResults []*SearchResult
strSearchResults, growingSegmentSearched, growingPartitionSearched, err := q.streaming.search(searchRequests, collection.id, searchMsg.PartitionIDs, channel, plan, travelTimestamp)
if err != nil {
return err
@ -1128,6 +1134,7 @@ func (q *queryCollection) search(msg queryMsg) error {
if err != nil {
return err
}
defer deleteMarshaledHits(marshaledHits)
hitsBlob, err := marshaledHits.getHitsBlob()
sp.LogFields(oplog.String("statistical time", "getHitsBlob end"))
@ -1216,13 +1223,7 @@ func (q *queryCollection) search(msg queryMsg) error {
}
tr.Record(fmt.Sprintf("publish search result, msgID = %d", searchMsg.ID()))
}
sp.LogFields(oplog.String("statistical time", "before free c++ memory"))
deleteSearchResults(searchResults)
deleteMarshaledHits(marshaledHits)
sp.LogFields(oplog.String("statistical time", "stats done"))
plan.delete()
searchReq.delete()
tr.Elapse(fmt.Sprintf("all done, msgID = %d", searchMsg.ID()))
return nil
}