diff --git a/cpp/core/jni/JniCommon.cc b/cpp/core/jni/JniCommon.cc index 9d435bb5923..3c17e1085bb 100644 --- a/cpp/core/jni/JniCommon.cc +++ b/cpp/core/jni/JniCommon.cc @@ -26,7 +26,7 @@ void gluten::JniCommonState::ensureInitialized(JNIEnv* env) { initialized_ = true; } -void gluten::JniCommonState::assertInitialized() { +void gluten::JniCommonState::assertInitialized() const { if (!initialized_) { throw gluten::GlutenException("Fatal: JniCommonState::Initialize(...) was not called before using the utility"); } @@ -95,7 +95,11 @@ gluten::JniColumnarBatchIterator::~JniColumnarBatchIterator() { attachCurrentThreadAsDaemonOrThrow(vm_, &env); env->DeleteGlobalRef(jColumnarBatchItr_); env->DeleteGlobalRef(serializedColumnarBatchIteratorClass_); - vm_->DetachCurrentThread(); + // Do NOT call DetachCurrentThread() here. + // libhdfs.so caches JNIEnv* in thread-local storage after AttachCurrentThread. + // If we detach, libhdfs's TLS cache becomes stale — the next HDFS call via + // libhdfs returns the stale env, causing SIGSEGV in jni_NewStringUTF. + // Daemon-attached threads are safe to leave attached; they won't block JVM shutdown. } std::shared_ptr gluten::JniColumnarBatchIterator::next() { diff --git a/cpp/core/jni/JniCommon.h b/cpp/core/jni/JniCommon.h index 783f8edfcc5..ee16412e3a1 100644 --- a/cpp/core/jni/JniCommon.h +++ b/cpp/core/jni/JniCommon.h @@ -21,6 +21,9 @@ #include #include #include +#include + +#include #include "compute/ProtobufUtils.h" #include "compute/Runtime.h" @@ -156,12 +159,17 @@ class JniCommonState { void ensureInitialized(JNIEnv* env); - void assertInitialized(); + void assertInitialized() const; void close(); jmethodID runtimeAwareCtxHandle(); + JavaVM* vm() const { + assertInitialized(); + return vm_; + } + private: void initialize(JNIEnv* env); @@ -179,6 +187,56 @@ inline JniCommonState* getJniCommonState() { return &jniCommonState; } +/// A folly::ThreadFactory for spill thread pools. Attaches each thread to the +/// JVM as a daemon at creation and calls DetachCurrentThread inside the thread +/// function body — after all work completes but before any pthread_key destructor +/// fires — to prevent unbounded JavaThread accumulation. +/// +/// INVARIANT: threads created by this factory must never call libhdfs. libhdfs +/// registers hdfsThreadDestructor via pthread_key on first HDFS call; that +/// destructor calls DetachCurrentThread at actual thread exit. Calling it +/// earlier (inside the thread body) would invalidate libhdfs's cached JNIEnv*, +/// causing SIGSEGV on the next HDFS call. +/// +/// REQUIRES: JniCommonState::ensureInitialized() must have been called before +/// constructing this factory (i.e. after JNI_OnLoad completes). +class JniAwareThreadFactory : public folly::ThreadFactory { + public: + JniAwareThreadFactory() : vm_(getJniCommonState()->vm()) {} + + std::thread newThread(folly::Func&& func) override { + return std::thread([vm = vm_, f = std::move(func)]() mutable { + JNIEnv* env = nullptr; + bool weAttached = (vm->GetEnv(reinterpret_cast(&env), jniVersion) == JNI_EDETACHED); + if (weAttached) { + if (vm->AttachCurrentThreadAsDaemon(reinterpret_cast(&env), nullptr) != JNI_OK) { + LOG(WARNING) << "JniAwareThreadFactory: failed to attach thread to JVM"; + weAttached = false; + } + } + // RAII guard: ensures DetachCurrentThread is called even if f() throws. + struct DetachGuard { + JavaVM* vm; + bool active; + ~DetachGuard() { + if (active) { + vm->DetachCurrentThread(); + } + } + } guard{vm, weAttached}; + f(); + }); + } + + const std::string& getNamePrefix() const override { + static const std::string kEmpty; + return kEmpty; + } + + private: + JavaVM* vm_; +}; + Runtime* getRuntime(JNIEnv* env, jobject runtimeAware); // Safe version of JNI {Get|Release}ArrayElements routines. diff --git a/cpp/core/jni/JniWrapper.cc b/cpp/core/jni/JniWrapper.cc index cfa06abfcb8..d30252ef4e5 100644 --- a/cpp/core/jni/JniWrapper.cc +++ b/cpp/core/jni/JniWrapper.cc @@ -103,7 +103,11 @@ class JavaInputStreamAdaptor final : public arrow::io::InputStream { env->CallVoidMethod(jniIn_, jniByteInputStreamClose); checkException(env); env->DeleteGlobalRef(jniIn_); - vm_->DetachCurrentThread(); + // Do NOT call DetachCurrentThread() here. + // libhdfs.so caches JNIEnv* in thread-local storage after AttachCurrentThread. + // If we detach, libhdfs's TLS cache becomes stale — the next HDFS call via + // libhdfs returns the stale env, causing SIGSEGV in jni_NewStringUTF. + // Daemon-attached threads are safe to leave attached; they won't block JVM shutdown. closed_ = true; return arrow::Status::OK(); } diff --git a/cpp/velox/compute/WholeStageResultIterator.cc b/cpp/velox/compute/WholeStageResultIterator.cc index 3f85dddb83f..4e0b0f3f40a 100644 --- a/cpp/velox/compute/WholeStageResultIterator.cc +++ b/cpp/velox/compute/WholeStageResultIterator.cc @@ -19,6 +19,7 @@ #include "VeloxPlanConverter.h" #include "VeloxRuntime.h" #include "config/VeloxConfig.h" +#include "jni/JniCommon.h" #include "utils/ConfigExtractor.h" #include "velox/connectors/hive/HiveConfig.h" #include "velox/connectors/hive/HiveConnectorSplit.h" @@ -91,7 +92,13 @@ WholeStageResultIterator::WholeStageResultIterator( spillStrategy_ = veloxCfg_->get(kSpillStrategy, kSpillStrategyDefaultValue); auto spillThreadNum = veloxCfg_->get(kSpillThreadNum, kSpillThreadNumDefaultValue); if (spillThreadNum > 0) { - spillExecutor_ = std::make_shared(spillThreadNum); + // INVARIANT: spillExecutor_ threads must never call libhdfs. + // JniAwareThreadFactory calls DetachCurrentThread at thread exit (inside the + // thread fn body, before any pthread_key destructor). If libhdfs were used on + // these threads, hdfsThreadDestructor would fire afterward with a stale JNIEnv*, + // causing SIGSEGV. Spill always uses local or heap-over-local filesystem. + spillExecutor_ = std::make_shared( + spillThreadNum, std::make_shared()); } getOrderedNodeIds(veloxPlan_, orderedNodeIds_);