You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@opendal.apache.org by xu...@apache.org on 2023/04/26 10:04:16 UTC
[incubator-opendal] branch main updated: feat(java): connect rust async/await with java future (#2112)
This is an automated email from the ASF dual-hosted git repository.
xuanwo pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/incubator-opendal.git
The following commit(s) were added to refs/heads/main by this push:
new 781dd1a8 feat(java): connect rust async/await with java future (#2112)
781dd1a8 is described below
commit 781dd1a831c9af08772e6f0232b2e0af8b1c4d14
Author: SSpirits <sh...@outlook.com>
AuthorDate: Wed Apr 26 18:04:11 2023 +0800
feat(java): connect rust async/await with java future (#2112)
* feat(java-binding): connect rust async/await with java future
* feat(java-binding): create global runtime upon loading the library
* fix(java-binding): remove getting runtime thread count from java property
---
Cargo.lock | 3 +
bindings/java/Cargo.toml | 6 +
bindings/java/src/lib.rs | 128 ++++++++++++++++++++-
.../src/main/java/org/apache/opendal/Operator.java | 18 ++-
.../java/org/apache/opendal/AsyncStepsTest.java | 11 ++
5 files changed, 156 insertions(+), 10 deletions(-)
diff --git a/Cargo.lock b/Cargo.lock
index 5ad02566..b97a917e 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -2445,7 +2445,10 @@ name = "opendal-java"
version = "0.1.0"
dependencies = [
"jni",
+ "num_cpus",
+ "once_cell",
"opendal",
+ "tokio",
]
[[package]]
diff --git a/bindings/java/Cargo.toml b/bindings/java/Cargo.toml
index af7a14f4..06b1fcc6 100644
--- a/bindings/java/Cargo.toml
+++ b/bindings/java/Cargo.toml
@@ -34,3 +34,9 @@ doc = false
[dependencies]
jni = "0.21.1"
opendal.workspace = true
+
+tokio = { version = "1", features = ["full"] }
+once_cell = "1.17.1"
+
+num_cpus = "1.15.0"
+
diff --git a/bindings/java/src/lib.rs b/bindings/java/src/lib.rs
index 487384f7..f67c9df9 100644
--- a/bindings/java/src/lib.rs
+++ b/bindings/java/src/lib.rs
@@ -15,20 +15,94 @@
// specific language governing permissions and limitations
// under the License.
+use std::cell::RefCell;
use std::collections::HashMap;
+use std::ffi::c_void;
use std::str::FromStr;
+use std::sync::Arc;
-use jni::objects::JClass;
use jni::objects::JMap;
use jni::objects::JObject;
use jni::objects::JString;
-use jni::sys::jboolean;
-use jni::sys::jlong;
-use jni::JNIEnv;
+use jni::objects::{JClass, JValue};
+use jni::sys::{jboolean, jint};
+use jni::sys::{jlong, JNI_VERSION_1_8};
+use jni::{JNIEnv, JavaVM};
+use once_cell::sync::OnceCell;
+use tokio::runtime::{Builder, Runtime};
+
use opendal::BlockingOperator;
use opendal::Operator;
use opendal::Scheme;
+static mut RUNTIME: OnceCell<Runtime> = OnceCell::new();
+
+thread_local! {
+ static JAVA_VM: RefCell<Option<Arc<JavaVM>>> = RefCell::new(None);
+ static JENV: RefCell<Option<*mut jni::sys::JNIEnv>> = RefCell::new(None);
+}
+
+fn get_system_property(env: &mut JNIEnv, key: &str) -> Result<String, jni::errors::Error> {
+ let system_class = env.find_class("java/lang/System")?;
+ let key = env.new_string(key)?;
+ let value = env.call_static_method(
+ system_class,
+ "getProperty",
+ "(Ljava/lang/String;)Ljava/lang/String;",
+ &[JValue::Object(key.as_ref())],
+ )?;
+
+ Ok(env
+ .get_string(JString::from(value.l().unwrap()).as_ref())?
+ .into())
+}
+
+/// # Safety
+///
+/// This function could be only called by java vm when load this lib.
+#[no_mangle]
+pub unsafe extern "system" fn JNI_OnLoad(vm: JavaVM, _: *mut c_void) -> jint {
+ // TODO: make this configurable in the future
+ let thread_count = num_cpus::get();
+
+ let java_vm = Arc::new(vm);
+ let runtime = Builder::new_multi_thread()
+ .worker_threads(thread_count)
+ .on_thread_start(move || {
+ JENV.with(|cell| {
+ let env = java_vm.attach_current_thread_as_daemon().unwrap();
+ *cell.borrow_mut() = Some(env.get_raw());
+ });
+ JAVA_VM.with(|cell| {
+ *cell.borrow_mut() = Some(java_vm.clone());
+ });
+ })
+ .on_thread_stop(move || {
+ JENV.with(|cell| {
+ *cell.borrow_mut() = None;
+ });
+ JAVA_VM.with(|cell| unsafe {
+ if let Some(vm) = cell.borrow_mut().take() {
+ vm.detach_current_thread();
+ }
+ });
+ })
+ .build()
+ .unwrap();
+ RUNTIME.set(runtime).unwrap();
+ JNI_VERSION_1_8
+}
+
+/// # Safety
+///
+/// This function could be only called by java vm when unload this lib.
+#[no_mangle]
+pub unsafe extern "system" fn JNI_OnUnload(_: JavaVM, _: *mut c_void) {
+ if let Some(runtime) = RUNTIME.take() {
+ runtime.shutdown_background();
+ }
+}
+
#[no_mangle]
pub extern "system" fn Java_org_apache_opendal_Operator_getOperator(
mut env: JNIEnv,
@@ -57,6 +131,52 @@ pub extern "system" fn Java_org_apache_opendal_Operator_getOperator(
}
}
+/// # Safety
+///
+/// This function should not be called before the Operator are ready.
+#[no_mangle]
+pub unsafe extern "system" fn Java_org_apache_opendal_Operator_asyncWrite(
+ mut env: JNIEnv,
+ _class: JClass,
+ ptr: *mut Operator,
+ file: JString,
+ content: JString,
+ future: JObject,
+) {
+ let op = &mut *ptr;
+
+ let file: String = env.get_string(&file).unwrap().into();
+ let content: String = env.get_string(&content).unwrap().into();
+ // keep the future alive, so that we can complete it later
+ // but this approach will be limited by global ref table size
+ let future = env.new_global_ref(future).unwrap();
+
+ let x = async move {
+ op.write(&file, content).await.unwrap();
+ JENV.with(|cell| {
+ let env_ptr = cell.borrow().unwrap();
+ let mut env = JNIEnv::from_raw(env_ptr).unwrap();
+
+ // build result
+ let boolean_class = env.find_class("java/lang/Boolean").unwrap();
+ let boolean = env
+ .get_static_field(boolean_class, "TRUE", "Ljava/lang/Boolean;")
+ .unwrap();
+
+ // complete the java future
+ let _ = env
+ .call_method(
+ future,
+ "complete",
+ "(Ljava/lang/Object;)Z",
+ &[boolean.borrow()],
+ )
+ .unwrap();
+ });
+ };
+ RUNTIME.get().unwrap().spawn(x);
+}
+
/// # Safety
///
/// This function should not be called before the Operator are ready.
diff --git a/bindings/java/src/main/java/org/apache/opendal/Operator.java b/bindings/java/src/main/java/org/apache/opendal/Operator.java
index a51d71e6..cbd7f4a2 100644
--- a/bindings/java/src/main/java/org/apache/opendal/Operator.java
+++ b/bindings/java/src/main/java/org/apache/opendal/Operator.java
@@ -21,9 +21,8 @@
package org.apache.opendal;
import io.questdb.jar.jni.JarJniLoader;
-
import java.util.Map;
-
+import java.util.concurrent.CompletableFuture;
public class Operator {
@@ -39,9 +38,9 @@ public class Operator {
static {
JarJniLoader.loadLib(
- Operator.class,
- ORG_APACHE_OPENDAL_RUST_LIBS,
- OPENDAL_JAVA);
+ Operator.class,
+ ORG_APACHE_OPENDAL_RUST_LIBS,
+ OPENDAL_JAVA);
}
private native long getOperator(String type, Map<String, String> params);
@@ -50,17 +49,24 @@ public class Operator {
private native void write(long ptr, String fileName, String content);
+ private native void asyncWrite(long ptr, String fileName, String content, CompletableFuture<Boolean> future);
+
private native String read(long ptr, String fileName);
private native void delete(long ptr, String fileName);
private native long stat(long ptr, String file);
-
public void write(String fileName, String content) {
write(this.ptr, fileName, content);
}
+ public CompletableFuture<Boolean> asyncWrite(String fileName, String content) {
+ CompletableFuture<Boolean> future = new CompletableFuture<>();
+ asyncWrite(this.ptr, fileName, content, future);
+ return future;
+ }
+
public String read(String s) {
return read(this.ptr, s);
}
diff --git a/bindings/java/src/test/java/org/apache/opendal/AsyncStepsTest.java b/bindings/java/src/test/java/org/apache/opendal/AsyncStepsTest.java
index fb3ed034..76633ae5 100644
--- a/bindings/java/src/test/java/org/apache/opendal/AsyncStepsTest.java
+++ b/bindings/java/src/test/java/org/apache/opendal/AsyncStepsTest.java
@@ -22,17 +22,28 @@ package org.apache.opendal;
import io.cucumber.java.en.Given;
import io.cucumber.java.en.Then;
import io.cucumber.java.en.When;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import static org.junit.jupiter.api.Assertions.*;
public class AsyncStepsTest {
+ Operator operator;
@Given("A new OpenDAL Async Operator")
public void a_new_open_dal_async_operator() {
+ Map<String, String> params = new HashMap<>();
+ params.put("root", "/tmp");
+ operator = new Operator("Memory", params);
}
@When("Async write path {string} with content {string}")
public void async_write_path_test_with_content_hello_world(String fileName, String content) {
+ CompletableFuture<Boolean> future = operator.asyncWrite(fileName, content);
+ Boolean result = future.join();
+ assertTrue(result);
}
@Then("The async file {string} should exist")