Memory leak caused by watch

[rleungx]

Bug Report

pd/pkg/mcs/metastorage/server/grpc_service.go

Lines 105 to 151 in 12d75c7

	for {
	select {
	case <-ctx.Done():
	return nil
	case res := <-watchChan:
	if res.Err() != nil {
	var resp meta_storagepb.WatchResponse
	if startRevision < res.CompactRevision {
	resp.Header = s.wrapErrorAndRevision(res.Header.GetRevision(), meta_storagepb.ErrorType_DATA_COMPACTED,
	fmt.Sprintf("required watch revision: %d is smaller than current compact/min revision %d.", startRevision, res.CompactRevision))
	resp.CompactRevision = res.CompactRevision
	} else {
	resp.Header = s.wrapErrorAndRevision(res.Header.GetRevision(), meta_storagepb.ErrorType_UNKNOWN,
	fmt.Sprintf("watch channel meet other error %s.", res.Err().Error()))
	}
	if err := server.Send(&resp); err != nil {
	return err
	}
	// Err() indicates that this WatchResponse holds a channel-closing error.
	return res.Err()
	}
	events := make([]*meta_storagepb.Event, 0, len(res.Events))
	for _, e := range res.Events {
	event := &meta_storagepb.Event{Kv: &meta_storagepb.KeyValue{
	Key: e.Kv.Key,
	Value: e.Kv.Value,
	ModRevision: e.Kv.ModRevision,
	CreateRevision: e.Kv.CreateRevision,
	Version: e.Kv.Version,
	Lease: e.Kv.Lease,
	}, Type: meta_storagepb.Event_EventType(e.Type)}
	if e.PrevKv != nil {
	event.PrevKv = &meta_storagepb.KeyValue{Key: e.PrevKv.Key, Value: e.PrevKv.Value}
	}
	events = append(events, event)
	}
	if len(events) > 0 {
	if err := server.Send(&meta_storagepb.WatchResponse{
	Header: &meta_storagepb.ResponseHeader{ClusterId: s.manager.ClusterID(), Revision: res.Header.GetRevision()},
	Events: events, CompactRevision: res.CompactRevision}); err != nil {
	return err
	}
	}
	}
	}
	}

It won't exit even if the client is gone. watch global config also has this problem.