From d183f4a43d8f2a6b9375c99346aa709074071681 Mon Sep 17 00:00:00 2001 From: Nick Craig-Wood Date: Fri, 24 Jul 2026 13:12:03 +0100 Subject: [PATCH] azureblob: list very large containers in parallel with list_parallelism Add a new "list_parallelism" advanced option (requires use_arrow_list) which splits a directory's blob-name keyspace into ranges and lists them concurrently using the Arrow startFrom/endBefore range parameters. Both recursive (ListR) and single-directory listings benefit. The keyspace is sharded with a single-case (digit+lowercase) character ladder. Single case is deliberate: the service validates startFrom / endBefore case-insensitively (while listing case-sensitively), so a mixed-case boundary pair is rejected with a 400; its endBefore filtering is byte-ordered, so the listing is exact. The service only honours endBefore on the Arrow listing path of accounts with Blob Listing with Apache Arrow enabled: elsewhere it either rejects the request with 400 OperationNotSupportedWithFeatureMissing or, if it fell back to XML, would ignore the shards' upper bounds. Both cases are detected and the parallel listing falls back to a sequential listing. Listing 100,000 objects from a well-connected server: config time vs XML xml 23.9s 1.0x use_arrow_list 6.9s 3.5x list_parallelism=2 4.4s 5.5x list_parallelism=4 2.8s 8.4x list_parallelism=6 2.2s 10.9x list_parallelism=10 1.5s 15.4x list_parallelism=20 1.4s 17.6x list_parallelism=30 1.1s 21.7x So Apache Arrow listing alone is ~3.5x faster than XML, and parallel listing takes that to ~20x. The ladder caps at 37 shards, so the benefit plateaus beyond list_parallelism=30. The option is hidden from the configurator, command line help and docs until the feature is officially released. --- backend/azureblob/arrow.go | 169 +++++++++++++++++++++++ backend/azureblob/arrow_internal_test.go | 88 ++++++++++++ backend/azureblob/azureblob.go | 77 ++++++++--- 3 files changed, 318 insertions(+), 16 deletions(-) create mode 100644 backend/azureblob/arrow.go 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