refactor: use a dedicated ThreadPool instead of std::async for parallel deletes

Addresses review feedback on apache#649: std::async spins up a fresh std::thread per
DeleteFiles call (and per worker), which thrashes thread creation when
CleanFiles invokes DeleteFiles 3-4 times in a row. Replace with a per-strategy
ThreadPool that owns its workers for the lifetime of the strategy.

- New util/thread_pool_internal.h / .cc: minimal worker pool with eager thread
  start, mutex+cv task queue, Submit returning a future<void>, and a RunAndWait
  fan-out helper for span-of-items workloads. Drained on destruction.
- FileCleanupStrategy now holds a ThreadPool sized once at construction
  (min(8, hardware_concurrency)). DeleteFiles short-circuits empty/single-item
  batches and otherwise delegates to pool_.RunAndWait. The pool member is
  declared last so workers are joined before file_io_ and delete_func_ are
  destroyed.
- Drops the RunInParallel free template, the per-call WorkerCount, and the
  <future>/<span> includes in expire_snapshots.cc.
- Adds util_test::ThreadPoolTest covering ctor validation, single submit,
  fan-out, empty no-op, observed concurrency, exception isolation, and dtor
  drain.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: shangxinli

src/iceberg/CMakeLists.txt

@@ -110,6 +110,7 @@ set(ICEBERG_SOURCES
     util/string_util.cc
     util/struct_like_set.cc
     util/temporal_util.cc
+    util/thread_pool_internal.cc
     util/timepoint.cc
     util/transform_util.cc
     util/truncate_util.cc

src/iceberg/test/CMakeLists.txt

@@ -132,6 +132,7 @@ add_iceberg_test(util_test
                  retry_util_test.cc
                  string_util_test.cc
                  struct_like_set_test.cc
+                 thread_pool_test.cc
                  transform_util_test.cc
                  truncate_util_test.cc
                  url_encoder_test.cc

src/iceberg/test/thread_pool_test.cc

@@ -0,0 +1,120 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#include <atomic>
+#include <chrono>
+#include <span>
+#include <stdexcept>
+#include <thread>
+#include <vector>
+
+#include <gtest/gtest.h>
+
+#include "iceberg/util/thread_pool_internal.h"
+
+namespace iceberg {
+
+TEST(ThreadPoolTest, ZeroWorkersThrows) {
+  EXPECT_THROW(ThreadPool(0), std::invalid_argument);
+}
+
+TEST(ThreadPoolTest, SubmitRunsTask) {
+  ThreadPool pool(2);
+  std::atomic<int> n{0};
+  auto fut = pool.Submit([&] { n.fetch_add(1, std::memory_order_relaxed); });
+  fut.wait();
+  EXPECT_EQ(n.load(), 1);
+}
+
+TEST(ThreadPoolTest, RunAndWaitProcessesEveryItem) {
+  ThreadPool pool(4);
+  std::vector<int> items(100);
+  for (int i = 0; i < 100; ++i) items[i] = i;
+
+  std::atomic<long> sum{0};
+  pool.RunAndWait<int>(std::span<const int>(items),
+                       [&](int v) { sum.fetch_add(v, std::memory_order_relaxed); });
+
+  // 0 + 1 + ... + 99 = 4950
+  EXPECT_EQ(sum.load(), 4950);
+}
+
+TEST(ThreadPoolTest, RunAndWaitEmptyIsNoOp) {
+  ThreadPool pool(2);
+  std::vector<int> items;
+  std::atomic<int> seen{0};
+  pool.RunAndWait<int>(std::span<const int>(items),
+                       [&](int) { seen.fetch_add(1, std::memory_order_relaxed); });
+  EXPECT_EQ(seen.load(), 0);
+}
+
+TEST(ThreadPoolTest, RunAndWaitExecutesConcurrently) {
+  // Use a barrier-style check: each task increments an in-flight counter, sleeps
+  // briefly, decrements, and records the peak. With multiple workers the peak
+  // should exceed 1.
+  constexpr int kWorkers = 4;
+  constexpr int kItems = 8;
+  ThreadPool pool(kWorkers);
+
+  std::vector<int> items(kItems, 0);
+  std::atomic<int> in_flight{0};
+  std::atomic<int> peak{0};
+
+  pool.RunAndWait<int>(std::span<const int>(items), [&](int) {
+    int now = in_flight.fetch_add(1, std::memory_order_acq_rel) + 1;
+    int prev = peak.load(std::memory_order_relaxed);
+    while (now > prev &&
+           !peak.compare_exchange_weak(prev, now, std::memory_order_relaxed)) {
+    }
+    std::this_thread::sleep_for(std::chrono::milliseconds(20));
+    in_flight.fetch_sub(1, std::memory_order_acq_rel);
+  });
+
+  EXPECT_GT(peak.load(), 1) << "expected concurrent execution across workers";
+}
+
+TEST(ThreadPoolTest, ExceptionInTaskDoesNotKillWorker) {
+  ThreadPool pool(1);
+  // packaged_task captures the exception into the future; the worker loop must
+  // continue and process the next submission.
+  auto bad = pool.Submit([] { throw std::runtime_error("boom"); });
+  EXPECT_THROW(bad.get(), std::runtime_error);
+
+  std::atomic<int> ok{0};
+  auto good = pool.Submit([&] { ok.store(1, std::memory_order_relaxed); });
+  good.wait();
+  EXPECT_EQ(ok.load(), 1);
+}
+
+TEST(ThreadPoolTest, DestructorJoinsAllPendingWork) {
+  std::atomic<int> done{0};
+  {
+    ThreadPool pool(2);
+    for (int i = 0; i < 16; ++i) {
+      // Discard futures: we rely on the destructor to drain queued work.
+      (void)pool.Submit([&] {
+        std::this_thread::sleep_for(std::chrono::milliseconds(5));
+        done.fetch_add(1, std::memory_order_relaxed);
+      });
+    }
+  }
+  EXPECT_EQ(done.load(), 16);
+}
+
+}  // namespace iceberg

src/iceberg/update/expire_snapshots.cc

@@ -22,11 +22,9 @@
 #include <algorithm>
 #include <chrono>
 #include <cstdint>
-#include <future>
 #include <iterator>
 #include <memory>
 #include <optional>
-#include <span>
 #include <string>
 #include <thread>
 #include <unordered_set>
@@ -44,6 +42,7 @@
 #include "iceberg/util/error_collector.h"
 #include "iceberg/util/macros.h"
 #include "iceberg/util/snapshot_util_internal.h"
+#include "iceberg/util/thread_pool_internal.h"
 
 namespace iceberg {
 
@@ -58,70 +57,28 @@ Result<std::shared_ptr<ManifestReader>> MakeManifestReader(
   return ManifestReader::Make(manifest, file_io, std::move(schema), std::move(spec));
 }
 
-/// \brief Cap on per-CleanFiles worker concurrency.
+/// \brief Cap on per-strategy worker concurrency.
 ///
 /// Java's RemoveSnapshots takes ExecutorServices from the table operations layer.
-/// C++ has no shared executor today, so each strategy spins up an ad-hoc pool via
-/// std::async. Cap concurrency to avoid swamping FileIO with hundreds of in-flight
-/// requests on hosts with very high core counts.
+/// C++ has no shared executor today, so each strategy owns a private ThreadPool.
+/// Cap concurrency to avoid swamping FileIO with hundreds of in-flight requests
+/// on hosts with very high core counts.
 constexpr std::size_t kMaxParallelism = 8;
 
-std::size_t WorkerCount(std::size_t item_count) {
-  if (item_count <= 1) return 1;
+std::size_t WorkerCount() {
   std::size_t hw = std::thread::hardware_concurrency();
   if (hw == 0) hw = 2;
-  return std::min({item_count, kMaxParallelism, hw});
-}
-
-/// \brief Run `work` over `items` using up to WorkerCount(items) std::async workers.
-///
-/// Each worker drains a contiguous slice of `items`. Exceptions thrown by `work` are
-/// swallowed -- callers that need to know whether a unit succeeded should record it
-/// inside `work` (e.g. via an atomic counter or a thread-safe collection).
-template <typename Item, typename Fn>
-void RunInParallel(std::span<const Item> items, Fn&& work) {
-  if (items.empty()) return;
-  std::size_t n = WorkerCount(items.size());
-  if (n <= 1) {
-    for (const auto& item : items) {
-      try {
-        work(item);
-      } catch (...) {
-        // best-effort
-      }
-    }
-    return;
-  }
-
-  std::vector<std::future<void>> futures;
-  futures.reserve(n);
-  std::size_t per = (items.size() + n - 1) / n;
-  for (std::size_t i = 0; i < n; ++i) {
-    std::size_t begin = i * per;
-    if (begin >= items.size()) break;
-    std::size_t end = std::min(begin + per, items.size());
-    auto slice = items.subspan(begin, end - begin);
-    futures.emplace_back(std::async(std::launch::async, [slice, &work]() {
-      for (const auto& item : slice) {
-        try {
-          work(item);
-        } catch (...) {
-          // best-effort: see RunInParallel doc
-        }
-      }
-    }));
-  }
-  for (auto& f : futures) {
-    f.wait();
-  }
+  return std::min(kMaxParallelism, hw);
 }
 
 /// \brief Abstract strategy for cleaning up files after snapshot expiration.
 class FileCleanupStrategy {
  public:
   FileCleanupStrategy(std::shared_ptr<FileIO> file_io,
                       std::function<void(const std::string&)> delete_func)
-      : file_io_(std::move(file_io)), delete_func_(std::move(delete_func)) {}
+      : file_io_(std::move(file_io)),
+        delete_func_(std::move(delete_func)),
+        pool_(WorkerCount()) {}
 
   virtual ~FileCleanupStrategy() = default;
 
@@ -174,16 +131,20 @@ class FileCleanupStrategy {
     }
   }
 
-  /// \brief Delete a batch of files, parallelized via RunInParallel.
+  /// \brief Delete a batch of files in parallel via the strategy's worker pool.
   ///
   /// TODO(shangxinli): When FileIO grows a SupportsBulkOperations-style
   /// `DeleteFiles(span<string>)` API, prefer the bulk path here (mirroring
   /// Java's FileCleanupStrategy.deleteFiles).
   void DeleteFiles(const std::unordered_set<std::string>& paths) {
     if (paths.empty()) return;
+    if (paths.size() == 1) {
+      DeleteFile(*paths.begin());
+      return;
+    }
     std::vector<std::string> as_vec(paths.begin(), paths.end());
-    RunInParallel<std::string>(std::span<const std::string>(as_vec),
-                               [this](const std::string& p) { DeleteFile(p); });
+    pool_.RunAndWait<std::string>(std::span<const std::string>(as_vec),
+                                  [this](const std::string& p) { DeleteFile(p); });
   }
 
   bool HasAnyStatisticsFiles(const TableMetadata& metadata) const {
@@ -225,6 +186,13 @@ class FileCleanupStrategy {
 
   std::shared_ptr<FileIO> file_io_;
   std::function<void(const std::string&)> delete_func_;
+  // Worker pool for parallel best-effort deletion. Must be declared after the
+  // members it does NOT touch -- the pool's worker threads only run tasks
+  // submitted via Submit/RunAndWait, which capture `this` and dereference
+  // file_io_ / delete_func_, so those members must outlive the pool. Since
+  // destruction order is reverse declaration order, listing pool_ last ensures
+  // workers are joined before file_io_ and delete_func_ are destroyed.
+  ThreadPool pool_;
 };
 
 /// \brief File cleanup strategy that determines safe deletions via full reachability.
@@ -233,7 +201,7 @@ class FileCleanupStrategy {
 /// still referenced by retained snapshots, then deletes orphaned manifests, data
 /// files, and manifest lists.
 ///
-/// TODO(shangxinli): Add multi-threaded manifest reading and file deletion support.
+/// TODO(shangxinli): Add multi-threaded manifest reading.
 class ReachableFileCleanup : public FileCleanupStrategy {
  public:
   using FileCleanupStrategy::FileCleanupStrategy;

src/iceberg/util/thread_pool_internal.cc

@@ -0,0 +1,84 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#include "iceberg/util/thread_pool_internal.h"
+
+#include <memory>
+#include <stdexcept>
+
+namespace iceberg {
+
+ThreadPool::ThreadPool(std::size_t num_workers) {
+  if (num_workers == 0) {
+    throw std::invalid_argument("ThreadPool num_workers must be > 0");
+  }
+  workers_.reserve(num_workers);
+  for (std::size_t i = 0; i < num_workers; ++i) {
+    workers_.emplace_back([this] { WorkerLoop(); });
+  }
+}
+
+ThreadPool::~ThreadPool() {
+  {
+    std::lock_guard<std::mutex> lock(mu_);
+    stop_ = true;
+  }
+  cv_.notify_all();
+  for (auto& w : workers_) {
+    if (w.joinable()) w.join();
+  }
+}
+
+std::future<void> ThreadPool::Submit(std::function<void()> task) {
+  auto pkg = std::make_shared<std::packaged_task<void()>>(std::move(task));
+  std::future<void> fut = pkg->get_future();
+  {
+    std::unique_lock<std::mutex> lock(mu_);
+    if (stop_) {
+      // Pool is shutting down. Run the task on the calling thread so the future
+      // becomes ready and callers don't deadlock waiting on it.
+      lock.unlock();
+      (*pkg)();
+      return fut;
+    }
+    queue_.emplace([pkg]() { (*pkg)(); });
+  }
+  cv_.notify_one();
+  return fut;
+}
+
+void ThreadPool::WorkerLoop() {
+  while (true) {
+    std::function<void()> task;
+    {
+      std::unique_lock<std::mutex> lock(mu_);
+      cv_.wait(lock, [this] { return stop_ || !queue_.empty(); });
+      if (queue_.empty()) {
+        // Drain mode: only exit once all queued work has been pulled.
+        return;
+      }
+      task = std::move(queue_.front());
+      queue_.pop();
+    }
+    // packaged_task captures exceptions into the future, so this won't throw.
+    task();
+  }
+}
+
+}  // namespace iceberg