Add Jvm object monitor and remove rust_object_field from Object/Thread (#138)

* Add object monitor to Jvm with object_wait, object_listen and object_notify

* Replace rust_object_field events with Jvm monitor in Object and Thread

Author: dlunch

Cargo.lock

@@ -14,7 +14,6 @@ tracing = { workspace = true }
 
 chrono = { version = "^0.4", default-features = false }
 encoding_rs = { version = "^0.8", features = ["alloc"], default-features = false }
-event-listener = { version = "^5.4", default-features = false }
 zip = { version = "^6.0", features = ["deflate"], default-features = false }
 url = { version = "^2.5", default-features = false }
 

java_runtime/Cargo.toml

@@ -1,13 +1,12 @@
 use core::{hash::BuildHasher, time::Duration};
 
-use alloc::{boxed::Box, format, sync::Arc, vec};
+use alloc::{boxed::Box, format, vec};
 
 use dyn_clone::clone_box;
-use event_listener::Event;
 use hashbrown::DefaultHashBuilder;
-use java_class_proto::{JavaFieldProto, JavaMethodProto};
+use java_class_proto::JavaMethodProto;
 use java_constants::MethodAccessFlags;
-use jvm::{Array, ClassInstance, ClassInstanceRef, Jvm, Result, runtime::JavaLangString};
+use jvm::{ClassInstance, ClassInstanceRef, Jvm, Result, runtime::JavaLangString};
 
 use crate::{Runtime, RuntimeClassProto, RuntimeContext, SpawnCallback, classes::java::lang::String};
 
@@ -34,7 +33,7 @@ impl Object {
                 JavaMethodProto::new("wait", "()V", Self::wait, Default::default()),
                 JavaMethodProto::new("finalize", "()V", Self::finalize, Default::default()),
             ],
-            fields: vec![JavaFieldProto::new("waitEvent", "[B", Default::default())],
+            fields: vec![],
             access_flags: Default::default(),
         }
     }
@@ -107,27 +106,15 @@ impl Object {
     async fn notify(jvm: &Jvm, _: &mut RuntimeContext, this: ClassInstanceRef<Self>) -> Result<()> {
         tracing::debug!("java.lang.Object::notify({:?})", &this);
 
-        let wait_event: ClassInstanceRef<Array<i8>> = jvm.get_field(&this, "waitEvent", "[B").await?;
-        if wait_event.is_null() {
-            return Ok(());
-        }
-
-        let wait_event: Arc<Event> = jvm.get_rust_object_field(&this, "waitEvent").await?;
-        wait_event.notify(1);
+        jvm.object_notify(&this, 1);
 
         Ok(())
     }
 
     async fn notify_all(jvm: &Jvm, _: &mut RuntimeContext, this: ClassInstanceRef<Self>) -> Result<()> {
         tracing::debug!("java.lang.Object::notifyAll({:?})", &this);
 
-        let wait_event: ClassInstanceRef<Array<i8>> = jvm.get_field(&this, "waitEvent", "[B").await?;
-        if wait_event.is_null() {
-            return Ok(());
-        }
-
-        let wait_event: Arc<Event> = jvm.get_rust_object_field(&this, "waitEvent").await?;
-        wait_event.notify(usize::MAX);
+        jvm.object_notify(&this, usize::MAX);
 
         Ok(())
     }
@@ -140,24 +127,21 @@ impl Object {
         Ok(())
     }
 
-    async fn wait_long_int(jvm: &Jvm, context: &mut RuntimeContext, mut this: ClassInstanceRef<Self>, millis: i64, nanos: i32) -> Result<()> {
+    async fn wait_long_int(jvm: &Jvm, context: &mut RuntimeContext, this: ClassInstanceRef<Self>, millis: i64, nanos: i32) -> Result<()> {
         tracing::debug!("java.lang.Object::wait({:?}, {:?}, {:?})", &this, millis, nanos);
 
-        let wait_event = Arc::new(Event::new());
-        jvm.put_rust_object_field(&mut this, "waitEvent", wait_event.clone()).await?;
-
-        struct Waiter {
+        struct TimeoutNotifier {
             timeout: i64,
-            wait_event: Arc<Event>,
+            jvm: Jvm,
+            this: Box<dyn ClassInstance>,
             context: Box<dyn Runtime>,
         }
 
         #[async_trait::async_trait]
-        impl SpawnCallback for Waiter {
+        impl SpawnCallback for TimeoutNotifier {
             async fn call(&self) -> Result<()> {
                 self.context.sleep(Duration::from_millis(self.timeout as _)).await;
-
-                self.wait_event.notify(1); // TODO this would notify other waiter
+                self.jvm.object_notify(&self.this, 1); // TODO this may wake an unrelated waiter
                 Ok(())
             }
         }
@@ -166,15 +150,16 @@ impl Object {
         if timeout != 0 {
             context.spawn(
                 jvm,
-                Box::new(Waiter {
+                Box::new(TimeoutNotifier {
                     timeout,
-                    wait_event: wait_event.clone(),
+                    jvm: jvm.clone(),
+                    this: this.clone().into(),
                     context: clone_box(context),
                 }),
             );
         }
 
-        wait_event.listen().await;
+        jvm.object_wait(&this).await?;
 
         Ok(())
     }

java_runtime/src/classes/java/lang/object.rs

@@ -1,11 +1,9 @@
-use alloc::{boxed::Box, sync::Arc, vec};
+use alloc::{boxed::Box, vec};
 use core::time::Duration;
 
-use event_listener::Event;
-
 use java_class_proto::{JavaFieldProto, JavaMethodProto};
 use java_constants::MethodAccessFlags;
-use jvm::{Array, ClassInstanceRef, Jvm, Result, runtime::JavaLangString};
+use jvm::{ClassInstanceRef, Jvm, Result, runtime::JavaLangString};
 
 use crate::{RuntimeClassProto, RuntimeContext, SpawnCallback, classes::java::lang::Runnable};
 
@@ -40,7 +38,7 @@ impl Thread {
             fields: vec![
                 JavaFieldProto::new("id", "J", Default::default()),
                 JavaFieldProto::new("target", "Ljava/lang/Runnable;", Default::default()),
-                JavaFieldProto::new("joinEvent", "[B", Default::default()),
+                JavaFieldProto::new("alive", "Z", Default::default()),
             ],
             access_flags: Default::default(),
         }
@@ -84,7 +82,6 @@ impl Thread {
         struct ThreadStartProxy {
             jvm: Jvm,
             thread_id: i32,
-            join_event: Arc<Event>,
             this: ClassInstanceRef<Thread>,
         }
 
@@ -129,14 +126,15 @@ impl Thread {
 
                 self.jvm.detach_thread()?;
 
-                self.join_event.notify(usize::MAX);
+                let mut this = self.this.clone();
+                self.jvm.put_field(&mut this, "alive", "Z", false).await.unwrap();
+                self.jvm.object_notify(&self.this, usize::MAX);
 
                 Ok(())
             }
         }
 
-        let join_event = Arc::new(Event::new());
-        jvm.put_rust_object_field(&mut this, "joinEvent", join_event.clone()).await?;
+        jvm.put_field(&mut this, "alive", "Z", true).await?;
 
         let id: i32 = jvm.invoke_virtual(&this, "hashCode", "()I", ()).await?;
 
@@ -145,7 +143,6 @@ impl Thread {
             Box::new(ThreadStartProxy {
                 jvm: jvm.clone(),
                 thread_id: id,
-                join_event,
                 this: this.clone(),
             }),
         );
@@ -164,27 +161,23 @@ impl Thread {
         Ok(())
     }
 
-    async fn join(jvm: &Jvm, _context: &mut RuntimeContext, mut this: ClassInstanceRef<Self>) -> Result<()> {
+    async fn join(jvm: &Jvm, _context: &mut RuntimeContext, this: ClassInstanceRef<Self>) -> Result<()> {
         tracing::debug!("java.lang.Thread::join({:?})", &this);
 
-        // TODO we don't have get same field twice
-        let raw_join_event: ClassInstanceRef<Array<i8>> = jvm.get_field(&this, "joinEvent", "[B").await?;
-        if raw_join_event.is_null() {
-            return Ok(()); // already joined or not started
+        loop {
+            let listener = jvm.object_listen(&this);
+            let alive: bool = jvm.get_field(&this, "alive", "Z").await?;
+            if !alive {
+                return Ok(());
+            }
+            listener.await;
         }
-
-        let join_event: Arc<Event> = jvm.get_rust_object_field(&this, "joinEvent").await?;
-        join_event.listen().await;
-
-        jvm.put_field(&mut this, "joinEvent", "[B", None).await?;
-
-        Ok(())
     }
 
-    async fn is_alive(_jvm: &Jvm, _: &mut RuntimeContext, this: ClassInstanceRef<Self>) -> Result<bool> {
-        tracing::warn!("stub java.lang.Thread::isAlive({:?})", &this);
-
-        Ok(true)
+    async fn is_alive(jvm: &Jvm, _: &mut RuntimeContext, this: ClassInstanceRef<Self>) -> Result<bool> {
+        tracing::debug!("java.lang.Thread::isAlive({:?})", &this);
+        let alive: bool = jvm.get_field(&this, "alive", "Z").await?;
+        Ok(alive)
     }
 
     async fn sleep(_: &Jvm, context: &mut RuntimeContext, duration: i64) -> Result<()> {

java_runtime/src/classes/java/lang/thread.rs

@@ -16,6 +16,8 @@ nom = { workspace = true }
 parking_lot = { workspace = true }
 tracing = { workspace = true }
 
+event-listener = { version = "^5.4", default-features = false }
+
 java_constants = { workspace = true }
 
 [dev-dependencies]

jvm/Cargo.toml

@@ -3,14 +3,16 @@
 use alloc::{borrow::ToOwned, boxed::Box, collections::BTreeMap, format, string::String, sync::Arc, vec::Vec};
 use core::{
     fmt::Debug,
+    hash::BuildHasher,
     iter,
     mem::{forget, size_of_val},
     sync::atomic::{AtomicBool, Ordering},
 };
 
 use bytemuck::cast_slice;
 use dyn_clone::clone_box;
-use hashbrown::HashSet;
+use event_listener::{Event, EventListener};
+use hashbrown::{DefaultHashBuilder, HashSet};
 use parking_lot::RwLock;
 
 use java_constants::{ClassAccessFlags, MethodAccessFlags};
@@ -36,6 +38,8 @@ struct JvmInner {
     classes: RwLock<BTreeMap<String, Class>>,
     threads: RwLock<BTreeMap<u64, JvmThread>>,
     all_objects: RwLock<HashSet<Box<dyn ClassInstance>>>,
+    monitors: RwLock<BTreeMap<u64, Arc<Event>>>,
+    monitor_hasher: DefaultHashBuilder,
     get_current_thread_id: Box<dyn Fn() -> u64 + Sync + Send>,
     bootstrap_class_loader: Box<dyn BootstrapClassLoader>,
     bootstrapping: AtomicBool,
@@ -57,6 +61,8 @@ impl Jvm {
                 classes: RwLock::new(BTreeMap::new()),
                 threads: RwLock::new(BTreeMap::new()),
                 all_objects: RwLock::new(HashSet::new()),
+                monitors: RwLock::new(BTreeMap::new()),
+                monitor_hasher: DefaultHashBuilder::default(),
                 get_current_thread_id: Box::new(get_current_thread_id),
                 bootstrap_class_loader: Box::new(bootstrap_class_loader),
                 bootstrapping: AtomicBool::new(true),
@@ -464,6 +470,21 @@ impl Jvm {
         self.inner.classes.read().get(class_name).cloned()
     }
 
+    pub fn object_listen(&self, obj: &Box<dyn ClassInstance>) -> EventListener {
+        self.get_or_create_monitor(obj).listen()
+    }
+
+    pub async fn object_wait(&self, obj: &Box<dyn ClassInstance>) -> Result<()> {
+        self.object_listen(obj).await;
+
+        Ok(())
+    }
+
+    pub fn object_notify(&self, obj: &Box<dyn ClassInstance>, count: usize) {
+        let monitor = self.get_or_create_monitor(obj);
+        monitor.notify(count);
+    }
+
     #[async_recursion::async_recursion]
     pub async fn resolve_class(&self, class_name: &str) -> Result<Class> {
         self.resolve_class_internal(class_name, None).await
@@ -720,6 +741,19 @@ impl Jvm {
         }
     }
 
+    fn get_or_create_monitor(&self, obj: &Box<dyn ClassInstance>) -> Arc<Event> {
+        let key = self.inner.monitor_hasher.hash_one(obj);
+
+        let monitors = self.inner.monitors.read();
+        if let Some(monitor) = monitors.get(&key) {
+            return monitor.clone();
+        }
+        drop(monitors);
+
+        let mut monitors = self.inner.monitors.write();
+        monitors.entry(key).or_insert_with(|| Arc::new(Event::new())).clone()
+    }
+
     pub(crate) fn find_field(&self, class: &dyn ClassDefinition, name: &str, descriptor: &str) -> Result<Option<Box<dyn Field>>> {
         let field = class.field(name, descriptor, false);