Skip to content

Commit cfef509

Browse files
taiyang-liAime
andauthored
[GLUTEN][CORE] Add TaskContext JNI callback for reading Spark task attempt id from native (#12435)
Instead of extending every JNI entry point (createRuntime / MemoryManager create/hold/release) to plumb the Spark task attempt id from Java down to C++ as an extra parameter, expose a small callback surface that native code uses on demand: - Java side: org.apache.gluten.task.TaskContextJniWrapper#currentTaskAttemptId() reads TaskContext.get().taskAttemptId() on the current thread and returns -1 when there is no task context. - C++ side: gluten::getCurrentSparkTaskAttemptId() attaches the current thread to the JVM as a daemon on demand and calls back into the Java helper via JNI. The class ref and method id are cached in function-local statics on first use. Because Spark's TaskContext is a per-thread ThreadLocal, this returns a meaningful value whenever the native call is running on an executor task thread (or any thread inheriting that ThreadLocal), which is exactly when backends need it. No signature change to Runtime / MemoryManager / RuntimeJniWrapper / NativeMemoryManagerJniWrapper. No behavior change for existing backends (Velox, ClickHouse) that do not query the task attempt id from native. Change-Id: I3185249796b0c396813dc39f54bd8e8b8589ca2a Co-authored-by: Aime <aime@bytedance.com>
1 parent bd2c1ce commit cfef509

5 files changed

Lines changed: 119 additions & 1 deletion

File tree

cpp/core/CMakeLists.txt

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -150,7 +150,8 @@ set(SPARK_COLUMNAR_PLUGIN_SRCS
150150
utils/StringUtil.cc
151151
utils/ObjectStore.cc
152152
jni/JniError.cc
153-
jni/JniCommon.cc)
153+
jni/JniCommon.cc
154+
jni/TaskContextJniWrapper.cc)
154155

155156
file(MAKE_DIRECTORY ${root_directory}/releases)
156157
add_library(gluten SHARED ${SPARK_COLUMNAR_PLUGIN_SRCS})

cpp/core/jni/JniCommon.h

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -163,6 +163,10 @@ class JniCommonState {
163163

164164
jmethodID runtimeAwareCtxHandle();
165165

166+
JavaVM* getJavaVM() const {
167+
return vm_;
168+
}
169+
166170
private:
167171
void initialize(JNIEnv* env);
168172

Lines changed: 38 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,38 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one or more
3+
* contributor license agreements. See the NOTICE file distributed with
4+
* this work for additional information regarding copyright ownership.
5+
* The ASF licenses this file to You under the Apache License, Version 2.0
6+
* (the "License"); you may not use this file except in compliance with
7+
* the License. You may obtain a copy of the License at
8+
*
9+
* http://www.apache.org/licenses/LICENSE-2.0
10+
*
11+
* Unless required by applicable law or agreed to in writing, software
12+
* distributed under the License is distributed on an "AS IS" BASIS,
13+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+
* See the License for the specific language governing permissions and
15+
* limitations under the License.
16+
*/
17+
#include "jni/TaskContextJniWrapper.h"
18+
19+
#include "jni/JniCommon.h"
20+
21+
namespace gluten {
22+
23+
int64_t getCurrentSparkTaskAttemptId() {
24+
JavaVM* vm = getJniCommonState()->getJavaVM();
25+
JNIEnv* env = nullptr;
26+
attachCurrentThreadAsDaemonOrThrow(vm, &env);
27+
28+
static jclass taskContextJniWrapperClass =
29+
createGlobalClassReferenceOrError(env, "Lorg/apache/gluten/task/TaskContextJniWrapper;");
30+
static jmethodID currentTaskAttemptIdMethod =
31+
getStaticMethodIdOrError(env, taskContextJniWrapperClass, "currentTaskAttemptId", "()J");
32+
33+
jlong id = env->CallStaticLongMethod(taskContextJniWrapperClass, currentTaskAttemptIdMethod);
34+
checkException(env);
35+
return static_cast<int64_t>(id);
36+
}
37+
38+
} // namespace gluten
Lines changed: 33 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,33 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one or more
3+
* contributor license agreements. See the NOTICE file distributed with
4+
* this work for additional information regarding copyright ownership.
5+
* The ASF licenses this file to You under the Apache License, Version 2.0
6+
* (the "License"); you may not use this file except in compliance with
7+
* the License. You may obtain a copy of the License at
8+
*
9+
* http://www.apache.org/licenses/LICENSE-2.0
10+
*
11+
* Unless required by applicable law or agreed to in writing, software
12+
* distributed under the License is distributed on an "AS IS" BASIS,
13+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+
* See the License for the specific language governing permissions and
15+
* limitations under the License.
16+
*/
17+
#pragma once
18+
19+
#include <cstdint>
20+
21+
namespace gluten {
22+
23+
/**
24+
* Returns the current thread's Spark task attempt id by calling back into
25+
* org.apache.gluten.task.TaskContextJniWrapper#currentTaskAttemptId via JNI.
26+
*
27+
* Returns -1 when there is no Spark TaskContext on the calling thread (driver,
28+
* standalone native worker, etc.). Safe to call from any thread; the helper
29+
* attaches the current thread to the JVM as a daemon on demand.
30+
*/
31+
int64_t getCurrentSparkTaskAttemptId();
32+
33+
} // namespace gluten
Lines changed: 42 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,42 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one or more
3+
* contributor license agreements. See the NOTICE file distributed with
4+
* this work for additional information regarding copyright ownership.
5+
* The ASF licenses this file to You under the Apache License, Version 2.0
6+
* (the "License"); you may not use this file except in compliance with
7+
* the License. You may obtain a copy of the License at
8+
*
9+
* http://www.apache.org/licenses/LICENSE-2.0
10+
*
11+
* Unless required by applicable law or agreed to in writing, software
12+
* distributed under the License is distributed on an "AS IS" BASIS,
13+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+
* See the License for the specific language governing permissions and
15+
* limitations under the License.
16+
*/
17+
package org.apache.gluten.task;
18+
19+
import org.apache.spark.TaskContext;
20+
21+
/**
22+
* Callback surface that native code uses to obtain the current thread's Spark task attempt id via
23+
* JNI, without having to plumb it explicitly through every JNI entry point.
24+
*
25+
* <p>Native side reads the value from {@link TaskContext#get()} at the moment it is needed. Since
26+
* {@link TaskContext} is a per-thread ThreadLocal, this only returns a meaningful value when the
27+
* native call is running on the executor task thread (or on a JVM thread inheriting the same
28+
* ThreadLocal). It returns {@code -1L} on the driver or on a native worker thread that has no
29+
* associated Spark task.
30+
*/
31+
public final class TaskContextJniWrapper {
32+
private TaskContextJniWrapper() {}
33+
34+
/**
35+
* Returns the current thread's Spark task attempt id, or {@code -1L} when there is no task
36+
* context on the calling thread.
37+
*/
38+
public static long currentTaskAttemptId() {
39+
TaskContext tc = TaskContext.get();
40+
return tc == null ? -1L : tc.taskAttemptId();
41+
}
42+
}

0 commit comments

Comments
 (0)