siso: add -reapi_walkdir_stream flag WalkDir is implemented by using GetTree gRPC call. GetTree is streaming call so it would hold gRPC stream long time, so gRPC stream would be saturated and other gRPC call need to wait for free gRPC stream. To compare performance, implement WalkDir by using BatchReadBlobs, which is unary call, so it would need more roundtrip than streaming call, but it would release gRPC stream soon, so reduce gRPC stream waiting time of other gRPC call. Bug: b/539291862 Change-Id: I6c40e39e65375d9124fc1e8f55f1e4b3a3e7a74e Reviewed-on: https://chromium-review.googlesource.com/c/build/+/8162420 Auto-Submit: Fumitoshi Ukai <ukai@google.com> Reviewed-by: Yuke Liao <liaoyuke@google.com> Commit-Queue: Yuke Liao <liaoyuke@google.com>
diff --git a/siso/reapi/reapi.go b/siso/reapi/reapi.go index a62b698..ef562fd 100644 --- a/siso/reapi/reapi.go +++ b/siso/reapi/reapi.go
@@ -103,6 +103,10 @@ // with WaitExecution to mitigate grpc/network issue. KeepExecStream bool + // Use GetTree for WalkDir. + // If false, use BatchReadBlobs instead. + WalkDirStream bool + ConnPool int KeepAliveParams keepalive.ClientParameters @@ -202,6 +206,8 @@ fs.BoolVar(&o.KeepExecStream, o.Prefix+"_keep_exec_stream", false, "keep Execute stream open as long as possible") + fs.BoolVar(&o.WalkDirStream, o.Prefix+"_walkdir_stream", false, "use GetTree for WalkDir") + fs.IntVar(&o.ConnPool, o.Prefix+"_grpc_conn_pool", 25, "grpc connection pool") // https://grpc.io/docs/guides/keepalive/#keepalive-configuration-specification
diff --git a/siso/reapi/walkdir.go b/siso/reapi/walkdir.go index 6074e95..b40d67f 100644 --- a/siso/reapi/walkdir.go +++ b/siso/reapi/walkdir.go
@@ -9,10 +9,13 @@ "errors" "fmt" "io" + "iter" "path" "time" log "github.com/golang/glog" + "google.golang.org/grpc/status" + "google.golang.org/protobuf/proto" "go.chromium.org/build/hashigo/digest" rpb "go.chromium.org/build/remote-apis/build/bazel/remote/execution/v2" @@ -32,42 +35,15 @@ ctx, span := trace.NewSpan(ctx, "reapi-walkdir") defer span.Close(nil) - casClient := rpb.NewContentAddressableStorageClient(c.casConn) - var pageToken string dirs := make(map[string]digest.Digest) // dname -> digest waiting := make(map[digest.Digest][]string) // digest -> dnames waiting[d] = append(waiting[d], "") known := make(map[digest.Digest]*rpb.Directory) - var apiDur, cbDur time.Duration - var nrecvs, ndirs int + var stats walkDirStats err := retry.Do(ctx, func() error { ctx, cancel := context.WithCancel(ctx) defer cancel() - stream, err := casClient.GetTree(ctx, &rpb.GetTreeRequest{ - InstanceName: c.opt.Instance, - RootDigest: d.Proto(), - PageToken: pageToken, - DigestFunction: c.digestFn.Value(), - }) - if err != nil { - return err - } for { - started := time.Now() - resp, err := stream.Recv() - dur := time.Since(started) - apiDur += dur - nrecvs++ - if errors.Is(err, io.EOF) { - // all directories received. - return nil - } - if err != nil { - return err - } - if log.V(1) { - clog.Infof(ctx, "recv %d %s", len(resp.Directories), dur) - } var handleDir func(d digest.Digest, dir *rpb.Directory) error handleDir = func(d digest.Digest, dir *rpb.Directory) error { if len(waiting[d]) == 0 { @@ -77,10 +53,10 @@ dnames, waiting[d] = waiting[d], nil for _, dname := range dnames { dirs[dname] = d - started = time.Now() - err = fn(dname, dir) - cbDur += time.Since(started) - ndirs++ + started := time.Now() + err := fn(dname, dir) + stats.cbDur += time.Since(started) + stats.ndirs++ if err != nil { return err } @@ -105,7 +81,10 @@ } return nil } - for _, dir := range resp.Directories { + for dir, err := range c.walkDirIter(ctx, d, &stats) { + if err != nil { + return err + } dd, derr := blob.FromProtoMessage(c.digestFn, dir) if derr != nil { return derr @@ -120,18 +99,142 @@ return err } } - pageToken = resp.NextPageToken + break } + return nil }) + clog.Infof(ctx, "walkdir %s %s", d, stats.String()) + return err +} + +type walkDirStats struct { + pageToken string + apiDur, cbDur time.Duration + nrecvs, ndirs int +} + +func (s *walkDirStats) String() string { var avgAPIDur time.Duration - if nrecvs > 0 { - avgAPIDur = apiDur / time.Duration(nrecvs) + if s.nrecvs > 0 { + avgAPIDur = s.apiDur / time.Duration(s.nrecvs) } var avgCBDur time.Duration - if ndirs > 0 { - avgCBDur = cbDur / time.Duration(ndirs) + if s.ndirs > 0 { + avgCBDur = s.cbDur / time.Duration(s.ndirs) } - clog.Infof(ctx, "walkdir %s api %d %s avg:%s dir %d cb %s avg:%s", - d, nrecvs, apiDur, avgAPIDur, ndirs, cbDur, avgCBDur) - return err + return fmt.Sprintf("api %d %s avg:%s dir %d cb %s avg:%s", + s.nrecvs, s.apiDur, avgAPIDur, s.ndirs, s.cbDur, avgCBDur) +} + +func (c *Client) walkDirIter(ctx context.Context, d digest.Digest, stats *walkDirStats) iter.Seq2[*rpb.Directory, error] { + if c.opt.WalkDirStream { + return c.getTreeIter(ctx, d, stats) + } + return c.readDirIter(ctx, d, stats) +} + +func (c *Client) getTreeIter(ctx context.Context, d digest.Digest, stats *walkDirStats) iter.Seq2[*rpb.Directory, error] { + casClient := rpb.NewContentAddressableStorageClient(c.casConn) + return func(yield func(*rpb.Directory, error) bool) { + ctx, cancel := context.WithCancel(ctx) + defer cancel() + stream, err := casClient.GetTree(ctx, &rpb.GetTreeRequest{ + InstanceName: c.opt.Instance, + RootDigest: d.Proto(), + PageToken: stats.pageToken, + DigestFunction: c.digestFn.Value(), + }) + if err != nil { + yield(nil, err) + return + } + for { + started := time.Now() + resp, err := stream.Recv() + dur := time.Since(started) + stats.apiDur += dur + stats.nrecvs++ + if log.V(1) { + clog.Infof(ctx, "recv %d %s", len(resp.GetDirectories()), err) + } + if errors.Is(err, io.EOF) { + return + } + if err != nil { + yield(nil, err) + return + } + for _, dir := range resp.Directories { + if !yield(dir, nil) { + return + } + } + stats.pageToken = resp.NextPageToken + } + } +} + +func (c *Client) readDirIter(ctx context.Context, d digest.Digest, stats *walkDirStats) iter.Seq2[*rpb.Directory, error] { + casClient := rpb.NewContentAddressableStorageClient(c.casConn) + var compressors []rpb.Compressor_Value + if c.opt.BatchCompressedBlob > 0 { + compressors = []rpb.Compressor_Value{ + rpb.Compressor_ZSTD, + rpb.Compressor_DEFLATE, + } + } + return func(yield func(*rpb.Directory, error) bool) { + pendings := []digest.Digest{d} + seen := make(map[digest.Digest]struct{}) + seen[d] = struct{}{} + for len(pendings) > 0 { + req := &rpb.BatchReadBlobsRequest{ + InstanceName: c.opt.Instance, + AcceptableCompressors: compressors, + DigestFunction: c.digestFn.Value(), + } + for _, d := range pendings { + req.Digests = append(req.Digests, d.Proto()) + } + pendings = pendings[:0] + started := time.Now() + resp, err := casClient.BatchReadBlobs(ctx, req) + dur := time.Since(started) + stats.apiDur += dur + stats.nrecvs++ + if err != nil { + yield(nil, err) + return + } + for _, rd := range resp.Responses { + dd := digest.FromProto(rd.Digest) + if rd.Status.GetCode() != 0 { + yield(nil, fmt.Errorf("blob %s resp: %w", dd, status.FromProto(rd.Status).Err())) + return + } + data, err := c.decodeForBatchRead(dd, rd.Data, rd.Compressor) + if err != nil { + yield(nil, fmt.Errorf("blob %s decode: %w", dd, err)) + return + } + dir := &rpb.Directory{} + err = proto.Unmarshal(data, dir) + if err != nil { + yield(nil, fmt.Errorf("blob %s unmarshal: %w", dd, err)) + return + } + if !yield(dir, nil) { + return + } + for _, subdir := range dir.Directories { + sd := digest.FromProto(subdir.Digest) + if _, ok := seen[sd]; ok { + continue + } + pendings = append(pendings, sd) + seen[sd] = struct{}{} + } + } + } + } }