This is an automated email from the ASF dual-hosted git repository.
zhouyuan pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/gluten.git
The following commit(s) were added to refs/heads/main by this push:
new f04968b108 Revert "[GLUTEN-12911][VL] Fix memory pool holding in async
thread (#12919)" (#13029)
f04968b108 is described below
commit f04968b1083b12c0a13581fdaa36003b3fd1a87f
Author: Yuan <[email protected]>
AuthorDate: Tue Sep 15 17:00:14 2026 +0100
Revert "[GLUTEN-12911][VL] Fix memory pool holding in async thread
(#12919)" (#13029)
This reverts commit 810810ce94b75f881205ab6d3e113fb409a26aaf.
---
cpp/velox/memory/GlutenDirectBufferedInput.h | 81 +++-------------------------
1 file changed, 6 insertions(+), 75 deletions(-)
diff --git a/cpp/velox/memory/GlutenDirectBufferedInput.h
b/cpp/velox/memory/GlutenDirectBufferedInput.h
index 7c33c8a879..d5b588770e 100644
--- a/cpp/velox/memory/GlutenDirectBufferedInput.h
+++ b/cpp/velox/memory/GlutenDirectBufferedInput.h
@@ -17,58 +17,11 @@
#pragma once
-#include <glog/logging.h>
-
#include "velox/dwio/common/DirectBufferedInput.h"
-#include "velox/dwio/common/ExecutorBarrier.h"
namespace gluten {
-namespace detail {
-
-// Owns the ExecutorBarrier that wraps the IO executor passed to
-// DirectBufferedInput. DirectBufferedInput::readRegions() enqueues an
-// AsyncLoadHolder closure per planned load, and that closure keeps a
-// shared_ptr on the reader's MemoryPool. Cancelling a load only flips its
-// state, it does not dequeue or destroy the closure, so the pool reference
-// survives until the executor happens to drain that entry. Routing all the
-// enqueues through a barrier lets the destructor wait for them explicitly.
-//
-// This must be listed as a base before DirectBufferedInput so that it is
-// constructed first (the barrier pointer is handed to the base constructor)
-// and destructed last (the barrier has to outlive any closure referring to
-// it).
-class ExecutorBarrierHolder {
- public:
- explicit ExecutorBarrierHolder(folly::Executor* executor) :
rawExecutor_(executor), barrier_(makeBarrier(executor)) {}
-
- protected:
- // The unwrapped executor, to be handed to clones instead of this object's
- // barrier.
- folly::Executor* rawExecutor() const {
- return rawExecutor_;
- }
-
- facebook::velox::dwio::common::ExecutorBarrier* barrier() const {
- return barrier_.get();
- }
-
- private:
- static std::unique_ptr<facebook::velox::dwio::common::ExecutorBarrier>
makeBarrier(folly::Executor* executor) {
- if (executor == nullptr) {
- return nullptr;
- }
- return
std::make_unique<facebook::velox::dwio::common::ExecutorBarrier>(folly::getKeepAliveToken(executor));
- }
-
- folly::Executor* const rawExecutor_;
- const std::unique_ptr<facebook::velox::dwio::common::ExecutorBarrier>
barrier_;
-};
-
-} // namespace detail
-
-class GlutenDirectBufferedInput : private detail::ExecutorBarrierHolder,
- public
facebook::velox::dwio::common::DirectBufferedInput {
+class GlutenDirectBufferedInput : public
facebook::velox::dwio::common::DirectBufferedInput {
public:
GlutenDirectBufferedInput(
std::shared_ptr<facebook::velox::ReadFile> readFile,
@@ -81,8 +34,7 @@ class GlutenDirectBufferedInput : private
detail::ExecutorBarrierHolder,
folly::Executor* executor,
const facebook::velox::io::ReaderOptions& readerOptions,
folly::F14FastMap<std::string, std::string> fileReadOps = {})
- : ExecutorBarrierHolder(executor),
- DirectBufferedInput(
+ : DirectBufferedInput(
std::move(readFile),
metricsLog,
std::move(fileNum),
@@ -90,16 +42,13 @@ class GlutenDirectBufferedInput : private
detail::ExecutorBarrierHolder,
std::move(groupId),
std::move(ioStatistics),
std::move(ioStats),
- barrier(),
+ executor,
readerOptions,
std::move(fileReadOps)) {}
~GlutenDirectBufferedInput() override {
requests_.clear();
// Cancel all the planned loads as soon as possible to avoid unnecessary
IO.
- // Only kPlanned loads may be cancelled: cancel() overwrites the state
- // unconditionally, so cancelling a kLoading load would hide it from the
- // wait below while its IO is still in flight.
for (auto& load : coalescedLoads_) {
if (load->state() ==
facebook::velox::cache::CoalescedLoad::State::kPlanned) {
load->cancel();
@@ -117,28 +66,11 @@ class GlutenDirectBufferedInput : private
detail::ExecutorBarrierHolder,
}
}
coalescedLoads_.clear();
- // The cancelled loads above are still referenced by the AsyncLoadHolder
- // closures queued on the executor, and those closures hold a shared_ptr on
- // the memory pool. Wait until the executor has run and destroyed them so
- // that the pool reference is released before this destructor returns,
- // instead of on an IO thread after the task and its memory manager are
- // gone.
- if (barrier() != nullptr) {
- try {
- barrier()->waitAll();
- } catch (const std::exception& e) {
- // waitAll() rethrows an exception raised by any of the loads. It must
- // not escape the destructor: the loads were cancelled anyway.
- LOG(WARNING) << "Async load failed while destructing
GlutenDirectBufferedInput: " << e.what();
- }
- }
}
std::unique_ptr<facebook::velox::dwio::common::BufferedInput> clone() const
override {
- // Pass the unwrapped executor: the clone has its own lifetime and must not
- // enqueue onto this object's barrier.
return std::unique_ptr<facebook::velox::dwio::common::BufferedInput>(new
GlutenDirectBufferedInput(
- input_, fileNum_, tracker_, groupId_, ioStatistics_, ioStats_,
rawExecutor(), options_));
+ input_, fileNum_, tracker_, groupId_, ioStatistics_, ioStats_,
executor_, options_));
}
private:
@@ -152,15 +84,14 @@ class GlutenDirectBufferedInput : private
detail::ExecutorBarrierHolder,
std::shared_ptr<facebook::velox::IoStats> ioStats,
folly::Executor* executor,
const facebook::velox::io::ReaderOptions& readerOptions)
- : ExecutorBarrierHolder(executor),
- DirectBufferedInput(
+ : DirectBufferedInput(
std::move(input),
std::move(fileNum),
std::move(tracker),
std::move(groupId),
std::move(ioStatistics),
std::move(ioStats),
- barrier(),
+ executor,
readerOptions) {}
};
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
