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..4eff4a7 100644 --- a/src/bagz_reader.h +++ b/src/bagz_reader.h @@ -62,6 +62,16 @@ class BagzReader { // many threads/fibers. int max_parallelism = 100; + // Maximum number of parallel file opens during the bulk-open phase + // (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. 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..2fa7652 100644 --- a/src/file/file_systems/gcs/gcs_file_system.cc +++ b/src/file/file_systems/gcs/gcs_file_system.cc @@ -50,7 +50,25 @@ namespace bagz { namespace { -constexpr int kMaxParallelism = 100; +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; @@ -219,7 +237,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 +246,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 : kDefaultBulkOpenMaxParallelism; if (absl::Status status = internal::ParallelDo( expanded_filespec.size(), [&expanded_filespec, &files_per_shard_spec, @@ -271,7 +292,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..a8430fc 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", + // 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) 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..6f95777 100644 --- a/src/python/bagz_reader.cc +++ b/src/python/bagz_reader.cc @@ -118,6 +118,12 @@ 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. 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"( @@ -504,25 +510,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"),