diff --git a/backend/azureblob/arrow.go b/backend/azureblob/arrow.go new file mode 100644 index 000000000..bfe74c278 --- /dev/null +++ b/backend/azureblob/arrow.go @@ -0,0 +1,169 @@ +//go:build !plan9 && !solaris && !js + +// Parallel listing support for azureblob, built on the Apache Arrow +// listing format. +// +// The Arrow listing exposes the server-side startFrom (inclusive) and endBefore +// (exclusive) name-range parameters, which let a listing be sharded by +// blob-name range and listed concurrently. + +package azureblob + +import ( + "context" + "errors" + "fmt" + "sync" + "sync/atomic" + + "github.com/Azure/azure-sdk-for-go/sdk/azcore" + "github.com/Azure/azure-sdk-for-go/sdk/storage/azblob/container" + "github.com/rclone/rclone/backend/azureblob/arrowlist" + "github.com/rclone/rclone/fs" + "golang.org/x/sync/errgroup" +) + +// listArrowParallel lists (containerName, directory) by splitting the blob name +// keyspace into shards and listing them concurrently. opts must already request +// the Arrow format; it is cloned per shard with StartFrom/EndBefore set. It +// returns the number of raw items seen. directory/prefix are already normalised +// by the caller. +func (f *Fs) listArrowParallel(ctx context.Context, containerName, directory, prefix string, addContainer bool, opts *arrowlist.ListBlobsHierarchyOptions, delimiter string, fn listFn) (int, error) { + // Parallel listing needs the arrowlist client (server-side endBefore is + // only honoured on the Arrow path) - without it fall back to sequential. + if _, err := f.arrowCntSVC(containerName); err != nil { + fs.Debugf(f, "Not using parallel Arrow listing: %v", err) + return f.listBlobsPager(ctx, containerName, directory, prefix, addContainer, opts, delimiter, fn) + } + + // Split the keyspace by a single-case character ladder (see + // arrowLadderBoundaries for why single-case matters). + points := arrowLadderBoundaries(directory, arrowShardTarget(f.opt.ListParallelism)) + if len(points) < 1 { + // Couldn't usefully split - list sequentially. + return f.listBlobsPager(ctx, containerName, directory, prefix, addContainer, opts, delimiter, fn) + } + + // Build half-open [startFrom, endBefore) shards from the sorted boundary + // points p0..pk: ["", p0), [p0, p1), ..., [pk, ""). + type shard struct{ startFrom, endBefore string } + shards := make([]shard, 0, len(points)+1) + prev := "" + for _, p := range points { + shards = append(shards, shard{prev, p}) + prev = p + } + shards = append(shards, shard{prev, ""}) + + // fn (and the list.Helper behind it) is not concurrency-safe, so serialise + // delivery. Only the cheap callback is locked, not the HTTP/decode work. + var fnMu sync.Mutex + safeFn := func(remote string, item *container.BlobItem, isDir bool) error { + fnMu.Lock() + defer fnMu.Unlock() + return fn(remote, item, isDir) + } + + // Schedule every shard; SetLimit bounds concurrency, so more shards than + // workers gives free load balancing as workers pick up the next shard. + var total int64 + g, gCtx := errgroup.WithContext(ctx) + g.SetLimit(f.opt.ListParallelism) + for _, sh := range shards { + if gCtx.Err() != nil { + break + } + g.Go(func() error { + // Clone opts per shard so concurrent shards don't share StartFrom/EndBefore. + shardOpts := *opts + if sh.startFrom != "" { + startFrom := sh.startFrom + shardOpts.StartFrom = &startFrom + } else { + shardOpts.StartFrom = nil + } + if sh.endBefore != "" { + endBefore := sh.endBefore + shardOpts.EndBefore = &endBefore + } else { + shardOpts.EndBefore = nil + } + n, err := f.listBlobsPager(gCtx, containerName, directory, prefix, addContainer, &shardOpts, delimiter, safeFn) + atomic.AddInt64(&total, int64(n)) + return err + }) + } + err := g.Wait() + found := int(atomic.LoadInt64(&total)) + if isEndBeforeUnsupported(err) { + // The account can't honour the shards' endBefore bounds, so parallel + // listing can't be used on it. + f.arrowXMLFallback.Store(true) + if found == 0 { + // No shard delivered anything yet (the normal case - Arrow listing + // enablement is per account so every shard fails on its first + // page) so it is safe to relist sequentially. + fs.Debugf(f, "Parallel Arrow listing not supported by this account (%v) - listing sequentially", err) + return f.listBlobsPager(ctx, containerName, directory, prefix, addContainer, opts, delimiter, fn) + } + // Some entries were already delivered so a rerun would duplicate them. + return found, fmt.Errorf("parallel listing aborted after partial results (retry, or unset list_parallelism): %w", err) + } + return found, err +} + +// isEndBeforeUnsupported reports whether err means the service can't honour +// the endBefore listing bound, so sharded parallel listing must not be used. +// +// This happens two ways: the arrowlist pager's ErrEndBeforeXMLFallback (the +// server answered XML with endBefore set), or the service rejecting the +// endBefore parameter with 400 OperationNotSupportedWithFeatureMissing "The +// requested operation is not allowed as EndBefore parameter support is +// missing on this account" - endBefore is only supported on the Arrow +// listing path of accounts with Blob Listing with Apache Arrow enabled. +func isEndBeforeUnsupported(err error) bool { + if errors.Is(err, arrowlist.ErrEndBeforeXMLFallback) { + return true + } + var respErr *azcore.ResponseError + return errors.As(err, &respErr) && respErr.ErrorCode == "OperationNotSupportedWithFeatureMissing" +} + +// arrowShardTarget is the number of ladder boundaries to use for the given +// parallelism (a few per worker for load balancing). +func arrowShardTarget(parallelism int) int { + return parallelism * 4 +} + +// arrowLadderBoundaries returns up to target sorted ENCODED-space boundaries of +// the form directory+c, used to split the keyspace into shards. +// +// The partition is always correct (no dup/gap) regardless of the alphabet: keys +// whose next byte is outside the alphabet simply fall into an adjacent shard; +// the alphabet only affects balance. A single-case (digit+lowercase) alphabet +// is deliberate: the boundaries are then monotonic under BOTH byte order (the +// listing order) AND the service's case-insensitive startFrom/endBefore +// validation, so server-side endBefore is accepted. (A mixed-case ladder gives +// pairs like [".../Z", ".../a") which are byte-ascending but rejected 400 as +// "endbefore precedes startfrom".) directory is already encoded; the alphabet +// is ASCII (encoder-invariant) and ascending, so the result is sorted. +func arrowLadderBoundaries(directory string, target int) []string { + const alphabet = "0123456789abcdefghijklmnopqrstuvwxyz" + if target < 1 { + return nil + } + if target > len(alphabet) { + target = len(alphabet) + } + points := make([]string, 0, target) + last := -1 + for i := 0; i < target; i++ { + idx := i * len(alphabet) / target + if idx == last { + continue + } + last = idx + points = append(points, directory+string(alphabet[idx])) + } + return points +} diff --git a/backend/azureblob/arrow_internal_test.go b/backend/azureblob/arrow_internal_test.go index 7bf76941c..3ac0744c3 100644 --- a/backend/azureblob/arrow_internal_test.go +++ b/backend/azureblob/arrow_internal_test.go @@ -4,9 +4,13 @@ package azureblob import ( "context" + "errors" + "fmt" "strings" "testing" + "github.com/Azure/azure-sdk-for-go/sdk/azcore" + "github.com/rclone/rclone/backend/azureblob/arrowlist" "github.com/rclone/rclone/fs" "github.com/rclone/rclone/fstest" "github.com/rclone/rclone/fstest/fstests" @@ -34,6 +38,14 @@ func (f *Fs) testArrowList(t *testing.T) { fn() } + // withParallel runs fn with ListParallelism temporarily set. + withParallel := func(n int, fn func()) { + old := f.opt.ListParallelism + f.opt.ListParallelism = n + defer func() { f.opt.ListParallelism = old }() + fn() + } + // Create a set of known objects under a unique directory: flat files plus // one in a subdirectory (so we exercise both the flat and delimited paths). dir := "arrow-list-" + random.String(8) @@ -130,4 +142,80 @@ func (f *Fs) testArrowList(t *testing.T) { } assert.True(t, foundSubDir, "expected %q to be listed as a directory", dir+"/sub") }) + + // Parallel listing must equal sequential listing, recursively and + // per-directory, including the "sub" subdirectory. + t.Run("ParallelEqualsSequential", func(t *testing.T) { + for _, recurse := range []bool{true, false} { + var seq, par map[string]int64 + var err error + withArrow(true, func() { + withParallel(0, func() { seq, err = listing(recurse) }) + require.NoError(t, err) + withParallel(8, func() { par, err = listing(recurse) }) + require.NoError(t, err) + }) + assert.Equal(t, seq, par, "recurse=%v: parallel listing differs from sequential", recurse) + } + }) + + // A tiny list_chunk forces each shard to span multiple Arrow pages, so this + // exercises pagination (startFrom+endBefore+marker resent together) inside a + // shard. The "sub/" boundary puts a.txt..e.txt in one shard => >1 page at + // chunk size 2. + t.Run("ParallelMultiPage", func(t *testing.T) { + oldChunk := f.opt.ListChunkSize + f.opt.ListChunkSize = 2 + defer func() { f.opt.ListChunkSize = oldChunk }() + var seq, par map[string]int64 + var err error + withArrow(true, func() { + withParallel(0, func() { seq, err = listing(true) }) + require.NoError(t, err) + withParallel(8, func() { par, err = listing(true) }) + require.NoError(t, err) + }) + assert.Equal(t, seq, par) + for remote, size := range wantSizes { + assert.Equal(t, size, par[remote], remote) + } + }) +} + +func TestArrowLadderBoundaries(t *testing.T) { + for _, tc := range []struct { + directory string + target int + }{ + {"", 8}, + {"dir/", 16}, + {"dir/", 100}, // capped at alphabet size + {"dir/", 1}, + } { + points := arrowLadderBoundaries(tc.directory, tc.target) + require.NotEmpty(t, points) + assert.LessOrEqual(t, len(points), 62) + for i, p := range points { + assert.True(t, strings.HasPrefix(p, tc.directory), "boundary %q lacks prefix %q", p, tc.directory) + if i > 0 { + assert.Less(t, points[i-1], p, "boundaries not strictly ascending") + } + } + } + assert.Nil(t, arrowLadderBoundaries("dir/", 0)) + // alphabet is single-case (digit+lowercase) so boundaries are monotonic + // under both byte order and case-insensitive order + pts := arrowLadderBoundaries("dir/", 36) + for _, p := range pts { + assert.Equal(t, p, strings.ToLower(p), "boundary must be lowercase: %q", p) + } +} + +func TestIsEndBeforeUnsupported(t *testing.T) { + assert.True(t, isEndBeforeUnsupported(arrowlist.ErrEndBeforeXMLFallback)) + assert.True(t, isEndBeforeUnsupported(fmt.Errorf("wrapped: %w", arrowlist.ErrEndBeforeXMLFallback))) + assert.True(t, isEndBeforeUnsupported(&azcore.ResponseError{ErrorCode: "OperationNotSupportedWithFeatureMissing"})) + assert.False(t, isEndBeforeUnsupported(&azcore.ResponseError{ErrorCode: "ContainerNotFound"})) + assert.False(t, isEndBeforeUnsupported(errors.New("some other error"))) + assert.False(t, isEndBeforeUnsupported(nil)) } diff --git a/backend/azureblob/azureblob.go b/backend/azureblob/azureblob.go index fbbed2c98..639275e0f 100644 --- a/backend/azureblob/azureblob.go +++ b/backend/azureblob/azureblob.go @@ -22,6 +22,7 @@ import ( "strconv" "strings" "sync" + "sync/atomic" "time" "github.com/Azure/azure-sdk-for-go/sdk/azcore" @@ -258,6 +259,22 @@ listing.`, Default: false, Advanced: true, Hide: fs.OptionHideBoth, + }, { + Name: "list_parallelism", + Help: `Number of parallel shards to list a directory with. + +EXPERIMENTAL. If set greater than 1, the blob name keyspace of each +directory is split into this many ranges which are listed concurrently +using the Arrow startFrom/endBefore range parameters. This can +dramatically speed up listing containers with millions of objects, for +both recursive (ListR) and single directory listings. + +This has no effect unless "use_arrow_list" is also set, as Arrow is the +only listing path that supports server-side name ranges. The default of +0 (or 1) lists sequentially.`, + Default: 0, + Advanced: true, + Hide: fs.OptionHideBoth, }, { Name: "access_tier", Help: `Access tier of blob: hot, cool, cold or archive. @@ -412,6 +429,7 @@ type Options struct { UploadConcurrency int `config:"upload_concurrency"` ListChunkSize uint `config:"list_chunk"` UseArrowList bool `config:"use_arrow_list"` + ListParallelism int `config:"list_parallelism"` AccessTier string `config:"access_tier"` ArchiveTierDelete bool `config:"archive_tier_delete"` DisableCheckSum bool `config:"disable_checksum"` @@ -435,6 +453,7 @@ type Fs struct { cntSVCcache map[string]*container.Client // reference to containerClient per container arrowCntSVCcache map[string]*arrowlist.Client // reference to arrowlist client per container arrowClientOpts *arrowlist.ClientOptions // client options for the arrowlist clients + arrowXMLFallback atomic.Bool // set once the server answers XML so parallel listing is not retried svc *service.Client // client to access azblob cred azcore.TokenCredential // how to generate tokens (may be nil) usingSharedKeyCred bool // set if using shared key credentials @@ -634,6 +653,9 @@ func NewFs(ctx context.Context, name, root string, m configmap.Mapper) (fs.Fs, e if opt.ListChunkSize > maxListChunkSize { return nil, fmt.Errorf("blob list size can't be greater than %v - was %v", maxListChunkSize, opt.ListChunkSize) } + if opt.ListParallelism > 1 && !opt.UseArrowList { + fs.Logf(nil, "azureblob: list_parallelism has no effect without use_arrow_list - listing sequentially") + } if opt.AccessTier == "" { opt.AccessTier = string(defaultAccessTier) @@ -1188,6 +1210,40 @@ func (f *Fs) list(ctx context.Context, containerName, directory, prefix string, opts.UseArrowFormat = to.Ptr(true) } + var foundItems int + var err error + if useArrow && f.opt.ListParallelism > 1 && !f.arrowXMLFallback.Load() { + // Arrow exposes server-side startFrom/endBefore name ranges, so the + // keyspace can be sharded and listed concurrently. + foundItems, err = f.listArrowParallel(ctx, containerName, directory, prefix, addContainer, opts, delimiter, fn) + } else { + foundItems, err = f.listBlobsPager(ctx, containerName, directory, prefix, addContainer, opts, delimiter, fn) + } + if err != nil { + return err + } + if f.opt.DirectoryMarkers && foundItems == 0 && directory != "" { + // Determine whether the directory exists or not by whether it has a marker + _, err := f.readMetaData(ctx, containerName, directory) + if err != nil { + if err == fs.ErrorObjectNotFound { + return fs.ErrorDirNotFound + } + return err + } + } + return nil +} + +// listBlobsPager runs the hierarchy listing described by opts, calling fn for +// each blob and subdirectory, and returns the number of raw items seen. +// delimiter selects flat (recurse) vs hierarchical listing. If opts requests +// the Apache Arrow format the listing goes through the arrowlist pager +// (falling back to the SDK pager if the credentials don't support it); if the +// service then answers with XML (Arrow listing not enabled) a debug message +// is logged. +func (f *Fs) listBlobsPager(ctx context.Context, containerName, directory, prefix string, addContainer bool, opts *arrowlist.ListBlobsHierarchyOptions, delimiter string, fn listFn) (foundItems int, err error) { + useArrow := opts.UseArrowFormat != nil && *opts.UseArrowFormat var pager *runtime.Pager[container.ListBlobsHierarchyResponse] if useArrow { arrowSVC, err := f.arrowCntSVC(containerName) @@ -1201,7 +1257,6 @@ func (f *Fs) list(ctx context.Context, containerName, directory, prefix string, if pager == nil { pager = f.cntSVC(containerName).NewListBlobsHierarchyPager(delimiter, &opts.ListBlobsHierarchyOptions) } - foundItems := 0 checkedArrow := false for pager.More() { var response container.ListBlobsHierarchyResponse @@ -1225,9 +1280,9 @@ func (f *Fs) list(ctx context.Context, containerName, directory, prefix string, if err != nil { // Check http error code along with service code, current SDK doesn't populate service code correctly sometimes if storageErr, ok := err.(*azcore.ResponseError); ok && (storageErr.ErrorCode == string(bloberror.ContainerNotFound) || storageErr.StatusCode == http.StatusNotFound) { - return fs.ErrorDirNotFound + return foundItems, fs.ErrorDirNotFound } - return err + return foundItems, err } // Advance marker to next // marker = response.NextMarker @@ -1268,7 +1323,7 @@ func (f *Fs) list(ctx context.Context, containerName, directory, prefix string, // Send object err = fn(remote, file, isDirectory) if err != nil { - return err + return foundItems, err } } // Send the subdirectories @@ -1302,21 +1357,11 @@ func (f *Fs) list(ctx context.Context, containerName, directory, prefix string, // Send object err = fn(remote, nil, true) if err != nil { - return err + return foundItems, err } } } - if f.opt.DirectoryMarkers && foundItems == 0 && directory != "" { - // Determine whether the directory exists or not by whether it has a marker - _, err := f.readMetaData(ctx, containerName, directory) - if err != nil { - if err == fs.ErrorObjectNotFound { - return fs.ErrorDirNotFound - } - return err - } - } - return nil + return foundItems, nil } // Convert a list item into a DirEntry