src/server/pkg/sync/sync.go
func (p *Puller) PullTree(client *pachclient.APIClient, root string, tree hashtree.HashTree, pipes bool, concurrency int) error {
	limiter := limit.New(concurrency)
	var eg errgroup.Group
	if err := tree.Walk("/", func(path string, node *hashtree.NodeProto) error {
		if node.FileNode != nil {
			path := filepath.Join(root, path)
			var hashes []string
			for _, object := range node.FileNode.Objects {
				hashes = append(hashes, object.Hash)
			}
			if pipes {
				return p.makePipe(path, func(w io.Writer) error {
					return client.GetObjects(hashes, 0, 0, uint64(node.SubtreeSize), w)
				})
			}
			limiter.Acquire()
			eg.Go(func() (retErr error) {
				defer limiter.Release()
				return p.makeFile(path, func(w io.Writer) error {
					return client.GetObjects(hashes, 0, 0, uint64(node.SubtreeSize), w)
				})
			})
		}
		return nil
	}); err != nil {
		return err
	}
	return eg.Wait()
}
func (p *Puller) PullTree(client *pachclient.APIClient, root string, tree hashtree.HashTree, pipes bool, concurrency int) error {
	limiter := limit.New(concurrency)
	var eg errgroup.Group
	if err := tree.Walk("/", func(path string, node *hashtree.NodeProto) error {
		if node.FileNode != nil {
			path := filepath.Join(root, path)
			var hashes []string
			for _, object := range node.FileNode.Objects {
				hashes = append(hashes, object.Hash)
			}
			if pipes {
				return p.makePipe(path, func(w io.Writer) error {
					return client.GetObjects(hashes, 0, 0, uint64(node.SubtreeSize), w)
				})
			}
			limiter.Acquire()
			eg.Go(func() (retErr error) {
				defer limiter.Release()
				return p.makeFile(path, func(w io.Writer) error {
					return client.GetObjects(hashes, 0, 0, uint64(node.SubtreeSize), w)
				})
			})
		}
		return nil
	}); err != nil {
		return err
	}
	return eg.Wait()
}
// PullTree pulls from a raw HashTree rather than a repo.