func (tbc *tsoBatchController) fetchPendingRequests(ctx context.Context) error {
	var firstTSORequest *tsoRequest
	select {
	case <-ctx.Done():
		return ctx.Err()
	case firstTSORequest = <-tbc.tsoRequestCh:
	}
	tbc.batchStartTime = time.Now()
	tbc.collectedRequestCount = 0
	tbc.pushRequest(firstTSORequest)
fetchPendingRequestsLoop:
	for tbc.collectedRequestCount < tbc.maxBatchSize {
		select {
		case tsoReq := <-tbc.tsoRequestCh:
			tbc.pushRequest(tsoReq)
		case <-ctx.Done():
			return ctx.Err()
		default:
			break fetchPendingRequestsLoop
		}
	}
	if tbc.collectedRequestCount >= tbc.maxBatchSize || tbc.maxBatchWaitInterval <= 0 {
		return nil
	}
	if tbc.collectedRequestCount < tbc.bestBatchSize {
		after := time.NewTimer(tbc.maxBatchWaitInterval)
		defer after.Stop()
		for tbc.collectedRequestCount < tbc.bestBatchSize {
			select {
			case tsoReq := <-tbc.tsoRequestCh:
				tbc.pushRequest(tsoReq)
			case <-ctx.Done():
				return ctx.Err()
			case <-after.C:
				return nil
			}
		}
	}
	for tbc.collectedRequestCount < tbc.maxBatchSize {
		select {
		case tsoReq := <-tbc.tsoRequestCh:
			tbc.pushRequest(tsoReq)
		case <-ctx.Done():
			return ctx.Err()
		default:
			return nil
		}
	}
	return nil
}
fetchPendingRequests will start a new round of the batch collecting from the channel.
It returns true if everything goes well, otherwise false which means we should stop the service.
package pd
import (
	"context"
	"fmt"
	"math/rand"
	"strings"
	"sync"
	"sync/atomic"
	"time"
	"github.com/opentracing/opentracing-go"
	"github.com/pingcap/errors"
	"github.com/pingcap/failpoint"
	"github.com/pingcap/kvproto/pkg/metapb"
	"github.com/pingcap/kvproto/pkg/pdpb"
	"github.com/pingcap/log"
	"github.com/tikv/pd/pkg/errs"
	"github.com/tikv/pd/pkg/grpcutil"
	"go.uber.org/zap"
	"google.golang.org/grpc"
	"google.golang.org/grpc/codes"
	healthpb "google.golang.org/grpc/health/grpc_health_v1"
	"google.golang.org/grpc/status"
)
type Region struct {
	Meta         *metapb.Region
	Leader       *metapb.Peer
	DownPeers    []*metapb.Peer
	PendingPeers []*metapb.Peer
}
type Client interface {
	GetClusterID(ctx context.Context) uint64
	GetAllMembers(ctx context.Context) ([]*pdpb.Member, error)
	GetLeaderAddr() string
	GetTS(ctx context.Context) (int64, int64, error)
	GetTSAsync(ctx context.Context) TSFuture
	GetLocalTS(ctx context.Context, dcLocation string) (int64, int64, error)
	GetLocalTSAsync(ctx context.Context, dcLocation string) TSFuture
	GetRegion(ctx context.Context, key []byte) (*Region, error)
	GetRegionFromMember(ctx context.Context, key []byte, memberURLs []string) (*Region, error)
	GetPrevRegion(ctx context.Context, key []byte) (*Region, error)
	GetRegionByID(ctx context.Context, regionID uint64) (*Region, error)
	ScanRegions(ctx context.Context, key, endKey []byte, limit int) ([]*Region, error)
	GetStore(ctx context.Context, storeID uint64) (*metapb.Store, error)
	GetAllStores(ctx context.Context, opts ...GetStoreOption) ([]*metapb.Store, error)
	UpdateGCSafePoint(ctx context.Context, safePoint uint64) (uint64, error)
	UpdateServiceGCSafePoint(ctx context.Context, serviceID string, ttl int64, safePoint uint64) (uint64, error)
	ScatterRegion(ctx context.Context, regionID uint64) error
	ScatterRegions(ctx context.Context, regionsID []uint64, opts ...RegionsOption) (*pdpb.ScatterRegionResponse, error)
	SplitRegions(ctx context.Context, splitKeys [][]byte, opts ...RegionsOption) (*pdpb.SplitRegionsResponse, error)
	GetOperator(ctx context.Context, regionID uint64) (*pdpb.GetOperatorResponse, error)
	Close()
}
type GetStoreOp struct {
	excludeTombstone bool
}
type GetStoreOption func(*GetStoreOp)
func WithExcludeTombstone() GetStoreOption {
	return func(op *GetStoreOp) { op.excludeTombstone = true }
}
type RegionsOp struct {
	group      string
	retryLimit uint64
}
type RegionsOption func(op *RegionsOp)
func WithGroup(group string) RegionsOption {
	return func(op *RegionsOp) { op.group = group }
}
func WithRetry(retry uint64) RegionsOption {
	return func(op *RegionsOp) { op.retryLimit = retry }
}
type tsoRequest struct {
	start      time.Time
	clientCtx  context.Context
	requestCtx context.Context
	done       chan error
	physical   int64
	logical    int64
	dcLocation string
}
type tsoBatchController struct {
	maxBatchSize         int
	maxBatchWaitInterval time.Duration
	bestBatchSize int
	tsoRequestCh          chan *tsoRequest
	collectedRequests     []*tsoRequest
	collectedRequestCount int
	batchStartTime time.Time
}
func newTSOBatchController(tsoRequestCh chan *tsoRequest, maxBatchSize int, maxBatchWaitInterval time.Duration) *tsoBatchController {
	return &tsoBatchController{
		maxBatchSize:          maxBatchSize,
		maxBatchWaitInterval:  maxBatchWaitInterval,
		bestBatchSize:         8,  
		tsoRequestCh:          tsoRequestCh,
		collectedRequests:     make([]*tsoRequest, maxBatchSize+1),
		collectedRequestCount: 0,
	}
}