From bd7de1b75a46d07212b2134fa44a2eca982b8923 Mon Sep 17 00:00:00 2001 From: Tobias Sargeant Date: Thu, 30 Apr 2026 22:03:29 +1000 Subject: [PATCH 1/3] Add bulk_open_max_parallelism option and plumb it through BulkOpenPRead MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The open phase has different concurrency economics from the read phase: * Each file open issues a fresh metadata request that requires DNS resolution. On macOS, libcurl uses a per-call threaded resolver (CURLRES_THREADED) — every getaddrinfo call spawns a fresh pthread for the lookup. Under bursts of concurrent opens at N=100 (the previous hard-coded fan-out), pthread_create() can return EAGAIN, libcurl emits "getaddrinfo() thread failed to start" and surfaces the failure to callers as CURLE_FAILED_INIT (curl error 2). The error is stochastic but reproducible at N >= ~117 shards on a default-configured macOS host. * The opens are GCS-latency bound; past ~16-32 in flight there is no further wall-clock benefit, only resolver-thread pressure. * The read phase reuses connections from the GCS client's pool, so it does not create resolver threads at the same rate and can safely run with the existing max_parallelism=100 default. This commit splits the two by: * Adding `Options::bulk_open_max_parallelism` (default 16) alongside the existing `max_parallelism` (default 100, retained for the read path). * Plumbing the new parameter through `file::BulkOpenPRead`, `FileSystem::BulkOpenPRead`, and the GCS / POSIX overrides + mock. * Routing `BagzReader::Open` to pass `options.bulk_open_max_parallelism` to `file::BulkOpenPRead`. * Exposing the option in the Python binding (kwargs and attribute). Values <= 0 mean "use the file-system default" (100), preserving the prior behaviour for any caller not passing the option. Trace evidence of the failure mode (libcurl 8.19.0, macOS arm64, captured via DYLD_INSERT_LIBRARIES interpose to force CURLOPT_VERBOSE): * getaddrinfo() thread failed to start * Could not resolve host: storage.googleapis.com * closing connection #N 15 thread-spawn failures in a single run × cascading retries until google-cloud-cpp's retry budget exhausts and one operation surfaces as the permanent error to the caller. --- src/bagz_reader.cc | 6 ++++-- src/bagz_reader.h | 20 +++++++++++++++++++ src/file/file.cc | 4 ++-- src/file/file.h | 9 ++++++++- src/file/file_system/file_system.cc | 9 ++++++--- src/file/file_system/file_system.h | 6 +++++- src/file/file_system/mock_file_system.h | 3 ++- src/file/file_systems/gcs/gcs_file_system.cc | 9 ++++++--- src/file/file_systems/gcs/gcs_file_system.h | 8 +++++++- .../file_systems/posix/posix_file_system.cc | 5 +++-- .../file_systems/posix/posix_file_system.h | 3 ++- src/python/bagz_reader.cc | 16 ++++++++++++--- 12 files changed, 78 insertions(+), 20 deletions(-) diff --git a/src/bagz_reader.cc b/src/bagz_reader.cc index 806626e..58c8bf1 100644 --- a/src/bagz_reader.cc +++ b/src/bagz_reader.cc @@ -608,7 +608,8 @@ absl::StatusOr BagzReader::Open(absl::string_view filespec, absl::Span> limits_files; if (options.limits_placement == LimitsPlacement::kSeparate) { all_files = file::BulkOpenPRead( - absl::StrCat(filespec, ",", internal::LimitsName(filespec))); + absl::StrCat(filespec, ",", internal::LimitsName(filespec)), + /*options=*/{}, options.bulk_open_max_parallelism); if (!all_files.ok()) { return all_files.status(); } @@ -616,7 +617,8 @@ absl::StatusOr BagzReader::Open(absl::string_view filespec, record_files = absl::MakeSpan(*all_files).subspan(0, num_files); limits_files = absl::MakeSpan(*all_files).subspan(num_files); } else { - all_files = file::BulkOpenPRead(filespec); + all_files = file::BulkOpenPRead(filespec, /*options=*/{}, + options.bulk_open_max_parallelism); if (!all_files.ok()) { return all_files.status(); } diff --git a/src/bagz_reader.h b/src/bagz_reader.h index 5ef99c5..1932679 100644 --- a/src/bagz_reader.h +++ b/src/bagz_reader.h @@ -62,6 +62,26 @@ class BagzReader { // many threads/fibers. int max_parallelism = 100; + // Maximum number of parallel file opens during the bulk-open phase + // (file_system->BulkOpenPRead). Kept lower than `max_parallelism` because + // the open phase has different concurrency economics than the read phase: + // + // * Each open issues a fresh GCS metadata request that requires a DNS + // resolution, and on macOS libcurl uses a per-call threaded resolver + // (`CURLRES_THREADED`). At high concurrency, rapid `pthread_create` + // calls for resolver threads can return EAGAIN, surfacing as + // `CURLE_FAILED_INIT` (curl error 2). The open-phase cap bounds that + // burst. + // * The opens are GCS-latency bound; past ~16-32 in flight there is no + // wall-clock benefit, only resolver-thread pressure. + // * The read phase reuses connections from the GCS client's pool, so it + // does not create resolver threads at the same rate and can safely + // run with `max_parallelism`. + // + // Values <= 0 mean "use the file-system default" (currently 100 in the + // base FileSystem implementation, retained for backward compatibility). + int bulk_open_max_parallelism = 16; + constexpr static size_t kDefaultReadAheadBytes = 1024 * 1024; // 1 MiB // Number of bytes to read ahead when iterating. std::optional read_ahead_bytes; diff --git a/src/file/file.cc b/src/file/file.cc index 07446a5..335b84c 100644 --- a/src/file/file.cc +++ b/src/file/file.cc @@ -107,7 +107,7 @@ absl::Status Delete(absl::string_view filename_with_prefix, absl::StatusOr>> BulkOpenPRead(absl::string_view file_spec_with_prefix, - absl::string_view options) { + absl::string_view options, int max_parallelism) { // Group files by prefix. Only groups files with contiguous prefixes. To // save book keeping. std::vector>> @@ -131,7 +131,7 @@ BulkOpenPRead(absl::string_view file_spec_with_prefix, for (const auto& [file_system, filenames] : filenames_from_prefix) { std::string file_spec = absl::StrJoin(filenames, ","); absl::StatusOr>> files = - file_system->BulkOpenPRead(file_spec, options); + file_system->BulkOpenPRead(file_spec, options, max_parallelism); if (!files.ok()) { return Error(files.status(), "BulkOpenPRead", file_spec_with_prefix); } diff --git a/src/file/file.h b/src/file/file.h index db3b565..b4776e6 100644 --- a/src/file/file.h +++ b/src/file/file.h @@ -81,9 +81,16 @@ absl::Status Delete(absl::string_view filename_with_prefix, // Comma separated file_specs are allowed to have different prefixes. // // `options` are passed to the underlying file system. +// +// `max_parallelism` caps the number of worker threads used to fan out the +// per-file open calls. Values <= 0 mean "use the file-system default" +// (currently 100). Lower values reduce concurrent libcurl resolver-thread +// creation on backends that route opens through HTTPS (notably the GCS +// backend on macOS), where `pthread_create` can return EAGAIN under burst +// load and surface to callers as `CURLE_FAILED_INIT`. absl::StatusOr>> BulkOpenPRead(absl::string_view file_spec_with_prefix, - absl::string_view options = {}); + absl::string_view options = {}, int max_parallelism = 0); } // namespace bagz::file diff --git a/src/file/file_system/file_system.cc b/src/file/file_system/file_system.cc index ed03850..16e3802 100644 --- a/src/file/file_system/file_system.cc +++ b/src/file/file_system/file_system.cc @@ -32,15 +32,18 @@ namespace bagz { namespace { -constexpr int kMaxParallelism = 100; +constexpr int kDefaultMaxParallelism = 100; } // namespace absl::StatusOr>> FileSystem::BulkOpenPRead(absl::string_view filespec_without_prefix, - absl::string_view options) const { + absl::string_view options, + int max_parallelism) const { std::vector filenames = ExpandShardSpec(filespec_without_prefix); std::vector> files(filenames.size()); + const int effective_parallelism = + max_parallelism > 0 ? max_parallelism : kDefaultMaxParallelism; if (absl::Status status = internal::ParallelDo( filenames.size(), [&](size_t file_index) -> absl::Status { @@ -54,7 +57,7 @@ FileSystem::BulkOpenPRead(absl::string_view filespec_without_prefix, files[file_index] = *std::move(file); return absl::OkStatus(); }, - /* max_parallelism */ kMaxParallelism, /*cpu_bound=*/false); + effective_parallelism, /*cpu_bound=*/false); !status.ok()) { return status; } diff --git a/src/file/file_system/file_system.h b/src/file/file_system/file_system.h index 9825054..5bad069 100644 --- a/src/file/file_system/file_system.h +++ b/src/file/file_system/file_system.h @@ -59,9 +59,13 @@ class FileSystem { // Opens a set of files for reading. See file_system/shard_spec.h for details // on the filespec format. + // + // `max_parallelism` caps the number of worker threads used to fan out + // per-file open calls. Values <= 0 mean "use the file-system default" + // (the base FileSystem implementation uses 100). virtual absl::StatusOr>> BulkOpenPRead(absl::string_view filespec_without_prefix, - absl::string_view options) const; + absl::string_view options, int max_parallelism = 0) const; }; } // namespace bagz diff --git a/src/file/file_system/mock_file_system.h b/src/file/file_system/mock_file_system.h index 2c5b28d..6543b78 100644 --- a/src/file/file_system/mock_file_system.h +++ b/src/file/file_system/mock_file_system.h @@ -177,7 +177,8 @@ class MockFileSystem : public FileSystem { MOCK_METHOD( absl::StatusOr>>, BulkOpenPRead, - (absl::string_view filespec_without_prefix, absl::string_view options), + (absl::string_view filespec_without_prefix, absl::string_view options, + int max_parallelism), (const, override)); }; diff --git a/src/file/file_systems/gcs/gcs_file_system.cc b/src/file/file_systems/gcs/gcs_file_system.cc index 7916675..7d24bd2 100644 --- a/src/file/file_systems/gcs/gcs_file_system.cc +++ b/src/file/file_systems/gcs/gcs_file_system.cc @@ -50,7 +50,7 @@ namespace bagz { namespace { -constexpr int kMaxParallelism = 100; +constexpr int kDefaultMaxParallelism = 100; namespace gc = ::google::cloud; namespace gcs = gc::storage; @@ -219,7 +219,8 @@ absl::Status GcsFileSystem::Delete(absl::string_view filename_without_prefix, absl::StatusOr>> GcsFileSystem::BulkOpenPRead(absl::string_view filespec_without_prefix, - absl::string_view options) const { + absl::string_view options, + int max_parallelism) const { std::vector expanded_filespec = ExpandShardSpec(filespec_without_prefix); @@ -227,6 +228,8 @@ GcsFileSystem::BulkOpenPRead(absl::string_view filespec_without_prefix, expanded_filespec.size()); gcs::Client* client = Client(); + const int effective_parallelism = + max_parallelism > 0 ? max_parallelism : kDefaultMaxParallelism; if (absl::Status status = internal::ParallelDo( expanded_filespec.size(), [&expanded_filespec, &files_per_shard_spec, @@ -271,7 +274,7 @@ GcsFileSystem::BulkOpenPRead(absl::string_view filespec_without_prefix, return absl::OkStatus(); }, - kMaxParallelism, /*cpu_bound=*/false); + effective_parallelism, /*cpu_bound=*/false); !status.ok()) { return status; } diff --git a/src/file/file_systems/gcs/gcs_file_system.h b/src/file/file_systems/gcs/gcs_file_system.h index 4ed1bb9..49e5ffd 100644 --- a/src/file/file_systems/gcs/gcs_file_system.h +++ b/src/file/file_systems/gcs/gcs_file_system.h @@ -65,9 +65,15 @@ class GcsFileSystem : public FileSystem { // on the filespec format. // `filename_without_prefix` should be the URI of the object on GCS without // the leading `gs:`. + // + // `max_parallelism` caps the number of worker threads used to issue + // ListObjects requests in parallel. Values <= 0 mean "use the default" + // (100). See bagz_reader.h's `Options::bulk_open_max_parallelism` for the + // motivation behind a tighter cap on macOS. absl::StatusOr>> BulkOpenPRead(absl::string_view filespec_without_prefix, - absl::string_view options) const override; + absl::string_view options, + int max_parallelism = 0) const override; private: google::cloud::storage::Client* absl_nonnull Client() const; diff --git a/src/file/file_systems/posix/posix_file_system.cc b/src/file/file_systems/posix/posix_file_system.cc index c1155b4..eced23a 100644 --- a/src/file/file_systems/posix/posix_file_system.cc +++ b/src/file/file_systems/posix/posix_file_system.cc @@ -200,7 +200,8 @@ absl::Status PosixFileSystem::Delete(absl::string_view filename, absl::StatusOr>> PosixFileSystem::BulkOpenPRead(absl::string_view filespec_without_prefix, - absl::string_view options) const { + absl::string_view options, + int max_parallelism) const { std::string filespec = CanonicaliseShardSpec( filespec_without_prefix, [](const std::string& pattern) { glob_t glob_result; @@ -213,7 +214,7 @@ PosixFileSystem::BulkOpenPRead(absl::string_view filespec_without_prefix, return std::string(glob_result.gl_pathv[glob_result.gl_pathc - 1]); }); - return FileSystem::BulkOpenPRead(filespec, options); + return FileSystem::BulkOpenPRead(filespec, options, max_parallelism); } } // namespace bagz diff --git a/src/file/file_systems/posix/posix_file_system.h b/src/file/file_systems/posix/posix_file_system.h index f317672..7d6fda7 100644 --- a/src/file/file_systems/posix/posix_file_system.h +++ b/src/file/file_systems/posix/posix_file_system.h @@ -48,7 +48,8 @@ class PosixFileSystem : public FileSystem { absl::StatusOr>> BulkOpenPRead(absl::string_view filespec_without_prefix, - absl::string_view options) const override; + absl::string_view options, + int max_parallelism = 0) const override; }; } // namespace bagz diff --git a/src/python/bagz_reader.cc b/src/python/bagz_reader.cc index 5770929..7a79acc 100644 --- a/src/python/bagz_reader.cc +++ b/src/python/bagz_reader.cc @@ -118,6 +118,10 @@ Options for creating the bagz.Reader. cache the limits in memory. max_parallelism: Maximum number of threads to use for operations that can be parallelized. + bulk_open_max_parallelism: Maximum number of parallel file opens during the + bulk-open phase. Defaults to 16 to avoid bursts of libcurl resolver-thread + creation that can cause CURLE_FAILED_INIT on macOS at high N. See + bagz_reader.h for details. )"; constexpr char kInitDoc[] = R"( @@ -504,25 +508,31 @@ void RegisterBagzReader(py::module& m) { .def( py::init([](ShardingLayout sharding_layout, LimitsPlacement limits_placement, Compression compression, - LimitsStorage limits_storage, int max_parallelism) { + LimitsStorage limits_storage, int max_parallelism, + int bulk_open_max_parallelism) { return BagzReader::Options{ .sharding_layout = sharding_layout, .limits_placement = limits_placement, .compression = compression, .limits_storage = limits_storage, .max_parallelism = max_parallelism, + .bulk_open_max_parallelism = bulk_open_max_parallelism, }; }), py::arg("sharding_layout") = BagzReader::Options{}.sharding_layout, py::arg("limits_placement") = BagzReader::Options{}.limits_placement, py::arg("compression") = BagzReader::Options{}.compression, py::arg("limits_storage") = BagzReader::Options{}.limits_storage, - py::arg("max_parallelism") = BagzReader::Options{}.max_parallelism) + py::arg("max_parallelism") = BagzReader::Options{}.max_parallelism, + py::arg("bulk_open_max_parallelism") = + BagzReader::Options{}.bulk_open_max_parallelism) .def_readwrite("sharding_layout", &BagzReader::Options::sharding_layout) .def_readwrite("limits_placement", &BagzReader::Options::limits_placement) .def_readwrite("compression", &BagzReader::Options::compression) .def_readwrite("limits_storage", &BagzReader::Options::limits_storage) - .def_readwrite("max_parallelism", &BagzReader::Options::max_parallelism); + .def_readwrite("max_parallelism", &BagzReader::Options::max_parallelism) + .def_readwrite("bulk_open_max_parallelism", + &BagzReader::Options::bulk_open_max_parallelism); reader .def(py::init(&Init), py::arg("file_spec"), From d795480ab200baab6c75402e9f152352b24ecfad Mon Sep 17 00:00:00 2001 From: Tobias Sargeant Date: Fri, 1 May 2026 09:18:08 +1000 Subject: [PATCH 2/3] Bump default bulk_open_max_parallelism from 16 to 32 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A parallelism sweep on a same-region GCE n2-standard-8 (1334 shards, 5 iters per setting) found: p min p50 p95 16 2.23s 2.25s 2.34s 32 1.67s 1.69s 1.71s <- floor 64 2.21s 2.23s 2.26s 100 2.31s 2.33s 4.09s p=32 is ~25% faster than p=16 and ~25% faster than p=64. Beyond 32 the curve regresses — connection-pool / libcurl-cache contention dominates the residual RTT savings. 16 was a conservative first guess; the data says we have headroom. Cross-region clients (~140ms RTT macOS->australia-southeast1) still prefer higher parallelism (the latency masks the worker-overhead cost), but 32 is within ~1s of optimum on a 1334-shard open and stays clear of the macOS pthread_create EAGAIN window that fires around p=64+. --- src/bagz_reader.h | 12 +++++++++--- src/python/bagz_reader.cc | 7 ++++--- 2 files changed, 13 insertions(+), 6 deletions(-) diff --git a/src/bagz_reader.h b/src/bagz_reader.h index 1932679..dd2f742 100644 --- a/src/bagz_reader.h +++ b/src/bagz_reader.h @@ -72,15 +72,21 @@ class BagzReader { // calls for resolver threads can return EAGAIN, surfacing as // `CURLE_FAILED_INIT` (curl error 2). The open-phase cap bounds that // burst. - // * The opens are GCS-latency bound; past ~16-32 in flight there is no - // wall-clock benefit, only resolver-thread pressure. + // * In-region the open phase is GCS-latency bound; past ~32 in flight + // there is no wall-clock benefit, only resolver-thread pressure. + // A 1334-shard sweep on a same-region GCE n2-standard-8 found p=32 + // ~25% faster than p=16, with p=64 and p=100 regressing. + // * Cross-region (high RTT) clients want more — RTT-bound work overlaps + // well — but 32 is a small ~1s cost relative to the optimum and stays + // clear of the pthread_create EAGAIN regime that triggers above ~p=64 + // on macOS. // * The read phase reuses connections from the GCS client's pool, so it // does not create resolver threads at the same rate and can safely // run with `max_parallelism`. // // Values <= 0 mean "use the file-system default" (currently 100 in the // base FileSystem implementation, retained for backward compatibility). - int bulk_open_max_parallelism = 16; + int bulk_open_max_parallelism = 32; constexpr static size_t kDefaultReadAheadBytes = 1024 * 1024; // 1 MiB // Number of bytes to read ahead when iterating. diff --git a/src/python/bagz_reader.cc b/src/python/bagz_reader.cc index 7a79acc..b5a1594 100644 --- a/src/python/bagz_reader.cc +++ b/src/python/bagz_reader.cc @@ -119,9 +119,10 @@ Options for creating the bagz.Reader. max_parallelism: Maximum number of threads to use for operations that can be parallelized. bulk_open_max_parallelism: Maximum number of parallel file opens during the - bulk-open phase. Defaults to 16 to avoid bursts of libcurl resolver-thread - creation that can cause CURLE_FAILED_INIT on macOS at high N. See - bagz_reader.h for details. + bulk-open phase. Defaults to 32 — the empirical sweet spot for in-region + GCS opens, while staying clear of the libcurl resolver-thread EAGAIN + regime that surfaces around p=64+ on macOS. See bagz_reader.h for + details. )"; constexpr char kInitDoc[] = R"( From f04552c82914c919c0131b6a8780b80eb7bee7d5 Mon Sep 17 00:00:00 2001 From: Tobias Sargeant Date: Fri, 1 May 2026 13:49:38 +1000 Subject: [PATCH 3/3] Move bulk-open parallelism default into the GCS driver. The 32-thread cap is justified by GCS-specific behaviour (DNS-resolution saturation past ~32 in flight, and fd pressure under burst load), so it belongs on the GCS backend rather than as a bagz-level default that silently constrains the posix path too. Bagz `bulk_open_max_parallelism` now defaults to 0 ("use the file-system default"), and `GcsFileSystem` picks 32 via a new `kDefaultBulkOpenMaxParallelism` alongside the existing `kDefaultMaxParallelism = 100` for reads. --- src/bagz_reader.h | 32 +++++--------------- src/file/file_systems/gcs/gcs_file_system.cc | 20 +++++++++++- src/file/file_systems/gcs/gcs_file_system.h | 6 ++-- src/python/bagz_reader.cc | 9 +++--- 4 files changed, 35 insertions(+), 32 deletions(-) diff --git a/src/bagz_reader.h b/src/bagz_reader.h index dd2f742..4eff4a7 100644 --- a/src/bagz_reader.h +++ b/src/bagz_reader.h @@ -63,30 +63,14 @@ class BagzReader { int max_parallelism = 100; // Maximum number of parallel file opens during the bulk-open phase - // (file_system->BulkOpenPRead). Kept lower than `max_parallelism` because - // the open phase has different concurrency economics than the read phase: - // - // * Each open issues a fresh GCS metadata request that requires a DNS - // resolution, and on macOS libcurl uses a per-call threaded resolver - // (`CURLRES_THREADED`). At high concurrency, rapid `pthread_create` - // calls for resolver threads can return EAGAIN, surfacing as - // `CURLE_FAILED_INIT` (curl error 2). The open-phase cap bounds that - // burst. - // * In-region the open phase is GCS-latency bound; past ~32 in flight - // there is no wall-clock benefit, only resolver-thread pressure. - // A 1334-shard sweep on a same-region GCE n2-standard-8 found p=32 - // ~25% faster than p=16, with p=64 and p=100 regressing. - // * Cross-region (high RTT) clients want more — RTT-bound work overlaps - // well — but 32 is a small ~1s cost relative to the optimum and stays - // clear of the pthread_create EAGAIN regime that triggers above ~p=64 - // on macOS. - // * The read phase reuses connections from the GCS client's pool, so it - // does not create resolver threads at the same rate and can safely - // run with `max_parallelism`. - // - // Values <= 0 mean "use the file-system default" (currently 100 in the - // base FileSystem implementation, retained for backward compatibility). - int bulk_open_max_parallelism = 32; + // (file_system->BulkOpenPRead). Values <= 0 mean "use the file-system + // default", which is the right choice for nearly all callers: each + // backend picks a default tuned to its own concurrency economics (e.g. + // the GCS backend caps lower than posix because past ~32 in flight the + // open phase is bound by DNS resolution rather than throughput, and + // higher parallelism only increases fd pressure). Override only if + // you have measured a benefit on your specific workload and backend. + int bulk_open_max_parallelism = 0; constexpr static size_t kDefaultReadAheadBytes = 1024 * 1024; // 1 MiB // Number of bytes to read ahead when iterating. diff --git a/src/file/file_systems/gcs/gcs_file_system.cc b/src/file/file_systems/gcs/gcs_file_system.cc index 7d24bd2..2fa7652 100644 --- a/src/file/file_systems/gcs/gcs_file_system.cc +++ b/src/file/file_systems/gcs/gcs_file_system.cc @@ -52,6 +52,24 @@ namespace { constexpr int kDefaultMaxParallelism = 100; +// Default cap on the number of parallel object opens during BulkOpenPRead. +// Kept well below `kDefaultMaxParallelism` because the open phase has +// different concurrency economics than the read phase: +// +// * Each open issues a fresh GCS metadata request that requires a DNS +// resolution. Past a certain point those resolutions saturate and +// additional parallelism just produces an address-resolution storm +// with no wall-clock benefit. In-region the open phase is GCS-latency +// bound; a 1334-shard sweep on a same-region GCE n2-standard-8 found +// p=32 ~25% faster than p=16, with p=64 and p=100 regressing. +// * Each in-flight open also holds a socket fd, so high parallelism +// raises the risk of fd exhaustion under burst load. The lower cap +// keeps headroom against the process fd limit. +// * The read phase reuses connections from the GCS client's pool, so it +// does not pay the same per-call resolution cost and can safely run +// with the larger `kDefaultMaxParallelism`. +constexpr int kDefaultBulkOpenMaxParallelism = 32; + namespace gc = ::google::cloud; namespace gcs = gc::storage; @@ -229,7 +247,7 @@ GcsFileSystem::BulkOpenPRead(absl::string_view filespec_without_prefix, gcs::Client* client = Client(); const int effective_parallelism = - max_parallelism > 0 ? max_parallelism : kDefaultMaxParallelism; + max_parallelism > 0 ? max_parallelism : kDefaultBulkOpenMaxParallelism; if (absl::Status status = internal::ParallelDo( expanded_filespec.size(), [&expanded_filespec, &files_per_shard_spec, diff --git a/src/file/file_systems/gcs/gcs_file_system.h b/src/file/file_systems/gcs/gcs_file_system.h index 49e5ffd..a8430fc 100644 --- a/src/file/file_systems/gcs/gcs_file_system.h +++ b/src/file/file_systems/gcs/gcs_file_system.h @@ -67,9 +67,9 @@ class GcsFileSystem : public FileSystem { // the leading `gs:`. // // `max_parallelism` caps the number of worker threads used to issue - // ListObjects requests in parallel. Values <= 0 mean "use the default" - // (100). See bagz_reader.h's `Options::bulk_open_max_parallelism` for the - // motivation behind a tighter cap on macOS. + // ListObjects requests in parallel. Values <= 0 mean "use the default", + // which is tuned in gcs_file_system.cc — see + // `kDefaultBulkOpenMaxParallelism` there for the rationale. absl::StatusOr>> BulkOpenPRead(absl::string_view filespec_without_prefix, absl::string_view options, diff --git a/src/python/bagz_reader.cc b/src/python/bagz_reader.cc index b5a1594..6f95777 100644 --- a/src/python/bagz_reader.cc +++ b/src/python/bagz_reader.cc @@ -119,10 +119,11 @@ Options for creating the bagz.Reader. max_parallelism: Maximum number of threads to use for operations that can be parallelized. bulk_open_max_parallelism: Maximum number of parallel file opens during the - bulk-open phase. Defaults to 32 — the empirical sweet spot for in-region - GCS opens, while staying clear of the libcurl resolver-thread EAGAIN - regime that surfaces around p=64+ on macOS. See bagz_reader.h for - details. + bulk-open phase. Values <= 0 (the default) mean "use the file-system + default", which is tuned per-backend (e.g. the GCS backend caps lower + than posix because the open phase saturates on DNS resolution past a + point and higher parallelism only adds fd pressure). Override only if + you have measured a benefit for your workload. )"; constexpr char kInitDoc[] = R"(