func (c *Client) unaryRPC(ctx context.Context, cfg *RPCConfig, f func(context.Context) error) error {
	return c.withRetries(ctx, func(ctx context.Context) error {
		ctx, cancel := context.WithTimeout(ctx, cfg.Timeout)
		defer cancel()
		return f(ctx)
	})
}
unaryRPC calls f with retries, and with per-RPC timeouts.
Does not limit concurrency.
It is useful when f calls an unary RPC.
package cas
import (
	"bufio"
	"bytes"
	"context"
	"fmt"
	"sync"
	"time"
	"github.com/pkg/errors"
	"golang.org/x/sync/semaphore"
	bsgrpc "google.golang.org/genproto/googleapis/bytestream"
	"google.golang.org/grpc"
	"github.com/bazelbuild/remote-apis-sdks/go/pkg/digest"
	"github.com/bazelbuild/remote-apis-sdks/go/pkg/retry"
	regrpc "github.com/bazelbuild/remote-apis/build/bazel/remote/execution/v2"
	repb "github.com/bazelbuild/remote-apis/build/bazel/remote/execution/v2"
)
type Client struct {
	conn *grpc.ClientConn
	InstanceName string
	Config ClientConfig
	byteStream bsgrpc.ByteStreamClient
	cas        regrpc.ContentAddressableStorageClient
	semFindMissingBlobs *semaphore.Weighted
	semBatchUpdateBlobs *semaphore.Weighted
	semByteStreamWrite  *semaphore.Weighted
	semFileIO *semaphore.Weighted
	semLargeFile *semaphore.Weighted
	fileBufReaders sync.Pool
	streamBufs sync.Pool
	testScheduleCheck func(ctx context.Context, item *uploadItem) error
}
type ClientConfig struct {
	FSConcurrency int
	FSLargeConcurrency int
	SmallFileThreshold int64
	LargeFileThreshold int64
	FileIOSize int64
	CompressedBytestreamThreshold int64
	FindMissingBlobs RPCConfig
	BatchUpdateBlobs RPCConfig
	ByteStreamWrite RPCConfig
	RetryPolicy retry.BackoffPolicy
	IgnoreCapabilities bool
}
type RPCConfig struct {
	Concurrency int
	MaxSizeBytes int
	MaxItems int
	Timeout time.Duration
}
func DefaultClientConfig() ClientConfig {
	return ClientConfig{
		FSConcurrency: 32,
		FSLargeConcurrency: 2,
		SmallFileThreshold: 1024 * 1024,        
		LargeFileThreshold: 256 * 1024 * 1024,  
		FileIOSize: 4 * 1024 * 1024,  
		FindMissingBlobs: RPCConfig{
			Concurrency: 256,  
			MaxItems:    1000,
			Timeout:     time.Minute,
		},
		BatchUpdateBlobs: RPCConfig{
			Concurrency: 256,
			MaxItems: 4000,
			MaxSizeBytes: 4 * 1024 * 1024,
			Timeout:      time.Minute,
		},
		ByteStreamWrite: RPCConfig{
			Concurrency: 256,
			MaxSizeBytes: 4 * 1024 * 1024,
			Timeout:      time.Minute,
		},
		CompressedBytestreamThreshold: -1,
		RetryPolicy: retry.ExponentialBackoff(225*time.Millisecond, 2*time.Second, retry.Attempts(6)),
	}
}
func (c *ClientConfig) Validate() error {
	switch {
	case c.FSConcurrency <= 0:
		return fmt.Errorf("FSConcurrency must be positive")
	case c.SmallFileThreshold < 0:
		return fmt.Errorf("SmallFileThreshold must be non-negative")
	case c.LargeFileThreshold <= 0:
		return fmt.Errorf("LargeFileThreshold must be positive")
	case c.SmallFileThreshold >= c.LargeFileThreshold:
		return fmt.Errorf("SmallFileThreshold must be smaller than LargeFileThreshold")
	case c.FileIOSize <= 0:
		return fmt.Errorf("FileIOSize must be positive")
	case c.FindMissingBlobs.MaxItems > 10000:
		return fmt.Errorf("FindMissingBlobs.MaxItems must <= 10000")
	}
	if err := c.FindMissingBlobs.validate(); err != nil {
		return errors.Wrap(err, "FindMissingBlobs")
	}
	if err := c.BatchUpdateBlobs.validate(); err != nil {
		return errors.Wrap(err, "BatchUpdateBlobs")
	}
	if err := c.ByteStreamWrite.validate(); err != nil {
		return errors.Wrap(err, "BatchUpdateBlobs")
	}
	return nil
}
func (c *RPCConfig) validate() error {
	switch {
	case c.Concurrency <= 0:
		return fmt.Errorf("Concurrency must be positive")
	case c.Timeout <= 0:
		return fmt.Errorf("Timeout must be positive")
	default:
		return nil
	}
}
func NewClient(ctx context.Context, conn *grpc.ClientConn, instanceName string) (*Client, error) {
	return NewClientWithConfig(ctx, conn, instanceName, DefaultClientConfig())
}
func NewClientWithConfig(ctx context.Context, conn *grpc.ClientConn, instanceName string, config ClientConfig) (*Client, error) {
	switch err := config.Validate(); {
	case err != nil:
		return nil, errors.Wrap(err, "invalid config")
	case conn == nil:
		return nil, fmt.Errorf("conn is unspecified")
	case instanceName == "":
		return nil, fmt.Errorf("instance name is unspecified")
	}
	client := &Client{
		InstanceName: instanceName,
		Config:       config,
		conn:         conn,
		byteStream:   bsgrpc.NewByteStreamClient(conn),
		cas:          regrpc.NewContentAddressableStorageClient(conn),
	}
	if !client.Config.IgnoreCapabilities {
		if err := client.checkCapabilities(ctx); err != nil {
			return nil, errors.Wrapf(err, "checking capabilities")
		}
	}
	client.init()
	return client, nil
}
var emptyReader = bytes.NewReader(nil)
func (c *Client) init() {
	c.semFindMissingBlobs = semaphore.NewWeighted(int64(c.Config.FindMissingBlobs.Concurrency))
	c.semBatchUpdateBlobs = semaphore.NewWeighted(int64(c.Config.BatchUpdateBlobs.Concurrency))
	c.semByteStreamWrite = semaphore.NewWeighted(int64(c.Config.ByteStreamWrite.Concurrency))
	c.semFileIO = semaphore.NewWeighted(int64(c.Config.FSConcurrency))
	c.semLargeFile = semaphore.NewWeighted(int64(c.Config.FSLargeConcurrency))
	c.fileBufReaders.New = func() interface{} {
		return bufio.NewReaderSize(emptyReader, int(c.Config.FileIOSize))
	}
	streamBufSize := 32 * 1024  
	if streamBufSize < c.Config.ByteStreamWrite.MaxSizeBytes {
		streamBufSize = int(c.Config.ByteStreamWrite.MaxSizeBytes)
	}
	c.streamBufs.New = func() interface{} {
		buf := make([]byte, streamBufSize)
		return &buf
	}
}