func (u *uploader) startProcessing(ctx context.Context, in *UploadInput) error {
	if !filepath.IsAbs(in.Path) {
		return errors.Errorf("%q is not absolute", in.Path)
	}
	if err := in.init(u); err != nil {
		return errors.WithStack(err)
	}
	u.wgFS.Add(1)
	u.eg.Go(func() error {
		defer u.wgFS.Done()
		log.Infof("start localEg %s", in.Path)
		localEg, ctx := errgroup.WithContext(ctx)
		var treeMu sync.Mutex
		for _, relPath := range in.cleanAllowlist {
			relPath := relPath
			localEg.Go(func() error {
				absPath := in.cleanPath
				info := in.pathInfo
				if relPath != "." {
					absPath = joinFilePathsFast(in.cleanPath, relPath)
					var err error
					if info, err = os.Lstat(absPath); err != nil {
						return errors.WithStack(err)
					}
				}
				switch dig, err := u.visitPath(ctx, absPath, info, in.Exclude); {
				case err != nil:
					return errors.Wrapf(err, "%q", absPath)
				case dig != nil:
					treeMu.Lock()
					in.tree[relPath] = dig
					treeMu.Unlock()
				}
				return nil
			})
		}
		if err := localEg.Wait(); err != nil {
			return errors.WithStack(err)
		}
		log.Infof("done localEg %s", in.Path)
		for _, item := range in.partialMerkleTree() {
			if err := u.scheduleCheck(ctx, item); err != nil {
				return err
			}
		}
		close(in.ensureDigestsComputedInited())
		return nil
	})
	return nil
}
startProcessing adds the item to the appropriate stage depending on its type.
package cas
import (
	"context"
	"fmt"
	"io"
	"io/ioutil"
	"os"
	"path/filepath"
	"regexp"
	"runtime/trace"
	"sort"
	"strings"
	"sync"
	"sync/atomic"
	"time"
	log "github.com/golang/glog"
	"github.com/golang/protobuf/proto"
	"github.com/klauspost/compress/zstd"
	"github.com/pborman/uuid"
	"github.com/pkg/errors"
	"golang.org/x/sync/errgroup"
	"google.golang.org/api/support/bundler"
	"google.golang.org/grpc/status"
	"github.com/bazelbuild/remote-apis-sdks/go/pkg/cache"
	"github.com/bazelbuild/remote-apis-sdks/go/pkg/digest"
	"github.com/bazelbuild/remote-apis-sdks/go/pkg/retry"
	repb "github.com/bazelbuild/remote-apis/build/bazel/remote/execution/v2"
	bspb "google.golang.org/genproto/googleapis/bytestream"
)
var zstdEncoders = sync.Pool{
	New: func() interface{} {
		enc, _ := zstd.NewWriter(nil)
		return enc
	},
}
type UploadInput struct {
	Path string
	Allowlist []string
	Exclude *regexp.Regexp
	cleanPath      string
	cleanAllowlist []string
	pathInfo os.FileInfo
	tree                map[string]*digested
	digestsComputed     chan struct{}
	digestsComputedInit sync.Once
	u                   *uploader
}
func (in *UploadInput) Digest(relPath string) (digest.Digest, error) {
	if in.cleanPath == "" {
		return digest.Digest{}, errors.Errorf("Digest called too soon")
	}
	relPath = filepath.Clean(relPath)
	if dig, ok := in.tree[relPath]; ok {
		return digest.NewFromProtoUnvalidated(dig.digest), nil
	}
	absPath := filepath.Join(in.cleanPath, relPath)
	info, err := os.Lstat(absPath)
	if err != nil {
		return digest.Digest{}, errors.WithStack(err)
	}
	key := makeFSCacheKey(absPath, info.Mode().IsRegular(), in.Exclude)
	switch val, err, loaded := in.u.fsCache.Load(key); {
	case !loaded:
		return digest.Digest{}, errors.Wrapf(ErrDigestUnknown, "digest not found for %#v", absPath)
	case err != nil:
		return digest.Digest{}, errors.WithStack(err)
	default:
		return digest.NewFromProtoUnvalidated(val.(*digested).digest), nil
	}
}
func (in *UploadInput) ensureDigestsComputedInited() chan struct{} {
	in.digestsComputedInit.Do(func() {
		in.digestsComputed = make(chan struct{})
	})
	return in.digestsComputed
}
func (in *UploadInput) DigestsComputed() <-chan struct{} {
	return in.ensureDigestsComputedInited()
}
var oneDot = []string{"."}
func (in *UploadInput) init(u *uploader) error {
	in.u = u
	if !filepath.IsAbs(in.Path) {
		return errors.Errorf("%q is not absolute", in.Path)
	}
	in.cleanPath = filepath.Clean(in.Path)
	var err error
	if in.pathInfo, err = os.Lstat(in.cleanPath); err != nil {
		return errors.WithStack(err)
	}
	in.tree = make(map[string]*digested, 1+len(in.Allowlist))
	switch {
	case len(in.Allowlist) == 0:
		in.cleanAllowlist = oneDot
	case in.pathInfo.Mode().IsRegular():
		return errors.Errorf("the Allowlist is not supported for regular files")
	default:
		in.cleanAllowlist = make([]string, len(in.Allowlist))
		for i, subPath := range in.Allowlist {
			if filepath.IsAbs(subPath) {
				return errors.Errorf("the allowlisted path %q is not relative", subPath)
			}
			cleanSubPath := filepath.Clean(subPath)
			if cleanSubPath == ".." || strings.HasPrefix(cleanSubPath, parentDirPrefix) {
				return errors.Errorf("the allowlisted path %q is not contained by %q", subPath, in.Path)
			}
			in.cleanAllowlist[i] = cleanSubPath
		}
	}
	return nil
}
func (in *UploadInput) partialMerkleTree() (added []*uploadItem) {
	children := map[string]map[string]struct{}{}
	for relPath := range in.tree {
		for relPath != "." {
			parent := dirNameRelFast(relPath)
			if childSet, ok := children[parent]; ok {
				childSet[relPath] = struct{}{}
			} else {
				children[parent] = map[string]struct{}{relPath: {}}
			}
			relPath = parent
		}
	}
	var dfs func(relPath string) proto.Message
	dfs = func(relPath string) proto.Message {
		if dig, ok := in.tree[relPath]; ok {
			return dig.dirEntry
		}
		dir := &repb.Directory{}
		for child := range children[relPath] {
			addDirEntry(dir, dfs(child))
		}
		absPath := joinFilePathsFast(in.cleanPath, relPath)
		item := uploadItemFromDirMsg(absPath, dir)  
		added = append(added, item)
		node := &repb.DirectoryNode{
			Name:   filepath.Base(absPath),
			Digest: item.Digest,
		}
		in.tree[relPath] = &digested{dirEntry: node, digest: item.Digest}
		return node
	}
	dfs(".")
	return added
}
type TransferStats struct {
	CacheHits   DigestStat
	CacheMisses DigestStat
	Streamed DigestStat  
	Batched  DigestStat  
}
type DigestStat struct {
	Digests int64  
	Bytes   int64  
}
type UploadOptions struct {
	PreserveSymlinks bool
	AllowDanglingSymlinks bool
	Prelude func(absPath string, mode os.FileMode) error
}
type digested struct {
	dirEntry proto.Message  
	digest   *repb.Digest   
}
var (
	ErrSkip = errors.New("skip file")
	ErrDigestUnknown = errors.New("the requested digest is unknown")
)
type UploadResult struct {
	Stats TransferStats
	u     *uploader
}
func (c *Client) Upload(ctx context.Context, opt UploadOptions, inputC <-chan *UploadInput) (*UploadResult, error) {
	eg, ctx := errgroup.WithContext(ctx)
	defer eg.Wait()
	u := &uploader{
		Client:        c,
		UploadOptions: opt,
		eg:            eg,
	}
	var wgChecks sync.WaitGroup
	u.checkBundler = bundler.NewBundler(&uploadItem{}, func(items interface{}) {
		wgChecks.Add(1)
		eg.Go(func() error {
			defer wgChecks.Done()
			return u.check(ctx, items.([]*uploadItem))
		})
	})
	u.checkBundler.BundleCountThreshold = u.Config.FindMissingBlobs.MaxItems
	u.batchBundler = bundler.NewBundler(&repb.BatchUpdateBlobsRequest_Request{}, func(subReq interface{}) {
		eg.Go(func() error {
			return u.uploadBatch(ctx, subReq.([]*repb.BatchUpdateBlobsRequest_Request))
		})
	})
	u.batchBundler.BundleByteLimit = c.Config.BatchUpdateBlobs.MaxSizeBytes - int(marshalledFieldSize(int64(len(c.InstanceName)))) - 1000
	u.batchBundler.BundleCountThreshold = c.Config.BatchUpdateBlobs.MaxItems
	eg.Go(func() error {
		defer func() {
			u.wgFS.Wait()
			u.checkBundler.Flush()  
			wgChecks.Wait()         
			u.batchBundler.Flush()  
		}()
		for {
			select {
			case <-ctx.Done():
				return ctx.Err()
			case in, ok := <-inputC:
				if !ok {
					return nil
				}
				log.Infof("start startProcessing %s", in.Path)
				if err := u.startProcessing(ctx, in); err != nil {
					return err
				}
				log.Infof("finish startProcessing %s", in.Path)
			}
		}
	})
	return &UploadResult{Stats: u.stats, u: u}, errors.WithStack(eg.Wait())
}
type uploader struct {
	*Client
	UploadOptions
	eg    *errgroup.Group
	stats TransferStats
	wgFS sync.WaitGroup
	fsCache cache.SingleFlight
	checkBundler *bundler.Bundler
	seenDigests  sync.Map  
	batchBundler *bundler.Bundler
}