Repository: milvus-io/milvus
File: internal/querynode/query_collection.go (Go)

Code before:
		return
	}

	//TODO:: can't add new vChannel to selectCase
	q.watcherSelectCase = make([]reflect.SelectCase, 0)
	log.Debug("register tSafe watcher and init watcher select case",
		zap.Any("collectionID", collection.ID()),
		zap.Any("dml channels", collection.getVChannels()),
	)
	for _, channel := range collection.getVChannels() {
		q.tSafeWatchers[channel] = newTSafeWatcher()
		q.streaming.tSafeReplica.registerTSafeWatcher(channel, q.tSafeWatchers[channel])
		q.watcherSelectCase = append(q.watcherSelectCase, reflect.SelectCase{
			Dir:  reflect.SelectRecv,
			Chan: reflect.ValueOf(q.tSafeWatchers[channel].watcherChan()),
		})
	}
}

func (q *queryCollection) addToUnsolvedMsg(msg queryMsg) {
	q.unsolvedMsgMu.Lock()
	defer q.unsolvedMsgMu.Unlock()
	q.unsolvedMsg = append(q.unsolvedMsg, msg)
}

func (q *queryCollection) popAllUnsolvedMsg() []queryMsg {
	q.unsolvedMsgMu.Lock()
	defer q.unsolvedMsgMu.Unlock()
	ret := make([]queryMsg, 0, len(q.unsolvedMsg))
	ret = append(ret, q.unsolvedMsg...)
	q.unsolvedMsg = q.unsolvedMsg[:0]
	return ret
}

func (q *queryCollection) waitNewTSafe() Timestamp {
	// block until any vChannel updating tSafe
	_, _, recvOK := reflect.Select(q.watcherSelectCase)
	if !recvOK {
		//log.Warn("tSafe has been closed", zap.Any("collectionID", q.collectionID))
		return Timestamp(math.MaxInt64)
	}
	//log.Debug("wait new tSafe", zap.Any("collectionID", s.collectionID))
	t := Timestamp(math.MaxInt64)
	for channel := range q.tSafeWatchers {
		ts := q.streaming.tSafeReplica.getTSafe(channel)
		if ts <= t {
			t = ts
		}
	}
	return t
}

Review comment (suggestion):
This place will cause go routine leaks
suggestion:
```
	for {
                select {
                case <-ctx.Done():
			return
		case <-q.tSafeWatchers[vChannel].watcherChan():
                         q.watcherSharedChan <- struct{}{}
		// TODO: check if channel is closed
                }
	}
```

Code after:

// registerCollectionTSafe registers tSafe watcher if vChannels exists
func (q *queryCollection) registerCollectionTSafe() error {
	collection, err := q.streaming.replica.getCollectionByID(q.collectionID)
	if err != nil {
		return err
	}

	log.Debug("register tSafe watcher and init watcher select case",
		zap.Any("collectionID", collection.ID()),
		zap.Any("dml channels", collection.getVChannels()),
	)
	for _, channel := range collection.getVChannels() {
		err = q.addTSafeWatcher(channel)
		if err != nil {
			return err
		}
	}
	return nil
}

func (q *queryCollection) addTSafeWatcher(vChannel Channel) error {
	q.tSafeWatchersMu.Lock()
	defer q.tSafeWatchersMu.Unlock()
	if _, ok := q.tSafeWatchers[vChannel]; ok {
		err := errors.New(fmt.Sprintln("tSafeWatcher of queryCollection has been exists, ",
			"collectionID = ", q.collectionID, ", ",
			"channel = ", vChannel))
		return err
	}
	q.tSafeWatchers[vChannel] = newTSafeWatcher()
	err := q.streaming.tSafeReplica.registerTSafeWatcher(vChannel, q.tSafeWatchers[vChannel])
	if err != nil {
		return err
	}
	log.Debug("add tSafeWatcher to queryCollection",
		zap.Any("collectionID", q.collectionID),
		zap.Any("channel", vChannel),
	)
	go q.startWatcher(q.tSafeWatchers[vChannel].watcherChan())
	return nil
}

// TODO: add stopWatcher(), add close() to tSafeWatcher
func (q *queryCollection) startWatcher(channel <-chan bool) {
	for {
		select {
		case <-q.releaseCtx.Done():
			return
		case <-channel:
			// TODO: check if channel is closed