func (s *casServer) BatchUpdateBlobs(
	ctx context.Context,
	req *remoteexecution.BatchUpdateBlobsRequest) (*remoteexecution.BatchUpdateBlobsResponse, error) {
	log.Debugf("Received CAS BatchUpdateBlobs request")
	if !s.IsInitialized() {
		return nil, status.Error(codes.Internal, "Server not initialized")
	}
	var (
		length  int64 = 0
		reqSize int64 = 0
		err     error = nil
	)
	defer func() {
		if err == nil {
			s.stat.Counter(stats.BzBatchUpdateSuccessCounter).Inc(1)
			if length > 0 {
				s.stat.Histogram(stats.BzBatchUpdateLengthHistogram).Update(length)
			}
		} else {
			s.stat.Counter(stats.BzBatchUpdateFailureCounter).Inc(1)
		}
	}()
	defer s.stat.Latency(stats.BzBatchUpdateLatency_ms).Time().Stop()
	for _, blobReq := range req.GetRequests() {
		d := blobReq.GetDigest()
		log.Debugf("updating: hash: %s, size: %d", d.GetHash(), d.GetSizeBytes())
		reqSize += d.GetSizeBytes()
	}
	if reqSize > bazel.BatchMaxCombinedSize {
		return nil, status.Error(codes.InvalidArgument, exceedBatchMaxMsg)
	}
	requestResource, err := s.getRequestResource(reqSize)
	if err != nil {
		return nil, status.Error(codes.Unavailable, unableToAllocMsg)
	}
	defer requestResource.Release()
	sem := make(chan struct{}, BatchParallelism)
	resultCh := make(chan *remoteexecution.BatchUpdateBlobsResponse_Response)
	var wg sync.WaitGroup
	res := &remoteexecution.BatchUpdateBlobsResponse{}
	length = int64(len(req.GetRequests()))
	log.Infof("Processing CAS BatchUpdateBlobs request of length: %d", length)
	go func() {
		for r := range resultCh {
			res.Responses = append(res.GetResponses(), r)
			<-sem
			wg.Done()
		}
	}()
	for _, blobReq := range req.GetRequests() {
		if err != nil {
			break
		}
		sem <- struct{}{}
		wg.Add(1)
		go func(r *remoteexecution.BatchUpdateBlobsRequest_Request) {
			writeRes := &remoteexecution.BatchUpdateBlobsResponse_Response{Digest: r.GetDigest()}
			if r.GetDigest().GetHash() == bazel.EmptySha {
				writeRes.Status = &google_rpc_status.Status{
					Code: int32(google_rpc_code.Code_OK),
				}
				resultCh <- writeRes
				return
			}
			if int64(len(r.GetData())) != r.GetDigest().GetSizeBytes() {
				log.Errorf("Data length/digest mismatch: %d/%d", len(r.GetData()), r.GetDigest().GetSizeBytes())
				writeRes.Status = &google_rpc_status.Status{
					Code:    int32(google_rpc_code.Code_INVALID_ARGUMENT),
					Message: fmt.Sprintf("Data to be written len: %d mismatch with request Digest size: %d", len(r.GetData()), r.GetDigest().GetSizeBytes()),
				}
				resultCh <- writeRes
				return
			}
			if bufferHash := fmt.Sprintf("%x", sha256.Sum256(r.GetData())); bufferHash != r.GetDigest().GetHash() {
				log.Errorf("Data hash/digest hash mismatch: %s/%s", bufferHash, r.GetDigest().GetHash())
				writeRes.Status = &google_rpc_status.Status{
					Code:    int32(google_rpc_code.Code_INVALID_ARGUMENT),
					Message: fmt.Sprintf("Data to be written did not hash to given Digest"),
				}
				resultCh <- writeRes
				return
			}
			storeName := bazel.DigestStoreName(r.GetDigest())
			buffer := bytes.NewReader(r.GetData())
			writeErr := s.writeToStore(storeName, buffer, int64(buffer.Len()))
			if writeErr != nil {
				writeRes.Status = &google_rpc_status.Status{
					Code:    int32(google_rpc_code.Code_INTERNAL),
					Message: writeErr.Error(),
				}
			} else {
				writeRes.Status = &google_rpc_status.Status{
					Code: int32(google_rpc_code.Code_OK),
				}
			}
			resultCh <- writeRes
		}(blobReq)
	}
	wg.Wait()
	close(resultCh)
	if err != nil {
		return nil, status.Error(codes.Internal, err.Error())
	}
	log.Infof("Finished processing CAS BatchUpdateBlobs Request of length: %d", length)
	return res, err
}
BatchUpdateBlobs writes a batch of inlined blob data into bundlestore with parallelism
package cas
import (
	"bytes"
	"crypto/sha256"
	"fmt"
	"io"
	"io/ioutil"
	"net"
	"sync"
	"time"
	"github.com/golang/protobuf/proto"
	log "github.com/sirupsen/logrus"
	"golang.org/x/net/context"
	"google.golang.org/genproto/googleapis/bytestream"
	google_rpc_code "google.golang.org/genproto/googleapis/rpc/code"
	google_rpc_status "google.golang.org/genproto/googleapis/rpc/status"
	"google.golang.org/grpc"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
	"github.com/twitter/scoot/bazel"
	remoteexecution "github.com/twitter/scoot/bazel/remoteexecution"
	"github.com/twitter/scoot/common/allocator"
	"github.com/twitter/scoot/common/stats"
	"github.com/twitter/scoot/snapshot/store"
)
type casServer struct {
	listener    net.Listener
	server      *grpc.Server
	storeConfig *store.StoreConfig
	stat        stats.StatsReceiver
	alloc       *allocator.AbstractAllocator
}
func MakeCASServer(gc *bazel.GRPCConfig, sc *store.StoreConfig, stat stats.StatsReceiver) *casServer {
	if gc == nil {
		return nil
	}
	a, err := allocator.NewAbstractAllocator(gc.ConcurrentReqSize)
	if err != nil {
		panic(err)
	}
	l, err := gc.NewListener()
	if err != nil {
		panic(err)
	}
	gs := gc.NewGRPCServer()
	g := casServer{
		listener:    l,
		server:      gs,
		storeConfig: sc,
		stat:        stat,
		alloc:       a,
	}
	remoteexecution.RegisterContentAddressableStorageServer(g.server, &g)
	remoteexecution.RegisterActionCacheServer(g.server, &g)
	bytestream.RegisterByteStreamServer(g.server, &g)
	return &g
}
func (s *casServer) IsInitialized() bool {
	if s == nil {
		return false
	} else if s.storeConfig == nil {
		return false
	}
	return true
}
func (s *casServer) Serve() error {
	log.Info("Serving GRPC CAS API on: ", s.listener.Addr())
	return s.server.Serve(s.listener)
}
func (s *casServer) FindMissingBlobs(
	ctx context.Context,
	req *remoteexecution.FindMissingBlobsRequest) (*remoteexecution.FindMissingBlobsResponse, error) {
	log.Debugf("Received CAS FindMissingBlobs request: %s", req)
	if !s.IsInitialized() {
		return nil, status.Error(codes.Internal, "Server not initialized")
	}
	var err error = nil
	var length int64 = 0
	defer func() {
		if err == nil {
			s.stat.Counter(stats.BzFindBlobsSuccessCounter).Inc(1)
			s.stat.Histogram(stats.BzFindBlobsLengthHistogram).Update(length)
		} else {
			s.stat.Counter(stats.BzFindBlobsFailureCounter).Inc(1)
		}
	}()
	defer s.stat.Latency(stats.BzFindBlobsLatency_ms).Time().Stop()
	sem := make(chan struct{}, BatchParallelism)
	resultCh := make(chan *remoteexecution.Digest)
	var wg sync.WaitGroup
	res := remoteexecution.FindMissingBlobsResponse{}
	length = int64(len(req.GetBlobDigests()))
	log.Infof("Processing CAS FindMissingBlobs request of length: %d", length)
	go func() {
		for d := range resultCh {
			if d != nil {
				res.MissingBlobDigests = append(res.MissingBlobDigests, d)
			}
			<-sem
			wg.Done()
		}
	}()
	for _, digest := range req.GetBlobDigests() {
		if err != nil {
			break
		}
		sem <- struct{}{}
		wg.Add(1)
		go func(d *remoteexecution.Digest) {
			if d.GetHash() == bazel.EmptySha {
				resultCh <- nil
				return
			}
			storeName := bazel.DigestStoreName(d)
			if exists, err := s.storeConfig.Store.Exists(storeName); err != nil {
				log.Errorf("Error checking existence of %s: %v", storeName, err)
				err = fmt.Errorf("Store failed checking existence of one or more digests")
			} else if !exists {
				resultCh <- d
				return
			}
			resultCh <- nil
		}(digest)
	}
	wg.Wait()
	close(resultCh)
	if err != nil {
		return nil, status.Error(codes.Internal, err.Error())
	}
	log.Infof("Returning CAS FindMissingBlobs missing digests: %s", res.MissingBlobDigests)
	return &res, nil
}