func (p *connPool) get(ctx context.Context) (*grpc.ClientConn, error) {
	p.mu.Lock()
	defer p.mu.Unlock()
	if len(p.conns) < p.cap {
		c, err := p.newConn(ctx)
		if err != nil {
			return nil, err
		}
		p.conns = append(p.conns, c)
		return c, nil
	}
	conn := p.conns[p.next]
	p.next = (p.next + 1) % p.cap
	return conn, nil
}
get tries to get an existing connection from the pool, or make a new one if the pool not full.
package backend
import (
	"bytes"
	"context"
	"encoding/binary"
	"encoding/json"
	"io"
	"os"
	"path/filepath"
	"sort"
	"strings"
	"sync"
	"time"
	"github.com/cockroachdb/pebble"
	"github.com/cockroachdb/pebble/sstable"
	"github.com/coreos/go-semver/semver"
	"github.com/google/btree"
	"github.com/google/uuid"
	split "github.com/pingcap/br/pkg/restore"
	"github.com/pingcap/br/pkg/utils"
	"github.com/pingcap/errors"
	"github.com/pingcap/failpoint"
	"github.com/pingcap/kvproto/pkg/errorpb"
	sst "github.com/pingcap/kvproto/pkg/import_sstpb"
	"github.com/pingcap/kvproto/pkg/kvrpcpb"
	"github.com/pingcap/kvproto/pkg/metapb"
	"github.com/pingcap/parser/model"
	"github.com/pingcap/tidb/table"
	"github.com/pingcap/tidb/tablecodec"
	"github.com/pingcap/tidb/util/codec"
	"github.com/pingcap/tidb/util/hack"
	pd "github.com/tikv/pd/client"
	"go.uber.org/atomic"
	"go.uber.org/multierr"
	"go.uber.org/zap"
	"golang.org/x/sync/errgroup"
	"google.golang.org/grpc"
	"google.golang.org/grpc/backoff"
	"google.golang.org/grpc/credentials"
	"google.golang.org/grpc/keepalive"
	"github.com/pingcap/tidb-lightning/lightning/common"
	"github.com/pingcap/tidb-lightning/lightning/config"
	"github.com/pingcap/tidb-lightning/lightning/glue"
	"github.com/pingcap/tidb-lightning/lightning/log"
	"github.com/pingcap/tidb-lightning/lightning/manual"
	"github.com/pingcap/tidb-lightning/lightning/metric"
	"github.com/pingcap/tidb-lightning/lightning/worker"
)
const (
	dialTimeout  = 5 * time.Second
	bigValueSize = 1 << 16  
	gRPCKeepAliveTime    = 10 * time.Second
	gRPCKeepAliveTimeout = 3 * time.Second
	gRPCBackOffMaxDelay  = 3 * time.Second
	LocalMemoryTableSize = config.LocalMemoryTableSize
	regionMaxKeyCount = 1_440_000
	propRangeIndex = "tikv.range_index"
	defaultPropSizeIndexDistance = 4 * 1024 * 1024  
	defaultPropKeysIndexDistance = 40 * 1024
	openFilesLowerThreshold = 128
)
var (
	localMinTiDBVersion = *semver.New("4.0.0")
	localMinTiKVVersion = *semver.New("4.0.0")
	localMinPDVersion   = *semver.New("4.0.0")
)
var (
	engineMetaKey      = []byte{0, 'm', 'e', 't', 'a'}
	normalIterStartKey = []byte{1}
)
type Range struct {
	start  []byte
	end    []byte
	length int
}
type localFileMeta struct {
	Ts uint64 `json:"ts"`
	Length atomic.Int64 `json:"length"`
	TotalSize atomic.Int64 `json:"total_size"`
}
type importMutexState uint32
const (
	importMutexStateImport importMutexState = 1 << iota
	importMutexStateFlush
	importMutexStateClose
	importMutexStateLocalIngest
)
type LocalFile struct {
	localFileMeta
	db           *pebble.DB
	Uuid         uuid.UUID
	localWriters sync.Map
	isImportingAtomic atomic.Uint32
	mutex             sync.Mutex
}
func (e *LocalFile) Close() error {
	log.L().Debug("closing local engine", zap.Stringer("engine", e.Uuid), zap.Stack("stack"))
	if e.db == nil {
		return nil
	}
	err := errors.Trace(e.db.Close())
	e.db = nil
	return err
}
func (e *LocalFile) Cleanup(dataDir string) error {
	dbPath := filepath.Join(dataDir, e.Uuid.String())
	return os.RemoveAll(dbPath)
}
func (e *LocalFile) Exist(dataDir string) error {
	dbPath := filepath.Join(dataDir, e.Uuid.String())
	if _, err := os.Stat(dbPath); err != nil {
		return err
	}
	return nil
}
func (e *LocalFile) getSizeProperties() (*sizeProperties, error) {
	sstables, err := e.db.SSTables(pebble.WithProperties())
	if err != nil {
		log.L().Warn("get table properties failed", zap.Stringer("engine", e.Uuid), log.ShortError(err))
		return nil, errors.Trace(err)
	}
	sizeProps := newSizeProperties()
	for _, level := range sstables {
		for _, info := range level {
			if prop, ok := info.Properties.UserProperties[propRangeIndex]; ok {
				data := hack.Slice(prop)
				rangeProps, err := decodeRangeProperties(data)
				if err != nil {
					log.L().Warn("decodeRangeProperties failed", zap.Stringer("engine", e.Uuid),
						zap.Stringer("fileNum", info.FileNum), log.ShortError(err))
					return nil, errors.Trace(err)
				}
				sizeProps.addAll(rangeProps)
			}
		}
	}
	return sizeProps, nil
}
func (e *LocalFile) isLocked() bool {
	return e.isImportingAtomic.Load() != 0
}
func (e *LocalFile) getEngineFileSize() EngineFileSize {
	metrics := e.db.Metrics()
	total := metrics.Total()
	var memSize int64
	e.localWriters.Range(func(k, v interface{}) bool {
		w := k.(*LocalWriter)
		memSize += w.writeBatch.totalSize
		if w.writer != nil {
			total.Size += int64(w.writer.writer.EstimatedSize())
		}
		return true
	})
	return EngineFileSize{
		UUID:        e.Uuid,
		DiskSize:    total.Size,
		MemSize:     memSize,
		IsImporting: e.isLocked(),
	}
}
func (e *LocalFile) lock(state importMutexState) {
	e.mutex.Lock()
	e.isImportingAtomic.Store(uint32(state))
}
func (e *LocalFile) lockUnless(newState, ignoreStateMask importMutexState) bool {
	curState := e.isImportingAtomic.Load()
	if curState&uint32(ignoreStateMask) != 0 {
		return false
	}
	e.lock(newState)
	return true
}
func (e *LocalFile) unlock() {
	if e == nil {
		return
	}
	e.isImportingAtomic.Store(0)
	e.mutex.Unlock()
}
func (e *LocalFile) flushLocalWriters(parentCtx context.Context) error {
	eg, ctx := errgroup.WithContext(parentCtx)
	e.localWriters.Range(func(k, v interface{}) bool {
		eg.Go(func() error {
			w := k.(*LocalWriter)
			replyErrCh := make(chan error, 1)
			w.flushChMutex.RLock()
			if w.flushCh != nil {
				w.flushCh <- replyErrCh
			} else {
				replyErrCh <- nil
			}
			w.flushChMutex.RUnlock()
			select {
			case <-ctx.Done():
				return ctx.Err()
			case err := <-replyErrCh:
				return err
			}
		})
		return true
	})
	return eg.Wait()
}
func (e *LocalFile) flushEngineWithoutLock(ctx context.Context) error {
	if err := e.flushLocalWriters(ctx); err != nil {
		return err
	}
	if err := e.saveEngineMeta(); err != nil {
		return err
	}
	flushFinishedCh, err := e.db.AsyncFlush()
	if err != nil {
		return err
	}
	select {
	case <-flushFinishedCh:
		return nil
	case <-ctx.Done():
		return ctx.Err()
	}
}
func (e *LocalFile) saveEngineMeta() error {
	jsonBytes, err := json.Marshal(&e.localFileMeta)
	if err != nil {
		return errors.Trace(err)
	}
	return errors.Trace(e.db.Set(engineMetaKey, jsonBytes, &pebble.WriteOptions{Sync: false}))
}
func (e *LocalFile) loadEngineMeta() {
	jsonBytes, closer, err := e.db.Get(engineMetaKey)
	if err != nil {
		log.L().Debug("local db missing engine meta", zap.Stringer("uuid", e.Uuid), zap.Error(err))
		return
	}
	defer closer.Close()
	err = json.Unmarshal(jsonBytes, &e.localFileMeta)
	if err != nil {
		log.L().Warn("local db failed to deserialize meta", zap.Stringer("uuid", e.Uuid), zap.ByteString("content", jsonBytes), zap.Error(err))
	}
}
type gRPCConns struct {
	mu    sync.Mutex
	conns map[uint64]*connPool
}
func (conns *gRPCConns) Close() {
	conns.mu.Lock()
	defer conns.mu.Unlock()
	for _, cp := range conns.conns {
		cp.Close()
	}
}
type local struct {
	engines sync.Map  
	conns    gRPCConns
	splitCli split.SplitClient
	tls      *common.TLS
	pdAddr   string
	g        glue.Glue
	localStoreDir   string
	regionSplitSize int64
	rangeConcurrency  *worker.Pool
	ingestConcurrency *worker.Pool
	batchWriteKVPairs int
	checkpointEnabled bool
	tcpConcurrency int
	maxOpenFiles   int
}
type connPool struct {
	mu sync.Mutex
	conns   []*grpc.ClientConn
	name    string
	next    int
	cap     int
	newConn func(ctx context.Context) (*grpc.ClientConn, error)
}
func (p *connPool) takeConns() (conns []*grpc.ClientConn) {
	p.mu.Lock()
	defer p.mu.Unlock()
	p.conns, conns = nil, p.conns
	p.next = 0
	return conns
}
func (p *connPool) Close() {
	for _, c := range p.takeConns() {
		if err := c.Close(); err != nil {
			log.L().Warn("failed to close clientConn", zap.String("target", c.Target()), log.ShortError(err))
		}
	}
}