Copilot commented on code in PR #12302:
URL: https://github.com/apache/gluten/pull/12302#discussion_r3419951736
##########
cpp/velox/compute/WholeStageResultIterator.h:
##########
@@ -137,12 +139,13 @@ class WholeStageResultIterator : public
SplitAwareColumnarBatchIterator {
#endif
const SparkTaskInfo taskInfo_;
folly::Executor* executor_;
- std::shared_ptr<facebook::velox::exec::Task> task_;
+ std::unique_ptr<facebook::velox::exec::TaskCursor> cursor_;
+ facebook::velox::exec::Task* task_ = nullptr;
std::shared_ptr<const facebook::velox::core::PlanNode> veloxPlan_;
/// Spill.
std::string spillStrategy_;
- folly::Executor* spillExecutor_ = nullptr;
+ folly::Executor* spillExecutor_;
Review Comment:
`spillExecutor_` is no longer default-initialized (previously `= nullptr`).
If any constructor path or usage assumes a null default, this is undefined
behavior. Initialize it to `nullptr` (or set it in the constructor initializer
list) to keep pointer state deterministic.
##########
cpp/velox/compute/WholeStageResultIterator.h:
##########
@@ -137,12 +139,13 @@ class WholeStageResultIterator : public
SplitAwareColumnarBatchIterator {
#endif
const SparkTaskInfo taskInfo_;
folly::Executor* executor_;
- std::shared_ptr<facebook::velox::exec::Task> task_;
+ std::unique_ptr<facebook::velox::exec::TaskCursor> cursor_;
+ facebook::velox::exec::Task* task_ = nullptr;
Review Comment:
`spillExecutor_` is no longer default-initialized (previously `= nullptr`).
If any constructor path or usage assumes a null default, this is undefined
behavior. Initialize it to `nullptr` (or set it in the constructor initializer
list) to keep pointer state deterministic.
##########
cpp/velox/compute/WholeStageResultIterator.cc:
##########
@@ -307,41 +307,24 @@ std::shared_ptr<velox::core::QueryCtx>
WholeStageResultIterator::createNewVeloxQ
}
std::shared_ptr<ColumnarBatch> WholeStageResultIterator::next() {
- if (task_->isFinished()) {
- return nullptr;
- }
- velox::RowVectorPtr vector;
while (true) {
- auto future = velox::ContinueFuture::makeEmpty();
- auto out = task_->next(&future);
- if (!future.valid()) {
- // Not need to wait. Break.
- vector = std::move(out);
- break;
+ if (!cursor_->moveNext()) {
+ return nullptr;
}
- // Velox suggested to wait. This might be because another thread (e.g.,
background io thread) is spilling the task.
- GLUTEN_CHECK(out == nullptr, "Expected to wait but still got non-null
output from Velox task");
- VLOG(2) << "Velox task " << task_->taskId()
- << " is busy when ::next() is called. Will wait and try again.
Task state: "
- << taskStateString(task_->state());
- future.wait();
- }
- if (vector == nullptr) {
- return nullptr;
- }
- uint64_t numRows = vector->size();
- if (numRows == 0) {
- return nullptr;
- }
-
- {
- ScopedTimer timer(&loadLazyVectorTime_);
- for (auto& child : vector->children()) {
- child->loadedVector();
+ RowVectorPtr vector = cursor_->current();
+ GLUTEN_CHECK(vector != nullptr, "Cursor returned null vector.");
+ uint64_t numRows = vector->size();
+ if (numRows == 0) {
+ continue;
+ }
Review Comment:
This changes behavior vs the previous implementation, which returned
`nullptr` when `numRows == 0`. Now it skips empty batches and continues. If
empty vectors can be produced mid-stream, the new behavior is likely correct;
however, the PR description says 'pure refactoring with no behavioral change'.
Please either restore the previous semantics or explicitly document/justify
this behavior change (and ideally add coverage for empty-batch handling).
##########
cpp/core/jni/JniCommon.h:
##########
@@ -549,3 +550,86 @@ class JavaRssClient : public RssClient {
jmethodID javaPushPartitionData_;
jbyteArray array_;
};
+
+/// Bridges gluten::ThreadInitializer callbacks to a Java-side
+/// NativeThreadInitializer instance via JNI.
+///
+/// On initialize(), attaches the current native thread to the JVM and calls
+/// into the Java initializer so it can install Spark TaskContext or other
+/// thread-local state. On destroy(), calls the Java destroy() method but
+/// does NOT detach the thread — the underlying JVM thread may be reused by
+/// the pool, and detaching prematurely could allow the Java Thread object
+/// to be garbage-collected.
+class SparkThreadInitializer final : public gluten::ThreadInitializer {
+ public:
+ /// @param vm The JavaVM pointer from JNI_OnLoad.
+ /// @param jInitializerLocalRef A local reference to a Java object
+ /// implementing org.apache.gluten.threads.NativeThreadInitializer.
+ /// A global reference is created internally.
+ SparkThreadInitializer(JavaVM* vm, jobject jInitializerLocalRef) : vm_(vm) {
+ JNIEnv* env;
+ attachCurrentThreadAsDaemonOrThrow(vm_, &env);
+ jInitializerGlobalRef_ = env->NewGlobalRef(jInitializerLocalRef);
+ GLUTEN_CHECK(jInitializerGlobalRef_ != nullptr, "Failed to create global
reference for native thread initializer.");
+ (void)initializeMethod(env);
+ }
+
+ SparkThreadInitializer(const SparkThreadInitializer&) = delete;
+ SparkThreadInitializer(SparkThreadInitializer&&) = delete;
+ SparkThreadInitializer& operator=(const SparkThreadInitializer&) = delete;
+ SparkThreadInitializer& operator=(SparkThreadInitializer&&) = delete;
+
+ ~SparkThreadInitializer() override {
+ JNIEnv* env;
+ if (vm_->GetEnv(reinterpret_cast<void**>(&env), jniVersion) != JNI_OK) {
+ LOG(WARNING) << "SparkThreadInitializer#~SparkThreadInitializer(): "
+ << "JNIEnv was not attached to current thread";
+ return;
+ }
+ env->DeleteGlobalRef(jInitializerGlobalRef_);
+ }
+
+ void initialize(const std::string& threadName) override {
+ JNIEnv* env;
+ attachCurrentThreadAsDaemonOrThrow(vm_, &env);
+ jstring jThreadName = env->NewStringUTF(threadName.c_str());
+ env->CallVoidMethod(jInitializerGlobalRef_, initializeMethod(env),
jThreadName);
+ env->DeleteLocalRef(jThreadName);
+ checkException(env);
+ }
+
+ void destroy(const std::string& threadName) override {
+ // IMPORTANT: Do not call vm_.DetachCurrentThread here, otherwise Java
side thread
+ // object might be dereferenced and garbage-collected, to break the reuse
of thread
+ // resources.
+ JNIEnv* env;
+ attachCurrentThreadAsDaemonOrThrow(vm_, &env);
+ jstring jThreadName = env->NewStringUTF(threadName.c_str());
+ env->CallVoidMethod(jInitializerGlobalRef_, destroyMethod(env),
jThreadName);
+ env->DeleteLocalRef(jThreadName);
+ checkException(env);
+ }
+
+ private:
+ jmethodID initializeMethod(JNIEnv* env) {
+ static jmethodID initializeMethod =
+ getMethodIdOrError(env, nativeThreadInitializerClass(env),
"initialize", "(Ljava/lang/String;)V");
+ return initializeMethod;
+ }
+
+ jmethodID destroyMethod(JNIEnv* env) {
+ static jmethodID destroyMethod =
+ getMethodIdOrError(env, nativeThreadInitializerClass(env), "destroy",
"(Ljava/lang/String;)V");
+ return destroyMethod;
+ }
+
+ jclass nativeThreadInitializerClass(JNIEnv* env) {
+ static jclass javaInitializerClass =
+ createGlobalClassReferenceOrError(env,
"Lorg/apache/gluten/threads/NativeThreadInitializer;");
+ return javaInitializerClass;
+ }
Review Comment:
`FindClass`-style lookups typically expect a binary name like
`org/apache/.../ClassName`, not a field descriptor like `Lorg/...;`. Using the
descriptor here is likely to fail class resolution at runtime. Switch to
`org/apache/gluten/threads/NativeThreadInitializer` (without `L` and `;`) if
`createGlobalClassReferenceOrError` ultimately uses `JNIEnv::FindClass`.
##########
cpp/core/compute/Runtime.cc:
##########
@@ -40,9 +40,10 @@ void Runtime::registerFactory(const std::string& kind,
Runtime::Factory factory,
Runtime* Runtime::create(
const std::string& kind,
MemoryManager* memoryManager,
+ ThreadManager* threadManager,
const std::unordered_map<std::string, std::string>& sessionConf) {
auto& factory = runtimeFactories().get(kind);
- return factory(kind, std::move(memoryManager), sessionConf);
+ return factory(kind, std::move(memoryManager), std::move(threadManager),
sessionConf);
Review Comment:
`std::move` on raw pointers is misleading (it does not transfer ownership
semantics beyond copying the value). Passing `memoryManager` and
`threadManager` directly would be clearer and avoids implying unique ownership
transfer.
##########
cpp/core/threads/ThreadInitializer.h:
##########
@@ -0,0 +1,54 @@
+/*
+ * 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.
+ */
+
+#pragma once
+
+#include <memory>
+
+namespace gluten {
+
+/// Lifecycle hook invoked on each worker thread managed by a ThreadManager.
+///
+/// When a thread pool (e.g., folly::CPUThreadPoolExecutor) spawns or reaps a
+/// thread, the ThreadInitializer gives the application a chance to attach
+/// per-thread context — such as JNI thread attachment or Spark TaskContext
+/// propagation — before the thread runs native work and to clean up after.
Review Comment:
In this PR, `ThreadInitializer` is invoked from `HookedExecutor::wrap()`
around each submitted task (not specifically on thread spawn/reap). Either
adjust this documentation to match the actual call pattern (per-task execution)
or rework the implementation to run on true thread lifecycle events (e.g.,
using a thread factory / thread hook) so the documented semantics remain
accurate.
##########
cpp/velox/operators/plannodes/CudfVectorStream.h:
##########
@@ -134,6 +134,11 @@ class CudfValueStreamNode final : public
facebook::velox::core::PlanNode {
std::shared_ptr<ResultIterator> iterator)
: facebook::velox::core::PlanNode(id), outputType_(outputType),
iterator_(std::move(iterator)) {}
+ // Only supports single thread because iterator_ is not guranteed
thread-safe.
Review Comment:
Correct typo: `guranteed` -> `guaranteed`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
